2 * Copyright (c) 2006 Ondrej Palkovsky
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
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.
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.
36 * Asynchronous library
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
42 * You should be able to write very simple multithreaded programs. The async
43 * framework will automatically take care of most of the synchronization
46 * Example of use (pseudo C):
48 * 1) Multithreaded client application
50 * fibril_create(fibril1, ...);
51 * fibril_create(fibril2, ...);
54 * int fibril1(void *arg)
56 * conn = async_connect_me_to(...);
58 * exch = async_exchange_begin(conn);
59 * c1 = async_send(exch);
60 * async_exchange_end(exch);
62 * exch = async_exchange_begin(conn);
63 * c2 = async_send(exch);
64 * async_exchange_end(exch);
72 * 2) Multithreaded server application
79 * port_handler(icallid, *icall)
82 * async_answer_0(icallid, ELIMIT);
85 * async_answer_0(icallid, EOK);
87 * callid = async_get_call(&call);
88 * somehow_handle_the_call(callid, call);
89 * async_answer_2(callid, 1, 2, 3);
91 * callid = async_get_call(&call);
100 #include "private/async.h"
104 #include <ipc/event.h>
107 #include <adt/hash_table.h>
108 #include <adt/list.h>
111 #include <sys/time.h>
112 #include <libarch/barrier.h>
119 #include <abi/mm/as.h>
120 #include "private/libc.h"
124 /** List of inactive exchanges */
127 /** Session interface */
130 /** Exchange management style */
133 /** Session identification */
136 /** First clone connection argument */
139 /** Second clone connection argument */
142 /** Third clone connection argument */
145 /** Exchange mutex */
146 fibril_mutex_t mutex
;
148 /** Number of opened exchanges */
151 /** Mutex for stateful connections */
152 fibril_mutex_t remote_state_mtx
;
154 /** Data for stateful connections */
155 void *remote_state_data
;
160 /** Link into list of inactive exchanges */
163 /** Link into global list of inactive exchanges */
166 /** Session pointer */
169 /** Exchange identification */
173 /** Async framework global futex */
174 futex_t async_futex
= FUTEX_INITIALIZER
;
176 /** Number of threads waiting for IPC in the kernel. */
177 atomic_t threads_in_ipc_wait
= { 0 };
179 /** Naming service session */
180 async_sess_t
*session_ns
;
194 /** If reply was received. */
197 /** If the message / reply should be discarded on arrival. */
200 /** If already destroyed. */
203 /** Pointer to where the answer data is stored. */
209 /* Client connection data */
213 task_id_t in_task_id
;
218 /* Server connection data */
222 /** Hash table link. */
225 /** Incoming client task ID. */
226 task_id_t in_task_id
;
228 /** Incoming phone hash. */
229 sysarg_t in_phone_hash
;
231 /** Link to the client tracking structure. */
234 /** Messages that should be delivered to this fibril. */
237 /** Identification of the opening call. */
240 /** Call data of the opening call. */
243 /** Identification of the closing call. */
244 ipc_callid_t close_callid
;
246 /** Fibril function that will be used to handle the connection. */
247 async_port_handler_t handler
;
253 /** Interface data */
260 /** Futex protecting the hash table */
263 /** Interface ports */
264 hash_table_t port_hash_table
;
266 /** Next available port ID */
267 port_id_t port_id_avail
;
277 /** Port connection handler */
278 async_port_handler_t handler
;
284 /* Notification data */
288 /** Notification method */
291 /** Notification handler */
292 async_notification_handler_t handler
;
294 /** Notification data */
298 /** Identifier of the incoming connection handled by the current fibril. */
299 static fibril_local connection_t
*fibril_connection
;
301 static void to_event_initialize(to_event_t
*to
)
303 struct timeval tv
= { 0, 0 };
306 to
->occurred
= false;
307 link_initialize(&to
->link
);
311 static void wu_event_initialize(wu_event_t
*wu
)
314 link_initialize(&wu
->link
);
317 void awaiter_initialize(awaiter_t
*aw
)
321 to_event_initialize(&aw
->to_event
);
322 wu_event_initialize(&aw
->wu_event
);
325 static amsg_t
*amsg_create(void)
327 amsg_t
*msg
= malloc(sizeof(amsg_t
));
331 msg
->destroyed
= false;
333 msg
->retval
= (sysarg_t
) EINVAL
;
334 awaiter_initialize(&msg
->wdata
);
340 static void amsg_destroy(amsg_t
*msg
)
342 assert(!msg
->destroyed
);
343 msg
->destroyed
= true;
347 static void *default_client_data_constructor(void)
352 static void default_client_data_destructor(void *data
)
356 static async_client_data_ctor_t async_client_data_create
=
357 default_client_data_constructor
;
358 static async_client_data_dtor_t async_client_data_destroy
=
359 default_client_data_destructor
;
361 void async_set_client_data_constructor(async_client_data_ctor_t ctor
)
363 assert(async_client_data_create
== default_client_data_constructor
);
364 async_client_data_create
= ctor
;
367 void async_set_client_data_destructor(async_client_data_dtor_t dtor
)
369 assert(async_client_data_destroy
== default_client_data_destructor
);
370 async_client_data_destroy
= dtor
;
373 /** Default fallback fibril function.
375 * This fallback fibril function gets called on incomming
376 * connections that do not have a specific handler defined.
378 * @param callid Hash of the incoming call.
379 * @param call Data of the incoming call.
380 * @param arg Local argument
383 static void default_fallback_port_handler(ipc_callid_t callid
, ipc_call_t
*call
,
386 ipc_answer_0(callid
, ENOENT
);
389 static async_port_handler_t fallback_port_handler
=
390 default_fallback_port_handler
;
391 static void *fallback_port_data
= NULL
;
393 static hash_table_t interface_hash_table
;
395 static size_t interface_key_hash(void *key
)
397 iface_t iface
= *(iface_t
*) key
;
401 static size_t interface_hash(const ht_link_t
*item
)
403 interface_t
*interface
= hash_table_get_inst(item
, interface_t
, link
);
404 return interface_key_hash(&interface
->iface
);
407 static bool interface_key_equal(void *key
, const ht_link_t
*item
)
409 iface_t iface
= *(iface_t
*) key
;
410 interface_t
*interface
= hash_table_get_inst(item
, interface_t
, link
);
411 return iface
== interface
->iface
;
414 /** Operations for the port hash table. */
415 static hash_table_ops_t interface_hash_table_ops
= {
416 .hash
= interface_hash
,
417 .key_hash
= interface_key_hash
,
418 .key_equal
= interface_key_equal
,
420 .remove_callback
= NULL
423 static size_t port_key_hash(void *key
)
425 port_id_t port_id
= *(port_id_t
*) key
;
429 static size_t port_hash(const ht_link_t
*item
)
431 port_t
*port
= hash_table_get_inst(item
, port_t
, link
);
432 return port_key_hash(&port
->id
);
435 static bool port_key_equal(void *key
, const ht_link_t
*item
)
437 port_id_t port_id
= *(port_id_t
*) key
;
438 port_t
*port
= hash_table_get_inst(item
, port_t
, link
);
439 return port_id
== port
->id
;
442 /** Operations for the port hash table. */
443 static hash_table_ops_t port_hash_table_ops
= {
445 .key_hash
= port_key_hash
,
446 .key_equal
= port_key_equal
,
448 .remove_callback
= NULL
451 static interface_t
*async_new_interface(iface_t iface
)
453 interface_t
*interface
=
454 (interface_t
*) malloc(sizeof(interface_t
));
458 bool ret
= hash_table_create(&interface
->port_hash_table
, 0, 0,
459 &port_hash_table_ops
);
465 interface
->iface
= iface
;
466 futex_initialize(&interface
->futex
, 1);
467 interface
->port_id_avail
= 0;
469 hash_table_insert(&interface_hash_table
, &interface
->link
);
474 static port_t
*async_new_port(interface_t
*interface
,
475 async_port_handler_t handler
, void *data
)
477 port_t
*port
= (port_t
*) malloc(sizeof(port_t
));
481 futex_down(&interface
->futex
);
483 port_id_t id
= interface
->port_id_avail
;
484 interface
->port_id_avail
++;
487 port
->handler
= handler
;
490 hash_table_insert(&interface
->port_hash_table
, &port
->link
);
492 futex_up(&interface
->futex
);
497 /** Mutex protecting inactive_exch_list and avail_phone_cv.
500 static FIBRIL_MUTEX_INITIALIZE(async_sess_mutex
);
502 /** List of all currently inactive exchanges.
505 static LIST_INITIALIZE(inactive_exch_list
);
507 /** Condition variable to wait for a phone to become available.
510 static FIBRIL_CONDVAR_INITIALIZE(avail_phone_cv
);
512 int async_create_port(iface_t iface
, async_port_handler_t handler
,
513 void *data
, port_id_t
*port_id
)
515 if ((iface
& IFACE_MOD_MASK
) == IFACE_MOD_CALLBACK
)
518 interface_t
*interface
;
520 futex_down(&async_futex
);
522 ht_link_t
*link
= hash_table_find(&interface_hash_table
, &iface
);
524 interface
= hash_table_get_inst(link
, interface_t
, link
);
526 interface
= async_new_interface(iface
);
529 futex_up(&async_futex
);
533 port_t
*port
= async_new_port(interface
, handler
, data
);
535 futex_up(&async_futex
);
541 futex_up(&async_futex
);
546 void async_set_fallback_port_handler(async_port_handler_t handler
, void *data
)
548 assert(handler
!= NULL
);
550 fallback_port_handler
= handler
;
551 fallback_port_data
= data
;
554 static hash_table_t client_hash_table
;
555 static hash_table_t conn_hash_table
;
556 static hash_table_t notification_hash_table
;
557 static LIST_INITIALIZE(timeout_list
);
559 static sysarg_t notification_avail
= 0;
561 static size_t client_key_hash(void *key
)
563 task_id_t in_task_id
= *(task_id_t
*) key
;
567 static size_t client_hash(const ht_link_t
*item
)
569 client_t
*client
= hash_table_get_inst(item
, client_t
, link
);
570 return client_key_hash(&client
->in_task_id
);
573 static bool client_key_equal(void *key
, const ht_link_t
*item
)
575 task_id_t in_task_id
= *(task_id_t
*) key
;
576 client_t
*client
= hash_table_get_inst(item
, client_t
, link
);
577 return in_task_id
== client
->in_task_id
;
580 /** Operations for the client hash table. */
581 static hash_table_ops_t client_hash_table_ops
= {
583 .key_hash
= client_key_hash
,
584 .key_equal
= client_key_equal
,
586 .remove_callback
= NULL
589 /** Compute hash into the connection hash table based on the source phone hash.
591 * @param key Pointer to source phone hash.
593 * @return Index into the connection hash table.
596 static size_t conn_key_hash(void *key
)
598 sysarg_t in_phone_hash
= *(sysarg_t
*) key
;
599 return in_phone_hash
;
602 static size_t conn_hash(const ht_link_t
*item
)
604 connection_t
*conn
= hash_table_get_inst(item
, connection_t
, link
);
605 return conn_key_hash(&conn
->in_phone_hash
);
608 static bool conn_key_equal(void *key
, const ht_link_t
*item
)
610 sysarg_t in_phone_hash
= *(sysarg_t
*) key
;
611 connection_t
*conn
= hash_table_get_inst(item
, connection_t
, link
);
612 return (in_phone_hash
== conn
->in_phone_hash
);
615 /** Operations for the connection hash table. */
616 static hash_table_ops_t conn_hash_table_ops
= {
618 .key_hash
= conn_key_hash
,
619 .key_equal
= conn_key_equal
,
621 .remove_callback
= NULL
624 static client_t
*async_client_get(task_id_t client_id
, bool create
)
626 client_t
*client
= NULL
;
628 futex_down(&async_futex
);
629 ht_link_t
*link
= hash_table_find(&client_hash_table
, &client_id
);
631 client
= hash_table_get_inst(link
, client_t
, link
);
632 atomic_inc(&client
->refcnt
);
634 client
= malloc(sizeof(client_t
));
636 client
->in_task_id
= client_id
;
637 client
->data
= async_client_data_create();
639 atomic_set(&client
->refcnt
, 1);
640 hash_table_insert(&client_hash_table
, &client
->link
);
644 futex_up(&async_futex
);
648 static void async_client_put(client_t
*client
)
652 futex_down(&async_futex
);
654 if (atomic_predec(&client
->refcnt
) == 0) {
655 hash_table_remove(&client_hash_table
, &client
->in_task_id
);
660 futex_up(&async_futex
);
664 async_client_data_destroy(client
->data
);
670 /** Wrapper for client connection fibril.
672 * When a new connection arrives, a fibril with this implementing
673 * function is created.
675 * @param arg Connection structure pointer.
677 * @return Always zero.
680 static int connection_fibril(void *arg
)
685 * Setup fibril-local connection pointer.
687 fibril_connection
= (connection_t
*) arg
;
690 * Add our reference for the current connection in the client task
691 * tracking structure. If this is the first reference, create and
692 * hash in a new tracking structure.
695 client_t
*client
= async_client_get(fibril_connection
->in_task_id
, true);
697 ipc_answer_0(fibril_connection
->callid
, ENOMEM
);
701 fibril_connection
->client
= client
;
704 * Call the connection handler function.
706 fibril_connection
->handler(fibril_connection
->callid
,
707 &fibril_connection
->call
, fibril_connection
->data
);
710 * Remove the reference for this client task connection.
712 async_client_put(client
);
715 * Remove myself from the connection hash table.
717 futex_down(&async_futex
);
718 hash_table_remove(&conn_hash_table
, &fibril_connection
->in_phone_hash
);
719 futex_up(&async_futex
);
722 * Answer all remaining messages with EHANGUP.
724 while (!list_empty(&fibril_connection
->msg_queue
)) {
726 list_get_instance(list_first(&fibril_connection
->msg_queue
),
729 list_remove(&msg
->link
);
730 ipc_answer_0(msg
->callid
, EHANGUP
);
735 * If the connection was hung-up, answer the last call,
736 * i.e. IPC_M_PHONE_HUNGUP.
738 if (fibril_connection
->close_callid
)
739 ipc_answer_0(fibril_connection
->close_callid
, EOK
);
741 free(fibril_connection
);
745 /** Create a new fibril for a new connection.
747 * Create new fibril for connection, fill in connection structures
748 * and insert it into the hash table, so that later we can easily
749 * do routing of messages to particular fibrils.
751 * @param in_task_id Identification of the incoming connection.
752 * @param in_phone_hash Identification of the incoming connection.
753 * @param callid Hash of the opening IPC_M_CONNECT_ME_TO call.
754 * If callid is zero, the connection was opened by
755 * accepting the IPC_M_CONNECT_TO_ME call and this
756 * function is called directly by the server.
757 * @param call Call data of the opening call.
758 * @param handler Connection handler.
759 * @param data Client argument to pass to the connection handler.
761 * @return New fibril id or NULL on failure.
764 static fid_t
async_new_connection(task_id_t in_task_id
, sysarg_t in_phone_hash
,
765 ipc_callid_t callid
, ipc_call_t
*call
, async_port_handler_t handler
,
768 connection_t
*conn
= malloc(sizeof(*conn
));
771 ipc_answer_0(callid
, ENOMEM
);
773 return (uintptr_t) NULL
;
776 conn
->in_task_id
= in_task_id
;
777 conn
->in_phone_hash
= in_phone_hash
;
778 list_initialize(&conn
->msg_queue
);
779 conn
->callid
= callid
;
780 conn
->close_callid
= 0;
781 conn
->handler
= handler
;
787 /* We will activate the fibril ASAP */
788 conn
->wdata
.active
= true;
789 conn
->wdata
.fid
= fibril_create(connection_fibril
, conn
);
791 if (conn
->wdata
.fid
== 0) {
795 ipc_answer_0(callid
, ENOMEM
);
797 return (uintptr_t) NULL
;
800 /* Add connection to the connection hash table */
802 futex_down(&async_futex
);
803 hash_table_insert(&conn_hash_table
, &conn
->link
);
804 futex_up(&async_futex
);
806 fibril_add_ready(conn
->wdata
.fid
);
808 return conn
->wdata
.fid
;
811 /** Wrapper for making IPC_M_CONNECT_TO_ME calls using the async framework.
813 * Ask through phone for a new connection to some service.
815 * @param exch Exchange for sending the message.
816 * @param iface Callback interface.
817 * @param arg1 User defined argument.
818 * @param arg2 User defined argument.
819 * @param handler Callback handler.
820 * @param data Handler data.
821 * @param port_id ID of the newly created port.
823 * @return Zero on success or a negative error code.
826 int async_create_callback_port(async_exch_t
*exch
, iface_t iface
, sysarg_t arg1
,
827 sysarg_t arg2
, async_port_handler_t handler
, void *data
, port_id_t
*port_id
)
829 if ((iface
& IFACE_MOD_CALLBACK
) != IFACE_MOD_CALLBACK
)
836 aid_t req
= async_send_3(exch
, IPC_M_CONNECT_TO_ME
, iface
, arg1
, arg2
,
840 async_wait_for(req
, &ret
);
844 sysarg_t phone_hash
= IPC_GET_ARG5(answer
);
845 interface_t
*interface
;
847 futex_down(&async_futex
);
849 ht_link_t
*link
= hash_table_find(&interface_hash_table
, &iface
);
851 interface
= hash_table_get_inst(link
, interface_t
, link
);
853 interface
= async_new_interface(iface
);
856 futex_up(&async_futex
);
860 port_t
*port
= async_new_port(interface
, handler
, data
);
862 futex_up(&async_futex
);
868 futex_up(&async_futex
);
870 fid_t fid
= async_new_connection(answer
.in_task_id
, phone_hash
,
871 0, NULL
, handler
, data
);
872 if (fid
== (uintptr_t) NULL
)
878 static size_t notification_key_hash(void *key
)
880 sysarg_t id
= *(sysarg_t
*) key
;
884 static size_t notification_hash(const ht_link_t
*item
)
886 notification_t
*notification
=
887 hash_table_get_inst(item
, notification_t
, link
);
888 return notification_key_hash(¬ification
->imethod
);
891 static bool notification_key_equal(void *key
, const ht_link_t
*item
)
893 sysarg_t id
= *(sysarg_t
*) key
;
894 notification_t
*notification
=
895 hash_table_get_inst(item
, notification_t
, link
);
896 return id
== notification
->imethod
;
899 /** Operations for the notification hash table. */
900 static hash_table_ops_t notification_hash_table_ops
= {
901 .hash
= notification_hash
,
902 .key_hash
= notification_key_hash
,
903 .key_equal
= notification_key_equal
,
905 .remove_callback
= NULL
908 /** Sort in current fibril's timeout request.
910 * @param wd Wait data of the current fibril.
913 void async_insert_timeout(awaiter_t
*wd
)
917 wd
->to_event
.occurred
= false;
918 wd
->to_event
.inlist
= true;
920 link_t
*tmp
= timeout_list
.head
.next
;
921 while (tmp
!= &timeout_list
.head
) {
923 = list_get_instance(tmp
, awaiter_t
, to_event
.link
);
925 if (tv_gteq(&cur
->to_event
.expires
, &wd
->to_event
.expires
))
931 list_insert_before(&wd
->to_event
.link
, tmp
);
934 /** Try to route a call to an appropriate connection fibril.
936 * If the proper connection fibril is found, a message with the call is added to
937 * its message queue. If the fibril was not active, it is activated and all
938 * timeouts are unregistered.
940 * @param callid Hash of the incoming call.
941 * @param call Data of the incoming call.
943 * @return False if the call doesn't match any connection.
944 * @return True if the call was passed to the respective connection fibril.
947 static bool route_call(ipc_callid_t callid
, ipc_call_t
*call
)
951 futex_down(&async_futex
);
953 ht_link_t
*link
= hash_table_find(&conn_hash_table
, &call
->in_phone_hash
);
955 futex_up(&async_futex
);
959 connection_t
*conn
= hash_table_get_inst(link
, connection_t
, link
);
961 msg_t
*msg
= malloc(sizeof(*msg
));
963 futex_up(&async_futex
);
967 msg
->callid
= callid
;
969 list_append(&msg
->link
, &conn
->msg_queue
);
971 if (IPC_GET_IMETHOD(*call
) == IPC_M_PHONE_HUNGUP
)
972 conn
->close_callid
= callid
;
974 /* If the connection fibril is waiting for an event, activate it */
975 if (!conn
->wdata
.active
) {
977 /* If in timeout list, remove it */
978 if (conn
->wdata
.to_event
.inlist
) {
979 conn
->wdata
.to_event
.inlist
= false;
980 list_remove(&conn
->wdata
.to_event
.link
);
983 conn
->wdata
.active
= true;
984 fibril_add_ready(conn
->wdata
.fid
);
987 futex_up(&async_futex
);
991 /** Process notification.
993 * @param callid Hash of the incoming call.
994 * @param call Data of the incoming call.
997 static void process_notification(ipc_callid_t callid
, ipc_call_t
*call
)
999 async_notification_handler_t handler
= NULL
;
1004 futex_down(&async_futex
);
1006 ht_link_t
*link
= hash_table_find(¬ification_hash_table
,
1007 &IPC_GET_IMETHOD(*call
));
1009 notification_t
*notification
=
1010 hash_table_get_inst(link
, notification_t
, link
);
1011 handler
= notification
->handler
;
1012 data
= notification
->data
;
1015 futex_up(&async_futex
);
1018 handler(callid
, call
, data
);
1021 /** Subscribe to IRQ notification.
1023 * @param inr IRQ number.
1024 * @param handler Notification handler.
1025 * @param data Notification handler client data.
1026 * @param ucode Top-half pseudocode handler.
1028 * @return IRQ capability handle on success.
1029 * @return Negative error code.
1032 int async_irq_subscribe(int inr
, async_notification_handler_t handler
,
1033 void *data
, const irq_code_t
*ucode
)
1035 notification_t
*notification
=
1036 (notification_t
*) malloc(sizeof(notification_t
));
1040 futex_down(&async_futex
);
1042 sysarg_t imethod
= notification_avail
;
1043 notification_avail
++;
1045 notification
->imethod
= imethod
;
1046 notification
->handler
= handler
;
1047 notification
->data
= data
;
1049 hash_table_insert(¬ification_hash_table
, ¬ification
->link
);
1051 futex_up(&async_futex
);
1053 return ipc_irq_subscribe(inr
, imethod
, ucode
);
1056 /** Unsubscribe from IRQ notification.
1058 * @param cap IRQ capability handle.
1060 * @return Zero on success or a negative error code.
1063 int async_irq_unsubscribe(int cap
)
1065 // TODO: Remove entry from hash table
1066 // to avoid memory leak
1068 return ipc_irq_unsubscribe(cap
);
1071 /** Subscribe to event notifications.
1073 * @param evno Event type to subscribe.
1074 * @param handler Notification handler.
1075 * @param data Notification handler client data.
1077 * @return Zero on success or a negative error code.
1080 int async_event_subscribe(event_type_t evno
,
1081 async_notification_handler_t handler
, void *data
)
1083 notification_t
*notification
=
1084 (notification_t
*) malloc(sizeof(notification_t
));
1088 futex_down(&async_futex
);
1090 sysarg_t imethod
= notification_avail
;
1091 notification_avail
++;
1093 notification
->imethod
= imethod
;
1094 notification
->handler
= handler
;
1095 notification
->data
= data
;
1097 hash_table_insert(¬ification_hash_table
, ¬ification
->link
);
1099 futex_up(&async_futex
);
1101 return ipc_event_subscribe(evno
, imethod
);
1104 /** Subscribe to task event notifications.
1106 * @param evno Event type to subscribe.
1107 * @param handler Notification handler.
1108 * @param data Notification handler client data.
1110 * @return Zero on success or a negative error code.
1113 int async_event_task_subscribe(event_task_type_t evno
,
1114 async_notification_handler_t handler
, void *data
)
1116 notification_t
*notification
=
1117 (notification_t
*) malloc(sizeof(notification_t
));
1121 futex_down(&async_futex
);
1123 sysarg_t imethod
= notification_avail
;
1124 notification_avail
++;
1126 notification
->imethod
= imethod
;
1127 notification
->handler
= handler
;
1128 notification
->data
= data
;
1130 hash_table_insert(¬ification_hash_table
, ¬ification
->link
);
1132 futex_up(&async_futex
);
1134 return ipc_event_task_subscribe(evno
, imethod
);
1137 /** Unmask event notifications.
1139 * @param evno Event type to unmask.
1141 * @return Value returned by the kernel.
1144 int async_event_unmask(event_type_t evno
)
1146 return ipc_event_unmask(evno
);
1149 /** Unmask task event notifications.
1151 * @param evno Event type to unmask.
1153 * @return Value returned by the kernel.
1156 int async_event_task_unmask(event_task_type_t evno
)
1158 return ipc_event_task_unmask(evno
);
1161 /** Return new incoming message for the current (fibril-local) connection.
1163 * @param call Storage where the incoming call data will be stored.
1164 * @param usecs Timeout in microseconds. Zero denotes no timeout.
1166 * @return If no timeout was specified, then a hash of the
1167 * incoming call is returned. If a timeout is specified,
1168 * then a hash of the incoming call is returned unless
1169 * the timeout expires prior to receiving a message. In
1170 * that case zero is returned.
1173 ipc_callid_t
async_get_call_timeout(ipc_call_t
*call
, suseconds_t usecs
)
1176 assert(fibril_connection
);
1179 * GCC 4.1.0 coughs on fibril_connection-> dereference.
1180 * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
1181 * I would never expect to find so many errors in
1184 connection_t
*conn
= fibril_connection
;
1186 futex_down(&async_futex
);
1189 getuptime(&conn
->wdata
.to_event
.expires
);
1190 tv_add_diff(&conn
->wdata
.to_event
.expires
, usecs
);
1192 conn
->wdata
.to_event
.inlist
= false;
1194 /* If nothing in queue, wait until something arrives */
1195 while (list_empty(&conn
->msg_queue
)) {
1196 if (conn
->close_callid
) {
1198 * Handle the case when the connection was already
1199 * closed by the client but the server did not notice
1200 * the first IPC_M_PHONE_HUNGUP call and continues to
1201 * call async_get_call_timeout(). Repeat
1202 * IPC_M_PHONE_HUNGUP until the caller notices.
1204 memset(call
, 0, sizeof(ipc_call_t
));
1205 IPC_SET_IMETHOD(*call
, IPC_M_PHONE_HUNGUP
);
1206 futex_up(&async_futex
);
1207 return conn
->close_callid
;
1211 async_insert_timeout(&conn
->wdata
);
1213 conn
->wdata
.active
= false;
1216 * Note: the current fibril will be rescheduled either due to a
1217 * timeout or due to an arriving message destined to it. In the
1218 * former case, handle_expired_timeouts() and, in the latter
1219 * case, route_call() will perform the wakeup.
1221 fibril_switch(FIBRIL_TO_MANAGER
);
1224 * Futex is up after getting back from async_manager.
1227 futex_down(&async_futex
);
1228 if ((usecs
) && (conn
->wdata
.to_event
.occurred
)
1229 && (list_empty(&conn
->msg_queue
))) {
1230 /* If we timed out -> exit */
1231 futex_up(&async_futex
);
1236 msg_t
*msg
= list_get_instance(list_first(&conn
->msg_queue
),
1238 list_remove(&msg
->link
);
1240 ipc_callid_t callid
= msg
->callid
;
1244 futex_up(&async_futex
);
1248 void *async_get_client_data(void)
1250 assert(fibril_connection
);
1251 return fibril_connection
->client
->data
;
1254 void *async_get_client_data_by_id(task_id_t client_id
)
1256 client_t
*client
= async_client_get(client_id
, false);
1260 if (!client
->data
) {
1261 async_client_put(client
);
1265 return client
->data
;
1268 void async_put_client_data_by_id(task_id_t client_id
)
1270 client_t
*client
= async_client_get(client_id
, false);
1273 assert(client
->data
);
1275 /* Drop the reference we got in async_get_client_data_by_hash(). */
1276 async_client_put(client
);
1278 /* Drop our own reference we got at the beginning of this function. */
1279 async_client_put(client
);
1282 static port_t
*async_find_port(iface_t iface
, port_id_t port_id
)
1284 port_t
*port
= NULL
;
1286 futex_down(&async_futex
);
1288 ht_link_t
*link
= hash_table_find(&interface_hash_table
, &iface
);
1290 interface_t
*interface
=
1291 hash_table_get_inst(link
, interface_t
, link
);
1293 link
= hash_table_find(&interface
->port_hash_table
, &port_id
);
1295 port
= hash_table_get_inst(link
, port_t
, link
);
1298 futex_up(&async_futex
);
1303 /** Handle a call that was received.
1305 * If the call has the IPC_M_CONNECT_ME_TO method, a new connection is created.
1306 * Otherwise the call is routed to its connection fibril.
1308 * @param callid Hash of the incoming call.
1309 * @param call Data of the incoming call.
1312 static void handle_call(ipc_callid_t callid
, ipc_call_t
*call
)
1316 /* Kernel notification */
1317 if ((callid
& IPC_CALLID_NOTIFICATION
)) {
1318 fibril_t
*fibril
= (fibril_t
*) __tcb_get()->fibril_data
;
1319 unsigned oldsw
= fibril
->switches
;
1321 process_notification(callid
, call
);
1323 if (oldsw
!= fibril
->switches
) {
1325 * The notification handler did not execute atomically
1326 * and so the current manager fibril assumed the role of
1327 * a notification fibril. While waiting for its
1328 * resources, it switched to another manager fibril that
1329 * had already existed or it created a new one. We
1330 * therefore know there is at least yet another
1331 * manager fibril that can take over. We now kill the
1332 * current 'notification' fibril to prevent fibril
1333 * population explosion.
1335 futex_down(&async_futex
);
1336 fibril_switch(FIBRIL_FROM_DEAD
);
1342 /* New connection */
1343 if (IPC_GET_IMETHOD(*call
) == IPC_M_CONNECT_ME_TO
) {
1344 iface_t iface
= (iface_t
) IPC_GET_ARG1(*call
);
1345 sysarg_t in_phone_hash
= IPC_GET_ARG5(*call
);
1347 async_notification_handler_t handler
= fallback_port_handler
;
1348 void *data
= fallback_port_data
;
1350 // TODO: Currently ignores all ports but the first one
1351 port_t
*port
= async_find_port(iface
, 0);
1353 handler
= port
->handler
;
1357 async_new_connection(call
->in_task_id
, in_phone_hash
, callid
,
1358 call
, handler
, data
);
1362 /* Try to route the call through the connection hash table */
1363 if (route_call(callid
, call
))
1366 /* Unknown call from unknown phone - hang it up */
1367 ipc_answer_0(callid
, EHANGUP
);
1370 /** Fire all timeouts that expired. */
1371 static void handle_expired_timeouts(void)
1376 futex_down(&async_futex
);
1378 link_t
*cur
= list_first(&timeout_list
);
1379 while (cur
!= NULL
) {
1381 list_get_instance(cur
, awaiter_t
, to_event
.link
);
1383 if (tv_gt(&waiter
->to_event
.expires
, &tv
))
1386 list_remove(&waiter
->to_event
.link
);
1387 waiter
->to_event
.inlist
= false;
1388 waiter
->to_event
.occurred
= true;
1391 * Redundant condition?
1392 * The fibril should not be active when it gets here.
1394 if (!waiter
->active
) {
1395 waiter
->active
= true;
1396 fibril_add_ready(waiter
->fid
);
1399 cur
= list_first(&timeout_list
);
1402 futex_up(&async_futex
);
1405 /** Endless loop dispatching incoming calls and answers.
1407 * @return Never returns.
1410 static int async_manager_worker(void)
1413 if (fibril_switch(FIBRIL_FROM_MANAGER
)) {
1414 futex_up(&async_futex
);
1416 * async_futex is always held when entering a manager
1422 futex_down(&async_futex
);
1424 suseconds_t timeout
;
1425 unsigned int flags
= SYNCH_FLAGS_NONE
;
1426 if (!list_empty(&timeout_list
)) {
1427 awaiter_t
*waiter
= list_get_instance(
1428 list_first(&timeout_list
), awaiter_t
, to_event
.link
);
1433 if (tv_gteq(&tv
, &waiter
->to_event
.expires
)) {
1434 futex_up(&async_futex
);
1435 handle_expired_timeouts();
1437 * Notice that even if the event(s) already
1438 * expired (and thus the other fibril was
1439 * supposed to be running already),
1440 * we check for incoming IPC.
1442 * Otherwise, a fibril that continuously
1443 * creates (almost) expired events could
1444 * prevent IPC retrieval from the kernel.
1447 flags
= SYNCH_FLAGS_NON_BLOCKING
;
1450 timeout
= tv_sub_diff(&waiter
->to_event
.expires
,
1452 futex_up(&async_futex
);
1455 futex_up(&async_futex
);
1456 timeout
= SYNCH_NO_TIMEOUT
;
1459 atomic_inc(&threads_in_ipc_wait
);
1462 ipc_callid_t callid
= ipc_wait_cycle(&call
, timeout
, flags
);
1464 atomic_dec(&threads_in_ipc_wait
);
1467 handle_expired_timeouts();
1471 if (callid
& IPC_CALLID_ANSWERED
)
1474 handle_call(callid
, &call
);
1480 /** Function to start async_manager as a standalone fibril.
1482 * When more kernel threads are used, one async manager should exist per thread.
1484 * @param arg Unused.
1485 * @return Never returns.
1488 static int async_manager_fibril(void *arg
)
1490 futex_up(&async_futex
);
1493 * async_futex is always locked when entering manager
1495 async_manager_worker();
1500 /** Add one manager to manager list. */
1501 void async_create_manager(void)
1503 fid_t fid
= fibril_create_generic(async_manager_fibril
, NULL
, PAGE_SIZE
);
1505 fibril_add_manager(fid
);
1508 /** Remove one manager from manager list */
1509 void async_destroy_manager(void)
1511 fibril_remove_manager();
1514 /** Initialize the async framework.
1517 void __async_init(void)
1519 if (!hash_table_create(&interface_hash_table
, 0, 0,
1520 &interface_hash_table_ops
))
1523 if (!hash_table_create(&client_hash_table
, 0, 0, &client_hash_table_ops
))
1526 if (!hash_table_create(&conn_hash_table
, 0, 0, &conn_hash_table_ops
))
1529 if (!hash_table_create(¬ification_hash_table
, 0, 0,
1530 ¬ification_hash_table_ops
))
1533 session_ns
= (async_sess_t
*) malloc(sizeof(async_sess_t
));
1534 if (session_ns
== NULL
)
1537 session_ns
->iface
= 0;
1538 session_ns
->mgmt
= EXCHANGE_ATOMIC
;
1539 session_ns
->phone
= PHONE_NS
;
1540 session_ns
->arg1
= 0;
1541 session_ns
->arg2
= 0;
1542 session_ns
->arg3
= 0;
1544 fibril_mutex_initialize(&session_ns
->remote_state_mtx
);
1545 session_ns
->remote_state_data
= NULL
;
1547 list_initialize(&session_ns
->exch_list
);
1548 fibril_mutex_initialize(&session_ns
->mutex
);
1549 atomic_set(&session_ns
->refcnt
, 0);
1552 /** Reply received callback.
1554 * This function is called whenever a reply for an asynchronous message sent out
1555 * by the asynchronous framework is received.
1557 * Notify the fibril which is waiting for this message that it has arrived.
1559 * @param arg Pointer to the asynchronous message record.
1560 * @param retval Value returned in the answer.
1561 * @param data Call data of the answer.
1564 void reply_received(void *arg
, int retval
, ipc_call_t
*data
)
1568 futex_down(&async_futex
);
1570 amsg_t
*msg
= (amsg_t
*) arg
;
1571 msg
->retval
= retval
;
1573 /* Copy data after futex_down, just in case the call was detached */
1574 if ((msg
->dataptr
) && (data
))
1575 *msg
->dataptr
= *data
;
1579 /* Remove message from timeout list */
1580 if (msg
->wdata
.to_event
.inlist
)
1581 list_remove(&msg
->wdata
.to_event
.link
);
1586 assert(msg
->wdata
.active
);
1588 } else if (!msg
->wdata
.active
) {
1589 msg
->wdata
.active
= true;
1590 fibril_add_ready(msg
->wdata
.fid
);
1593 futex_up(&async_futex
);
1596 /** Send message and return id of the sent message.
1598 * The return value can be used as input for async_wait() to wait for
1601 * @param exch Exchange for sending the message.
1602 * @param imethod Service-defined interface and method.
1603 * @param arg1 Service-defined payload argument.
1604 * @param arg2 Service-defined payload argument.
1605 * @param arg3 Service-defined payload argument.
1606 * @param arg4 Service-defined payload argument.
1607 * @param dataptr If non-NULL, storage where the reply data will be
1610 * @return Hash of the sent message or 0 on error.
1613 aid_t
async_send_fast(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
,
1614 sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
, ipc_call_t
*dataptr
)
1619 amsg_t
*msg
= amsg_create();
1623 msg
->dataptr
= dataptr
;
1624 msg
->wdata
.active
= true;
1626 ipc_call_async_4(exch
->phone
, imethod
, arg1
, arg2
, arg3
, arg4
, msg
,
1632 /** Send message and return id of the sent message
1634 * The return value can be used as input for async_wait() to wait for
1637 * @param exch Exchange for sending the message.
1638 * @param imethod Service-defined interface and method.
1639 * @param arg1 Service-defined payload argument.
1640 * @param arg2 Service-defined payload argument.
1641 * @param arg3 Service-defined payload argument.
1642 * @param arg4 Service-defined payload argument.
1643 * @param arg5 Service-defined payload argument.
1644 * @param dataptr If non-NULL, storage where the reply data will be
1647 * @return Hash of the sent message or 0 on error.
1650 aid_t
async_send_slow(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
,
1651 sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
, sysarg_t arg5
,
1652 ipc_call_t
*dataptr
)
1657 amsg_t
*msg
= amsg_create();
1661 msg
->dataptr
= dataptr
;
1662 msg
->wdata
.active
= true;
1664 ipc_call_async_5(exch
->phone
, imethod
, arg1
, arg2
, arg3
, arg4
, arg5
,
1665 msg
, reply_received
);
1670 /** Wait for a message sent by the async framework.
1672 * @param amsgid Hash of the message to wait for.
1673 * @param retval Pointer to storage where the retval of the answer will
1677 void async_wait_for(aid_t amsgid
, sysarg_t
*retval
)
1681 amsg_t
*msg
= (amsg_t
*) amsgid
;
1683 futex_down(&async_futex
);
1685 assert(!msg
->forget
);
1686 assert(!msg
->destroyed
);
1689 futex_up(&async_futex
);
1693 msg
->wdata
.fid
= fibril_get_id();
1694 msg
->wdata
.active
= false;
1695 msg
->wdata
.to_event
.inlist
= false;
1697 /* Leave the async_futex locked when entering this function */
1698 fibril_switch(FIBRIL_TO_MANAGER
);
1700 /* Futex is up automatically after fibril_switch */
1704 *retval
= msg
->retval
;
1709 /** Wait for a message sent by the async framework, timeout variant.
1711 * If the wait times out, the caller may choose to either wait again by calling
1712 * async_wait_for() or async_wait_timeout(), or forget the message via
1715 * @param amsgid Hash of the message to wait for.
1716 * @param retval Pointer to storage where the retval of the answer will
1718 * @param timeout Timeout in microseconds.
1720 * @return Zero on success, ETIMEOUT if the timeout has expired.
1723 int async_wait_timeout(aid_t amsgid
, sysarg_t
*retval
, suseconds_t timeout
)
1727 amsg_t
*msg
= (amsg_t
*) amsgid
;
1729 futex_down(&async_futex
);
1731 assert(!msg
->forget
);
1732 assert(!msg
->destroyed
);
1735 futex_up(&async_futex
);
1740 * Negative timeout is converted to zero timeout to avoid
1741 * using tv_add with negative augmenter.
1746 getuptime(&msg
->wdata
.to_event
.expires
);
1747 tv_add_diff(&msg
->wdata
.to_event
.expires
, timeout
);
1750 * Current fibril is inserted as waiting regardless of the
1751 * "size" of the timeout.
1753 * Checking for msg->done and immediately bailing out when
1754 * timeout == 0 would mean that the manager fibril would never
1755 * run (consider single threaded program).
1756 * Thus the IPC answer would be never retrieved from the kernel.
1758 * Notice that the actual delay would be very small because we
1759 * - switch to manager fibril
1760 * - the manager sees expired timeout
1761 * - and thus adds us back to ready queue
1762 * - manager switches back to some ready fibril
1763 * (prior it, it checks for incoming IPC).
1766 msg
->wdata
.fid
= fibril_get_id();
1767 msg
->wdata
.active
= false;
1768 async_insert_timeout(&msg
->wdata
);
1770 /* Leave the async_futex locked when entering this function */
1771 fibril_switch(FIBRIL_TO_MANAGER
);
1773 /* Futex is up automatically after fibril_switch */
1780 *retval
= msg
->retval
;
1787 /** Discard the message / reply on arrival.
1789 * The message will be marked to be discarded once the reply arrives in
1790 * reply_received(). It is not allowed to call async_wait_for() or
1791 * async_wait_timeout() on this message after a call to this function.
1793 * @param amsgid Hash of the message to forget.
1795 void async_forget(aid_t amsgid
)
1797 amsg_t
*msg
= (amsg_t
*) amsgid
;
1800 assert(!msg
->forget
);
1801 assert(!msg
->destroyed
);
1803 futex_down(&async_futex
);
1808 msg
->dataptr
= NULL
;
1812 futex_up(&async_futex
);
1815 /** Wait for specified time.
1817 * The current fibril is suspended but the thread continues to execute.
1819 * @param timeout Duration of the wait in microseconds.
1822 void async_usleep(suseconds_t timeout
)
1824 amsg_t
*msg
= amsg_create();
1828 msg
->wdata
.fid
= fibril_get_id();
1830 getuptime(&msg
->wdata
.to_event
.expires
);
1831 tv_add_diff(&msg
->wdata
.to_event
.expires
, timeout
);
1833 futex_down(&async_futex
);
1835 async_insert_timeout(&msg
->wdata
);
1837 /* Leave the async_futex locked when entering this function */
1838 fibril_switch(FIBRIL_TO_MANAGER
);
1840 /* Futex is up automatically after fibril_switch() */
1845 /** Pseudo-synchronous message sending - fast version.
1847 * Send message asynchronously and return only after the reply arrives.
1849 * This function can only transfer 4 register payload arguments. For
1850 * transferring more arguments, see the slower async_req_slow().
1852 * @param exch Exchange for sending the message.
1853 * @param imethod Interface and method of the call.
1854 * @param arg1 Service-defined payload argument.
1855 * @param arg2 Service-defined payload argument.
1856 * @param arg3 Service-defined payload argument.
1857 * @param arg4 Service-defined payload argument.
1858 * @param r1 If non-NULL, storage for the 1st reply argument.
1859 * @param r2 If non-NULL, storage for the 2nd reply argument.
1860 * @param r3 If non-NULL, storage for the 3rd reply argument.
1861 * @param r4 If non-NULL, storage for the 4th reply argument.
1862 * @param r5 If non-NULL, storage for the 5th reply argument.
1864 * @return Return code of the reply or a negative error code.
1867 sysarg_t
async_req_fast(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
,
1868 sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
, sysarg_t
*r1
, sysarg_t
*r2
,
1869 sysarg_t
*r3
, sysarg_t
*r4
, sysarg_t
*r5
)
1875 aid_t aid
= async_send_4(exch
, imethod
, arg1
, arg2
, arg3
, arg4
,
1879 async_wait_for(aid
, &rc
);
1882 *r1
= IPC_GET_ARG1(result
);
1885 *r2
= IPC_GET_ARG2(result
);
1888 *r3
= IPC_GET_ARG3(result
);
1891 *r4
= IPC_GET_ARG4(result
);
1894 *r5
= IPC_GET_ARG5(result
);
1899 /** Pseudo-synchronous message sending - slow version.
1901 * Send message asynchronously and return only after the reply arrives.
1903 * @param exch Exchange for sending the message.
1904 * @param imethod Interface and method of the call.
1905 * @param arg1 Service-defined payload argument.
1906 * @param arg2 Service-defined payload argument.
1907 * @param arg3 Service-defined payload argument.
1908 * @param arg4 Service-defined payload argument.
1909 * @param arg5 Service-defined payload argument.
1910 * @param r1 If non-NULL, storage for the 1st reply argument.
1911 * @param r2 If non-NULL, storage for the 2nd reply argument.
1912 * @param r3 If non-NULL, storage for the 3rd reply argument.
1913 * @param r4 If non-NULL, storage for the 4th reply argument.
1914 * @param r5 If non-NULL, storage for the 5th reply argument.
1916 * @return Return code of the reply or a negative error code.
1919 sysarg_t
async_req_slow(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
,
1920 sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
, sysarg_t arg5
, sysarg_t
*r1
,
1921 sysarg_t
*r2
, sysarg_t
*r3
, sysarg_t
*r4
, sysarg_t
*r5
)
1927 aid_t aid
= async_send_5(exch
, imethod
, arg1
, arg2
, arg3
, arg4
, arg5
,
1931 async_wait_for(aid
, &rc
);
1934 *r1
= IPC_GET_ARG1(result
);
1937 *r2
= IPC_GET_ARG2(result
);
1940 *r3
= IPC_GET_ARG3(result
);
1943 *r4
= IPC_GET_ARG4(result
);
1946 *r5
= IPC_GET_ARG5(result
);
1951 void async_msg_0(async_exch_t
*exch
, sysarg_t imethod
)
1954 ipc_call_async_0(exch
->phone
, imethod
, NULL
, NULL
);
1957 void async_msg_1(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
)
1960 ipc_call_async_1(exch
->phone
, imethod
, arg1
, NULL
, NULL
);
1963 void async_msg_2(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
,
1967 ipc_call_async_2(exch
->phone
, imethod
, arg1
, arg2
, NULL
, NULL
);
1970 void async_msg_3(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
,
1971 sysarg_t arg2
, sysarg_t arg3
)
1974 ipc_call_async_3(exch
->phone
, imethod
, arg1
, arg2
, arg3
, NULL
,
1978 void async_msg_4(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
,
1979 sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
)
1982 ipc_call_async_4(exch
->phone
, imethod
, arg1
, arg2
, arg3
, arg4
,
1986 void async_msg_5(async_exch_t
*exch
, sysarg_t imethod
, sysarg_t arg1
,
1987 sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
, sysarg_t arg5
)
1990 ipc_call_async_5(exch
->phone
, imethod
, arg1
, arg2
, arg3
, arg4
,
1994 sysarg_t
async_answer_0(ipc_callid_t callid
, sysarg_t retval
)
1996 return ipc_answer_0(callid
, retval
);
1999 sysarg_t
async_answer_1(ipc_callid_t callid
, sysarg_t retval
, sysarg_t arg1
)
2001 return ipc_answer_1(callid
, retval
, arg1
);
2004 sysarg_t
async_answer_2(ipc_callid_t callid
, sysarg_t retval
, sysarg_t arg1
,
2007 return ipc_answer_2(callid
, retval
, arg1
, arg2
);
2010 sysarg_t
async_answer_3(ipc_callid_t callid
, sysarg_t retval
, sysarg_t arg1
,
2011 sysarg_t arg2
, sysarg_t arg3
)
2013 return ipc_answer_3(callid
, retval
, arg1
, arg2
, arg3
);
2016 sysarg_t
async_answer_4(ipc_callid_t callid
, sysarg_t retval
, sysarg_t arg1
,
2017 sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
)
2019 return ipc_answer_4(callid
, retval
, arg1
, arg2
, arg3
, arg4
);
2022 sysarg_t
async_answer_5(ipc_callid_t callid
, sysarg_t retval
, sysarg_t arg1
,
2023 sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
, sysarg_t arg5
)
2025 return ipc_answer_5(callid
, retval
, arg1
, arg2
, arg3
, arg4
, arg5
);
2028 int async_forward_fast(ipc_callid_t callid
, async_exch_t
*exch
,
2029 sysarg_t imethod
, sysarg_t arg1
, sysarg_t arg2
, unsigned int mode
)
2034 return ipc_forward_fast(callid
, exch
->phone
, imethod
, arg1
, arg2
, mode
);
2037 int async_forward_slow(ipc_callid_t callid
, async_exch_t
*exch
,
2038 sysarg_t imethod
, sysarg_t arg1
, sysarg_t arg2
, sysarg_t arg3
,
2039 sysarg_t arg4
, sysarg_t arg5
, unsigned int mode
)
2044 return ipc_forward_slow(callid
, exch
->phone
, imethod
, arg1
, arg2
, arg3
,
2048 /** Wrapper for making IPC_M_CONNECT_TO_ME calls using the async framework.
2050 * Ask through phone for a new connection to some service.
2052 * @param exch Exchange for sending the message.
2053 * @param arg1 User defined argument.
2054 * @param arg2 User defined argument.
2055 * @param arg3 User defined argument.
2057 * @return Zero on success or a negative error code.
2060 int async_connect_to_me(async_exch_t
*exch
, sysarg_t arg1
, sysarg_t arg2
,
2067 aid_t req
= async_send_3(exch
, IPC_M_CONNECT_TO_ME
, arg1
, arg2
, arg3
,
2071 async_wait_for(req
, &rc
);
2078 static int async_connect_me_to_internal(int phone
, sysarg_t arg1
, sysarg_t arg2
,
2079 sysarg_t arg3
, sysarg_t arg4
)
2083 amsg_t
*msg
= amsg_create();
2087 msg
->dataptr
= &result
;
2088 msg
->wdata
.active
= true;
2090 ipc_call_async_4(phone
, IPC_M_CONNECT_ME_TO
, arg1
, arg2
, arg3
, arg4
,
2091 msg
, reply_received
);
2094 async_wait_for((aid_t
) msg
, &rc
);
2099 return (int) IPC_GET_ARG5(result
);
2102 /** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
2104 * Ask through for a new connection to some service.
2106 * @param mgmt Exchange management style.
2107 * @param exch Exchange for sending the message.
2108 * @param arg1 User defined argument.
2109 * @param arg2 User defined argument.
2110 * @param arg3 User defined argument.
2112 * @return New session on success or NULL on error.
2115 async_sess_t
*async_connect_me_to(exch_mgmt_t mgmt
, async_exch_t
*exch
,
2116 sysarg_t arg1
, sysarg_t arg2
, sysarg_t arg3
)
2123 async_sess_t
*sess
= (async_sess_t
*) malloc(sizeof(async_sess_t
));
2129 int phone
= async_connect_me_to_internal(exch
->phone
, arg1
, arg2
, arg3
,
2139 sess
->phone
= phone
;
2144 fibril_mutex_initialize(&sess
->remote_state_mtx
);
2145 sess
->remote_state_data
= NULL
;
2147 list_initialize(&sess
->exch_list
);
2148 fibril_mutex_initialize(&sess
->mutex
);
2149 atomic_set(&sess
->refcnt
, 0);
2154 /** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
2156 * Ask through phone for a new connection to some service and block until
2159 * @param exch Exchange for sending the message.
2160 * @param iface Connection interface.
2161 * @param arg2 User defined argument.
2162 * @param arg3 User defined argument.
2164 * @return New session on success or NULL on error.
2167 async_sess_t
*async_connect_me_to_iface(async_exch_t
*exch
, iface_t iface
,
2168 sysarg_t arg2
, sysarg_t arg3
)
2175 async_sess_t
*sess
= (async_sess_t
*) malloc(sizeof(async_sess_t
));
2181 int phone
= async_connect_me_to_internal(exch
->phone
, iface
, arg2
,
2189 sess
->iface
= iface
;
2190 sess
->phone
= phone
;
2195 fibril_mutex_initialize(&sess
->remote_state_mtx
);
2196 sess
->remote_state_data
= NULL
;
2198 list_initialize(&sess
->exch_list
);
2199 fibril_mutex_initialize(&sess
->mutex
);
2200 atomic_set(&sess
->refcnt
, 0);
2205 /** Set arguments for new connections.
2207 * FIXME This is an ugly hack to work around the problem that parallel
2208 * exchanges are implemented using parallel connections. When we create
2209 * a callback session, the framework does not know arguments for the new
2212 * The proper solution seems to be to implement parallel exchanges using
2215 void async_sess_args_set(async_sess_t
*sess
, sysarg_t arg1
, sysarg_t arg2
,
2223 /** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
2225 * Ask through phone for a new connection to some service and block until
2228 * @param mgmt Exchange management style.
2229 * @param exch Exchange for sending the message.
2230 * @param arg1 User defined argument.
2231 * @param arg2 User defined argument.
2232 * @param arg3 User defined argument.
2234 * @return New session on success or NULL on error.
2237 async_sess_t
*async_connect_me_to_blocking(exch_mgmt_t mgmt
, async_exch_t
*exch
,
2238 sysarg_t arg1
, sysarg_t arg2
, sysarg_t arg3
)
2245 async_sess_t
*sess
= (async_sess_t
*) malloc(sizeof(async_sess_t
));
2251 int phone
= async_connect_me_to_internal(exch
->phone
, arg1
, arg2
, arg3
,
2262 sess
->phone
= phone
;
2267 fibril_mutex_initialize(&sess
->remote_state_mtx
);
2268 sess
->remote_state_data
= NULL
;
2270 list_initialize(&sess
->exch_list
);
2271 fibril_mutex_initialize(&sess
->mutex
);
2272 atomic_set(&sess
->refcnt
, 0);
2277 /** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
2279 * Ask through phone for a new connection to some service and block until
2282 * @param exch Exchange for sending the message.
2283 * @param iface Connection interface.
2284 * @param arg2 User defined argument.
2285 * @param arg3 User defined argument.
2287 * @return New session on success or NULL on error.
2290 async_sess_t
*async_connect_me_to_blocking_iface(async_exch_t
*exch
, iface_t iface
,
2291 sysarg_t arg2
, sysarg_t arg3
)
2298 async_sess_t
*sess
= (async_sess_t
*) malloc(sizeof(async_sess_t
));
2304 int phone
= async_connect_me_to_internal(exch
->phone
, iface
, arg2
,
2305 arg3
, IPC_FLAG_BLOCKING
);
2312 sess
->iface
= iface
;
2313 sess
->phone
= phone
;
2318 fibril_mutex_initialize(&sess
->remote_state_mtx
);
2319 sess
->remote_state_data
= NULL
;
2321 list_initialize(&sess
->exch_list
);
2322 fibril_mutex_initialize(&sess
->mutex
);
2323 atomic_set(&sess
->refcnt
, 0);
2328 /** Connect to a task specified by id.
2331 async_sess_t
*async_connect_kbox(task_id_t id
)
2333 async_sess_t
*sess
= (async_sess_t
*) malloc(sizeof(async_sess_t
));
2339 int phone
= ipc_connect_kbox(id
);
2347 sess
->mgmt
= EXCHANGE_ATOMIC
;
2348 sess
->phone
= phone
;
2353 fibril_mutex_initialize(&sess
->remote_state_mtx
);
2354 sess
->remote_state_data
= NULL
;
2356 list_initialize(&sess
->exch_list
);
2357 fibril_mutex_initialize(&sess
->mutex
);
2358 atomic_set(&sess
->refcnt
, 0);
2363 static int async_hangup_internal(int phone
)
2365 return ipc_hangup(phone
);
2368 /** Wrapper for ipc_hangup.
2370 * @param sess Session to hung up.
2372 * @return Zero on success or a negative error code.
2375 int async_hangup(async_sess_t
*sess
)
2381 if (atomic_get(&sess
->refcnt
) > 0)
2384 fibril_mutex_lock(&async_sess_mutex
);
2386 int rc
= async_hangup_internal(sess
->phone
);
2388 while (!list_empty(&sess
->exch_list
)) {
2389 exch
= (async_exch_t
*)
2390 list_get_instance(list_first(&sess
->exch_list
),
2391 async_exch_t
, sess_link
);
2393 list_remove(&exch
->sess_link
);
2394 list_remove(&exch
->global_link
);
2395 async_hangup_internal(exch
->phone
);
2401 fibril_mutex_unlock(&async_sess_mutex
);
2406 /** Interrupt one thread of this task from waiting for IPC. */
2407 void async_poke(void)
2412 /** Start new exchange in a session.
2414 * @param session Session.
2416 * @return New exchange or NULL on error.
2419 async_exch_t
*async_exchange_begin(async_sess_t
*sess
)
2424 exch_mgmt_t mgmt
= sess
->mgmt
;
2425 if (sess
->iface
!= 0)
2426 mgmt
= sess
->iface
& IFACE_EXCHANGE_MASK
;
2428 async_exch_t
*exch
= NULL
;
2430 fibril_mutex_lock(&async_sess_mutex
);
2432 if (!list_empty(&sess
->exch_list
)) {
2434 * There are inactive exchanges in the session.
2436 exch
= (async_exch_t
*)
2437 list_get_instance(list_first(&sess
->exch_list
),
2438 async_exch_t
, sess_link
);
2440 list_remove(&exch
->sess_link
);
2441 list_remove(&exch
->global_link
);
2444 * There are no available exchanges in the session.
2447 if ((mgmt
== EXCHANGE_ATOMIC
) ||
2448 (mgmt
== EXCHANGE_SERIALIZE
)) {
2449 exch
= (async_exch_t
*) malloc(sizeof(async_exch_t
));
2451 link_initialize(&exch
->sess_link
);
2452 link_initialize(&exch
->global_link
);
2454 exch
->phone
= sess
->phone
;
2456 } else if (mgmt
== EXCHANGE_PARALLEL
) {
2461 * Make a one-time attempt to connect a new data phone.
2463 phone
= async_connect_me_to_internal(sess
->phone
, sess
->arg1
,
2464 sess
->arg2
, sess
->arg3
, 0);
2466 exch
= (async_exch_t
*) malloc(sizeof(async_exch_t
));
2468 link_initialize(&exch
->sess_link
);
2469 link_initialize(&exch
->global_link
);
2471 exch
->phone
= phone
;
2473 async_hangup_internal(phone
);
2474 } else if (!list_empty(&inactive_exch_list
)) {
2476 * We did not manage to connect a new phone. But we
2477 * can try to close some of the currently inactive
2478 * connections in other sessions and try again.
2480 exch
= (async_exch_t
*)
2481 list_get_instance(list_first(&inactive_exch_list
),
2482 async_exch_t
, global_link
);
2484 list_remove(&exch
->sess_link
);
2485 list_remove(&exch
->global_link
);
2486 async_hangup_internal(exch
->phone
);
2491 * Wait for a phone to become available.
2493 fibril_condvar_wait(&avail_phone_cv
, &async_sess_mutex
);
2499 fibril_mutex_unlock(&async_sess_mutex
);
2502 atomic_inc(&sess
->refcnt
);
2504 if (mgmt
== EXCHANGE_SERIALIZE
)
2505 fibril_mutex_lock(&sess
->mutex
);
2511 /** Finish an exchange.
2513 * @param exch Exchange to finish.
2516 void async_exchange_end(async_exch_t
*exch
)
2521 async_sess_t
*sess
= exch
->sess
;
2522 assert(sess
!= NULL
);
2524 exch_mgmt_t mgmt
= sess
->mgmt
;
2525 if (sess
->iface
!= 0)
2526 mgmt
= sess
->iface
& IFACE_EXCHANGE_MASK
;
2528 atomic_dec(&sess
->refcnt
);
2530 if (mgmt
== EXCHANGE_SERIALIZE
)
2531 fibril_mutex_unlock(&sess
->mutex
);
2533 fibril_mutex_lock(&async_sess_mutex
);
2535 list_append(&exch
->sess_link
, &sess
->exch_list
);
2536 list_append(&exch
->global_link
, &inactive_exch_list
);
2537 fibril_condvar_signal(&avail_phone_cv
);
2539 fibril_mutex_unlock(&async_sess_mutex
);
2542 /** Wrapper for IPC_M_SHARE_IN calls using the async framework.
2544 * @param exch Exchange for sending the message.
2545 * @param size Size of the destination address space area.
2546 * @param arg User defined argument.
2547 * @param flags Storage for the received flags. Can be NULL.
2548 * @param dst Address of the storage for the destination address space area
2549 * base address. Cannot be NULL.
2551 * @return Zero on success or a negative error code from errno.h.
2554 int async_share_in_start(async_exch_t
*exch
, size_t size
, sysarg_t arg
,
2555 unsigned int *flags
, void **dst
)
2560 sysarg_t _flags
= 0;
2561 sysarg_t _dst
= (sysarg_t
) -1;
2562 int res
= async_req_2_4(exch
, IPC_M_SHARE_IN
, (sysarg_t
) size
,
2563 arg
, NULL
, &_flags
, NULL
, &_dst
);
2566 *flags
= (unsigned int) _flags
;
2568 *dst
= (void *) _dst
;
2572 /** Wrapper for receiving the IPC_M_SHARE_IN calls using the async framework.
2574 * This wrapper only makes it more comfortable to receive IPC_M_SHARE_IN
2575 * calls so that the user doesn't have to remember the meaning of each IPC
2578 * So far, this wrapper is to be used from within a connection fibril.
2580 * @param callid Storage for the hash of the IPC_M_SHARE_IN call.
2581 * @param size Destination address space area size.
2583 * @return True on success, false on failure.
2586 bool async_share_in_receive(ipc_callid_t
*callid
, size_t *size
)
2592 *callid
= async_get_call(&data
);
2594 if (IPC_GET_IMETHOD(data
) != IPC_M_SHARE_IN
)
2597 *size
= (size_t) IPC_GET_ARG1(data
);
2601 /** Wrapper for answering the IPC_M_SHARE_IN calls using the async framework.
2603 * This wrapper only makes it more comfortable to answer IPC_M_SHARE_IN
2604 * calls so that the user doesn't have to remember the meaning of each IPC
2607 * @param callid Hash of the IPC_M_DATA_READ call to answer.
2608 * @param src Source address space base.
2609 * @param flags Flags to be used for sharing. Bits can be only cleared.
2611 * @return Zero on success or a value from @ref errno.h on failure.
2614 int async_share_in_finalize(ipc_callid_t callid
, void *src
, unsigned int flags
)
2616 return ipc_answer_3(callid
, EOK
, (sysarg_t
) src
, (sysarg_t
) flags
,
2617 (sysarg_t
) __entry
);
2620 /** Wrapper for IPC_M_SHARE_OUT calls using the async framework.
2622 * @param exch Exchange for sending the message.
2623 * @param src Source address space area base address.
2624 * @param flags Flags to be used for sharing. Bits can be only cleared.
2626 * @return Zero on success or a negative error code from errno.h.
2629 int async_share_out_start(async_exch_t
*exch
, void *src
, unsigned int flags
)
2634 return async_req_3_0(exch
, IPC_M_SHARE_OUT
, (sysarg_t
) src
, 0,
2638 /** Wrapper for receiving the IPC_M_SHARE_OUT calls using the async framework.
2640 * This wrapper only makes it more comfortable to receive IPC_M_SHARE_OUT
2641 * calls so that the user doesn't have to remember the meaning of each IPC
2644 * So far, this wrapper is to be used from within a connection fibril.
2646 * @param callid Storage for the hash of the IPC_M_SHARE_OUT call.
2647 * @param size Storage for the source address space area size.
2648 * @param flags Storage for the sharing flags.
2650 * @return True on success, false on failure.
2653 bool async_share_out_receive(ipc_callid_t
*callid
, size_t *size
, unsigned int *flags
)
2660 *callid
= async_get_call(&data
);
2662 if (IPC_GET_IMETHOD(data
) != IPC_M_SHARE_OUT
)
2665 *size
= (size_t) IPC_GET_ARG2(data
);
2666 *flags
= (unsigned int) IPC_GET_ARG3(data
);
2670 /** Wrapper for answering the IPC_M_SHARE_OUT calls using the async framework.
2672 * This wrapper only makes it more comfortable to answer IPC_M_SHARE_OUT
2673 * calls so that the user doesn't have to remember the meaning of each IPC
2676 * @param callid Hash of the IPC_M_DATA_WRITE call to answer.
2677 * @param dst Address of the storage for the destination address space area
2680 * @return Zero on success or a value from @ref errno.h on failure.
2683 int async_share_out_finalize(ipc_callid_t callid
, void **dst
)
2685 return ipc_answer_2(callid
, EOK
, (sysarg_t
) __entry
, (sysarg_t
) dst
);
2688 /** Start IPC_M_DATA_READ using the async framework.
2690 * @param exch Exchange for sending the message.
2691 * @param dst Address of the beginning of the destination buffer.
2692 * @param size Size of the destination buffer (in bytes).
2693 * @param dataptr Storage of call data (arg 2 holds actual data size).
2695 * @return Hash of the sent message or 0 on error.
2698 aid_t
async_data_read(async_exch_t
*exch
, void *dst
, size_t size
,
2699 ipc_call_t
*dataptr
)
2701 return async_send_2(exch
, IPC_M_DATA_READ
, (sysarg_t
) dst
,
2702 (sysarg_t
) size
, dataptr
);
2705 /** Wrapper for IPC_M_DATA_READ calls using the async framework.
2707 * @param exch Exchange for sending the message.
2708 * @param dst Address of the beginning of the destination buffer.
2709 * @param size Size of the destination buffer.
2711 * @return Zero on success or a negative error code from errno.h.
2714 int async_data_read_start(async_exch_t
*exch
, void *dst
, size_t size
)
2719 return async_req_2_0(exch
, IPC_M_DATA_READ
, (sysarg_t
) dst
,
2723 /** Wrapper for receiving the IPC_M_DATA_READ calls using the async framework.
2725 * This wrapper only makes it more comfortable to receive IPC_M_DATA_READ
2726 * calls so that the user doesn't have to remember the meaning of each IPC
2729 * So far, this wrapper is to be used from within a connection fibril.
2731 * @param callid Storage for the hash of the IPC_M_DATA_READ.
2732 * @param size Storage for the maximum size. Can be NULL.
2734 * @return True on success, false on failure.
2737 bool async_data_read_receive(ipc_callid_t
*callid
, size_t *size
)
2740 return async_data_read_receive_call(callid
, &data
, size
);
2743 /** Wrapper for receiving the IPC_M_DATA_READ calls using the async framework.
2745 * This wrapper only makes it more comfortable to receive IPC_M_DATA_READ
2746 * calls so that the user doesn't have to remember the meaning of each IPC
2749 * So far, this wrapper is to be used from within a connection fibril.
2751 * @param callid Storage for the hash of the IPC_M_DATA_READ.
2752 * @param size Storage for the maximum size. Can be NULL.
2754 * @return True on success, false on failure.
2757 bool async_data_read_receive_call(ipc_callid_t
*callid
, ipc_call_t
*data
,
2763 *callid
= async_get_call(data
);
2765 if (IPC_GET_IMETHOD(*data
) != IPC_M_DATA_READ
)
2769 *size
= (size_t) IPC_GET_ARG2(*data
);
2774 /** Wrapper for answering the IPC_M_DATA_READ calls using the async framework.
2776 * This wrapper only makes it more comfortable to answer IPC_M_DATA_READ
2777 * calls so that the user doesn't have to remember the meaning of each IPC
2780 * @param callid Hash of the IPC_M_DATA_READ call to answer.
2781 * @param src Source address for the IPC_M_DATA_READ call.
2782 * @param size Size for the IPC_M_DATA_READ call. Can be smaller than
2783 * the maximum size announced by the sender.
2785 * @return Zero on success or a value from @ref errno.h on failure.
2788 int async_data_read_finalize(ipc_callid_t callid
, const void *src
, size_t size
)
2790 return ipc_answer_2(callid
, EOK
, (sysarg_t
) src
, (sysarg_t
) size
);
2793 /** Wrapper for forwarding any read request
2796 int async_data_read_forward_fast(async_exch_t
*exch
, sysarg_t imethod
,
2797 sysarg_t arg1
, sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
,
2798 ipc_call_t
*dataptr
)
2803 ipc_callid_t callid
;
2804 if (!async_data_read_receive(&callid
, NULL
)) {
2805 ipc_answer_0(callid
, EINVAL
);
2809 aid_t msg
= async_send_fast(exch
, imethod
, arg1
, arg2
, arg3
, arg4
,
2812 ipc_answer_0(callid
, EINVAL
);
2816 int retval
= ipc_forward_fast(callid
, exch
->phone
, 0, 0, 0,
2817 IPC_FF_ROUTE_FROM_ME
);
2818 if (retval
!= EOK
) {
2820 ipc_answer_0(callid
, retval
);
2825 async_wait_for(msg
, &rc
);
2830 /** Wrapper for IPC_M_DATA_WRITE calls using the async framework.
2832 * @param exch Exchange for sending the message.
2833 * @param src Address of the beginning of the source buffer.
2834 * @param size Size of the source buffer.
2836 * @return Zero on success or a negative error code from errno.h.
2839 int async_data_write_start(async_exch_t
*exch
, const void *src
, size_t size
)
2844 return async_req_2_0(exch
, IPC_M_DATA_WRITE
, (sysarg_t
) src
,
2848 /** Wrapper for receiving the IPC_M_DATA_WRITE calls using the async framework.
2850 * This wrapper only makes it more comfortable to receive IPC_M_DATA_WRITE
2851 * calls so that the user doesn't have to remember the meaning of each IPC
2854 * So far, this wrapper is to be used from within a connection fibril.
2856 * @param callid Storage for the hash of the IPC_M_DATA_WRITE.
2857 * @param size Storage for the suggested size. May be NULL.
2859 * @return True on success, false on failure.
2862 bool async_data_write_receive(ipc_callid_t
*callid
, size_t *size
)
2865 return async_data_write_receive_call(callid
, &data
, size
);
2868 /** Wrapper for receiving the IPC_M_DATA_WRITE calls using the async framework.
2870 * This wrapper only makes it more comfortable to receive IPC_M_DATA_WRITE
2871 * calls so that the user doesn't have to remember the meaning of each IPC
2874 * So far, this wrapper is to be used from within a connection fibril.
2876 * @param callid Storage for the hash of the IPC_M_DATA_WRITE.
2877 * @param data Storage for the ipc call data.
2878 * @param size Storage for the suggested size. May be NULL.
2880 * @return True on success, false on failure.
2883 bool async_data_write_receive_call(ipc_callid_t
*callid
, ipc_call_t
*data
,
2889 *callid
= async_get_call(data
);
2891 if (IPC_GET_IMETHOD(*data
) != IPC_M_DATA_WRITE
)
2895 *size
= (size_t) IPC_GET_ARG2(*data
);
2900 /** Wrapper for answering the IPC_M_DATA_WRITE calls using the async framework.
2902 * This wrapper only makes it more comfortable to answer IPC_M_DATA_WRITE
2903 * calls so that the user doesn't have to remember the meaning of each IPC
2906 * @param callid Hash of the IPC_M_DATA_WRITE call to answer.
2907 * @param dst Final destination address for the IPC_M_DATA_WRITE call.
2908 * @param size Final size for the IPC_M_DATA_WRITE call.
2910 * @return Zero on success or a value from @ref errno.h on failure.
2913 int async_data_write_finalize(ipc_callid_t callid
, void *dst
, size_t size
)
2915 return ipc_answer_2(callid
, EOK
, (sysarg_t
) dst
, (sysarg_t
) size
);
2918 /** Wrapper for receiving binary data or strings
2920 * This wrapper only makes it more comfortable to use async_data_write_*
2921 * functions to receive binary data or strings.
2923 * @param data Pointer to data pointer (which should be later disposed
2924 * by free()). If the operation fails, the pointer is not
2926 * @param nullterm If true then the received data is always zero terminated.
2927 * This also causes to allocate one extra byte beyond the
2928 * raw transmitted data.
2929 * @param min_size Minimum size (in bytes) of the data to receive.
2930 * @param max_size Maximum size (in bytes) of the data to receive. 0 means
2932 * @param granulariy If non-zero then the size of the received data has to
2933 * be divisible by this value.
2934 * @param received If not NULL, the size of the received data is stored here.
2936 * @return Zero on success or a value from @ref errno.h on failure.
2939 int async_data_write_accept(void **data
, const bool nullterm
,
2940 const size_t min_size
, const size_t max_size
, const size_t granularity
,
2945 ipc_callid_t callid
;
2947 if (!async_data_write_receive(&callid
, &size
)) {
2948 ipc_answer_0(callid
, EINVAL
);
2952 if (size
< min_size
) {
2953 ipc_answer_0(callid
, EINVAL
);
2957 if ((max_size
> 0) && (size
> max_size
)) {
2958 ipc_answer_0(callid
, EINVAL
);
2962 if ((granularity
> 0) && ((size
% granularity
) != 0)) {
2963 ipc_answer_0(callid
, EINVAL
);
2970 arg_data
= malloc(size
+ 1);
2972 arg_data
= malloc(size
);
2974 if (arg_data
== NULL
) {
2975 ipc_answer_0(callid
, ENOMEM
);
2979 int rc
= async_data_write_finalize(callid
, arg_data
, size
);
2986 ((char *) arg_data
)[size
] = 0;
2989 if (received
!= NULL
)
2995 /** Wrapper for voiding any data that is about to be received
2997 * This wrapper can be used to void any pending data
2999 * @param retval Error value from @ref errno.h to be returned to the caller.
3002 void async_data_write_void(sysarg_t retval
)
3004 ipc_callid_t callid
;
3005 async_data_write_receive(&callid
, NULL
);
3006 ipc_answer_0(callid
, retval
);
3009 /** Wrapper for forwarding any data that is about to be received
3012 int async_data_write_forward_fast(async_exch_t
*exch
, sysarg_t imethod
,
3013 sysarg_t arg1
, sysarg_t arg2
, sysarg_t arg3
, sysarg_t arg4
,
3014 ipc_call_t
*dataptr
)
3019 ipc_callid_t callid
;
3020 if (!async_data_write_receive(&callid
, NULL
)) {
3021 ipc_answer_0(callid
, EINVAL
);
3025 aid_t msg
= async_send_fast(exch
, imethod
, arg1
, arg2
, arg3
, arg4
,
3028 ipc_answer_0(callid
, EINVAL
);
3032 int retval
= ipc_forward_fast(callid
, exch
->phone
, 0, 0, 0,
3033 IPC_FF_ROUTE_FROM_ME
);
3034 if (retval
!= EOK
) {
3036 ipc_answer_0(callid
, retval
);
3041 async_wait_for(msg
, &rc
);
3046 /** Wrapper for receiving the IPC_M_CONNECT_TO_ME calls.
3048 * If the current call is IPC_M_CONNECT_TO_ME then a new
3049 * async session is created for the accepted phone.
3051 * @param mgmt Exchange management style.
3053 * @return New async session.
3054 * @return NULL on failure.
3057 async_sess_t
*async_callback_receive(exch_mgmt_t mgmt
)
3059 /* Accept the phone */
3061 ipc_callid_t callid
= async_get_call(&call
);
3062 int phone
= (int) IPC_GET_ARG5(call
);
3064 if ((IPC_GET_IMETHOD(call
) != IPC_M_CONNECT_TO_ME
) ||
3066 async_answer_0(callid
, EINVAL
);
3070 async_sess_t
*sess
= (async_sess_t
*) malloc(sizeof(async_sess_t
));
3072 async_answer_0(callid
, ENOMEM
);
3078 sess
->phone
= phone
;
3083 fibril_mutex_initialize(&sess
->remote_state_mtx
);
3084 sess
->remote_state_data
= NULL
;
3086 list_initialize(&sess
->exch_list
);
3087 fibril_mutex_initialize(&sess
->mutex
);
3088 atomic_set(&sess
->refcnt
, 0);
3090 /* Acknowledge the connected phone */
3091 async_answer_0(callid
, EOK
);
3096 /** Wrapper for receiving the IPC_M_CONNECT_TO_ME calls.
3098 * If the call is IPC_M_CONNECT_TO_ME then a new
3099 * async session is created. However, the phone is
3100 * not accepted automatically.
3102 * @param mgmt Exchange management style.
3103 * @param call Call data.
3105 * @return New async session.
3106 * @return NULL on failure.
3107 * @return NULL if the call is not IPC_M_CONNECT_TO_ME.
3110 async_sess_t
*async_callback_receive_start(exch_mgmt_t mgmt
, ipc_call_t
*call
)
3112 int phone
= (int) IPC_GET_ARG5(*call
);
3114 if ((IPC_GET_IMETHOD(*call
) != IPC_M_CONNECT_TO_ME
) ||
3118 async_sess_t
*sess
= (async_sess_t
*) malloc(sizeof(async_sess_t
));
3124 sess
->phone
= phone
;
3129 fibril_mutex_initialize(&sess
->remote_state_mtx
);
3130 sess
->remote_state_data
= NULL
;
3132 list_initialize(&sess
->exch_list
);
3133 fibril_mutex_initialize(&sess
->mutex
);
3134 atomic_set(&sess
->refcnt
, 0);
3139 int async_state_change_start(async_exch_t
*exch
, sysarg_t arg1
, sysarg_t arg2
,
3140 sysarg_t arg3
, async_exch_t
*other_exch
)
3142 return async_req_5_0(exch
, IPC_M_STATE_CHANGE_AUTHORIZE
,
3143 arg1
, arg2
, arg3
, 0, other_exch
->phone
);
3146 bool async_state_change_receive(ipc_callid_t
*callid
, sysarg_t
*arg1
,
3147 sysarg_t
*arg2
, sysarg_t
*arg3
)
3152 *callid
= async_get_call(&call
);
3154 if (IPC_GET_IMETHOD(call
) != IPC_M_STATE_CHANGE_AUTHORIZE
)
3158 *arg1
= IPC_GET_ARG1(call
);
3160 *arg2
= IPC_GET_ARG2(call
);
3162 *arg3
= IPC_GET_ARG3(call
);
3167 int async_state_change_finalize(ipc_callid_t callid
, async_exch_t
*other_exch
)
3169 return ipc_answer_1(callid
, EOK
, other_exch
->phone
);
3172 /** Lock and get session remote state
3174 * Lock and get the local replica of the remote state
3175 * in stateful sessions. The call should be paired
3176 * with async_remote_state_release*().
3178 * @param[in] sess Stateful session.
3180 * @return Local replica of the remote state.
3183 void *async_remote_state_acquire(async_sess_t
*sess
)
3185 fibril_mutex_lock(&sess
->remote_state_mtx
);
3186 return sess
->remote_state_data
;
3189 /** Update the session remote state
3191 * Update the local replica of the remote state
3192 * in stateful sessions. The remote state must
3193 * be already locked.
3195 * @param[in] sess Stateful session.
3196 * @param[in] state New local replica of the remote state.
3199 void async_remote_state_update(async_sess_t
*sess
, void *state
)
3201 assert(fibril_mutex_is_locked(&sess
->remote_state_mtx
));
3202 sess
->remote_state_data
= state
;
3205 /** Release the session remote state
3207 * Unlock the local replica of the remote state
3208 * in stateful sessions.
3210 * @param[in] sess Stateful session.
3213 void async_remote_state_release(async_sess_t
*sess
)
3215 assert(fibril_mutex_is_locked(&sess
->remote_state_mtx
));
3217 fibril_mutex_unlock(&sess
->remote_state_mtx
);
3220 /** Release the session remote state and end an exchange
3222 * Unlock the local replica of the remote state
3223 * in stateful sessions. This is convenience function
3224 * which gets the session pointer from the exchange
3225 * and also ends the exchange.
3227 * @param[in] exch Stateful session's exchange.
3230 void async_remote_state_release_exchange(async_exch_t
*exch
)
3235 async_sess_t
*sess
= exch
->sess
;
3236 assert(fibril_mutex_is_locked(&sess
->remote_state_mtx
));
3238 async_exchange_end(exch
);
3239 fibril_mutex_unlock(&sess
->remote_state_mtx
);
3242 void *async_as_area_create(void *base
, size_t size
, unsigned int flags
,
3243 async_sess_t
*pager
, sysarg_t id1
, sysarg_t id2
, sysarg_t id3
)
3245 as_area_pager_info_t pager_info
= {
3246 .pager
= pager
->phone
,
3251 return as_area_create(base
, size
, flags
, &pager_info
);