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
RevLine 
[06502f7d]1/*
[df4ed85]2 * Copyright (c) 2006 Ondrej Palkovsky
[06502f7d]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.
[b2951e2]27 */
28
[a46da63]29/** @addtogroup libc
[b2951e2]30 * @{
31 */
32/** @file
[c07544d3]33 */
[06502f7d]34
[80649a91]35/**
36 * Asynchronous library
37 *
[c07544d3]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.
[80649a91]41 *
[79ae36dd]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.
[80649a91]45 *
[9591265]46 * Example of use (pseudo C):
[c07544d3]47 *
[80649a91]48 * 1) Multithreaded client application
[9591265]49 *
[c07544d3]50 * fibril_create(fibril1, ...);
51 * fibril_create(fibril2, ...);
52 * ...
53 *
54 * int fibril1(void *arg)
55 * {
[79ae36dd]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 *
[c07544d3]66 * async_wait_for(c1);
67 * async_wait_for(c2);
68 * ...
69 * }
[80649a91]70 *
71 *
72 * 2) Multithreaded server application
73 *
[c07544d3]74 * main()
75 * {
76 * async_manager();
77 * }
78 *
79 * my_client_connection(icallid, *icall)
80 * {
81 * if (want_refuse) {
[64d2b10]82 * async_answer_0(icallid, ELIMIT);
[c07544d3]83 * return;
84 * }
[64d2b10]85 * async_answer_0(icallid, EOK);
[80649a91]86 *
[c07544d3]87 * callid = async_get_call(&call);
[0772aff]88 * somehow_handle_the_call(callid, call);
[64d2b10]89 * async_answer_2(callid, 1, 2, 3);
[53ca318]90 *
[c07544d3]91 * callid = async_get_call(&call);
92 * ...
93 * }
[a2cd194]94 *
[80649a91]95 */
[9591265]96
[64d2b10]97#define LIBC_ASYNC_C_
98#include <ipc/ipc.h>
[80649a91]99#include <async.h>
[b76a7329]100#include "private/async.h"
[64d2b10]101#undef LIBC_ASYNC_C_
102
103#include <futex.h>
[bc1f1c2]104#include <fibril.h>
[d9c8c81]105#include <adt/hash_table.h>
106#include <adt/list.h>
[80649a91]107#include <assert.h>
108#include <errno.h>
[daa90e8]109#include <sys/time.h>
[c0699467]110#include <libarch/barrier.h>
[0cc4313]111#include <bool.h>
[c7bbf029]112#include <malloc.h>
[79ae36dd]113#include <mem.h>
114#include <stdlib.h>
[e2ab36f1]115#include <macros.h>
[80649a91]116
[79ae36dd]117
[5da7199]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
[79ae36dd]166/** Async framework global futex */
[fc42b28]167atomic_t async_futex = FUTEX_INITIALIZER;
[80649a91]168
[8619f25]169/** Number of threads waiting for IPC in the kernel. */
170atomic_t threads_in_ipc_wait = { 0 };
171
[79ae36dd]172/** Naming service session */
173async_sess_t *session_ns;
[01ff41c]174
[79ae36dd]175/** Call data */
[80649a91]176typedef struct {
177 link_t link;
[79ae36dd]178
[80649a91]179 ipc_callid_t callid;
180 ipc_call_t call;
181} msg_t;
182
[5da7199]183/** Message data */
184typedef struct {
185 awaiter_t wdata;
186
187 /** If reply was received. */
188 bool done;
[47c9a8c]189
190 /** If the message / reply should be discarded on arrival. */
191 bool forget;
192
193 /** If already destroyed. */
194 bool destroyed;
[5da7199]195
196 /** Pointer to where the answer data is stored. */
197 ipc_call_t *dataptr;
198
199 sysarg_t retval;
200} amsg_t;
201
[79ae36dd]202/* Client connection data */
[c80fdd0]203typedef struct {
204 link_t link;
[79ae36dd]205
[649f087]206 task_id_t in_task_id;
[79ae36dd]207 atomic_t refcnt;
[c80fdd0]208 void *data;
209} client_t;
210
[79ae36dd]211/* Server connection data */
[80649a91]212typedef struct {
[49d072e]213 awaiter_t wdata;
[c07544d3]214
[e70bfa5]215 /** Hash table link. */
216 link_t link;
[c07544d3]217
[e2ab36f1]218 /** Incoming client task ID. */
219 task_id_t in_task_id;
[79ae36dd]220
[e70bfa5]221 /** Incoming phone hash. */
[96b02eb9]222 sysarg_t in_phone_hash;
[c07544d3]223
[23882034]224 /** Link to the client tracking structure. */
225 client_t *client;
[47b7006]226
[e70bfa5]227 /** Messages that should be delivered to this fibril. */
[b72efe8]228 list_t msg_queue;
[c07544d3]229
[e70bfa5]230 /** Identification of the opening call. */
[80649a91]231 ipc_callid_t callid;
[e70bfa5]232 /** Call data of the opening call. */
[80649a91]233 ipc_call_t call;
[9934f7d]234 /** Local argument or NULL if none. */
235 void *carg;
[c07544d3]236
[e70bfa5]237 /** Identification of the closing call. */
238 ipc_callid_t close_callid;
[c07544d3]239
[e70bfa5]240 /** Fibril function that will be used to handle the connection. */
[9934f7d]241 async_client_conn_t cfibril;
[80649a91]242} connection_t;
243
[bc1f1c2]244/** Identifier of the incoming connection handled by the current fibril. */
[79ae36dd]245static fibril_local connection_t *fibril_connection;
[e70bfa5]246
[47c9a8c]247static void to_event_initialize(to_event_t *to)
248{
[aeeddeb]249 struct timeval tv = { 0, 0 };
[47c9a8c]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
[46eec3b]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{
[f302586]311 assert(async_client_data_create == default_client_data_constructor);
[46eec3b]312 async_client_data_create = ctor;
313}
314
315void async_set_client_data_destructor(async_client_data_dtor_t dtor)
316{
[f302586]317 assert(async_client_data_destroy == default_client_data_destructor);
[46eec3b]318 async_client_data_destroy = dtor;
319}
320
[47b7006]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 *
[3815efb]325 * @param callid Hash of the incoming call.
326 * @param call Data of the incoming call.
327 * @param arg Local argument
[47b7006]328 *
329 */
[9934f7d]330static void default_client_connection(ipc_callid_t callid, ipc_call_t *call,
331 void *arg)
[47b7006]332{
333 ipc_answer_0(callid, ENOENT);
334}
[36c9234]335
[47b7006]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 *
[3815efb]340 * @param callid Hash of the incoming call.
341 * @param call Data of the incoming call.
342 * @param arg Local argument.
[47b7006]343 *
344 */
345static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call)
346{
347}
348
[79ae36dd]349static async_client_conn_t client_connection = default_client_connection;
[9934f7d]350static async_interrupt_handler_t interrupt_received = default_interrupt_received;
[da0c91e7]351
[79ae36dd]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{
[f302586]359 assert(client_connection == default_client_connection);
[79ae36dd]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 */
[9934f7d]368void async_set_interrupt_received(async_interrupt_handler_t intr)
[79ae36dd]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
[c80fdd0]388static hash_table_t client_hash_table;
[c07544d3]389static hash_table_t conn_hash_table;
390static LIST_INITIALIZE(timeout_list);
391
[0ca7286]392static size_t client_key_hash(unsigned long key[])
[c80fdd0]393{
394 assert(key);
[0ca7286]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);
[79ae36dd]402
[0ca7286]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);
[c80fdd0]409}
410
[0ca7286]411static bool client_match(unsigned long key[], size_t keys, const link_t *item)
[c80fdd0]412{
[79ae36dd]413 assert(key);
[e2ab36f1]414 assert(keys == 2);
[79ae36dd]415 assert(item);
416
[47b7006]417 client_t *client = hash_table_get_instance(item, client_t, link);
[e2ab36f1]418 return (key[0] == LOWER32(client->in_task_id) &&
419 (key[1] == UPPER32(client->in_task_id)));
[c80fdd0]420}
421
422
423/** Operations for the client hash table. */
[0ca7286]424static hash_table_ops_t client_hash_table_ops = {
[c80fdd0]425 .hash = client_hash,
[0ca7286]426 .key_hash = client_key_hash,
427 .match = client_match,
428 .equal = 0,
429 .remove_callback = 0
[c80fdd0]430};
[80649a91]431
[e70bfa5]432/** Compute hash into the connection hash table based on the source phone hash.
433 *
[c07544d3]434 * @param key Pointer to source phone hash.
435 *
436 * @return Index into the connection hash table.
[e70bfa5]437 *
438 */
[0ca7286]439static size_t conn_key_hash(unsigned long key[])
[450cd3a]440{
[80649a91]441 assert(key);
[0ca7286]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);
[450cd3a]450}
[06502f7d]451
[e70bfa5]452/** Compare hash table item with a key.
453 *
[c07544d3]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.
[e70bfa5]459 *
460 */
[0ca7286]461static bool conn_match(unsigned long key[], size_t keys, const link_t *item)
[450cd3a]462{
[79ae36dd]463 assert(key);
464 assert(item);
465
[47b7006]466 connection_t *conn = hash_table_get_instance(item, connection_t, link);
467 return (key[0] == conn->in_phone_hash);
[450cd3a]468}
[06502f7d]469
[0ca7286]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
[80649a91]478static void conn_remove(link_t *item)
[450cd3a]479{
480}
481
[e70bfa5]482/** Operations for the connection hash table. */
[0ca7286]483static hash_table_ops_t conn_hash_table_ops = {
[80649a91]484 .hash = conn_hash,
[0ca7286]485 .key_hash = conn_key_hash,
486 .match = conn_match,
487 .equal = conn_equal,
[80649a91]488 .remove_callback = conn_remove
489};
490
[e70bfa5]491/** Sort in current fibril's timeout request.
[49d072e]492 *
[c07544d3]493 * @param wd Wait data of the current fibril.
494 *
[49d072e]495 */
[b6ee5b1]496void async_insert_timeout(awaiter_t *wd)
[49d072e]497{
[79ae36dd]498 assert(wd);
499
[f53cc81]500 wd->to_event.occurred = false;
501 wd->to_event.inlist = true;
[c07544d3]502
[b72efe8]503 link_t *tmp = timeout_list.head.next;
504 while (tmp != &timeout_list.head) {
[47b7006]505 awaiter_t *cur
506 = list_get_instance(tmp, awaiter_t, to_event.link);
[c07544d3]507
[f53cc81]508 if (tv_gteq(&cur->to_event.expires, &wd->to_event.expires))
[49d072e]509 break;
[47b7006]510
[49d072e]511 tmp = tmp->next;
512 }
[c07544d3]513
[b72efe8]514 list_insert_before(&wd->to_event.link, tmp);
[49d072e]515}
516
[e70bfa5]517/** Try to route a call to an appropriate connection fibril.
[80649a91]518 *
[36c9234]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 *
[c07544d3]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.
[47b7006]527 * @return True if the call was passed to the respective connection fibril.
[36c9234]528 *
[80649a91]529 */
[c07544d3]530static bool route_call(ipc_callid_t callid, ipc_call_t *call)
[450cd3a]531{
[79ae36dd]532 assert(call);
533
[01ff41c]534 futex_down(&async_futex);
[c07544d3]535
536 unsigned long key = call->in_phone_hash;
537 link_t *hlp = hash_table_find(&conn_hash_table, &key);
538
[80649a91]539 if (!hlp) {
[01ff41c]540 futex_up(&async_futex);
[c07544d3]541 return false;
[450cd3a]542 }
[c07544d3]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
[80649a91]552 msg->callid = callid;
553 msg->call = *call;
554 list_append(&msg->link, &conn->msg_queue);
[c07544d3]555
[228e490]556 if (IPC_GET_IMETHOD(*call) == IPC_M_PHONE_HUNGUP)
[41269bd]557 conn->close_callid = callid;
[80649a91]558
[36c9234]559 /* If the connection fibril is waiting for an event, activate it */
[49d072e]560 if (!conn->wdata.active) {
[c07544d3]561
[49d072e]562 /* If in timeout list, remove it */
[f53cc81]563 if (conn->wdata.to_event.inlist) {
564 conn->wdata.to_event.inlist = false;
565 list_remove(&conn->wdata.to_event.link);
[49d072e]566 }
[c07544d3]567
568 conn->wdata.active = true;
[bc1f1c2]569 fibril_add_ready(conn->wdata.fid);
[80649a91]570 }
[c07544d3]571
[01ff41c]572 futex_up(&async_futex);
[c07544d3]573 return true;
574}
[80649a91]575
[c07544d3]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{
[79ae36dd]588 assert(arg);
589
[c07544d3]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{
[79ae36dd]610 assert(call);
611
[c07544d3]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);
[86d7bfa]624 if (fid == 0) {
625 free(msg);
626 futex_up(&async_futex);
627 return false;
628 }
629
[c07544d3]630 fibril_add_ready(fid);
631
632 futex_up(&async_futex);
633 return true;
[80649a91]634}
635
[e70bfa5]636/** Return new incoming message for the current (fibril-local) connection.
637 *
[c07544d3]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.
[e70bfa5]646 *
647 */
[49d072e]648ipc_callid_t async_get_call_timeout(ipc_call_t *call, suseconds_t usecs)
[80649a91]649{
[79ae36dd]650 assert(call);
651 assert(fibril_connection);
[c07544d3]652
653 /* Why doing this?
[79ae36dd]654 * GCC 4.1.0 coughs on fibril_connection-> dereference.
[6c46350]655 * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
[c07544d3]656 * I would never expect to find so many errors in
657 * a compiler.
[6c46350]658 */
[79ae36dd]659 connection_t *conn = fibril_connection;
[c07544d3]660
[01ff41c]661 futex_down(&async_futex);
[c07544d3]662
[49d072e]663 if (usecs) {
[f53cc81]664 gettimeofday(&conn->wdata.to_event.expires, NULL);
665 tv_add(&conn->wdata.to_event.expires, usecs);
[c07544d3]666 } else
[f53cc81]667 conn->wdata.to_event.inlist = false;
[c07544d3]668
[e70bfa5]669 /* If nothing in queue, wait until something arrives */
[6c46350]670 while (list_empty(&conn->msg_queue)) {
[8c8f8d6]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
[47b7006]677 * IPC_M_PHONE_HUNGUP until the caller notices.
[8c8f8d6]678 */
679 memset(call, 0, sizeof(ipc_call_t));
[228e490]680 IPC_SET_IMETHOD(*call, IPC_M_PHONE_HUNGUP);
[8c8f8d6]681 futex_up(&async_futex);
682 return conn->close_callid;
683 }
[47b7006]684
[085bd54]685 if (usecs)
[b6ee5b1]686 async_insert_timeout(&conn->wdata);
[c07544d3]687
688 conn->wdata.active = false;
689
[c7509e5]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 */
[116d3f6f]696 fibril_switch(FIBRIL_TO_MANAGER);
[c07544d3]697
[e70bfa5]698 /*
[c07544d3]699 * Futex is up after getting back from async_manager.
700 * Get it again.
[c7509e5]701 */
[49d072e]702 futex_down(&async_futex);
[f53cc81]703 if ((usecs) && (conn->wdata.to_event.occurred)
[c07544d3]704 && (list_empty(&conn->msg_queue))) {
[e70bfa5]705 /* If we timed out -> exit */
[49d072e]706 futex_up(&async_futex);
707 return 0;
708 }
[450cd3a]709 }
710
[b72efe8]711 msg_t *msg = list_get_instance(list_first(&conn->msg_queue), msg_t, link);
[80649a91]712 list_remove(&msg->link);
[c07544d3]713
714 ipc_callid_t callid = msg->callid;
[80649a91]715 *call = msg->call;
716 free(msg);
717
[01ff41c]718 futex_up(&async_futex);
[80649a91]719 return callid;
720}
721
[e2ab36f1]722static client_t *async_client_get(task_id_t client_id, bool create)
[26fbb7bb]723{
[e2ab36f1]724 unsigned long key[2] = {
725 LOWER32(client_id),
726 UPPER32(client_id),
727 };
[26fbb7bb]728 client_t *client = NULL;
729
730 futex_down(&async_futex);
[e2ab36f1]731 link_t *lnk = hash_table_find(&client_hash_table, key);
[26fbb7bb]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) {
[e2ab36f1]738 client->in_task_id = client_id;
[26fbb7bb]739 client->data = async_client_data_create();
740
741 atomic_set(&client->refcnt, 1);
[0ca7286]742 hash_table_insert(&client_hash_table, &client->link);
[26fbb7bb]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;
[e2ab36f1]753 unsigned long key[2] = {
754 LOWER32(client->in_task_id),
755 UPPER32(client->in_task_id)
756 };
[26fbb7bb]757
758 futex_down(&async_futex);
759
760 if (atomic_predec(&client->refcnt) == 0) {
[e2ab36f1]761 hash_table_remove(&client_hash_table, key, 2);
[26fbb7bb]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
[455f190]776void *async_get_client_data(void)
777{
778 assert(fibril_connection);
779 return fibril_connection->client->data;
780}
781
[e2ab36f1]782void *async_get_client_data_by_id(task_id_t client_id)
[455f190]783{
[e2ab36f1]784 client_t *client = async_client_get(client_id, false);
[455f190]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
[e2ab36f1]795void async_put_client_data_by_id(task_id_t client_id)
[455f190]796{
[e2ab36f1]797 client_t *client = async_client_get(client_id, false);
[455f190]798
799 assert(client);
800 assert(client->data);
801
[cdc8ee2d]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. */
[455f190]806 async_client_put(client);
807}
808
[f2f0392]809/** Wrapper for client connection fibril.
810 *
[36c9234]811 * When a new connection arrives, a fibril with this implementing function is
[f2f0392]812 * created. It calls client_connection() and does the final cleanup.
[a2cd194]813 *
[c07544d3]814 * @param arg Connection structure pointer.
815 *
816 * @return Always zero.
[a2cd194]817 *
818 */
[c07544d3]819static int connection_fibril(void *arg)
[80649a91]820{
[79ae36dd]821 assert(arg);
822
[c07544d3]823 /*
[c80fdd0]824 * Setup fibril-local connection pointer.
[c07544d3]825 */
[79ae36dd]826 fibril_connection = (connection_t *) arg;
[47b7006]827
[c80fdd0]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 */
[26fbb7bb]833
[e2ab36f1]834 client_t *client = async_client_get(fibril_connection->in_task_id, true);
[26fbb7bb]835 if (!client) {
836 ipc_answer_0(fibril_connection->callid, ENOMEM);
837 return 0;
[c80fdd0]838 }
[26fbb7bb]839
[79ae36dd]840 fibril_connection->client = client;
[47b7006]841
[c80fdd0]842 /*
843 * Call the connection handler function.
844 */
[79ae36dd]845 fibril_connection->cfibril(fibril_connection->callid,
[9934f7d]846 &fibril_connection->call, fibril_connection->carg);
[a46da63]847
[c80fdd0]848 /*
849 * Remove the reference for this client task connection.
850 */
[26fbb7bb]851 async_client_put(client);
[47b7006]852
[c80fdd0]853 /*
854 * Remove myself from the connection hash table.
855 */
856 futex_down(&async_futex);
[26fbb7bb]857 unsigned long key = fibril_connection->in_phone_hash;
[a2cd194]858 hash_table_remove(&conn_hash_table, &key, 1);
[01ff41c]859 futex_up(&async_futex);
[a46da63]860
[c80fdd0]861 /*
862 * Answer all remaining messages with EHANGUP.
863 */
[79ae36dd]864 while (!list_empty(&fibril_connection->msg_queue)) {
[47b7006]865 msg_t *msg =
[b72efe8]866 list_get_instance(list_first(&fibril_connection->msg_queue),
867 msg_t, link);
[c07544d3]868
[a2cd194]869 list_remove(&msg->link);
[b74959bd]870 ipc_answer_0(msg->callid, EHANGUP);
[a2cd194]871 free(msg);
872 }
[c07544d3]873
[c80fdd0]874 /*
875 * If the connection was hung-up, answer the last call,
876 * i.e. IPC_M_PHONE_HUNGUP.
877 */
[79ae36dd]878 if (fibril_connection->close_callid)
879 ipc_answer_0(fibril_connection->close_callid, EOK);
[a46da63]880
[79ae36dd]881 free(fibril_connection);
[a46da63]882 return 0;
[80649a91]883}
884
[f2f0392]885/** Create a new fibril for a new connection.
[80649a91]886 *
[79ae36dd]887 * Create new fibril for connection, fill in connection structures and insert
[f2f0392]888 * it into the hash table, so that later we can easily do routing of messages to
889 * particular fibrils.
[53ca318]890 *
[e2ab36f1]891 * @param in_task_id Identification of the incoming connection.
[c07544d3]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.
[3815efb]900 * @param carg Extra argument to pass to the connection fibril
[c07544d3]901 *
902 * @return New fibril id or NULL on failure.
[36c9234]903 *
[80649a91]904 */
[e2ab36f1]905fid_t async_new_connection(task_id_t in_task_id, sysarg_t in_phone_hash,
[3c22f70]906 ipc_callid_t callid, ipc_call_t *call,
[9934f7d]907 async_client_conn_t cfibril, void *carg)
[80649a91]908{
[c07544d3]909 connection_t *conn = malloc(sizeof(*conn));
[80649a91]910 if (!conn) {
[6675c70]911 if (callid)
[b74959bd]912 ipc_answer_0(callid, ENOMEM);
[47b7006]913
[0b4a67a]914 return (uintptr_t) NULL;
[80649a91]915 }
[c07544d3]916
[e2ab36f1]917 conn->in_task_id = in_task_id;
[44c6d88d]918 conn->in_phone_hash = in_phone_hash;
[80649a91]919 list_initialize(&conn->msg_queue);
920 conn->callid = callid;
[c4702804]921 conn->close_callid = 0;
[9934f7d]922 conn->carg = carg;
[c07544d3]923
[eaf34f7]924 if (call)
925 conn->call = *call;
[6b21292]926
[c07544d3]927 /* We will activate the fibril ASAP */
928 conn->wdata.active = true;
929 conn->cfibril = cfibril;
[bc1f1c2]930 conn->wdata.fid = fibril_create(connection_fibril, conn);
[c07544d3]931
[86d7bfa]932 if (conn->wdata.fid == 0) {
[80649a91]933 free(conn);
[86d7bfa]934
[6675c70]935 if (callid)
[b74959bd]936 ipc_answer_0(callid, ENOMEM);
[86d7bfa]937
[0b4a67a]938 return (uintptr_t) NULL;
[80649a91]939 }
[6b21292]940
[36c9234]941 /* Add connection to the connection hash table */
[c07544d3]942
[01ff41c]943 futex_down(&async_futex);
[0ca7286]944 hash_table_insert(&conn_hash_table, &conn->link);
[01ff41c]945 futex_up(&async_futex);
[6b21292]946
[bc1f1c2]947 fibril_add_ready(conn->wdata.fid);
[6b21292]948
[bc1f1c2]949 return conn->wdata.fid;
[80649a91]950}
951
[36c9234]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 *
[c07544d3]957 * @param callid Hash of the incoming call.
958 * @param call Data of the incoming call.
[6b21292]959 *
[36c9234]960 */
[80649a91]961static void handle_call(ipc_callid_t callid, ipc_call_t *call)
962{
[79ae36dd]963 assert(call);
964
[47b7006]965 /* Unrouted call - take some default action */
[15039b67]966 if ((callid & IPC_CALLID_NOTIFICATION)) {
[c07544d3]967 process_notification(callid, call);
[47b7006]968 return;
[6b21292]969 }
970
[228e490]971 switch (IPC_GET_IMETHOD(*call)) {
[6aae539d]972 case IPC_M_CLONE_ESTABLISH:
[80649a91]973 case IPC_M_CONNECT_ME_TO:
[47b7006]974 /* Open new connection with fibril, etc. */
[e2ab36f1]975 async_new_connection(call->in_task_id, IPC_GET_ARG5(*call),
[9934f7d]976 callid, call, client_connection, NULL);
[47b7006]977 return;
[80649a91]978 }
[6b21292]979
[36c9234]980 /* Try to route the call through the connection hash table */
[44c6d88d]981 if (route_call(callid, call))
[47b7006]982 return;
[6b21292]983
[44c6d88d]984 /* Unknown call from unknown phone - hang it up */
[b74959bd]985 ipc_answer_0(callid, EHANGUP);
[450cd3a]986}
987
[f2f0392]988/** Fire all timeouts that expired. */
[c042bdd]989static void handle_expired_timeouts(void)
990{
991 struct timeval tv;
[36c9234]992 gettimeofday(&tv, NULL);
[c07544d3]993
[c042bdd]994 futex_down(&async_futex);
[c07544d3]995
[b72efe8]996 link_t *cur = list_first(&timeout_list);
997 while (cur != NULL) {
[47b7006]998 awaiter_t *waiter =
999 list_get_instance(cur, awaiter_t, to_event.link);
[c07544d3]1000
[f53cc81]1001 if (tv_gt(&waiter->to_event.expires, &tv))
[c042bdd]1002 break;
[47b7006]1003
[f53cc81]1004 list_remove(&waiter->to_event.link);
1005 waiter->to_event.inlist = false;
1006 waiter->to_event.occurred = true;
[c07544d3]1007
[36c9234]1008 /*
[c07544d3]1009 * Redundant condition?
1010 * The fibril should not be active when it gets here.
[c042bdd]1011 */
[49d072e]1012 if (!waiter->active) {
[c07544d3]1013 waiter->active = true;
[bc1f1c2]1014 fibril_add_ready(waiter->fid);
[c042bdd]1015 }
[b72efe8]1016
1017 cur = list_first(&timeout_list);
[c042bdd]1018 }
[c07544d3]1019
[c042bdd]1020 futex_up(&async_futex);
1021}
1022
[36c9234]1023/** Endless loop dispatching incoming calls and answers.
1024 *
[c07544d3]1025 * @return Never returns.
1026 *
[36c9234]1027 */
[085bd54]1028static int async_manager_worker(void)
[80649a91]1029{
[c07544d3]1030 while (true) {
[116d3f6f]1031 if (fibril_switch(FIBRIL_FROM_MANAGER)) {
[47b7006]1032 futex_up(&async_futex);
[36c9234]1033 /*
1034 * async_futex is always held when entering a manager
1035 * fibril.
[a46da63]1036 */
[80649a91]1037 continue;
1038 }
[c07544d3]1039
[c042bdd]1040 futex_down(&async_futex);
[c07544d3]1041
1042 suseconds_t timeout;
[1db6dfd]1043 unsigned int flags = SYNCH_FLAGS_NONE;
[c042bdd]1044 if (!list_empty(&timeout_list)) {
[b72efe8]1045 awaiter_t *waiter = list_get_instance(
1046 list_first(&timeout_list), awaiter_t, to_event.link);
[c07544d3]1047
1048 struct timeval tv;
[bc1f1c2]1049 gettimeofday(&tv, NULL);
[c07544d3]1050
[f53cc81]1051 if (tv_gteq(&tv, &waiter->to_event.expires)) {
[6c46350]1052 futex_up(&async_futex);
[c042bdd]1053 handle_expired_timeouts();
[1db6dfd]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 {
[47b7006]1068 timeout = tv_sub(&waiter->to_event.expires, &tv);
[1db6dfd]1069 futex_up(&async_futex);
1070 }
1071 } else {
1072 futex_up(&async_futex);
[0b99e40]1073 timeout = SYNCH_NO_TIMEOUT;
[1db6dfd]1074 }
[47b7006]1075
[8619f25]1076 atomic_inc(&threads_in_ipc_wait);
[c07544d3]1077
1078 ipc_call_t call;
[1db6dfd]1079 ipc_callid_t callid = ipc_wait_cycle(&call, timeout, flags);
[c07544d3]1080
[8619f25]1081 atomic_dec(&threads_in_ipc_wait);
[47b7006]1082
[0b99e40]1083 if (!callid) {
[c042bdd]1084 handle_expired_timeouts();
[0b99e40]1085 continue;
1086 }
[c07544d3]1087
1088 if (callid & IPC_CALLID_ANSWERED)
[80649a91]1089 continue;
[c07544d3]1090
[80649a91]1091 handle_call(callid, &call);
1092 }
[a46da63]1093
1094 return 0;
[80649a91]1095}
1096
[36c9234]1097/** Function to start async_manager as a standalone fibril.
[c07544d3]1098 *
[36c9234]1099 * When more kernel threads are used, one async manager should exist per thread.
1100 *
[c07544d3]1101 * @param arg Unused.
1102 * @return Never returns.
[36c9234]1103 *
[a2cd194]1104 */
[9591265]1105static int async_manager_fibril(void *arg)
[80649a91]1106{
[a46da63]1107 futex_up(&async_futex);
[c07544d3]1108
[36c9234]1109 /*
1110 * async_futex is always locked when entering manager
1111 */
[085bd54]1112 async_manager_worker();
[a46da63]1113
1114 return 0;
[80649a91]1115}
[450cd3a]1116
[36c9234]1117/** Add one manager to manager list. */
[80649a91]1118void async_create_manager(void)
[450cd3a]1119{
[c07544d3]1120 fid_t fid = fibril_create(async_manager_fibril, NULL);
[86d7bfa]1121 if (fid != 0)
1122 fibril_add_manager(fid);
[80649a91]1123}
1124
1125/** Remove one manager from manager list */
1126void async_destroy_manager(void)
1127{
[bc1f1c2]1128 fibril_remove_manager();
[80649a91]1129}
1130
[36c9234]1131/** Initialize the async framework.
1132 *
1133 */
[47b7006]1134void __async_init(void)
[80649a91]1135{
[0ca7286]1136 if (!hash_table_create(&client_hash_table, 0, 2, &client_hash_table_ops))
[47b7006]1137 abort();
[80649a91]1138
[0ca7286]1139 if (!hash_table_create(&conn_hash_table, 0, 1, &conn_hash_table_ops))
[47b7006]1140 abort();
[79ae36dd]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
[58cbf8d5]1152 fibril_mutex_initialize(&session_ns->remote_state_mtx);
1153 session_ns->remote_state_data = NULL;
1154
[79ae36dd]1155 list_initialize(&session_ns->exch_list);
1156 fibril_mutex_initialize(&session_ns->mutex);
1157 atomic_set(&session_ns->refcnt, 0);
[450cd3a]1158}
[01ff41c]1159
[36c9234]1160/** Reply received callback.
[01ff41c]1161 *
[36c9234]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 *
[c07544d3]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.
[47b7006]1170 *
[01ff41c]1171 */
[79ae36dd]1172void reply_received(void *arg, int retval, ipc_call_t *data)
[01ff41c]1173{
[79ae36dd]1174 assert(arg);
1175
[9db9b10]1176 futex_down(&async_futex);
1177
[c07544d3]1178 amsg_t *msg = (amsg_t *) arg;
[01ff41c]1179 msg->retval = retval;
[c07544d3]1180
[36c9234]1181 /* Copy data after futex_down, just in case the call was detached */
[9db9b10]1182 if ((msg->dataptr) && (data))
[c07544d3]1183 *msg->dataptr = *data;
1184
[c042bdd]1185 write_barrier();
[c07544d3]1186
[c042bdd]1187 /* Remove message from timeout list */
[f53cc81]1188 if (msg->wdata.to_event.inlist)
1189 list_remove(&msg->wdata.to_event.link);
[c07544d3]1190
1191 msg->done = true;
[47c9a8c]1192
1193 if (msg->forget) {
1194 assert(msg->wdata.active);
1195 amsg_destroy(msg);
1196 } else if (!msg->wdata.active) {
[c07544d3]1197 msg->wdata.active = true;
[bc1f1c2]1198 fibril_add_ready(msg->wdata.fid);
[01ff41c]1199 }
[47c9a8c]1200
[01ff41c]1201 futex_up(&async_futex);
1202}
1203
[36c9234]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.
[01ff41c]1208 *
[79ae36dd]1209 * @param exch Exchange for sending the message.
1210 * @param imethod Service-defined interface and method.
[c07544d3]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.
[36c9234]1219 *
[01ff41c]1220 */
[79ae36dd]1221aid_t async_send_fast(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
[96b02eb9]1222 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, ipc_call_t *dataptr)
[01ff41c]1223{
[79ae36dd]1224 if (exch == NULL)
1225 return 0;
[c07544d3]1226
[47c9a8c]1227 amsg_t *msg = amsg_create();
[79ae36dd]1228 if (msg == NULL)
[c07544d3]1229 return 0;
[6b21292]1230
[01ff41c]1231 msg->dataptr = dataptr;
[c07544d3]1232 msg->wdata.active = true;
1233
[79ae36dd]1234 ipc_call_async_4(exch->phone, imethod, arg1, arg2, arg3, arg4, msg,
[c07544d3]1235 reply_received, true);
[6b21292]1236
[01ff41c]1237 return (aid_t) msg;
1238}
1239
[90f5d64]1240/** Send message and return id of the sent message
1241 *
[36c9234]1242 * The return value can be used as input for async_wait() to wait for
1243 * completion.
1244 *
[79ae36dd]1245 * @param exch Exchange for sending the message.
1246 * @param imethod Service-defined interface and method.
[c07544d3]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.
[36c9234]1256 *
[90f5d64]1257 */
[79ae36dd]1258aid_t async_send_slow(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
[96b02eb9]1259 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5,
[0cc4313]1260 ipc_call_t *dataptr)
[90f5d64]1261{
[79ae36dd]1262 if (exch == NULL)
1263 return 0;
1264
[47c9a8c]1265 amsg_t *msg = amsg_create();
[79ae36dd]1266 if (msg == NULL)
[c07544d3]1267 return 0;
1268
[90f5d64]1269 msg->dataptr = dataptr;
[c07544d3]1270 msg->wdata.active = true;
[6b21292]1271
[79ae36dd]1272 ipc_call_async_5(exch->phone, imethod, arg1, arg2, arg3, arg4, arg5,
1273 msg, reply_received, true);
[6b21292]1274
[90f5d64]1275 return (aid_t) msg;
1276}
1277
[36c9234]1278/** Wait for a message sent by the async framework.
[01ff41c]1279 *
[c07544d3]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 *
[01ff41c]1284 */
[96b02eb9]1285void async_wait_for(aid_t amsgid, sysarg_t *retval)
[01ff41c]1286{
[79ae36dd]1287 assert(amsgid);
1288
[01ff41c]1289 amsg_t *msg = (amsg_t *) amsgid;
[c07544d3]1290
[01ff41c]1291 futex_down(&async_futex);
[47c9a8c]1292
1293 assert(!msg->forget);
1294 assert(!msg->destroyed);
1295
[01ff41c]1296 if (msg->done) {
1297 futex_up(&async_futex);
1298 goto done;
1299 }
[c07544d3]1300
[bc1f1c2]1301 msg->wdata.fid = fibril_get_id();
[c07544d3]1302 msg->wdata.active = false;
[f53cc81]1303 msg->wdata.to_event.inlist = false;
[c07544d3]1304
[36c9234]1305 /* Leave the async_futex locked when entering this function */
[116d3f6f]1306 fibril_switch(FIBRIL_TO_MANAGER);
[c07544d3]1307
1308 /* Futex is up automatically after fibril_switch */
1309
[01ff41c]1310done:
1311 if (retval)
1312 *retval = msg->retval;
[c07544d3]1313
[47c9a8c]1314 amsg_destroy(msg);
[01ff41c]1315}
[0b99e40]1316
[36c9234]1317/** Wait for a message sent by the async framework, timeout variant.
[47c9a8c]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().
[c042bdd]1322 *
[c07544d3]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.
[c042bdd]1329 *
1330 */
[96b02eb9]1331int async_wait_timeout(aid_t amsgid, sysarg_t *retval, suseconds_t timeout)
[c042bdd]1332{
[79ae36dd]1333 assert(amsgid);
1334
[c042bdd]1335 amsg_t *msg = (amsg_t *) amsgid;
[1db6dfd]1336
[c042bdd]1337 futex_down(&async_futex);
[47c9a8c]1338
1339 assert(!msg->forget);
1340 assert(!msg->destroyed);
1341
[c042bdd]1342 if (msg->done) {
1343 futex_up(&async_futex);
1344 goto done;
1345 }
[c07544d3]1346
[1db6dfd]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
[f53cc81]1354 gettimeofday(&msg->wdata.to_event.expires, NULL);
1355 tv_add(&msg->wdata.to_event.expires, timeout);
[c07544d3]1356
[1db6dfd]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 */
[bc1f1c2]1374 msg->wdata.fid = fibril_get_id();
[c07544d3]1375 msg->wdata.active = false;
[b6ee5b1]1376 async_insert_timeout(&msg->wdata);
[c07544d3]1377
[36c9234]1378 /* Leave the async_futex locked when entering this function */
[116d3f6f]1379 fibril_switch(FIBRIL_TO_MANAGER);
[c07544d3]1380
1381 /* Futex is up automatically after fibril_switch */
1382
[c042bdd]1383 if (!msg->done)
1384 return ETIMEOUT;
[c07544d3]1385
[c042bdd]1386done:
1387 if (retval)
1388 *retval = msg->retval;
[c07544d3]1389
[47c9a8c]1390 amsg_destroy(msg);
[c07544d3]1391
[c042bdd]1392 return 0;
1393}
[47c9a8c]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);
[375e501]1412 if (msg->done) {
[47c9a8c]1413 amsg_destroy(msg);
[375e501]1414 } else {
1415 msg->dataptr = NULL;
[47c9a8c]1416 msg->forget = true;
[375e501]1417 }
[47c9a8c]1418 futex_up(&async_futex);
1419}
[0b99e40]1420
[36c9234]1421/** Wait for specified time.
[44c6d88d]1422 *
[36c9234]1423 * The current fibril is suspended but the thread continues to execute.
1424 *
[c07544d3]1425 * @param timeout Duration of the wait in microseconds.
1426 *
[44c6d88d]1427 */
1428void async_usleep(suseconds_t timeout)
1429{
[47c9a8c]1430 amsg_t *msg = amsg_create();
[44c6d88d]1431 if (!msg)
1432 return;
[6b21292]1433
[bc1f1c2]1434 msg->wdata.fid = fibril_get_id();
[6b21292]1435
[f53cc81]1436 gettimeofday(&msg->wdata.to_event.expires, NULL);
1437 tv_add(&msg->wdata.to_event.expires, timeout);
[6b21292]1438
[44c6d88d]1439 futex_down(&async_futex);
[c07544d3]1440
[b6ee5b1]1441 async_insert_timeout(&msg->wdata);
[c07544d3]1442
[36c9234]1443 /* Leave the async_futex locked when entering this function */
[116d3f6f]1444 fibril_switch(FIBRIL_TO_MANAGER);
[c07544d3]1445
1446 /* Futex is up automatically after fibril_switch() */
1447
[47c9a8c]1448 amsg_destroy(msg);
[44c6d88d]1449}
[da0c91e7]1450
[0cc4313]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 *
[79ae36dd]1458 * @param exch Exchange for sending the message.
1459 * @param imethod Interface and method of the call.
[c07544d3]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 *
[0cc4313]1472 */
[79ae36dd]1473sysarg_t async_req_fast(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
[96b02eb9]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)
[085bd54]1476{
[79ae36dd]1477 if (exch == NULL)
1478 return ENOENT;
1479
[0cc4313]1480 ipc_call_t result;
[79ae36dd]1481 aid_t aid = async_send_4(exch, imethod, arg1, arg2, arg3, arg4,
[0cc4313]1482 &result);
[c07544d3]1483
[96b02eb9]1484 sysarg_t rc;
[79ae36dd]1485 async_wait_for(aid, &rc);
[c07544d3]1486
1487 if (r1)
[0cc4313]1488 *r1 = IPC_GET_ARG1(result);
[c07544d3]1489
[0cc4313]1490 if (r2)
1491 *r2 = IPC_GET_ARG2(result);
[c07544d3]1492
[0cc4313]1493 if (r3)
1494 *r3 = IPC_GET_ARG3(result);
[c07544d3]1495
[0cc4313]1496 if (r4)
1497 *r4 = IPC_GET_ARG4(result);
[c07544d3]1498
[0cc4313]1499 if (r5)
1500 *r5 = IPC_GET_ARG5(result);
[c07544d3]1501
[0cc4313]1502 return rc;
[085bd54]1503}
1504
[0cc4313]1505/** Pseudo-synchronous message sending - slow version.
1506 *
1507 * Send message asynchronously and return only after the reply arrives.
1508 *
[79ae36dd]1509 * @param exch Exchange for sending the message.
1510 * @param imethod Interface and method of the call.
[c07544d3]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 *
[0cc4313]1524 */
[79ae36dd]1525sysarg_t async_req_slow(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
[96b02eb9]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)
[085bd54]1528{
[79ae36dd]1529 if (exch == NULL)
1530 return ENOENT;
1531
[0cc4313]1532 ipc_call_t result;
[79ae36dd]1533 aid_t aid = async_send_5(exch, imethod, arg1, arg2, arg3, arg4, arg5,
[0cc4313]1534 &result);
[c07544d3]1535
[96b02eb9]1536 sysarg_t rc;
[79ae36dd]1537 async_wait_for(aid, &rc);
[c07544d3]1538
1539 if (r1)
[0cc4313]1540 *r1 = IPC_GET_ARG1(result);
[c07544d3]1541
[0cc4313]1542 if (r2)
1543 *r2 = IPC_GET_ARG2(result);
[c07544d3]1544
[0cc4313]1545 if (r3)
1546 *r3 = IPC_GET_ARG3(result);
[c07544d3]1547
[0cc4313]1548 if (r4)
1549 *r4 = IPC_GET_ARG4(result);
[c07544d3]1550
[0cc4313]1551 if (r5)
1552 *r5 = IPC_GET_ARG5(result);
[c07544d3]1553
[0cc4313]1554 return rc;
[085bd54]1555}
[b2951e2]1556
[79ae36dd]1557void async_msg_0(async_exch_t *exch, sysarg_t imethod)
[64d2b10]1558{
[79ae36dd]1559 if (exch != NULL)
1560 ipc_call_async_0(exch->phone, imethod, NULL, NULL, true);
[64d2b10]1561}
1562
[79ae36dd]1563void async_msg_1(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1)
[64d2b10]1564{
[79ae36dd]1565 if (exch != NULL)
1566 ipc_call_async_1(exch->phone, imethod, arg1, NULL, NULL, true);
[64d2b10]1567}
1568
[79ae36dd]1569void async_msg_2(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1570 sysarg_t arg2)
[64d2b10]1571{
[79ae36dd]1572 if (exch != NULL)
1573 ipc_call_async_2(exch->phone, imethod, arg1, arg2, NULL, NULL,
1574 true);
[64d2b10]1575}
1576
[79ae36dd]1577void async_msg_3(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1578 sysarg_t arg2, sysarg_t arg3)
[64d2b10]1579{
[79ae36dd]1580 if (exch != NULL)
1581 ipc_call_async_3(exch->phone, imethod, arg1, arg2, arg3, NULL,
1582 NULL, true);
[64d2b10]1583}
1584
[79ae36dd]1585void async_msg_4(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1,
1586 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4)
[64d2b10]1587{
[79ae36dd]1588 if (exch != NULL)
1589 ipc_call_async_4(exch->phone, imethod, arg1, arg2, arg3, arg4,
1590 NULL, NULL, true);
[64d2b10]1591}
1592
[79ae36dd]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)
[64d2b10]1595{
[79ae36dd]1596 if (exch != NULL)
1597 ipc_call_async_5(exch->phone, imethod, arg1, arg2, arg3, arg4,
1598 arg5, NULL, NULL, true);
[64d2b10]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
[79ae36dd]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)
[64d2b10]1637{
[79ae36dd]1638 if (exch == NULL)
1639 return ENOENT;
1640
1641 return ipc_forward_fast(callid, exch->phone, imethod, arg1, arg2, mode);
[64d2b10]1642}
1643
[79ae36dd]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)
[64d2b10]1647{
[79ae36dd]1648 if (exch == NULL)
1649 return ENOENT;
1650
1651 return ipc_forward_slow(callid, exch->phone, imethod, arg1, arg2, arg3,
1652 arg4, arg5, mode);
[64d2b10]1653}
1654
[007e6efa]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 *
[79ae36dd]1659 * @param exch Exchange for sending the message.
[007e6efa]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 *
[79ae36dd]1665 * @return Zero on success or a negative error code.
[007e6efa]1666 *
1667 */
[79ae36dd]1668int async_connect_to_me(async_exch_t *exch, sysarg_t arg1, sysarg_t arg2,
[9934f7d]1669 sysarg_t arg3, async_client_conn_t client_receiver, void *carg)
[007e6efa]1670{
[79ae36dd]1671 if (exch == NULL)
1672 return ENOENT;
1673
[007e6efa]1674 sysarg_t phone_hash;
[ab34cc9]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);
[007e6efa]1682 if (rc != EOK)
[ab34cc9]1683 return (int) rc;
1684
1685 phone_hash = IPC_GET_ARG5(answer);
[36b16bc]1686
[007e6efa]1687 if (client_receiver != NULL)
[ab34cc9]1688 async_new_connection(answer.in_task_id, phone_hash, 0, NULL,
[9934f7d]1689 client_receiver, carg);
[007e6efa]1690
1691 return EOK;
1692}
1693
[6aae539d]1694/** Wrapper for making IPC_M_CLONE_ESTABLISH calls using the async framework.
[007e6efa]1695 *
[6aae539d]1696 * Ask for a cloned connection to some service.
[f74392f]1697 *
[79ae36dd]1698 * @param mgmt Exchange management style.
1699 * @param exch Exchange for sending the message.
[007e6efa]1700 *
[79ae36dd]1701 * @return New session on success or NULL on error.
[f74392f]1702 *
1703 */
[6aae539d]1704async_sess_t *async_clone_establish(exch_mgmt_t mgmt, async_exch_t *exch)
[79ae36dd]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
[47c9a8c]1719 amsg_t *msg = amsg_create();
1720 if (!msg) {
[79ae36dd]1721 free(sess);
1722 errno = ENOMEM;
1723 return NULL;
1724 }
1725
1726 msg->dataptr = &result;
1727 msg->wdata.active = true;
1728
[6aae539d]1729 ipc_call_async_0(exch->phone, IPC_M_CLONE_ESTABLISH, msg,
[79ae36dd]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
[58cbf8d5]1755 fibril_mutex_initialize(&sess->remote_state_mtx);
1756 sess->remote_state_data = NULL;
1757
[79ae36dd]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)
[f74392f]1767{
[79ae36dd]1768 ipc_call_t result;
1769
[47c9a8c]1770 amsg_t *msg = amsg_create();
1771 if (!msg)
[79ae36dd]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);
[f74392f]1782
[007e6efa]1783 if (rc != EOK)
[f74392f]1784 return rc;
[007e6efa]1785
[79ae36dd]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
[58cbf8d5]1831 fibril_mutex_initialize(&sess->remote_state_mtx);
1832 sess->remote_state_data = NULL;
1833
[79ae36dd]1834 list_initialize(&sess->exch_list);
1835 fibril_mutex_initialize(&sess->mutex);
1836 atomic_set(&sess->refcnt, 0);
1837
1838 return sess;
[f74392f]1839}
1840
[93ad49a8]1841/** Set arguments for new connections.
[0f4532e]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
[93ad49a8]1845 * a callback session, the framework does not know arguments for the new
1846 * connections.
[0f4532e]1847 *
1848 * The proper solution seems to be to implement parallel exchanges using
1849 * tagging.
1850 */
[93ad49a8]1851void async_sess_args_set(async_sess_t *sess, sysarg_t arg1, sysarg_t arg2,
1852 sysarg_t arg3)
[0f4532e]1853{
[93ad49a8]1854 sess->arg1 = arg1;
1855 sess->arg2 = arg2;
1856 sess->arg3 = arg3;
[0f4532e]1857}
1858
[f74392f]1859/** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework.
[007e6efa]1860 *
[f74392f]1861 * Ask through phone for a new connection to some service and block until
1862 * success.
1863 *
[79ae36dd]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.
[007e6efa]1869 *
[79ae36dd]1870 * @return New session on success or NULL on error.
[f74392f]1871 *
1872 */
[79ae36dd]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)
[f74392f]1875{
[79ae36dd]1876 if (exch == NULL) {
1877 errno = ENOENT;
1878 return NULL;
1879 }
[f74392f]1880
[79ae36dd]1881 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t));
1882 if (sess == NULL) {
1883 errno = ENOMEM;
1884 return NULL;
1885 }
[007e6efa]1886
[79ae36dd]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
[58cbf8d5]1902 fibril_mutex_initialize(&sess->remote_state_mtx);
1903 sess->remote_state_data = NULL;
1904
[79ae36dd]1905 list_initialize(&sess->exch_list);
1906 fibril_mutex_initialize(&sess->mutex);
1907 atomic_set(&sess->refcnt, 0);
1908
1909 return sess;
[f74392f]1910}
1911
[64d2b10]1912/** Connect to a task specified by id.
1913 *
1914 */
[79ae36dd]1915async_sess_t *async_connect_kbox(task_id_t id)
[64d2b10]1916{
[79ae36dd]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
[58cbf8d5]1936 fibril_mutex_initialize(&sess->remote_state_mtx);
1937 sess->remote_state_data = NULL;
1938
[79ae36dd]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);
[64d2b10]1949}
1950
1951/** Wrapper for ipc_hangup.
1952 *
[79ae36dd]1953 * @param sess Session to hung up.
[64d2b10]1954 *
1955 * @return Zero on success or a negative error code.
1956 *
1957 */
[79ae36dd]1958int async_hangup(async_sess_t *sess)
[64d2b10]1959{
[36e2b55]1960 async_exch_t *exch;
1961
[79ae36dd]1962 assert(sess);
1963
1964 if (atomic_get(&sess->refcnt) > 0)
1965 return EBUSY;
1966
[36e2b55]1967 fibril_mutex_lock(&async_sess_mutex);
[972c60ce]1968
[cff3fb6]1969 int rc = async_hangup_internal(sess->phone);
[36e2b55]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 }
[4c50c8d]1981
1982 free(sess);
[36e2b55]1983
1984 fibril_mutex_unlock(&async_sess_mutex);
1985
[79ae36dd]1986 return rc;
[64d2b10]1987}
1988
1989/** Interrupt one thread of this task from waiting for IPC. */
1990void async_poke(void)
1991{
1992 ipc_poke();
1993}
1994
[79ae36dd]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 *)
[b72efe8]2016 list_get_instance(list_first(&sess->exch_list),
2017 async_exch_t, sess_link);
2018
[79ae36dd]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) {
[b72efe8]2030 link_initialize(&exch->sess_link);
2031 link_initialize(&exch->global_link);
[79ae36dd]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) {
[b72efe8]2048 link_initialize(&exch->sess_link);
2049 link_initialize(&exch->global_link);
[79ae36dd]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 *)
[b72efe8]2061 list_get_instance(list_first(&inactive_exch_list),
2062 async_exch_t, global_link);
2063
[79ae36dd]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
[1c6436a]2103 atomic_dec(&sess->refcnt);
2104
[79ae36dd]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
[47b7006]2117/** Wrapper for IPC_M_SHARE_IN calls using the async framework.
2118 *
[79ae36dd]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.
[fbcdeb8]2123 * @param dst Destination address space area base. Cannot be NULL.
[0da4e41]2124 *
[47b7006]2125 * @return Zero on success or a negative error code from errno.h.
[0da4e41]2126 *
2127 */
[fbcdeb8]2128int async_share_in_start(async_exch_t *exch, size_t size, sysarg_t arg,
2129 unsigned int *flags, void **dst)
[0da4e41]2130{
[79ae36dd]2131 if (exch == NULL)
2132 return ENOENT;
2133
[fbcdeb8]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);
[47b7006]2138
[0da4e41]2139 if (flags)
[fbcdeb8]2140 *flags = (unsigned int) _flags;
[47b7006]2141
[fbcdeb8]2142 *dst = (void *) _dst;
[0da4e41]2143 return res;
2144}
2145
2146/** Wrapper for receiving the IPC_M_SHARE_IN calls using the async framework.
2147 *
[47b7006]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.
[0da4e41]2151 *
2152 * So far, this wrapper is to be used from within a connection fibril.
2153 *
[47b7006]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.
[0da4e41]2158 *
2159 */
[47b7006]2160bool async_share_in_receive(ipc_callid_t *callid, size_t *size)
[0da4e41]2161{
2162 assert(callid);
2163 assert(size);
[47b7006]2164
2165 ipc_call_t data;
[0da4e41]2166 *callid = async_get_call(&data);
[47b7006]2167
[228e490]2168 if (IPC_GET_IMETHOD(data) != IPC_M_SHARE_IN)
[47b7006]2169 return false;
2170
[fbcdeb8]2171 *size = (size_t) IPC_GET_ARG1(data);
[47b7006]2172 return true;
[0da4e41]2173}
2174
2175/** Wrapper for answering the IPC_M_SHARE_IN calls using the async framework.
2176 *
[fbcdeb8]2177 * This wrapper only makes it more comfortable to answer IPC_M_SHARE_IN
[47b7006]2178 * calls so that the user doesn't have to remember the meaning of each IPC
2179 * argument.
[0da4e41]2180 *
[47b7006]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.
[0da4e41]2186 *
2187 */
[47b7006]2188int async_share_in_finalize(ipc_callid_t callid, void *src, unsigned int flags)
[0da4e41]2189{
2190 return ipc_share_in_finalize(callid, src, flags);
2191}
2192
[47b7006]2193/** Wrapper for IPC_M_SHARE_OUT calls using the async framework.
[0da4e41]2194 *
[79ae36dd]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.
[47b7006]2198 *
2199 * @return Zero on success or a negative error code from errno.h.
[0da4e41]2200 *
2201 */
[79ae36dd]2202int async_share_out_start(async_exch_t *exch, void *src, unsigned int flags)
[0da4e41]2203{
[79ae36dd]2204 if (exch == NULL)
2205 return ENOENT;
2206
2207 return async_req_3_0(exch, IPC_M_SHARE_OUT, (sysarg_t) src, 0,
[96b02eb9]2208 (sysarg_t) flags);
[0da4e41]2209}
2210
2211/** Wrapper for receiving the IPC_M_SHARE_OUT calls using the async framework.
2212 *
[47b7006]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.
[0da4e41]2216 *
2217 * So far, this wrapper is to be used from within a connection fibril.
2218 *
[47b7006]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.
[0da4e41]2224 *
2225 */
[47b7006]2226bool async_share_out_receive(ipc_callid_t *callid, size_t *size, unsigned int *flags)
[0da4e41]2227{
2228 assert(callid);
2229 assert(size);
2230 assert(flags);
[47b7006]2231
2232 ipc_call_t data;
[0da4e41]2233 *callid = async_get_call(&data);
[47b7006]2234
[228e490]2235 if (IPC_GET_IMETHOD(data) != IPC_M_SHARE_OUT)
[47b7006]2236 return false;
2237
[0da4e41]2238 *size = (size_t) IPC_GET_ARG2(data);
[47b7006]2239 *flags = (unsigned int) IPC_GET_ARG3(data);
2240 return true;
[0da4e41]2241}
2242
2243/** Wrapper for answering the IPC_M_SHARE_OUT calls using the async framework.
2244 *
[47b7006]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.
[0da4e41]2248 *
[47b7006]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.
[0da4e41]2253 *
2254 */
[fbcdeb8]2255int async_share_out_finalize(ipc_callid_t callid, void **dst)
[0da4e41]2256{
2257 return ipc_share_out_finalize(callid, dst);
2258}
2259
[8bf1eeb]2260/** Start IPC_M_DATA_READ using the async framework.
2261 *
[79ae36dd]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).
[8bf1eeb]2265 * @param dataptr Storage of call data (arg 2 holds actual data size).
[79ae36dd]2266 *
[8bf1eeb]2267 * @return Hash of the sent message or 0 on error.
[79ae36dd]2268 *
[8bf1eeb]2269 */
[79ae36dd]2270aid_t async_data_read(async_exch_t *exch, void *dst, size_t size,
2271 ipc_call_t *dataptr)
[8bf1eeb]2272{
[79ae36dd]2273 return async_send_2(exch, IPC_M_DATA_READ, (sysarg_t) dst,
[8bf1eeb]2274 (sysarg_t) size, dataptr);
2275}
2276
[47b7006]2277/** Wrapper for IPC_M_DATA_READ calls using the async framework.
[0da4e41]2278 *
[79ae36dd]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.
[47b7006]2282 *
2283 * @return Zero on success or a negative error code from errno.h.
[0da4e41]2284 *
2285 */
[79ae36dd]2286int async_data_read_start(async_exch_t *exch, void *dst, size_t size)
[0da4e41]2287{
[79ae36dd]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);
[0da4e41]2293}
2294
2295/** Wrapper for receiving the IPC_M_DATA_READ calls using the async framework.
2296 *
[47b7006]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.
[0da4e41]2300 *
2301 * So far, this wrapper is to be used from within a connection fibril.
2302 *
[47b7006]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.
[0da4e41]2307 *
2308 */
[47b7006]2309bool async_data_read_receive(ipc_callid_t *callid, size_t *size)
[0da4e41]2310{
2311 assert(callid);
[47b7006]2312
2313 ipc_call_t data;
[0da4e41]2314 *callid = async_get_call(&data);
[47b7006]2315
[228e490]2316 if (IPC_GET_IMETHOD(data) != IPC_M_DATA_READ)
[47b7006]2317 return false;
2318
[0da4e41]2319 if (size)
2320 *size = (size_t) IPC_GET_ARG2(data);
[47b7006]2321
2322 return true;
[0da4e41]2323}
2324
2325/** Wrapper for answering the IPC_M_DATA_READ calls using the async framework.
2326 *
[47b7006]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.
[0da4e41]2330 *
[47b7006]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.
[0da4e41]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
[b4cbef1]2344/** Wrapper for forwarding any read request
2345 *
2346 */
[79ae36dd]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)
[b4cbef1]2350{
[79ae36dd]2351 if (exch == NULL)
2352 return ENOENT;
2353
[b4cbef1]2354 ipc_callid_t callid;
2355 if (!async_data_read_receive(&callid, NULL)) {
2356 ipc_answer_0(callid, EINVAL);
2357 return EINVAL;
2358 }
2359
[79ae36dd]2360 aid_t msg = async_send_fast(exch, imethod, arg1, arg2, arg3, arg4,
[b4cbef1]2361 dataptr);
2362 if (msg == 0) {
2363 ipc_answer_0(callid, EINVAL);
2364 return EINVAL;
2365 }
2366
[79ae36dd]2367 int retval = ipc_forward_fast(callid, exch->phone, 0, 0, 0,
[b4cbef1]2368 IPC_FF_ROUTE_FROM_ME);
2369 if (retval != EOK) {
[ab9f443]2370 async_forget(msg);
[b4cbef1]2371 ipc_answer_0(callid, retval);
2372 return retval;
2373 }
2374
[96b02eb9]2375 sysarg_t rc;
[b4cbef1]2376 async_wait_for(msg, &rc);
2377
2378 return (int) rc;
2379}
2380
[47b7006]2381/** Wrapper for IPC_M_DATA_WRITE calls using the async framework.
[0da4e41]2382 *
[79ae36dd]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.
[b4cbef1]2386 *
2387 * @return Zero on success or a negative error code from errno.h.
[0da4e41]2388 *
2389 */
[79ae36dd]2390int async_data_write_start(async_exch_t *exch, const void *src, size_t size)
[0da4e41]2391{
[79ae36dd]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);
[0da4e41]2397}
2398
2399/** Wrapper for receiving the IPC_M_DATA_WRITE calls using the async framework.
2400 *
[47b7006]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.
[0da4e41]2404 *
2405 * So far, this wrapper is to be used from within a connection fibril.
2406 *
[47b7006]2407 * @param callid Storage for the hash of the IPC_M_DATA_WRITE.
2408 * @param size Storage for the suggested size. May be NULL.
[b4cbef1]2409 *
[47b7006]2410 * @return True on success, false on failure.
[0da4e41]2411 *
2412 */
[47b7006]2413bool async_data_write_receive(ipc_callid_t *callid, size_t *size)
[0da4e41]2414{
2415 assert(callid);
[b4cbef1]2416
[47b7006]2417 ipc_call_t data;
[0da4e41]2418 *callid = async_get_call(&data);
[47b7006]2419
[228e490]2420 if (IPC_GET_IMETHOD(data) != IPC_M_DATA_WRITE)
[47b7006]2421 return false;
[b4cbef1]2422
[0da4e41]2423 if (size)
2424 *size = (size_t) IPC_GET_ARG2(data);
[b4cbef1]2425
[47b7006]2426 return true;
[0da4e41]2427}
2428
2429/** Wrapper for answering the IPC_M_DATA_WRITE calls using the async framework.
2430 *
[47b7006]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.
[0da4e41]2434 *
[b4cbef1]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.
[0da4e41]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
[eda925a]2447/** Wrapper for receiving binary data or strings
[8aa42e3]2448 *
2449 * This wrapper only makes it more comfortable to use async_data_write_*
[eda925a]2450 * functions to receive binary data or strings.
[8aa42e3]2451 *
[472c09d]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.
[eda925a]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.
[b4cbef1]2458 * @param min_size Minimum size (in bytes) of the data to receive.
[472c09d]2459 * @param max_size Maximum size (in bytes) of the data to receive. 0 means
2460 * no limit.
[eda925a]2461 * @param granulariy If non-zero then the size of the received data has to
[472c09d]2462 * be divisible by this value.
2463 * @param received If not NULL, the size of the received data is stored here.
[8aa42e3]2464 *
2465 * @return Zero on success or a value from @ref errno.h on failure.
2466 *
2467 */
[eda925a]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)
[8aa42e3]2471{
[79ae36dd]2472 assert(data);
2473
[8aa42e3]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
[b4cbef1]2481 if (size < min_size) {
2482 ipc_answer_0(callid, EINVAL);
2483 return EINVAL;
2484 }
2485
[8aa42e3]2486 if ((max_size > 0) && (size > max_size)) {
2487 ipc_answer_0(callid, EINVAL);
2488 return EINVAL;
2489 }
2490
[472c09d]2491 if ((granularity > 0) && ((size % granularity) != 0)) {
2492 ipc_answer_0(callid, EINVAL);
2493 return EINVAL;
2494 }
2495
[eda925a]2496 void *_data;
2497
2498 if (nullterm)
2499 _data = malloc(size + 1);
2500 else
2501 _data = malloc(size);
2502
[472c09d]2503 if (_data == NULL) {
[8aa42e3]2504 ipc_answer_0(callid, ENOMEM);
2505 return ENOMEM;
2506 }
2507
[472c09d]2508 int rc = async_data_write_finalize(callid, _data, size);
[8aa42e3]2509 if (rc != EOK) {
[472c09d]2510 free(_data);
[8aa42e3]2511 return rc;
2512 }
2513
[eda925a]2514 if (nullterm)
2515 ((char *) _data)[size] = 0;
[8aa42e3]2516
[eda925a]2517 *data = _data;
[472c09d]2518 if (received != NULL)
2519 *received = size;
2520
[8aa42e3]2521 return EOK;
2522}
2523
[b4cbef1]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 */
[47b7006]2531void async_data_write_void(sysarg_t retval)
[b4cbef1]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 */
[79ae36dd]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)
[b4cbef1]2544{
[79ae36dd]2545 if (exch == NULL)
2546 return ENOENT;
2547
[b4cbef1]2548 ipc_callid_t callid;
2549 if (!async_data_write_receive(&callid, NULL)) {
2550 ipc_answer_0(callid, EINVAL);
2551 return EINVAL;
2552 }
2553
[79ae36dd]2554 aid_t msg = async_send_fast(exch, imethod, arg1, arg2, arg3, arg4,
[b4cbef1]2555 dataptr);
2556 if (msg == 0) {
2557 ipc_answer_0(callid, EINVAL);
2558 return EINVAL;
2559 }
2560
[79ae36dd]2561 int retval = ipc_forward_fast(callid, exch->phone, 0, 0, 0,
[b4cbef1]2562 IPC_FF_ROUTE_FROM_ME);
2563 if (retval != EOK) {
[ab9f443]2564 async_forget(msg);
[b4cbef1]2565 ipc_answer_0(callid, retval);
2566 return retval;
2567 }
2568
[96b02eb9]2569 sysarg_t rc;
[b4cbef1]2570 async_wait_for(msg, &rc);
2571
2572 return (int) rc;
2573}
2574
[79ae36dd]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
[58cbf8d5]2621 fibril_mutex_initialize(&sess->remote_state_mtx);
2622 sess->remote_state_data = NULL;
2623
[79ae36dd]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 *
[8869f7b]2641 * @return New async session.
2642 * @return NULL on failure.
[79ae36dd]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
[58cbf8d5]2670 fibril_mutex_initialize(&sess->remote_state_mtx);
2671 sess->remote_state_data = NULL;
2672
[79ae36dd]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
[8869f7b]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
[58cbf8d5]2715 fibril_mutex_initialize(&sess->remote_state_mtx);
2716 sess->remote_state_data = NULL;
2717
[8869f7b]2718 list_initialize(&sess->exch_list);
2719 fibril_mutex_initialize(&sess->mutex);
2720 atomic_set(&sess->refcnt, 0);
2721
2722 return sess;
2723}
2724
[2c4aa39]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
[58cbf8d5]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
[a46da63]2828/** @}
[b2951e2]2829 */
Note: See TracBrowser for help on using the repository browser.