source: mainline/uspace/lib/c/generic/async.c@ 0ca7286

lfn serial ticket/834-toolchain-update topic/msim-upgrade topic/simplify-dev-export
Last change on this file since 0ca7286 was 0ca7286, checked in by Adam Hraska <adam.hraska+hos@…>, 13 years ago

Added resizing to user space (single-threaded) hash_table. Resizes in a way to mitigate effects of bad hash functions. Change of interface affected many files.

  • Property mode set to 100644
File size: 68.0 KB
Line 
1/*
2 * Copyright (c) 2006 Ondrej Palkovsky
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
7 * are met:
8 *
9 * - Redistributions of source code must retain the above copyright
10 * notice, this list of conditions and the following disclaimer.
11 * - Redistributions in binary form must reproduce the above copyright
12 * notice, this list of conditions and the following disclaimer in the
13 * documentation and/or other materials provided with the distribution.
14 * - The name of the author may not be used to endorse or promote products
15 * derived from this software without specific prior written permission.
16 *
17 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
18 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
19 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
20 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
21 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
22 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
26 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27 */
28
29/** @addtogroup libc
30 * @{
31 */
32/** @file
33 */
34
35/**
36 * Asynchronous library
37 *
38 * The aim of this library is to provide a facility for writing programs which
39 * utilize the asynchronous nature of HelenOS IPC, yet using a normal way of
40 * programming.
41 *
42 * You should be able to write very simple multithreaded programs. The async
43 * framework will automatically take care of most of the synchronization
44 * problems.
45 *
46 * Example of use (pseudo C):
47 *
48 * 1) Multithreaded client application
49 *
50 * fibril_create(fibril1, ...);
51 * fibril_create(fibril2, ...);
52 * ...
53 *
54 * int fibril1(void *arg)
55 * {
56 * conn = async_connect_me_to(...);
57 *
58 * exch = async_exchange_begin(conn);
59 * c1 = async_send(exch);
60 * async_exchange_end(exch);
61 *
62 * exch = async_exchange_begin(conn);
63 * c2 = async_send(exch);
64 * async_exchange_end(exch);
65 *
66 * async_wait_for(c1);
67 * async_wait_for(c2);
68 * ...
69 * }
70 *
71 *
72 * 2) Multithreaded server application
73 *
74 * main()
75 * {
76 * async_manager();
77 * }
78 *
79 * my_client_connection(icallid, *icall)
80 * {
81 * if (want_refuse) {
82 * async_answer_0(icallid, ELIMIT);
83 * return;
84 * }
85 * async_answer_0(icallid, EOK);
86 *
87 * callid = async_get_call(&call);
88 * somehow_handle_the_call(callid, call);
89 * async_answer_2(callid, 1, 2, 3);
90 *
91 * callid = async_get_call(&call);
92 * ...
93 * }
94 *
95 */
96
97#define LIBC_ASYNC_C_
98#include <ipc/ipc.h>
99#include <async.h>
100#include "private/async.h"
101#undef LIBC_ASYNC_C_
102
103#include <futex.h>
104#include <fibril.h>
105#include <adt/hash_table.h>
106#include <adt/list.h>
107#include <assert.h>
108#include <errno.h>
109#include <sys/time.h>
110#include <libarch/barrier.h>
111#include <bool.h>
112#include <malloc.h>
113#include <mem.h>
114#include <stdlib.h>
115#include <macros.h>
116
117
118/** Session data */
119struct async_sess {
120 /** List of inactive exchanges */
121 list_t exch_list;
122
123 /** Exchange management style */
124 exch_mgmt_t mgmt;
125
126 /** Session identification */
127 int phone;
128
129 /** First clone connection argument */
130 sysarg_t arg1;
131
132 /** Second clone connection argument */
133 sysarg_t arg2;
134
135 /** Third clone connection argument */
136 sysarg_t arg3;
137
138 /** Exchange mutex */
139 fibril_mutex_t mutex;
140
141 /** Number of opened exchanges */
142 atomic_t refcnt;
143
144 /** Mutex for stateful connections */
145 fibril_mutex_t remote_state_mtx;
146
147 /** Data for stateful connections */
148 void *remote_state_data;
149};
150
151/** Exchange data */
152struct async_exch {
153 /** Link into list of inactive exchanges */
154 link_t sess_link;
155
156 /** Link into global list of inactive exchanges */
157 link_t global_link;
158
159 /** Session pointer */
160 async_sess_t *sess;
161
162 /** Exchange identification */
163 int phone;
164};
165
166/** Async framework global futex */
167atomic_t async_futex = FUTEX_INITIALIZER;
168
169/** Number of threads waiting for IPC in the kernel. */
170atomic_t threads_in_ipc_wait = { 0 };
171
172/** Naming service session */
173async_sess_t *session_ns;
174
175/** Call data */
176typedef struct {
177 link_t link;
178
179 ipc_callid_t callid;
180 ipc_call_t call;
181} msg_t;
182
183/** Message data */
184typedef struct {
185 awaiter_t wdata;
186
187 /** If reply was received. */
188 bool done;
189
190 /** If the message / reply should be discarded on arrival. */
191 bool forget;
192
193 /** If already destroyed. */
194 bool destroyed;
195
196 /** Pointer to where the answer data is stored. */
197 ipc_call_t *dataptr;
198
199 sysarg_t retval;
200} amsg_t;
201
202/* Client connection data */
203typedef struct {
204 link_t link;
205
206 task_id_t in_task_id;
207 atomic_t refcnt;
208 void *data;
209} client_t;
210
211/* Server connection data */
212typedef struct {
213 awaiter_t wdata;
214
215 /** Hash table link. */
216 link_t link;
217
218 /** Incoming client task ID. */
219 task_id_t in_task_id;
220
221 /** Incoming phone hash. */
222 sysarg_t in_phone_hash;
223
224 /** Link to the client tracking structure. */
225 client_t *client;
226
227 /** Messages that should be delivered to this fibril. */
228 list_t msg_queue;
229
230 /** Identification of the opening call. */
231 ipc_callid_t callid;
232 /** Call data of the opening call. */
233 ipc_call_t call;
234 /** Local argument or NULL if none. */
235 void *carg;
236
237 /** Identification of the closing call. */
238 ipc_callid_t close_callid;
239
240 /** Fibril function that will be used to handle the connection. */
241 async_client_conn_t cfibril;
242} connection_t;
243
244/** Identifier of the incoming connection handled by the current fibril. */
245static fibril_local connection_t *fibril_connection;
246
247static void to_event_initialize(to_event_t *to)
248{
249 struct timeval tv = { 0, 0 };
250
251 to->inlist = false;
252 to->occurred = false;
253 link_initialize(&to->link);
254 to->expires = tv;
255}
256
257static void wu_event_initialize(wu_event_t *wu)
258{
259 wu->inlist = false;
260 link_initialize(&wu->link);
261}
262
263void awaiter_initialize(awaiter_t *aw)
264{
265 aw->fid = 0;
266 aw->active = false;
267 to_event_initialize(&aw->to_event);
268 wu_event_initialize(&aw->wu_event);
269}
270
271static amsg_t *amsg_create(void)
272{
273 amsg_t *msg;
274
275 msg = malloc(sizeof(amsg_t));
276 if (msg) {
277 msg->done = false;
278 msg->forget = false;
279 msg->destroyed = false;
280 msg->dataptr = NULL;
281 msg->retval = (sysarg_t) EINVAL;
282 awaiter_initialize(&msg->wdata);
283 }
284
285 return msg;
286}
287
288static void amsg_destroy(amsg_t *msg)
289{
290 assert(!msg->destroyed);
291 msg->destroyed = true;
292 free(msg);
293}
294
295static void *default_client_data_constructor(void)
296{
297 return NULL;
298}
299
300static void default_client_data_destructor(void *data)
301{
302}
303
304static async_client_data_ctor_t async_client_data_create =
305 default_client_data_constructor;
306static async_client_data_dtor_t async_client_data_destroy =
307 default_client_data_destructor;
308
309void async_set_client_data_constructor(async_client_data_ctor_t ctor)
310{
311 assert(async_client_data_create == default_client_data_constructor);
312 async_client_data_create = ctor;
313}
314
315void async_set_client_data_destructor(async_client_data_dtor_t dtor)
316{
317 assert(async_client_data_destroy == default_client_data_destructor);
318 async_client_data_destroy = dtor;
319}
320
321/** Default fibril function that gets called to handle new connection.
322 *
323 * This function is defined as a weak symbol - to be redefined in user code.
324 *
325 * @param callid Hash of the incoming call.
326 * @param call Data of the incoming call.
327 * @param arg Local argument
328 *
329 */
330static void default_client_connection(ipc_callid_t callid, ipc_call_t *call,
331 void *arg)
332{
333 ipc_answer_0(callid, ENOENT);
334}
335
336/** Default fibril function that gets called to handle interrupt notifications.
337 *
338 * This function is defined as a weak symbol - to be redefined in user code.
339 *
340 * @param callid Hash of the incoming call.
341 * @param call Data of the incoming call.
342 * @param arg Local argument.
343 *
344 */
345static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call)
346{
347}
348
349static async_client_conn_t client_connection = default_client_connection;
350static async_interrupt_handler_t interrupt_received = default_interrupt_received;
351
352/** Setter for client_connection function pointer.
353 *
354 * @param conn Function that will implement a new connection fibril.
355 *
356 */
357void async_set_client_connection(async_client_conn_t conn)
358{
359 assert(client_connection == default_client_connection);
360 client_connection = conn;
361}
362
363/** Setter for interrupt_received function pointer.
364 *
365 * @param intr Function that will implement a new interrupt
366 * notification fibril.
367 */
368void async_set_interrupt_received(async_interrupt_handler_t intr)
369{
370 interrupt_received = intr;
371}
372
373/** Mutex protecting inactive_exch_list and avail_phone_cv.
374 *
375 */
376static FIBRIL_MUTEX_INITIALIZE(async_sess_mutex);
377
378/** List of all currently inactive exchanges.
379 *
380 */
381static LIST_INITIALIZE(inactive_exch_list);
382
383/** Condition variable to wait for a phone to become available.
384 *
385 */
386static FIBRIL_CONDVAR_INITIALIZE(avail_phone_cv);
387
388static hash_table_t client_hash_table;
389static hash_table_t conn_hash_table;
390static LIST_INITIALIZE(timeout_list);
391
392static size_t client_key_hash(unsigned long key[])
393{
394 assert(key);
395 /* LOWER32(in_task_id) */
396 return key[0] >> 4;
397}
398
399static size_t client_hash(const link_t *item)
400{
401 client_t *client = hash_table_get_instance(item, client_t, link);
402
403 unsigned long key[2] = {
404 LOWER32(client->in_task_id),
405 UPPER32(client->in_task_id),
406 };
407
408 return client_key_hash(key);
409}
410
411static bool client_match(unsigned long key[], size_t keys, const link_t *item)
412{
413 assert(key);
414 assert(keys == 2);
415 assert(item);
416
417 client_t *client = hash_table_get_instance(item, client_t, link);
418 return (key[0] == LOWER32(client->in_task_id) &&
419 (key[1] == UPPER32(client->in_task_id)));
420}
421
422
423/** Operations for the client hash table. */
424static hash_table_ops_t client_hash_table_ops = {
425 .hash = client_hash,
426 .key_hash = client_key_hash,
427 .match = client_match,
428 .equal = 0,
429 .remove_callback = 0
430};
431
432/** Compute hash into the connection hash table based on the source phone hash.
433 *
434 * @param key Pointer to source phone hash.
435 *
436 * @return Index into the connection hash table.
437 *
438 */
439static size_t conn_key_hash(unsigned long key[])
440{
441 assert(key);
442 return key[0] >> 4;
443}
444
445static size_t conn_hash(const link_t *item)
446{
447 connection_t *conn = hash_table_get_instance(item, connection_t, link);
448 unsigned long key = conn->in_phone_hash;
449 return conn_key_hash(&key);
450}
451
452/** Compare hash table item with a key.
453 *
454 * @param key Array containing the source phone hash as the only item.
455 * @param keys Expected 1 but ignored.
456 * @param item Connection hash table item.
457 *
458 * @return True on match, false otherwise.
459 *
460 */
461static bool conn_match(unsigned long key[], size_t keys, const link_t *item)
462{
463 assert(key);
464 assert(item);
465
466 connection_t *conn = hash_table_get_instance(item, connection_t, link);
467 return (key[0] == conn->in_phone_hash);
468}
469
470static bool conn_equal(const link_t *item1, const link_t *item2)
471{
472 connection_t *c1 = hash_table_get_instance(item1, connection_t, link);
473 connection_t *c2 = hash_table_get_instance(item2, connection_t, link);
474
475 return c1->in_phone_hash == c2->in_phone_hash;
476}
477
478static void conn_remove(link_t *item)
479{
480}
481
482/** Operations for the connection hash table. */
483static hash_table_ops_t conn_hash_table_ops = {
484 .hash = conn_hash,
485 .key_hash = conn_key_hash,
486 .match = conn_match,
487 .equal = conn_equal,
488 .remove_callback = conn_remove
489};
490
491/** Sort in current fibril's timeout request.
492 *
493 * @param wd Wait data of the current fibril.
494 *
495 */
496void async_insert_timeout(awaiter_t *wd)
497{
498 assert(wd);
499
500 wd->to_event.occurred = false;
501 wd->to_event.inlist = true;
502
503 link_t *tmp = timeout_list.head.next;
504 while (tmp != &timeout_list.head) {
505 awaiter_t *cur
506 = list_get_instance(tmp, awaiter_t, to_event.link);
507
508 if (tv_gteq(&cur->to_event.expires, &wd->to_event.expires))
509 break;
510
511 tmp = tmp->next;
512 }
513
514 list_insert_before(&wd->to_event.link, tmp);
515}
516
517/** Try to route a call to an appropriate connection fibril.
518 *
519 * If the proper connection fibril is found, a message with the call is added to
520 * its message queue. If the fibril was not active, it is activated and all
521 * timeouts are unregistered.
522 *
523 * @param callid Hash of the incoming call.
524 * @param call Data of the incoming call.
525 *
526 * @return False if the call doesn't match any connection.
527 * @return True if the call was passed to the respective connection fibril.
528 *
529 */
530static bool route_call(ipc_callid_t callid, ipc_call_t *call)
531{
532 assert(call);
533
534 futex_down(&async_futex);
535
536 unsigned long key = call->in_phone_hash;
537 link_t *hlp = hash_table_find(&conn_hash_table, &key);
538
539 if (!hlp) {
540 futex_up(&async_futex);
541 return false;
542 }
543
544 connection_t *conn = hash_table_get_instance(hlp, connection_t, link);
545
546 msg_t *msg = malloc(sizeof(*msg));
547 if (!msg) {
548 futex_up(&async_futex);
549 return false;
550 }
551
552 msg->callid = callid;
553 msg->call = *call;
554 list_append(&msg->link, &conn->msg_queue);
555
556 if (IPC_GET_IMETHOD(*call) == IPC_M_PHONE_HUNGUP)
557 conn->close_callid = callid;
558
559 /* If the connection fibril is waiting for an event, activate it */
560 if (!conn->wdata.active) {
561
562 /* If in timeout list, remove it */
563 if (conn->wdata.to_event.inlist) {
564 conn->wdata.to_event.inlist = false;
565 list_remove(&conn->wdata.to_event.link);
566 }
567
568 conn->wdata.active = true;
569 fibril_add_ready(conn->wdata.fid);
570 }
571
572 futex_up(&async_futex);
573 return true;
574}
575
576/** Notification fibril.
577 *
578 * When a notification arrives, a fibril with this implementing function is
579 * created. It calls interrupt_received() and does the final cleanup.
580 *
581 * @param arg Message structure pointer.
582 *
583 * @return Always zero.
584 *
585 */
586static int notification_fibril(void *arg)
587{
588 assert(arg);
589
590 msg_t *msg = (msg_t *) arg;
591 interrupt_received(msg->callid, &msg->call);
592
593 free(msg);
594 return 0;
595}
596
597/** Process interrupt notification.
598 *
599 * A new fibril is created which would process the notification.
600 *
601 * @param callid Hash of the incoming call.
602 * @param call Data of the incoming call.
603 *
604 * @return False if an error occured.
605 * True if the call was passed to the notification fibril.
606 *
607 */
608static bool process_notification(ipc_callid_t callid, ipc_call_t *call)
609{
610 assert(call);
611
612 futex_down(&async_futex);
613
614 msg_t *msg = malloc(sizeof(*msg));
615 if (!msg) {
616 futex_up(&async_futex);
617 return false;
618 }
619
620 msg->callid = callid;
621 msg->call = *call;
622
623 fid_t fid = fibril_create(notification_fibril, msg);
624 if (fid == 0) {
625 free(msg);
626 futex_up(&async_futex);
627 return false;
628 }
629
630 fibril_add_ready(fid);
631
632 futex_up(&async_futex);
633 return true;
634}
635
636/** Return new incoming message for the current (fibril-local) connection.
637 *
638 * @param call Storage where the incoming call data will be stored.
639 * @param usecs Timeout in microseconds. Zero denotes no timeout.
640 *
641 * @return If no timeout was specified, then a hash of the
642 * incoming call is returned. If a timeout is specified,
643 * then a hash of the incoming call is returned unless
644 * the timeout expires prior to receiving a message. In
645 * that case zero is returned.
646 *
647 */
648ipc_callid_t async_get_call_timeout(ipc_call_t *call, suseconds_t usecs)
649{
650 assert(call);
651 assert(fibril_connection);
652
653 /* Why doing this?
654 * GCC 4.1.0 coughs on fibril_connection-> dereference.
655 * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
656 * I would never expect to find so many errors in
657 * a compiler.
658 */
659 connection_t *conn = fibril_connection;
660
661 futex_down(&async_futex);
662
663 if (usecs) {
664 gettimeofday(&conn->wdata.to_event.expires, NULL);
665 tv_add(&conn->wdata.to_event.expires, usecs);
666 } else
667 conn->wdata.to_event.inlist = false;
668
669 /* If nothing in queue, wait until something arrives */
670 while (list_empty(&conn->msg_queue)) {
671 if (conn->close_callid) {
672 /*
673 * Handle the case when the connection was already
674 * closed by the client but the server did not notice
675 * the first IPC_M_PHONE_HUNGUP call and continues to
676 * call async_get_call_timeout(). Repeat
677 * IPC_M_PHONE_HUNGUP until the caller notices.
678 */
679 memset(call, 0, sizeof(ipc_call_t));
680 IPC_SET_IMETHOD(*call, IPC_M_PHONE_HUNGUP);
681 futex_up(&async_futex);
682 return conn->close_callid;
683 }
684
685 if (usecs)
686 async_insert_timeout(&conn->wdata);
687
688 conn->wdata.active = false;
689
690 /*
691 * Note: the current fibril will be rescheduled either due to a
692 * timeout or due to an arriving message destined to it. In the
693 * former case, handle_expired_timeouts() and, in the latter
694 * case, route_call() will perform the wakeup.
695 */
696 fibril_switch(FIBRIL_TO_MANAGER);
697
698 /*
699 * Futex is up after getting back from async_manager.
700 * Get it again.
701 */
702 futex_down(&async_futex);
703 if ((usecs) && (conn->wdata.to_event.occurred)
704 && (list_empty(&conn->msg_queue))) {
705 /* If we timed out -> exit */
706 futex_up(&async_futex);
707 return 0;
708 }
709 }
710
711 msg_t *msg = list_get_instance(list_first(&conn->msg_queue), msg_t, link);
712 list_remove(&msg->link);
713
714 ipc_callid_t callid = msg->callid;
715 *call = msg->call;
716 free(msg);
717
718 futex_up(&async_futex);
719 return callid;
720}
721
722static client_t *async_client_get(task_id_t client_id, bool create)
723{
724 unsigned long key[2] = {
725 LOWER32(client_id),
726 UPPER32(client_id),
727 };
728 client_t *client = NULL;
729
730 futex_down(&async_futex);
731 link_t *lnk = hash_table_find(&client_hash_table, key);
732 if (lnk) {
733 client = hash_table_get_instance(lnk, client_t, link);
734 atomic_inc(&client->refcnt);
735 } else if (create) {
736 client = malloc(sizeof(client_t));
737 if (client) {
738 client->in_task_id = client_id;
739 client->data = async_client_data_create();
740
741 atomic_set(&client->refcnt, 1);
742 hash_table_insert(&client_hash_table, &client->link);
743 }
744 }
745
746 futex_up(&async_futex);
747 return client;
748}
749
750static void async_client_put(client_t *client)
751{
752 bool destroy;
753 unsigned long key[2] = {
754 LOWER32(client->in_task_id),
755 UPPER32(client->in_task_id)
756 };
757
758 futex_down(&async_futex);
759
760 if (atomic_predec(&client->refcnt) == 0) {
761 hash_table_remove(&client_hash_table, key, 2);
762 destroy = true;
763 } else
764 destroy = false;
765
766 futex_up(&async_futex);
767
768 if (destroy) {
769 if (client->data)
770 async_client_data_destroy(client->data);
771
772 free(client);
773 }
774}
775
776void *async_get_client_data(void)
777{
778 assert(fibril_connection);
779 return fibril_connection->client->data;
780}
781
782void *async_get_client_data_by_id(task_id_t client_id)
783{
784 client_t *client = async_client_get(client_id, false);
785 if (!client)
786 return NULL;
787 if (!client->data) {
788 async_client_put(client);
789 return NULL;
790 }
791
792 return client->data;
793}
794
795void async_put_client_data_by_id(task_id_t client_id)
796{
797 client_t *client = async_client_get(client_id, false);
798
799 assert(client);
800 assert(client->data);
801
802 /* Drop the reference we got in async_get_client_data_by_hash(). */
803 async_client_put(client);
804
805 /* Drop our own reference we got at the beginning of this function. */
806 async_client_put(client);
807}
808
809/** Wrapper for client connection fibril.
810 *
811 * When a new connection arrives, a fibril with this implementing function is
812 * created. It calls client_connection() and does the final cleanup.
813 *
814 * @param arg Connection structure pointer.
815 *
816 * @return Always zero.
817 *
818 */
819static int connection_fibril(void *arg)
820{
821 assert(arg);
822
823 /*
824 * Setup fibril-local connection pointer.
825 */
826 fibril_connection = (connection_t *) arg;
827
828 /*
829 * Add our reference for the current connection in the client task
830 * tracking structure. If this is the first reference, create and
831 * hash in a new tracking structure.
832 */
833
834 client_t *client = async_client_get(fibril_connection->in_task_id, true);
835 if (!client) {
836 ipc_answer_0(fibril_connection->callid, ENOMEM);
837 return 0;
838 }
839
840 fibril_connection->client = client;
841
842 /*
843 * Call the connection handler function.
844 */
845 fibril_connection->cfibril(fibril_connection->callid,
846 &fibril_connection->call, fibril_connection->carg);
847
848 /*
849 * Remove the reference for this client task connection.
850 */
851 async_client_put(client);
852
853 /*
854 * Remove myself from the connection hash table.
855 */
856 futex_down(&async_futex);
857 unsigned long key = fibril_connection->in_phone_hash;
858 hash_table_remove(&conn_hash_table, &key, 1);
859 futex_up(&async_futex);
860
861 /*
862 * Answer all remaining messages with EHANGUP.
863 */
864 while (!list_empty(&fibril_connection->msg_queue)) {
865 msg_t *msg =
866 list_get_instance(list_first(&fibril_connection->msg_queue),
867 msg_t, link);
868
869 list_remove(&msg->link);
870 ipc_answer_0(msg->callid, EHANGUP);
871 free(msg);
872 }
873
874 /*
875 * If the connection was hung-up, answer the last call,
876 * i.e. IPC_M_PHONE_HUNGUP.
877 */
878 if (fibril_connection->close_callid)
879 ipc_answer_0(fibril_connection->close_callid, EOK);
880
881 free(fibril_connection);
882 return 0;
883}
884
885/** Create a new fibril for a new connection.
886 *
887 * Create new fibril for connection, fill in connection structures and insert
888 * it into the hash table, so that later we can easily do routing of messages to
889 * particular fibrils.
890 *
891 * @param in_task_id Identification of the incoming connection.
892 * @param in_phone_hash Identification of the incoming connection.
893 * @param callid Hash of the opening IPC_M_CONNECT_ME_TO call.
894 * If callid is zero, the connection was opened by
895 * accepting the IPC_M_CONNECT_TO_ME call and this function
896 * is called directly by the server.
897 * @param call Call data of the opening call.
898 * @param cfibril Fibril function that should be called upon opening the
899 * connection.
900 * @param carg Extra argument to pass to the connection fibril
901 *
902 * @return New fibril id or NULL on failure.
903 *
904 */
905fid_t async_new_connection(task_id_t in_task_id, sysarg_t in_phone_hash,
906 ipc_callid_t callid, ipc_call_t *call,
907 async_client_conn_t cfibril, void *carg)
908{
909 connection_t *conn = malloc(sizeof(*conn));
910 if (!conn) {
911 if (callid)
912 ipc_answer_0(callid, ENOMEM);
913
914 return (uintptr_t) NULL;
915 }
916
917 conn->in_task_id = in_task_id;
918 conn->in_phone_hash = in_phone_hash;
919 list_initialize(&conn->msg_queue);
920 conn->callid = callid;
921 conn->close_callid = 0;
922 conn->carg = carg;
923
924 if (call)
925 conn->call = *call;
926
927 /* We will activate the fibril ASAP */
928 conn->wdata.active = true;
929 conn->cfibril = cfibril;
930 conn->wdata.fid = fibril_create(connection_fibril, conn);
931
932 if (conn->wdata.fid == 0) {
933 free(conn);
934
935 if (callid)
936 ipc_answer_0(callid, ENOMEM);
937
938 return (uintptr_t) NULL;
939 }
940
941 /* Add connection to the connection hash table */
942
943 futex_down(&async_futex);
944 hash_table_insert(&conn_hash_table, &conn->link);
945 futex_up(&async_futex);
946
947 fibril_add_ready(conn->wdata.fid);
948
949 return conn->wdata.fid;
950}
951
952/** Handle a call that was received.
953 *
954 * If the call has the IPC_M_CONNECT_ME_TO method, a new connection is created.
955 * Otherwise the call is routed to its connection fibril.
956 *
957 * @param callid Hash of the incoming call.
958 * @param call Data of the incoming call.
959 *
960 */
961static void handle_call(ipc_callid_t callid, ipc_call_t *call)
962{
963 assert(call);
964
965 /* Unrouted call - take some default action */
966 if ((callid & IPC_CALLID_NOTIFICATION)) {
967 process_notification(callid, call);
968 return;
969 }
970
971 switch (IPC_GET_IMETHOD(*call)) {
972 case IPC_M_CLONE_ESTABLISH:
973 case IPC_M_CONNECT_ME_TO:
974 /* Open new connection with fibril, etc. */
975 async_new_connection(call->in_task_id, IPC_GET_ARG5(*call),
976 callid, call, client_connection, NULL);
977 return;
978 }
979
980 /* Try to route the call through the connection hash table */
981 if (route_call(callid, call))
982 return;
983
984 /* Unknown call from unknown phone - hang it up */
985 ipc_answer_0(callid, EHANGUP);
986}
987
988/** Fire all timeouts that expired. */
989static void handle_expired_timeouts(void)
990{
991 struct timeval tv;
992 gettimeofday(&tv, NULL);
993
994 futex_down(&async_futex);
995
996 link_t *cur = list_first(&timeout_list);
997 while (cur != NULL) {
998 awaiter_t *waiter =
999 list_get_instance(cur, awaiter_t, to_event.link);
1000
1001 if (tv_gt(&waiter->to_event.expires, &tv))
1002 break;
1003
1004 list_remove(&waiter->to_event.link);
1005 waiter->to_event.inlist = false;
1006 waiter->to_event.occurred = true;
1007
1008 /*
1009 * Redundant condition?
1010 * The fibril should not be active when it gets here.
1011 */
1012 if (!waiter->active) {
1013 waiter->active = true;
1014 fibril_add_ready(waiter->fid);
1015 }
1016
1017 cur = list_first(&timeout_list);
1018 }
1019
1020 futex_up(&async_futex);
1021}
1022
1023/** Endless loop dispatching incoming calls and answers.
1024 *
1025 * @return Never returns.
1026 *
1027 */
1028static int async_manager_worker(void)
1029{
1030 while (true) {
1031 if (fibril_switch(FIBRIL_FROM_MANAGER)) {
1032 futex_up(&async_futex);
1033 /*
1034 * async_futex is always held when entering a manager
1035 * fibril.
1036 */
1037 continue;
1038 }
1039
1040 futex_down(&async_futex);
1041
1042 suseconds_t timeout;
1043 unsigned int flags = SYNCH_FLAGS_NONE;
1044 if (!list_empty(&timeout_list)) {
1045 awaiter_t *waiter = list_get_instance(
1046 list_first(&timeout_list), awaiter_t, to_event.link);
1047
1048 struct timeval tv;
1049 gettimeofday(&tv, NULL);
1050
1051 if (tv_gteq(&tv, &waiter->to_event.expires)) {
1052 futex_up(&async_futex);
1053 handle_expired_timeouts();
1054 /*
1055 * Notice that even if the event(s) already
1056 * expired (and thus the other fibril was
1057 * supposed to be running already),
1058 * we check for incoming IPC.
1059 *
1060 * Otherwise, a fibril that continuously
1061 * creates (almost) expired events could
1062 * prevent IPC retrieval from the kernel.
1063 */
1064 timeout = 0;
1065 flags = SYNCH_FLAGS_NON_BLOCKING;
1066
1067 } else {
1068 timeout = tv_sub(&waiter->to_event.expires, &tv);
1069 futex_up(&async_futex);
1070 }
1071 } else {
1072 futex_up(&async_futex);
1073 timeout = SYNCH_NO_TIMEOUT;
1074 }
1075
1076 atomic_inc(&threads_in_ipc_wait);
1077
1078 ipc_call_t call;
1079 ipc_callid_t callid = ipc_wait_cycle(&call, timeout, flags);
1080
1081 atomic_dec(&threads_in_ipc_wait);
1082
1083 if (!callid) {
1084 handle_expired_timeouts();
1085 continue;
1086 }
1087
1088 if (callid & IPC_CALLID_ANSWERED)
1089 continue;
1090
1091 handle_call(callid, &call);
1092 }
1093
1094 return 0;
1095}
1096
1097/** Function to start async_manager as a standalone fibril.
1098 *
1099 * When more kernel threads are used, one async manager should exist per thread.
1100 *
1101 * @param arg Unused.
1102 * @return Never returns.
1103 *
1104 */
1105static int async_manager_fibril(void *arg)
1106{
1107 futex_up(&async_futex);
1108
1109 /*
1110 * async_futex is always locked when entering manager
1111 */
1112 async_manager_worker();
1113
1114 return 0;
1115}
1116
1117/** Add one manager to manager list. */
1118void async_create_manager(void)
1119{
1120 fid_t fid = fibril_create(async_manager_fibril, NULL);
1121 if (fid != 0)
1122 fibril_add_manager(fid);
1123}
1124
1125/** Remove one manager from manager list */
1126void async_destroy_manager(void)
1127{
1128 fibril_remove_manager();
1129}
1130
1131/** Initialize the async framework.
1132 *
1133 */
1134void __async_init(void)
1135{
1136 if (!hash_table_create(&client_hash_table, 0, 2, &client_hash_table_ops))
1137 abort();
1138
1139 if (!hash_table_create(&conn_hash_table, 0, 1, &conn_hash_table_ops))
1140 abort();
1141
1142 session_ns = (async_sess_t *) malloc(sizeof(async_sess_t));
1143 if (session_ns == NULL)
1144 abort();
1145
1146 session_ns->mgmt = EXCHANGE_ATOMIC;
1147 session_ns->phone = PHONE_NS;
1148 session_ns->arg1 = 0;
1149 session_ns->arg2 = 0;
1150 session_ns->arg3 = 0;
1151
1152 fibril_mutex_initialize(&session_ns->remote_state_mtx);
1153 session_ns->remote_state_data = NULL;
1154
1155 list_initialize(&session_ns->exch_list);
1156 fibril_mutex_initialize(&session_ns->mutex);
1157 atomic_set(&session_ns->refcnt, 0);
1158}
1159
1160/** Reply received callback.
1161 *
1162 * This function is called whenever a reply for an asynchronous message sent out
1163 * by the asynchronous framework is received.
1164 *
1165 * Notify the fibril which is waiting for this message that it has arrived.
1166 *
1167 * @param arg Pointer to the asynchronous message record.
1168 * @param retval Value returned in the answer.
1169 * @param data Call data of the answer.
1170 *
1171 */
1172void reply_received(void *arg, int retval, ipc_call_t *data)
1173{
1174 assert(arg);
1175
1176 futex_down(&async_futex);
1177
1178 amsg_t *msg = (amsg_t *) arg;
1179 msg->retval = retval;
1180
1181 /* Copy data after futex_down, just in case the call was detached */
1182 if ((msg->dataptr) && (data))
1183 *msg->dataptr = *data;
1184
1185 write_barrier();
1186
1187 /* Remove message from timeout list */
1188 if (msg->wdata.to_event.inlist)
1189 list_remove(&msg->wdata.to_event.link);
1190
1191 msg->done = true;
1192
1193 if (msg->forget) {
1194 assert(msg->wdata.active);
1195 amsg_destroy(msg);
1196 } else if (!msg->wdata.active) {
1197 msg->wdata.active = true;
1198 fibril_add_ready(msg->wdata.fid);
1199 }
1200
1201 futex_up(&async_futex);
1202}
1203
1204/** Send message and return id of the sent message.
1205 *
1206 * The return value can be used as input for async_wait() to wait for
1207 * completion.
1208 *
1209 * @param exch Exchange for sending the message.
1210 * @param imethod Service-defined interface and method.
1211 * @param arg1 Service-defined payload argument.
1212 * @param arg2 Service-defined payload argument.
1213 * @param arg3 Service-defined payload argument.
1214 * @param arg4 Service-defined payload argument.
1215 * @param dataptr If non-NULL, storage where the reply data will be
1216 * stored.
1217 *
1218 * @return Hash of the sent message or 0 on error.
1219 *
1220 */
1221aid_t async_send_fast(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1222 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, ipc_call_t *dataptr)
1223{
1224 if (exch == NULL)
1225 return 0;
1226
1227 amsg_t *msg = amsg_create();
1228 if (msg == NULL)
1229 return 0;
1230
1231 msg->dataptr = dataptr;
1232 msg->wdata.active = true;
1233
1234 ipc_call_async_4(exch->phone, imethod, arg1, arg2, arg3, arg4, msg,
1235 reply_received, true);
1236
1237 return (aid_t) msg;
1238}
1239
1240/** Send message and return id of the sent message
1241 *
1242 * The return value can be used as input for async_wait() to wait for
1243 * completion.
1244 *
1245 * @param exch Exchange for sending the message.
1246 * @param imethod Service-defined interface and method.
1247 * @param arg1 Service-defined payload argument.
1248 * @param arg2 Service-defined payload argument.
1249 * @param arg3 Service-defined payload argument.
1250 * @param arg4 Service-defined payload argument.
1251 * @param arg5 Service-defined payload argument.
1252 * @param dataptr If non-NULL, storage where the reply data will be
1253 * stored.
1254 *
1255 * @return Hash of the sent message or 0 on error.
1256 *
1257 */
1258aid_t async_send_slow(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1259 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5,
1260 ipc_call_t *dataptr)
1261{
1262 if (exch == NULL)
1263 return 0;
1264
1265 amsg_t *msg = amsg_create();
1266 if (msg == NULL)
1267 return 0;
1268
1269 msg->dataptr = dataptr;
1270 msg->wdata.active = true;
1271
1272 ipc_call_async_5(exch->phone, imethod, arg1, arg2, arg3, arg4, arg5,
1273 msg, reply_received, true);
1274
1275 return (aid_t) msg;
1276}
1277
1278/** Wait for a message sent by the async framework.
1279 *
1280 * @param amsgid Hash of the message to wait for.
1281 * @param retval Pointer to storage where the retval of the answer will
1282 * be stored.
1283 *
1284 */
1285void async_wait_for(aid_t amsgid, sysarg_t *retval)
1286{
1287 assert(amsgid);
1288
1289 amsg_t *msg = (amsg_t *) amsgid;
1290
1291 futex_down(&async_futex);
1292
1293 assert(!msg->forget);
1294 assert(!msg->destroyed);
1295
1296 if (msg->done) {
1297 futex_up(&async_futex);
1298 goto done;
1299 }
1300
1301 msg->wdata.fid = fibril_get_id();
1302 msg->wdata.active = false;
1303 msg->wdata.to_event.inlist = false;
1304
1305 /* Leave the async_futex locked when entering this function */
1306 fibril_switch(FIBRIL_TO_MANAGER);
1307
1308 /* Futex is up automatically after fibril_switch */
1309
1310done:
1311 if (retval)
1312 *retval = msg->retval;
1313
1314 amsg_destroy(msg);
1315}
1316
1317/** Wait for a message sent by the async framework, timeout variant.
1318 *
1319 * If the wait times out, the caller may choose to either wait again by calling
1320 * async_wait_for() or async_wait_timeout(), or forget the message via
1321 * async_forget().
1322 *
1323 * @param amsgid Hash of the message to wait for.
1324 * @param retval Pointer to storage where the retval of the answer will
1325 * be stored.
1326 * @param timeout Timeout in microseconds.
1327 *
1328 * @return Zero on success, ETIMEOUT if the timeout has expired.
1329 *
1330 */
1331int async_wait_timeout(aid_t amsgid, sysarg_t *retval, suseconds_t timeout)
1332{
1333 assert(amsgid);
1334
1335 amsg_t *msg = (amsg_t *) amsgid;
1336
1337 futex_down(&async_futex);
1338
1339 assert(!msg->forget);
1340 assert(!msg->destroyed);
1341
1342 if (msg->done) {
1343 futex_up(&async_futex);
1344 goto done;
1345 }
1346
1347 /*
1348 * Negative timeout is converted to zero timeout to avoid
1349 * using tv_add with negative augmenter.
1350 */
1351 if (timeout < 0)
1352 timeout = 0;
1353
1354 gettimeofday(&msg->wdata.to_event.expires, NULL);
1355 tv_add(&msg->wdata.to_event.expires, timeout);
1356
1357 /*
1358 * Current fibril is inserted as waiting regardless of the
1359 * "size" of the timeout.
1360 *
1361 * Checking for msg->done and immediately bailing out when
1362 * timeout == 0 would mean that the manager fibril would never
1363 * run (consider single threaded program).
1364 * Thus the IPC answer would be never retrieved from the kernel.
1365 *
1366 * Notice that the actual delay would be very small because we
1367 * - switch to manager fibril
1368 * - the manager sees expired timeout
1369 * - and thus adds us back to ready queue
1370 * - manager switches back to some ready fibril
1371 * (prior it, it checks for incoming IPC).
1372 *
1373 */
1374 msg->wdata.fid = fibril_get_id();
1375 msg->wdata.active = false;
1376 async_insert_timeout(&msg->wdata);
1377
1378 /* Leave the async_futex locked when entering this function */
1379 fibril_switch(FIBRIL_TO_MANAGER);
1380
1381 /* Futex is up automatically after fibril_switch */
1382
1383 if (!msg->done)
1384 return ETIMEOUT;
1385
1386done:
1387 if (retval)
1388 *retval = msg->retval;
1389
1390 amsg_destroy(msg);
1391
1392 return 0;
1393}
1394
1395/** Discard the message / reply on arrival.
1396 *
1397 * The message will be marked to be discarded once the reply arrives in
1398 * reply_received(). It is not allowed to call async_wait_for() or
1399 * async_wait_timeout() on this message after a call to this function.
1400 *
1401 * @param amsgid Hash of the message to forget.
1402 */
1403void async_forget(aid_t amsgid)
1404{
1405 amsg_t *msg = (amsg_t *) amsgid;
1406
1407 assert(msg);
1408 assert(!msg->forget);
1409 assert(!msg->destroyed);
1410
1411 futex_down(&async_futex);
1412 if (msg->done) {
1413 amsg_destroy(msg);
1414 } else {
1415 msg->dataptr = NULL;
1416 msg->forget = true;
1417 }
1418 futex_up(&async_futex);
1419}
1420
1421/** Wait for specified time.
1422 *
1423 * The current fibril is suspended but the thread continues to execute.
1424 *
1425 * @param timeout Duration of the wait in microseconds.
1426 *
1427 */
1428void async_usleep(suseconds_t timeout)
1429{
1430 amsg_t *msg = amsg_create();
1431 if (!msg)
1432 return;
1433
1434 msg->wdata.fid = fibril_get_id();
1435
1436 gettimeofday(&msg->wdata.to_event.expires, NULL);
1437 tv_add(&msg->wdata.to_event.expires, timeout);
1438
1439 futex_down(&async_futex);
1440
1441 async_insert_timeout(&msg->wdata);
1442
1443 /* Leave the async_futex locked when entering this function */
1444 fibril_switch(FIBRIL_TO_MANAGER);
1445
1446 /* Futex is up automatically after fibril_switch() */
1447
1448 amsg_destroy(msg);
1449}
1450
1451/** Pseudo-synchronous message sending - fast version.
1452 *
1453 * Send message asynchronously and return only after the reply arrives.
1454 *
1455 * This function can only transfer 4 register payload arguments. For
1456 * transferring more arguments, see the slower async_req_slow().
1457 *
1458 * @param exch Exchange for sending the message.
1459 * @param imethod Interface and method of the call.
1460 * @param arg1 Service-defined payload argument.
1461 * @param arg2 Service-defined payload argument.
1462 * @param arg3 Service-defined payload argument.
1463 * @param arg4 Service-defined payload argument.
1464 * @param r1 If non-NULL, storage for the 1st reply argument.
1465 * @param r2 If non-NULL, storage for the 2nd reply argument.
1466 * @param r3 If non-NULL, storage for the 3rd reply argument.
1467 * @param r4 If non-NULL, storage for the 4th reply argument.
1468 * @param r5 If non-NULL, storage for the 5th reply argument.
1469 *
1470 * @return Return code of the reply or a negative error code.
1471 *
1472 */
1473sysarg_t async_req_fast(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1474 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t *r1, sysarg_t *r2,
1475 sysarg_t *r3, sysarg_t *r4, sysarg_t *r5)
1476{
1477 if (exch == NULL)
1478 return ENOENT;
1479
1480 ipc_call_t result;
1481 aid_t aid = async_send_4(exch, imethod, arg1, arg2, arg3, arg4,
1482 &result);
1483
1484 sysarg_t rc;
1485 async_wait_for(aid, &rc);
1486
1487 if (r1)
1488 *r1 = IPC_GET_ARG1(result);
1489
1490 if (r2)
1491 *r2 = IPC_GET_ARG2(result);
1492
1493 if (r3)
1494 *r3 = IPC_GET_ARG3(result);
1495
1496 if (r4)
1497 *r4 = IPC_GET_ARG4(result);
1498
1499 if (r5)
1500 *r5 = IPC_GET_ARG5(result);
1501
1502 return rc;
1503}
1504
1505/** Pseudo-synchronous message sending - slow version.
1506 *
1507 * Send message asynchronously and return only after the reply arrives.
1508 *
1509 * @param exch Exchange for sending the message.
1510 * @param imethod Interface and method of the call.
1511 * @param arg1 Service-defined payload argument.
1512 * @param arg2 Service-defined payload argument.
1513 * @param arg3 Service-defined payload argument.
1514 * @param arg4 Service-defined payload argument.
1515 * @param arg5 Service-defined payload argument.
1516 * @param r1 If non-NULL, storage for the 1st reply argument.
1517 * @param r2 If non-NULL, storage for the 2nd reply argument.
1518 * @param r3 If non-NULL, storage for the 3rd reply argument.
1519 * @param r4 If non-NULL, storage for the 4th reply argument.
1520 * @param r5 If non-NULL, storage for the 5th reply argument.
1521 *
1522 * @return Return code of the reply or a negative error code.
1523 *
1524 */
1525sysarg_t async_req_slow(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1526 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5, sysarg_t *r1,
1527 sysarg_t *r2, sysarg_t *r3, sysarg_t *r4, sysarg_t *r5)
1528{
1529 if (exch == NULL)
1530 return ENOENT;
1531
1532 ipc_call_t result;
1533 aid_t aid = async_send_5(exch, imethod, arg1, arg2, arg3, arg4, arg5,
1534 &result);
1535
1536 sysarg_t rc;
1537 async_wait_for(aid, &rc);
1538
1539 if (r1)
1540 *r1 = IPC_GET_ARG1(result);
1541
1542 if (r2)
1543 *r2 = IPC_GET_ARG2(result);
1544
1545 if (r3)
1546 *r3 = IPC_GET_ARG3(result);
1547
1548 if (r4)
1549 *r4 = IPC_GET_ARG4(result);
1550
1551 if (r5)
1552 *r5 = IPC_GET_ARG5(result);
1553
1554 return rc;
1555}
1556
1557void async_msg_0(async_exch_t *exch, sysarg_t imethod)
1558{
1559 if (exch != NULL)
1560 ipc_call_async_0(exch->phone, imethod, NULL, NULL, true);
1561}
1562
1563void async_msg_1(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1)
1564{
1565 if (exch != NULL)
1566 ipc_call_async_1(exch->phone, imethod, arg1, NULL, NULL, true);
1567}
1568
1569void async_msg_2(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1570 sysarg_t arg2)
1571{
1572 if (exch != NULL)
1573 ipc_call_async_2(exch->phone, imethod, arg1, arg2, NULL, NULL,
1574 true);
1575}
1576
1577void async_msg_3(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1578 sysarg_t arg2, sysarg_t arg3)
1579{
1580 if (exch != NULL)
1581 ipc_call_async_3(exch->phone, imethod, arg1, arg2, arg3, NULL,
1582 NULL, true);
1583}
1584
1585void async_msg_4(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1586 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4)
1587{
1588 if (exch != NULL)
1589 ipc_call_async_4(exch->phone, imethod, arg1, arg2, arg3, arg4,
1590 NULL, NULL, true);
1591}
1592
1593void async_msg_5(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1594 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5)
1595{
1596 if (exch != NULL)
1597 ipc_call_async_5(exch->phone, imethod, arg1, arg2, arg3, arg4,
1598 arg5, NULL, NULL, true);
1599}
1600
1601sysarg_t async_answer_0(ipc_callid_t callid, sysarg_t retval)
1602{
1603 return ipc_answer_0(callid, retval);
1604}
1605
1606sysarg_t async_answer_1(ipc_callid_t callid, sysarg_t retval, sysarg_t arg1)
1607{
1608 return ipc_answer_1(callid, retval, arg1);
1609}
1610
1611sysarg_t async_answer_2(ipc_callid_t callid, sysarg_t retval, sysarg_t arg1,
1612 sysarg_t arg2)
1613{
1614 return ipc_answer_2(callid, retval, arg1, arg2);
1615}
1616
1617sysarg_t async_answer_3(ipc_callid_t callid, sysarg_t retval, sysarg_t arg1,
1618 sysarg_t arg2, sysarg_t arg3)
1619{
1620 return ipc_answer_3(callid, retval, arg1, arg2, arg3);
1621}
1622
1623sysarg_t async_answer_4(ipc_callid_t callid, sysarg_t retval, sysarg_t arg1,
1624 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4)
1625{
1626 return ipc_answer_4(callid, retval, arg1, arg2, arg3, arg4);
1627}
1628
1629sysarg_t async_answer_5(ipc_callid_t callid, sysarg_t retval, sysarg_t arg1,
1630 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5)
1631{
1632 return ipc_answer_5(callid, retval, arg1, arg2, arg3, arg4, arg5);
1633}
1634
1635int async_forward_fast(ipc_callid_t callid, async_exch_t *exch,
1636 sysarg_t imethod, sysarg_t arg1, sysarg_t arg2, unsigned int mode)
1637{
1638 if (exch == NULL)
1639 return ENOENT;
1640
1641 return ipc_forward_fast(callid, exch->phone, imethod, arg1, arg2, mode);
1642}
1643
1644int async_forward_slow(ipc_callid_t callid, async_exch_t *exch,
1645 sysarg_t imethod, sysarg_t arg1, sysarg_t arg2, sysarg_t arg3,
1646 sysarg_t arg4, sysarg_t arg5, unsigned int mode)
1647{
1648 if (exch == NULL)
1649 return ENOENT;
1650
1651 return ipc_forward_slow(callid, exch->phone, imethod, arg1, arg2, arg3,
1652 arg4, arg5, mode);
1653}
1654
1655/** Wrapper for making IPC_M_CONNECT_TO_ME calls using the async framework.
1656 *
1657 * Ask through phone for a new connection to some service.
1658 *
1659 * @param exch Exchange for sending the message.
1660 * @param arg1 User defined argument.
1661 * @param arg2 User defined argument.
1662 * @param arg3 User defined argument.
1663 * @param client_receiver Connection handing routine.
1664 *
1665 * @return Zero on success or a negative error code.
1666 *
1667 */
1668int async_connect_to_me(async_exch_t *exch, sysarg_t arg1, sysarg_t arg2,
1669 sysarg_t arg3, async_client_conn_t client_receiver, void *carg)
1670{
1671 if (exch == NULL)
1672 return ENOENT;
1673
1674 sysarg_t phone_hash;
1675 sysarg_t rc;
1676
1677 aid_t req;
1678 ipc_call_t answer;
1679 req = async_send_3(exch, IPC_M_CONNECT_TO_ME, arg1, arg2, arg3,
1680 &answer);
1681 async_wait_for(req, &rc);
1682 if (rc != EOK)
1683 return (int) rc;
1684
1685 phone_hash = IPC_GET_ARG5(answer);
1686
1687 if (client_receiver != NULL)
1688 async_new_connection(answer.in_task_id, phone_hash, 0, NULL,
1689 client_receiver, carg);
1690
1691 return EOK;
1692}
1693
1694/** Wrapper for making IPC_M_CLONE_ESTABLISH calls using the async framework.
1695 *
1696 * Ask for a cloned connection to some service.
1697 *
1698 * @param mgmt Exchange management style.
1699 * @param exch Exchange for sending the message.
1700 *
1701 * @return New session on success or NULL on error.
1702 *
1703 */
1704async_sess_t *async_clone_establish(exch_mgmt_t mgmt, async_exch_t *exch)
1705{
1706 if (exch == NULL) {
1707 errno = ENOENT;
1708 return NULL;
1709 }
1710
1711 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
1712 if (sess == NULL) {
1713 errno = ENOMEM;
1714 return NULL;
1715 }
1716
1717 ipc_call_t result;
1718
1719 amsg_t *msg = amsg_create();
1720 if (!msg) {
1721 free(sess);
1722 errno = ENOMEM;
1723 return NULL;
1724 }
1725
1726 msg->dataptr = &result;
1727 msg->wdata.active = true;
1728
1729 ipc_call_async_0(exch->phone, IPC_M_CLONE_ESTABLISH, msg,
1730 reply_received, true);
1731
1732 sysarg_t rc;
1733 async_wait_for((aid_t) msg, &rc);
1734
1735 if (rc != EOK) {
1736 errno = rc;
1737 free(sess);
1738 return NULL;
1739 }
1740
1741 int phone = (int) IPC_GET_ARG5(result);
1742
1743 if (phone < 0) {
1744 errno = phone;
1745 free(sess);
1746 return NULL;
1747 }
1748
1749 sess->mgmt = mgmt;
1750 sess->phone = phone;
1751 sess->arg1 = 0;
1752 sess->arg2 = 0;
1753 sess->arg3 = 0;
1754
1755 fibril_mutex_initialize(&sess->remote_state_mtx);
1756 sess->remote_state_data = NULL;
1757
1758 list_initialize(&sess->exch_list);
1759 fibril_mutex_initialize(&sess->mutex);
1760 atomic_set(&sess->refcnt, 0);
1761
1762 return sess;
1763}
1764
1765static int async_connect_me_to_internal(int phone, sysarg_t arg1, sysarg_t arg2,
1766 sysarg_t arg3, sysarg_t arg4)
1767{
1768 ipc_call_t result;
1769
1770 amsg_t *msg = amsg_create();
1771 if (!msg)
1772 return ENOENT;
1773
1774 msg->dataptr = &result;
1775 msg->wdata.active = true;
1776
1777 ipc_call_async_4(phone, IPC_M_CONNECT_ME_TO, arg1, arg2, arg3, arg4,
1778 msg, reply_received, true);
1779
1780 sysarg_t rc;
1781 async_wait_for((aid_t) msg, &rc);
1782
1783 if (rc != EOK)
1784 return rc;
1785
1786 return (int) IPC_GET_ARG5(result);
1787}
1788
1789/** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
1790 *
1791 * Ask through for a new connection to some service.
1792 *
1793 * @param mgmt Exchange management style.
1794 * @param exch Exchange for sending the message.
1795 * @param arg1 User defined argument.
1796 * @param arg2 User defined argument.
1797 * @param arg3 User defined argument.
1798 *
1799 * @return New session on success or NULL on error.
1800 *
1801 */
1802async_sess_t *async_connect_me_to(exch_mgmt_t mgmt, async_exch_t *exch,
1803 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3)
1804{
1805 if (exch == NULL) {
1806 errno = ENOENT;
1807 return NULL;
1808 }
1809
1810 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
1811 if (sess == NULL) {
1812 errno = ENOMEM;
1813 return NULL;
1814 }
1815
1816 int phone = async_connect_me_to_internal(exch->phone, arg1, arg2, arg3,
1817 0);
1818
1819 if (phone < 0) {
1820 errno = phone;
1821 free(sess);
1822 return NULL;
1823 }
1824
1825 sess->mgmt = mgmt;
1826 sess->phone = phone;
1827 sess->arg1 = arg1;
1828 sess->arg2 = arg2;
1829 sess->arg3 = arg3;
1830
1831 fibril_mutex_initialize(&sess->remote_state_mtx);
1832 sess->remote_state_data = NULL;
1833
1834 list_initialize(&sess->exch_list);
1835 fibril_mutex_initialize(&sess->mutex);
1836 atomic_set(&sess->refcnt, 0);
1837
1838 return sess;
1839}
1840
1841/** Set arguments for new connections.
1842 *
1843 * FIXME This is an ugly hack to work around the problem that parallel
1844 * exchanges are implemented using parallel connections. When we create
1845 * a callback session, the framework does not know arguments for the new
1846 * connections.
1847 *
1848 * The proper solution seems to be to implement parallel exchanges using
1849 * tagging.
1850 */
1851void async_sess_args_set(async_sess_t *sess, sysarg_t arg1, sysarg_t arg2,
1852 sysarg_t arg3)
1853{
1854 sess->arg1 = arg1;
1855 sess->arg2 = arg2;
1856 sess->arg3 = arg3;
1857}
1858
1859/** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
1860 *
1861 * Ask through phone for a new connection to some service and block until
1862 * success.
1863 *
1864 * @param mgmt Exchange management style.
1865 * @param exch Exchange for sending the message.
1866 * @param arg1 User defined argument.
1867 * @param arg2 User defined argument.
1868 * @param arg3 User defined argument.
1869 *
1870 * @return New session on success or NULL on error.
1871 *
1872 */
1873async_sess_t *async_connect_me_to_blocking(exch_mgmt_t mgmt, async_exch_t *exch,
1874 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3)
1875{
1876 if (exch == NULL) {
1877 errno = ENOENT;
1878 return NULL;
1879 }
1880
1881 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
1882 if (sess == NULL) {
1883 errno = ENOMEM;
1884 return NULL;
1885 }
1886
1887 int phone = async_connect_me_to_internal(exch->phone, arg1, arg2, arg3,
1888 IPC_FLAG_BLOCKING);
1889
1890 if (phone < 0) {
1891 errno = phone;
1892 free(sess);
1893 return NULL;
1894 }
1895
1896 sess->mgmt = mgmt;
1897 sess->phone = phone;
1898 sess->arg1 = arg1;
1899 sess->arg2 = arg2;
1900 sess->arg3 = arg3;
1901
1902 fibril_mutex_initialize(&sess->remote_state_mtx);
1903 sess->remote_state_data = NULL;
1904
1905 list_initialize(&sess->exch_list);
1906 fibril_mutex_initialize(&sess->mutex);
1907 atomic_set(&sess->refcnt, 0);
1908
1909 return sess;
1910}
1911
1912/** Connect to a task specified by id.
1913 *
1914 */
1915async_sess_t *async_connect_kbox(task_id_t id)
1916{
1917 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
1918 if (sess == NULL) {
1919 errno = ENOMEM;
1920 return NULL;
1921 }
1922
1923 int phone = ipc_connect_kbox(id);
1924 if (phone < 0) {
1925 errno = phone;
1926 free(sess);
1927 return NULL;
1928 }
1929
1930 sess->mgmt = EXCHANGE_ATOMIC;
1931 sess->phone = phone;
1932 sess->arg1 = 0;
1933 sess->arg2 = 0;
1934 sess->arg3 = 0;
1935
1936 fibril_mutex_initialize(&sess->remote_state_mtx);
1937 sess->remote_state_data = NULL;
1938
1939 list_initialize(&sess->exch_list);
1940 fibril_mutex_initialize(&sess->mutex);
1941 atomic_set(&sess->refcnt, 0);
1942
1943 return sess;
1944}
1945
1946static int async_hangup_internal(int phone)
1947{
1948 return ipc_hangup(phone);
1949}
1950
1951/** Wrapper for ipc_hangup.
1952 *
1953 * @param sess Session to hung up.
1954 *
1955 * @return Zero on success or a negative error code.
1956 *
1957 */
1958int async_hangup(async_sess_t *sess)
1959{
1960 async_exch_t *exch;
1961
1962 assert(sess);
1963
1964 if (atomic_get(&sess->refcnt) > 0)
1965 return EBUSY;
1966
1967 fibril_mutex_lock(&async_sess_mutex);
1968
1969 int rc = async_hangup_internal(sess->phone);
1970
1971 while (!list_empty(&sess->exch_list)) {
1972 exch = (async_exch_t *)
1973 list_get_instance(list_first(&sess->exch_list),
1974 async_exch_t, sess_link);
1975
1976 list_remove(&exch->sess_link);
1977 list_remove(&exch->global_link);
1978 async_hangup_internal(exch->phone);
1979 free(exch);
1980 }
1981
1982 free(sess);
1983
1984 fibril_mutex_unlock(&async_sess_mutex);
1985
1986 return rc;
1987}
1988
1989/** Interrupt one thread of this task from waiting for IPC. */
1990void async_poke(void)
1991{
1992 ipc_poke();
1993}
1994
1995/** Start new exchange in a session.
1996 *
1997 * @param session Session.
1998 *
1999 * @return New exchange or NULL on error.
2000 *
2001 */
2002async_exch_t *async_exchange_begin(async_sess_t *sess)
2003{
2004 if (sess == NULL)
2005 return NULL;
2006
2007 async_exch_t *exch;
2008
2009 fibril_mutex_lock(&async_sess_mutex);
2010
2011 if (!list_empty(&sess->exch_list)) {
2012 /*
2013 * There are inactive exchanges in the session.
2014 */
2015 exch = (async_exch_t *)
2016 list_get_instance(list_first(&sess->exch_list),
2017 async_exch_t, sess_link);
2018
2019 list_remove(&exch->sess_link);
2020 list_remove(&exch->global_link);
2021 } else {
2022 /*
2023 * There are no available exchanges in the session.
2024 */
2025
2026 if ((sess->mgmt == EXCHANGE_ATOMIC) ||
2027 (sess->mgmt == EXCHANGE_SERIALIZE)) {
2028 exch = (async_exch_t *) malloc(sizeof(async_exch_t));
2029 if (exch != NULL) {
2030 link_initialize(&exch->sess_link);
2031 link_initialize(&exch->global_link);
2032 exch->sess = sess;
2033 exch->phone = sess->phone;
2034 }
2035 } else { /* EXCHANGE_PARALLEL */
2036 /*
2037 * Make a one-time attempt to connect a new data phone.
2038 */
2039
2040 int phone;
2041
2042retry:
2043 phone = async_connect_me_to_internal(sess->phone, sess->arg1,
2044 sess->arg2, sess->arg3, 0);
2045 if (phone >= 0) {
2046 exch = (async_exch_t *) malloc(sizeof(async_exch_t));
2047 if (exch != NULL) {
2048 link_initialize(&exch->sess_link);
2049 link_initialize(&exch->global_link);
2050 exch->sess = sess;
2051 exch->phone = phone;
2052 } else
2053 async_hangup_internal(phone);
2054 } else if (!list_empty(&inactive_exch_list)) {
2055 /*
2056 * We did not manage to connect a new phone. But we
2057 * can try to close some of the currently inactive
2058 * connections in other sessions and try again.
2059 */
2060 exch = (async_exch_t *)
2061 list_get_instance(list_first(&inactive_exch_list),
2062 async_exch_t, global_link);
2063
2064 list_remove(&exch->sess_link);
2065 list_remove(&exch->global_link);
2066 async_hangup_internal(exch->phone);
2067 free(exch);
2068 goto retry;
2069 } else {
2070 /*
2071 * Wait for a phone to become available.
2072 */
2073 fibril_condvar_wait(&avail_phone_cv, &async_sess_mutex);
2074 goto retry;
2075 }
2076 }
2077 }
2078
2079 fibril_mutex_unlock(&async_sess_mutex);
2080
2081 if (exch != NULL) {
2082 atomic_inc(&sess->refcnt);
2083
2084 if (sess->mgmt == EXCHANGE_SERIALIZE)
2085 fibril_mutex_lock(&sess->mutex);
2086 }
2087
2088 return exch;
2089}
2090
2091/** Finish an exchange.
2092 *
2093 * @param exch Exchange to finish.
2094 *
2095 */
2096void async_exchange_end(async_exch_t *exch)
2097{
2098 if (exch == NULL)
2099 return;
2100
2101 async_sess_t *sess = exch->sess;
2102
2103 atomic_dec(&sess->refcnt);
2104
2105 if (sess->mgmt == EXCHANGE_SERIALIZE)
2106 fibril_mutex_unlock(&sess->mutex);
2107
2108 fibril_mutex_lock(&async_sess_mutex);
2109
2110 list_append(&exch->sess_link, &sess->exch_list);
2111 list_append(&exch->global_link, &inactive_exch_list);
2112 fibril_condvar_signal(&avail_phone_cv);
2113
2114 fibril_mutex_unlock(&async_sess_mutex);
2115}
2116
2117/** Wrapper for IPC_M_SHARE_IN calls using the async framework.
2118 *
2119 * @param exch Exchange for sending the message.
2120 * @param size Size of the destination address space area.
2121 * @param arg User defined argument.
2122 * @param flags Storage for the received flags. Can be NULL.
2123 * @param dst Destination address space area base. Cannot be NULL.
2124 *
2125 * @return Zero on success or a negative error code from errno.h.
2126 *
2127 */
2128int async_share_in_start(async_exch_t *exch, size_t size, sysarg_t arg,
2129 unsigned int *flags, void **dst)
2130{
2131 if (exch == NULL)
2132 return ENOENT;
2133
2134 sysarg_t _flags = 0;
2135 sysarg_t _dst = (sysarg_t) -1;
2136 int res = async_req_2_4(exch, IPC_M_SHARE_IN, (sysarg_t) size,
2137 arg, NULL, &_flags, NULL, &_dst);
2138
2139 if (flags)
2140 *flags = (unsigned int) _flags;
2141
2142 *dst = (void *) _dst;
2143 return res;
2144}
2145
2146/** Wrapper for receiving the IPC_M_SHARE_IN calls using the async framework.
2147 *
2148 * This wrapper only makes it more comfortable to receive IPC_M_SHARE_IN
2149 * calls so that the user doesn't have to remember the meaning of each IPC
2150 * argument.
2151 *
2152 * So far, this wrapper is to be used from within a connection fibril.
2153 *
2154 * @param callid Storage for the hash of the IPC_M_SHARE_IN call.
2155 * @param size Destination address space area size.
2156 *
2157 * @return True on success, false on failure.
2158 *
2159 */
2160bool async_share_in_receive(ipc_callid_t *callid, size_t *size)
2161{
2162 assert(callid);
2163 assert(size);
2164
2165 ipc_call_t data;
2166 *callid = async_get_call(&data);
2167
2168 if (IPC_GET_IMETHOD(data) != IPC_M_SHARE_IN)
2169 return false;
2170
2171 *size = (size_t) IPC_GET_ARG1(data);
2172 return true;
2173}
2174
2175/** Wrapper for answering the IPC_M_SHARE_IN calls using the async framework.
2176 *
2177 * This wrapper only makes it more comfortable to answer IPC_M_SHARE_IN
2178 * calls so that the user doesn't have to remember the meaning of each IPC
2179 * argument.
2180 *
2181 * @param callid Hash of the IPC_M_DATA_READ call to answer.
2182 * @param src Source address space base.
2183 * @param flags Flags to be used for sharing. Bits can be only cleared.
2184 *
2185 * @return Zero on success or a value from @ref errno.h on failure.
2186 *
2187 */
2188int async_share_in_finalize(ipc_callid_t callid, void *src, unsigned int flags)
2189{
2190 return ipc_share_in_finalize(callid, src, flags);
2191}
2192
2193/** Wrapper for IPC_M_SHARE_OUT calls using the async framework.
2194 *
2195 * @param exch Exchange for sending the message.
2196 * @param src Source address space area base address.
2197 * @param flags Flags to be used for sharing. Bits can be only cleared.
2198 *
2199 * @return Zero on success or a negative error code from errno.h.
2200 *
2201 */
2202int async_share_out_start(async_exch_t *exch, void *src, unsigned int flags)
2203{
2204 if (exch == NULL)
2205 return ENOENT;
2206
2207 return async_req_3_0(exch, IPC_M_SHARE_OUT, (sysarg_t) src, 0,
2208 (sysarg_t) flags);
2209}
2210
2211/** Wrapper for receiving the IPC_M_SHARE_OUT calls using the async framework.
2212 *
2213 * This wrapper only makes it more comfortable to receive IPC_M_SHARE_OUT
2214 * calls so that the user doesn't have to remember the meaning of each IPC
2215 * argument.
2216 *
2217 * So far, this wrapper is to be used from within a connection fibril.
2218 *
2219 * @param callid Storage for the hash of the IPC_M_SHARE_OUT call.
2220 * @param size Storage for the source address space area size.
2221 * @param flags Storage for the sharing flags.
2222 *
2223 * @return True on success, false on failure.
2224 *
2225 */
2226bool async_share_out_receive(ipc_callid_t *callid, size_t *size, unsigned int *flags)
2227{
2228 assert(callid);
2229 assert(size);
2230 assert(flags);
2231
2232 ipc_call_t data;
2233 *callid = async_get_call(&data);
2234
2235 if (IPC_GET_IMETHOD(data) != IPC_M_SHARE_OUT)
2236 return false;
2237
2238 *size = (size_t) IPC_GET_ARG2(data);
2239 *flags = (unsigned int) IPC_GET_ARG3(data);
2240 return true;
2241}
2242
2243/** Wrapper for answering the IPC_M_SHARE_OUT calls using the async framework.
2244 *
2245 * This wrapper only makes it more comfortable to answer IPC_M_SHARE_OUT
2246 * calls so that the user doesn't have to remember the meaning of each IPC
2247 * argument.
2248 *
2249 * @param callid Hash of the IPC_M_DATA_WRITE call to answer.
2250 * @param dst Destination address space area base address.
2251 *
2252 * @return Zero on success or a value from @ref errno.h on failure.
2253 *
2254 */
2255int async_share_out_finalize(ipc_callid_t callid, void **dst)
2256{
2257 return ipc_share_out_finalize(callid, dst);
2258}
2259
2260/** Start IPC_M_DATA_READ using the async framework.
2261 *
2262 * @param exch Exchange for sending the message.
2263 * @param dst Address of the beginning of the destination buffer.
2264 * @param size Size of the destination buffer (in bytes).
2265 * @param dataptr Storage of call data (arg 2 holds actual data size).
2266 *
2267 * @return Hash of the sent message or 0 on error.
2268 *
2269 */
2270aid_t async_data_read(async_exch_t *exch, void *dst, size_t size,
2271 ipc_call_t *dataptr)
2272{
2273 return async_send_2(exch, IPC_M_DATA_READ, (sysarg_t) dst,
2274 (sysarg_t) size, dataptr);
2275}
2276
2277/** Wrapper for IPC_M_DATA_READ calls using the async framework.
2278 *
2279 * @param exch Exchange for sending the message.
2280 * @param dst Address of the beginning of the destination buffer.
2281 * @param size Size of the destination buffer.
2282 *
2283 * @return Zero on success or a negative error code from errno.h.
2284 *
2285 */
2286int async_data_read_start(async_exch_t *exch, void *dst, size_t size)
2287{
2288 if (exch == NULL)
2289 return ENOENT;
2290
2291 return async_req_2_0(exch, IPC_M_DATA_READ, (sysarg_t) dst,
2292 (sysarg_t) size);
2293}
2294
2295/** Wrapper for receiving the IPC_M_DATA_READ calls using the async framework.
2296 *
2297 * This wrapper only makes it more comfortable to receive IPC_M_DATA_READ
2298 * calls so that the user doesn't have to remember the meaning of each IPC
2299 * argument.
2300 *
2301 * So far, this wrapper is to be used from within a connection fibril.
2302 *
2303 * @param callid Storage for the hash of the IPC_M_DATA_READ.
2304 * @param size Storage for the maximum size. Can be NULL.
2305 *
2306 * @return True on success, false on failure.
2307 *
2308 */
2309bool async_data_read_receive(ipc_callid_t *callid, size_t *size)
2310{
2311 assert(callid);
2312
2313 ipc_call_t data;
2314 *callid = async_get_call(&data);
2315
2316 if (IPC_GET_IMETHOD(data) != IPC_M_DATA_READ)
2317 return false;
2318
2319 if (size)
2320 *size = (size_t) IPC_GET_ARG2(data);
2321
2322 return true;
2323}
2324
2325/** Wrapper for answering the IPC_M_DATA_READ calls using the async framework.
2326 *
2327 * This wrapper only makes it more comfortable to answer IPC_M_DATA_READ
2328 * calls so that the user doesn't have to remember the meaning of each IPC
2329 * argument.
2330 *
2331 * @param callid Hash of the IPC_M_DATA_READ call to answer.
2332 * @param src Source address for the IPC_M_DATA_READ call.
2333 * @param size Size for the IPC_M_DATA_READ call. Can be smaller than
2334 * the maximum size announced by the sender.
2335 *
2336 * @return Zero on success or a value from @ref errno.h on failure.
2337 *
2338 */
2339int async_data_read_finalize(ipc_callid_t callid, const void *src, size_t size)
2340{
2341 return ipc_data_read_finalize(callid, src, size);
2342}
2343
2344/** Wrapper for forwarding any read request
2345 *
2346 */
2347int async_data_read_forward_fast(async_exch_t *exch, sysarg_t imethod,
2348 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3, sysarg_t arg4,
2349 ipc_call_t *dataptr)
2350{
2351 if (exch == NULL)
2352 return ENOENT;
2353
2354 ipc_callid_t callid;
2355 if (!async_data_read_receive(&callid, NULL)) {
2356 ipc_answer_0(callid, EINVAL);
2357 return EINVAL;
2358 }
2359
2360 aid_t msg = async_send_fast(exch, imethod, arg1, arg2, arg3, arg4,
2361 dataptr);
2362 if (msg == 0) {
2363 ipc_answer_0(callid, EINVAL);
2364 return EINVAL;
2365 }
2366
2367 int retval = ipc_forward_fast(callid, exch->phone, 0, 0, 0,
2368 IPC_FF_ROUTE_FROM_ME);
2369 if (retval != EOK) {
2370 async_forget(msg);
2371 ipc_answer_0(callid, retval);
2372 return retval;
2373 }
2374
2375 sysarg_t rc;
2376 async_wait_for(msg, &rc);
2377
2378 return (int) rc;
2379}
2380
2381/** Wrapper for IPC_M_DATA_WRITE calls using the async framework.
2382 *
2383 * @param exch Exchange for sending the message.
2384 * @param src Address of the beginning of the source buffer.
2385 * @param size Size of the source buffer.
2386 *
2387 * @return Zero on success or a negative error code from errno.h.
2388 *
2389 */
2390int async_data_write_start(async_exch_t *exch, const void *src, size_t size)
2391{
2392 if (exch == NULL)
2393 return ENOENT;
2394
2395 return async_req_2_0(exch, IPC_M_DATA_WRITE, (sysarg_t) src,
2396 (sysarg_t) size);
2397}
2398
2399/** Wrapper for receiving the IPC_M_DATA_WRITE calls using the async framework.
2400 *
2401 * This wrapper only makes it more comfortable to receive IPC_M_DATA_WRITE
2402 * calls so that the user doesn't have to remember the meaning of each IPC
2403 * argument.
2404 *
2405 * So far, this wrapper is to be used from within a connection fibril.
2406 *
2407 * @param callid Storage for the hash of the IPC_M_DATA_WRITE.
2408 * @param size Storage for the suggested size. May be NULL.
2409 *
2410 * @return True on success, false on failure.
2411 *
2412 */
2413bool async_data_write_receive(ipc_callid_t *callid, size_t *size)
2414{
2415 assert(callid);
2416
2417 ipc_call_t data;
2418 *callid = async_get_call(&data);
2419
2420 if (IPC_GET_IMETHOD(data) != IPC_M_DATA_WRITE)
2421 return false;
2422
2423 if (size)
2424 *size = (size_t) IPC_GET_ARG2(data);
2425
2426 return true;
2427}
2428
2429/** Wrapper for answering the IPC_M_DATA_WRITE calls using the async framework.
2430 *
2431 * This wrapper only makes it more comfortable to answer IPC_M_DATA_WRITE
2432 * calls so that the user doesn't have to remember the meaning of each IPC
2433 * argument.
2434 *
2435 * @param callid Hash of the IPC_M_DATA_WRITE call to answer.
2436 * @param dst Final destination address for the IPC_M_DATA_WRITE call.
2437 * @param size Final size for the IPC_M_DATA_WRITE call.
2438 *
2439 * @return Zero on success or a value from @ref errno.h on failure.
2440 *
2441 */
2442int async_data_write_finalize(ipc_callid_t callid, void *dst, size_t size)
2443{
2444 return ipc_data_write_finalize(callid, dst, size);
2445}
2446
2447/** Wrapper for receiving binary data or strings
2448 *
2449 * This wrapper only makes it more comfortable to use async_data_write_*
2450 * functions to receive binary data or strings.
2451 *
2452 * @param data Pointer to data pointer (which should be later disposed
2453 * by free()). If the operation fails, the pointer is not
2454 * touched.
2455 * @param nullterm If true then the received data is always zero terminated.
2456 * This also causes to allocate one extra byte beyond the
2457 * raw transmitted data.
2458 * @param min_size Minimum size (in bytes) of the data to receive.
2459 * @param max_size Maximum size (in bytes) of the data to receive. 0 means
2460 * no limit.
2461 * @param granulariy If non-zero then the size of the received data has to
2462 * be divisible by this value.
2463 * @param received If not NULL, the size of the received data is stored here.
2464 *
2465 * @return Zero on success or a value from @ref errno.h on failure.
2466 *
2467 */
2468int async_data_write_accept(void **data, const bool nullterm,
2469 const size_t min_size, const size_t max_size, const size_t granularity,
2470 size_t *received)
2471{
2472 assert(data);
2473
2474 ipc_callid_t callid;
2475 size_t size;
2476 if (!async_data_write_receive(&callid, &size)) {
2477 ipc_answer_0(callid, EINVAL);
2478 return EINVAL;
2479 }
2480
2481 if (size < min_size) {
2482 ipc_answer_0(callid, EINVAL);
2483 return EINVAL;
2484 }
2485
2486 if ((max_size > 0) && (size > max_size)) {
2487 ipc_answer_0(callid, EINVAL);
2488 return EINVAL;
2489 }
2490
2491 if ((granularity > 0) && ((size % granularity) != 0)) {
2492 ipc_answer_0(callid, EINVAL);
2493 return EINVAL;
2494 }
2495
2496 void *_data;
2497
2498 if (nullterm)
2499 _data = malloc(size + 1);
2500 else
2501 _data = malloc(size);
2502
2503 if (_data == NULL) {
2504 ipc_answer_0(callid, ENOMEM);
2505 return ENOMEM;
2506 }
2507
2508 int rc = async_data_write_finalize(callid, _data, size);
2509 if (rc != EOK) {
2510 free(_data);
2511 return rc;
2512 }
2513
2514 if (nullterm)
2515 ((char *) _data)[size] = 0;
2516
2517 *data = _data;
2518 if (received != NULL)
2519 *received = size;
2520
2521 return EOK;
2522}
2523
2524/** Wrapper for voiding any data that is about to be received
2525 *
2526 * This wrapper can be used to void any pending data
2527 *
2528 * @param retval Error value from @ref errno.h to be returned to the caller.
2529 *
2530 */
2531void async_data_write_void(sysarg_t retval)
2532{
2533 ipc_callid_t callid;
2534 async_data_write_receive(&callid, NULL);
2535 ipc_answer_0(callid, retval);
2536}
2537
2538/** Wrapper for forwarding any data that is about to be received
2539 *
2540 */
2541int async_data_write_forward_fast(async_exch_t *exch, sysarg_t imethod,
2542 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3, sysarg_t arg4,
2543 ipc_call_t *dataptr)
2544{
2545 if (exch == NULL)
2546 return ENOENT;
2547
2548 ipc_callid_t callid;
2549 if (!async_data_write_receive(&callid, NULL)) {
2550 ipc_answer_0(callid, EINVAL);
2551 return EINVAL;
2552 }
2553
2554 aid_t msg = async_send_fast(exch, imethod, arg1, arg2, arg3, arg4,
2555 dataptr);
2556 if (msg == 0) {
2557 ipc_answer_0(callid, EINVAL);
2558 return EINVAL;
2559 }
2560
2561 int retval = ipc_forward_fast(callid, exch->phone, 0, 0, 0,
2562 IPC_FF_ROUTE_FROM_ME);
2563 if (retval != EOK) {
2564 async_forget(msg);
2565 ipc_answer_0(callid, retval);
2566 return retval;
2567 }
2568
2569 sysarg_t rc;
2570 async_wait_for(msg, &rc);
2571
2572 return (int) rc;
2573}
2574
2575/** Wrapper for sending an exchange over different exchange for cloning
2576 *
2577 * @param exch Exchange to be used for sending.
2578 * @param clone_exch Exchange to be cloned.
2579 *
2580 */
2581int async_exchange_clone(async_exch_t *exch, async_exch_t *clone_exch)
2582{
2583 return async_req_1_0(exch, IPC_M_CONNECTION_CLONE, clone_exch->phone);
2584}
2585
2586/** Wrapper for receiving the IPC_M_CONNECTION_CLONE calls.
2587 *
2588 * If the current call is IPC_M_CONNECTION_CLONE then a new
2589 * async session is created for the accepted phone.
2590 *
2591 * @param mgmt Exchange management style.
2592 *
2593 * @return New async session or NULL on failure.
2594 *
2595 */
2596async_sess_t *async_clone_receive(exch_mgmt_t mgmt)
2597{
2598 /* Accept the phone */
2599 ipc_call_t call;
2600 ipc_callid_t callid = async_get_call(&call);
2601 int phone = (int) IPC_GET_ARG1(call);
2602
2603 if ((IPC_GET_IMETHOD(call) != IPC_M_CONNECTION_CLONE) ||
2604 (phone < 0)) {
2605 async_answer_0(callid, EINVAL);
2606 return NULL;
2607 }
2608
2609 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
2610 if (sess == NULL) {
2611 async_answer_0(callid, ENOMEM);
2612 return NULL;
2613 }
2614
2615 sess->mgmt = mgmt;
2616 sess->phone = phone;
2617 sess->arg1 = 0;
2618 sess->arg2 = 0;
2619 sess->arg3 = 0;
2620
2621 fibril_mutex_initialize(&sess->remote_state_mtx);
2622 sess->remote_state_data = NULL;
2623
2624 list_initialize(&sess->exch_list);
2625 fibril_mutex_initialize(&sess->mutex);
2626 atomic_set(&sess->refcnt, 0);
2627
2628 /* Acknowledge the cloned phone */
2629 async_answer_0(callid, EOK);
2630
2631 return sess;
2632}
2633
2634/** Wrapper for receiving the IPC_M_CONNECT_TO_ME calls.
2635 *
2636 * If the current call is IPC_M_CONNECT_TO_ME then a new
2637 * async session is created for the accepted phone.
2638 *
2639 * @param mgmt Exchange management style.
2640 *
2641 * @return New async session.
2642 * @return NULL on failure.
2643 *
2644 */
2645async_sess_t *async_callback_receive(exch_mgmt_t mgmt)
2646{
2647 /* Accept the phone */
2648 ipc_call_t call;
2649 ipc_callid_t callid = async_get_call(&call);
2650 int phone = (int) IPC_GET_ARG5(call);
2651
2652 if ((IPC_GET_IMETHOD(call) != IPC_M_CONNECT_TO_ME) ||
2653 (phone < 0)) {
2654 async_answer_0(callid, EINVAL);
2655 return NULL;
2656 }
2657
2658 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
2659 if (sess == NULL) {
2660 async_answer_0(callid, ENOMEM);
2661 return NULL;
2662 }
2663
2664 sess->mgmt = mgmt;
2665 sess->phone = phone;
2666 sess->arg1 = 0;
2667 sess->arg2 = 0;
2668 sess->arg3 = 0;
2669
2670 fibril_mutex_initialize(&sess->remote_state_mtx);
2671 sess->remote_state_data = NULL;
2672
2673 list_initialize(&sess->exch_list);
2674 fibril_mutex_initialize(&sess->mutex);
2675 atomic_set(&sess->refcnt, 0);
2676
2677 /* Acknowledge the connected phone */
2678 async_answer_0(callid, EOK);
2679
2680 return sess;
2681}
2682
2683/** Wrapper for receiving the IPC_M_CONNECT_TO_ME calls.
2684 *
2685 * If the call is IPC_M_CONNECT_TO_ME then a new
2686 * async session is created. However, the phone is
2687 * not accepted automatically.
2688 *
2689 * @param mgmt Exchange management style.
2690 * @param call Call data.
2691 *
2692 * @return New async session.
2693 * @return NULL on failure.
2694 * @return NULL if the call is not IPC_M_CONNECT_TO_ME.
2695 *
2696 */
2697async_sess_t *async_callback_receive_start(exch_mgmt_t mgmt, ipc_call_t *call)
2698{
2699 int phone = (int) IPC_GET_ARG5(*call);
2700
2701 if ((IPC_GET_IMETHOD(*call) != IPC_M_CONNECT_TO_ME) ||
2702 (phone < 0))
2703 return NULL;
2704
2705 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
2706 if (sess == NULL)
2707 return NULL;
2708
2709 sess->mgmt = mgmt;
2710 sess->phone = phone;
2711 sess->arg1 = 0;
2712 sess->arg2 = 0;
2713 sess->arg3 = 0;
2714
2715 fibril_mutex_initialize(&sess->remote_state_mtx);
2716 sess->remote_state_data = NULL;
2717
2718 list_initialize(&sess->exch_list);
2719 fibril_mutex_initialize(&sess->mutex);
2720 atomic_set(&sess->refcnt, 0);
2721
2722 return sess;
2723}
2724
2725int async_state_change_start(async_exch_t *exch, sysarg_t arg1, sysarg_t arg2,
2726 sysarg_t arg3, async_exch_t *other_exch)
2727{
2728 return async_req_5_0(exch, IPC_M_STATE_CHANGE_AUTHORIZE,
2729 arg1, arg2, arg3, 0, other_exch->phone);
2730}
2731
2732bool async_state_change_receive(ipc_callid_t *callid, sysarg_t *arg1,
2733 sysarg_t *arg2, sysarg_t *arg3)
2734{
2735 assert(callid);
2736
2737 ipc_call_t call;
2738 *callid = async_get_call(&call);
2739
2740 if (IPC_GET_IMETHOD(call) != IPC_M_STATE_CHANGE_AUTHORIZE)
2741 return false;
2742
2743 if (arg1)
2744 *arg1 = IPC_GET_ARG1(call);
2745 if (arg2)
2746 *arg2 = IPC_GET_ARG2(call);
2747 if (arg3)
2748 *arg3 = IPC_GET_ARG3(call);
2749
2750 return true;
2751}
2752
2753int async_state_change_finalize(ipc_callid_t callid, async_exch_t *other_exch)
2754{
2755 return ipc_answer_1(callid, EOK, other_exch->phone);
2756}
2757
2758/** Lock and get session remote state
2759 *
2760 * Lock and get the local replica of the remote state
2761 * in stateful sessions. The call should be paired
2762 * with async_remote_state_release*().
2763 *
2764 * @param[in] sess Stateful session.
2765 *
2766 * @return Local replica of the remote state.
2767 *
2768 */
2769void *async_remote_state_acquire(async_sess_t *sess)
2770{
2771 fibril_mutex_lock(&sess->remote_state_mtx);
2772 return sess->remote_state_data;
2773}
2774
2775/** Update the session remote state
2776 *
2777 * Update the local replica of the remote state
2778 * in stateful sessions. The remote state must
2779 * be already locked.
2780 *
2781 * @param[in] sess Stateful session.
2782 * @param[in] state New local replica of the remote state.
2783 *
2784 */
2785void async_remote_state_update(async_sess_t *sess, void *state)
2786{
2787 assert(fibril_mutex_is_locked(&sess->remote_state_mtx));
2788 sess->remote_state_data = state;
2789}
2790
2791/** Release the session remote state
2792 *
2793 * Unlock the local replica of the remote state
2794 * in stateful sessions.
2795 *
2796 * @param[in] sess Stateful session.
2797 *
2798 */
2799void async_remote_state_release(async_sess_t *sess)
2800{
2801 assert(fibril_mutex_is_locked(&sess->remote_state_mtx));
2802
2803 fibril_mutex_unlock(&sess->remote_state_mtx);
2804}
2805
2806/** Release the session remote state and end an exchange
2807 *
2808 * Unlock the local replica of the remote state
2809 * in stateful sessions. This is convenience function
2810 * which gets the session pointer from the exchange
2811 * and also ends the exchange.
2812 *
2813 * @param[in] exch Stateful session's exchange.
2814 *
2815 */
2816void async_remote_state_release_exchange(async_exch_t *exch)
2817{
2818 if (exch == NULL)
2819 return;
2820
2821 async_sess_t *sess = exch->sess;
2822 assert(fibril_mutex_is_locked(&sess->remote_state_mtx));
2823
2824 async_exchange_end(exch);
2825 fibril_mutex_unlock(&sess->remote_state_mtx);
2826}
2827
2828/** @}
2829 */
Note: See TracBrowser for help on using the repository browser.