source: mainline/libc/generic/async.c@ 51dbadf3

lfn serial ticket/834-toolchain-update topic/msim-upgrade topic/simplify-dev-export
Last change on this file since 51dbadf3 was 51dbadf3, checked in by Ondrej Palkovsky <ondrap@…>, 19 years ago

Added basic klog.
Added ipc tester.
TODO add serializing functions to psthread, so that the lines in klog won't intermix.

  • Property mode set to 100644
File size: 19.0 KB
Line 
1/*
2 * Copyright (C) 2006 Ondrej Palkovsky
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
7 * are met:
8 *
9 * - Redistributions of source code must retain the above copyright
10 * notice, this list of conditions and the following disclaimer.
11 * - Redistributions in binary form must reproduce the above copyright
12 * notice, this list of conditions and the following disclaimer in the
13 * documentation and/or other materials provided with the distribution.
14 * - The name of the author may not be used to endorse or promote products
15 * derived from this software without specific prior written permission.
16 *
17 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
18 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
19 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
20 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
21 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
22 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
26 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27 */
28
29/**
30 * Asynchronous library
31 *
32 * The aim of this library is facilitating writing programs utilizing
33 * the asynchronous nature of Helenos IPC, yet using a normal way
34 * of programming.
35 *
36 * You should be able to write very simple multithreaded programs,
37 * the async framework will automatically take care of most synchronization
38 * problems.
39 *
40 * Default semantics:
41 * - send() - send asynchronously. If the kernel refuses to send more
42 * messages, [ try to get responses from kernel, if nothing
43 * found, might try synchronous ]
44 *
45 * Example of use:
46 *
47 * 1) Multithreaded client application
48 * create_thread(thread1);
49 * create_thread(thread2);
50 * ...
51 *
52 * thread1() {
53 * conn = ipc_connect_me_to();
54 * c1 = send(conn);
55 * c2 = send(conn);
56 * wait_for(c1);
57 * wait_for(c2);
58 * }
59 *
60 *
61 * 2) Multithreaded server application
62 * main() {
63 * async_manager();
64 * }
65 *
66 *
67 * client_connection(icallid, *icall) {
68 * if (want_refuse) {
69 * ipc_answer_fast(icallid, ELIMIT, 0, 0);
70 * return;
71 * }
72 * ipc_answer_fast(icallid, 0, 0, 0);
73 *
74 * callid = async_get_call(&call);
75 * handle(callid, call);
76 * ipc_answer_fast(callid, 1,2,3);
77 *
78 * callid = async_get_call(&call);
79 * ....
80 * }
81 *
82 * TODO: Detaching/joining dead psthreads?
83 */
84#include <futex.h>
85#include <async.h>
86#include <psthread.h>
87#include <stdio.h>
88#include <libadt/hash_table.h>
89#include <libadt/list.h>
90#include <ipc/ipc.h>
91#include <assert.h>
92#include <errno.h>
93#include <time.h>
94#include <arch/barrier.h>
95
96atomic_t async_futex = FUTEX_INITIALIZER;
97static hash_table_t conn_hash_table;
98static LIST_INITIALIZE(timeout_list);
99
100typedef struct {
101 struct timeval expires; /**< Expiration time for waiting thread */
102 int inlist; /**< If true, this struct is in timeout list */
103 link_t link;
104
105 pstid_t ptid; /**< Thread waiting for this message */
106 int active; /**< If this thread is currently active */
107 int timedout; /**< If true, we timed out */
108} awaiter_t;
109
110typedef struct {
111 awaiter_t wdata;
112
113 int done; /**< If reply was received */
114 ipc_call_t *dataptr; /**< Pointer where the answer data
115 * is stored */
116 ipcarg_t retval;
117} amsg_t;
118
119typedef struct {
120 link_t link;
121 ipc_callid_t callid;
122 ipc_call_t call;
123} msg_t;
124
125typedef struct {
126 awaiter_t wdata;
127
128 link_t link; /**< Hash table link */
129 ipcarg_t in_phone_hash; /**< Incoming phone hash. */
130 link_t msg_queue; /**< Messages that should be delivered to this thread */
131 /* Structures for connection opening packet */
132 ipc_callid_t callid;
133 ipc_call_t call;
134 void (*cthread)(ipc_callid_t,ipc_call_t *);
135} connection_t;
136
137
138__thread connection_t *PS_connection;
139
140static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);
141static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call);
142static async_client_conn_t client_connection = default_client_connection;
143static async_client_conn_t interrupt_received = default_interrupt_received;
144
145/** Add microseconds to give timeval */
146static void tv_add(struct timeval *tv, suseconds_t usecs)
147{
148 tv->tv_sec += usecs / 1000000;
149 tv->tv_usec += usecs % 1000000;
150 if (tv->tv_usec > 1000000) {
151 tv->tv_sec++;
152 tv->tv_usec -= 1000000;
153 }
154}
155
156/** Subtract 2 timevals, return microseconds difference */
157static suseconds_t tv_sub(struct timeval *tv1, struct timeval *tv2)
158{
159 suseconds_t result;
160
161 result = tv1->tv_usec - tv2->tv_usec;
162 result += (tv1->tv_sec - tv2->tv_sec) * 1000000;
163
164 return result;
165}
166
167/** Compare timeval
168 *
169 * @return 1 if tv1 > tv2, otherwise 0
170 */
171static int tv_gt(struct timeval *tv1, struct timeval *tv2)
172{
173 if (tv1->tv_sec > tv2->tv_sec)
174 return 1;
175 if (tv1->tv_sec == tv2->tv_sec && tv1->tv_usec > tv2->tv_usec)
176 return 1;
177 return 0;
178}
179static int tv_gteq(struct timeval *tv1, struct timeval *tv2)
180{
181 if (tv1->tv_sec > tv2->tv_sec)
182 return 1;
183 if (tv1->tv_sec == tv2->tv_sec && tv1->tv_usec >= tv2->tv_usec)
184 return 1;
185 return 0;
186}
187
188/* Hash table functions */
189#define CONN_HASH_TABLE_CHAINS 32
190
191static hash_index_t conn_hash(unsigned long *key)
192{
193 assert(key);
194 return ((*key) >> 4) % CONN_HASH_TABLE_CHAINS;
195}
196
197static int conn_compare(unsigned long key[], hash_count_t keys, link_t *item)
198{
199 connection_t *hs;
200
201 hs = hash_table_get_instance(item, connection_t, link);
202
203 return key[0] == hs->in_phone_hash;
204}
205
206static void conn_remove(link_t *item)
207{
208 free(hash_table_get_instance(item, connection_t, link));
209}
210
211
212/** Operations for NS hash table. */
213static hash_table_operations_t conn_hash_table_ops = {
214 .hash = conn_hash,
215 .compare = conn_compare,
216 .remove_callback = conn_remove
217};
218
219/** Insert sort timeout msg into timeouts list
220 *
221 */
222static void insert_timeout(awaiter_t *wd)
223{
224 link_t *tmp;
225 awaiter_t *cur;
226
227 wd->timedout = 0;
228
229 tmp = timeout_list.next;
230 while (tmp != &timeout_list) {
231 cur = list_get_instance(tmp, awaiter_t, link);
232 if (tv_gteq(&cur->expires, &wd->expires))
233 break;
234 tmp = tmp->next;
235 }
236 list_append(&wd->link, tmp);
237}
238
239/*************************************************/
240
241/** Try to route a call to an appropriate connection thread
242 *
243 */
244static int route_call(ipc_callid_t callid, ipc_call_t *call)
245{
246 connection_t *conn;
247 msg_t *msg;
248 link_t *hlp;
249 unsigned long key;
250
251 futex_down(&async_futex);
252
253 key = call->in_phone_hash;
254 hlp = hash_table_find(&conn_hash_table, &key);
255 if (!hlp) {
256 futex_up(&async_futex);
257 return 0;
258 }
259 conn = hash_table_get_instance(hlp, connection_t, link);
260
261 msg = malloc(sizeof(*msg));
262 msg->callid = callid;
263 msg->call = *call;
264 list_append(&msg->link, &conn->msg_queue);
265
266 /* If the call is waiting for event, run it */
267 if (!conn->wdata.active) {
268 /* If in timeout list, remove it */
269 if (conn->wdata.inlist) {
270 conn->wdata.inlist = 0;
271 list_remove(&conn->wdata.link);
272 }
273 conn->wdata.active = 1;
274 psthread_add_ready(conn->wdata.ptid);
275 }
276
277 futex_up(&async_futex);
278
279 return 1;
280}
281
282/** Return new incoming message for current(thread-local) connection */
283ipc_callid_t async_get_call_timeout(ipc_call_t *call, suseconds_t usecs)
284{
285 msg_t *msg;
286 ipc_callid_t callid;
287 connection_t *conn;
288
289 assert(PS_connection);
290 /* GCC 4.1.0 coughs on PS_connection-> dereference,
291 * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
292 * I would never expect to find so many errors in
293 * compiler *($&$(*&$
294 */
295 conn = PS_connection;
296
297 if (usecs < 0) /* TODO: let it get through the ipc_call once */
298 return 0;
299
300 futex_down(&async_futex);
301
302 if (usecs) {
303 gettimeofday(&conn->wdata.expires, NULL);
304 tv_add(&conn->wdata.expires, usecs);
305 } else {
306 conn->wdata.inlist = 0;
307 }
308 /* If nothing in queue, wait until something appears */
309 while (list_empty(&conn->msg_queue)) {
310 if (usecs) {
311 conn->wdata.inlist = 1;
312 insert_timeout(&conn->wdata);
313 }
314 conn->wdata.active = 0;
315 psthread_schedule_next_adv(PS_TO_MANAGER);
316 /* Futex is up after getting back from async_manager
317 * get it again */
318 futex_down(&async_futex);
319 if (usecs && conn->wdata.timedout && \
320 list_empty(&conn->msg_queue)) {
321 /* If we timed out-> exit */
322 futex_up(&async_futex);
323 return 0;
324 }
325 }
326
327 msg = list_get_instance(conn->msg_queue.next, msg_t, link);
328 list_remove(&msg->link);
329 callid = msg->callid;
330 *call = msg->call;
331 free(msg);
332
333 futex_up(&async_futex);
334 return callid;
335}
336
337/** Thread function that gets created on new connection
338 *
339 * This function is defined as a weak symbol - to be redefined in
340 * user code.
341 */
342static void default_client_connection(ipc_callid_t callid, ipc_call_t *call)
343{
344 ipc_answer_fast(callid, ENOENT, 0, 0);
345}
346static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call)
347{
348}
349
350/** Wrapper for client connection thread
351 *
352 * When new connection arrives, thread with this function is created.
353 * It calls client_connection and does final cleanup.
354 *
355 * @parameter arg Connection structure pointer
356 */
357static int connection_thread(void *arg)
358{
359 unsigned long key;
360 msg_t *msg;
361
362 /* Setup thread local connection pointer */
363 PS_connection = (connection_t *)arg;
364 PS_connection->cthread(PS_connection->callid, &PS_connection->call);
365
366 /* Remove myself from connection hash table */
367 futex_down(&async_futex);
368 key = PS_connection->in_phone_hash;
369 hash_table_remove(&conn_hash_table, &key, 1);
370 futex_up(&async_futex);
371 /* Answer all remaining messages with ehangup */
372 while (!list_empty(&PS_connection->msg_queue)) {
373 msg = list_get_instance(PS_connection->msg_queue.next, msg_t, link);
374 list_remove(&msg->link);
375 ipc_answer_fast(msg->callid, EHANGUP, 0, 0);
376 free(msg);
377 }
378}
379
380/** Create new thread for a new connection
381 *
382 * Creates new thread for connection, fills in connection
383 * structures and inserts it into the hash table, so that
384 * later we can easily do routing of messages to particular
385 * threads.
386 *
387 * @param in_phone_hash Identification of the incoming connection
388 * @param callid Callid of the IPC_M_CONNECT_ME_TO packet
389 * @param call Call data of the opening packet
390 * @param cthread Thread function that should be called upon
391 * opening the connection
392 * @return New thread id
393 */
394pstid_t async_new_connection(ipcarg_t in_phone_hash,ipc_callid_t callid,
395 ipc_call_t *call,
396 void (*cthread)(ipc_callid_t,ipc_call_t *))
397{
398 pstid_t ptid;
399 connection_t *conn;
400 unsigned long key;
401
402 conn = malloc(sizeof(*conn));
403 if (!conn) {
404 ipc_answer_fast(callid, ENOMEM, 0, 0);
405 return NULL;
406 }
407 conn->in_phone_hash = in_phone_hash;
408 list_initialize(&conn->msg_queue);
409 conn->callid = callid;
410 if (call)
411 conn->call = *call;
412 conn->wdata.active = 1; /* We will activate it asap */
413 conn->cthread = cthread;
414
415 conn->wdata.ptid = psthread_create(connection_thread, conn);
416 if (!conn->wdata.ptid) {
417 free(conn);
418 ipc_answer_fast(callid, ENOMEM, 0, 0);
419 return NULL;
420 }
421 /* Add connection to hash table */
422 key = conn->in_phone_hash;
423 futex_down(&async_futex);
424 hash_table_insert(&conn_hash_table, &key, &conn->link);
425 futex_up(&async_futex);
426
427 psthread_add_ready(conn->wdata.ptid);
428
429 return conn->wdata.ptid;
430}
431
432/** Handle call that was received */
433static void handle_call(ipc_callid_t callid, ipc_call_t *call)
434{
435 /* Unrouted call - do some default behaviour */
436 switch (IPC_GET_METHOD(*call)) {
437 case IPC_M_INTERRUPT:
438 (*interrupt_received)(callid,call);
439 return;
440 case IPC_M_CONNECT_ME_TO:
441 /* Open new connection with thread etc. */
442 async_new_connection(IPC_GET_ARG3(*call), callid, call, client_connection);
443 return;
444 }
445
446 /* Try to route call through connection tables */
447 if (route_call(callid, call))
448 return;
449
450 /* Unknown call from unknown phone - hang it up */
451 ipc_answer_fast(callid, EHANGUP, 0, 0);
452}
453
454/** Fire all timeouts that expired
455 *
456 */
457static void handle_expired_timeouts(void)
458{
459 struct timeval tv;
460 awaiter_t *waiter;
461 link_t *cur;
462
463 gettimeofday(&tv,NULL);
464 futex_down(&async_futex);
465
466 cur = timeout_list.next;
467 while (cur != &timeout_list) {
468 waiter = list_get_instance(cur,awaiter_t,link);
469 if (tv_gt(&waiter->expires, &tv))
470 break;
471 cur = cur->next;
472 list_remove(&waiter->link);
473 waiter->inlist = 0;
474 waiter->timedout = 1;
475 /* Redundant condition? The thread should not
476 * be active when it gets here.
477 */
478 if (!waiter->active) {
479 waiter->active = 1;
480 psthread_add_ready(waiter->ptid);
481 }
482 }
483
484 futex_up(&async_futex);
485}
486
487/** Endless loop dispatching incoming calls and answers */
488int async_manager(void)
489{
490 ipc_call_t call;
491 ipc_callid_t callid;
492 int timeout;
493 awaiter_t *waiter;
494 struct timeval tv;
495
496 while (1) {
497 if (psthread_schedule_next_adv(PS_FROM_MANAGER)) {
498 futex_up(&async_futex); /* async_futex is always held
499 * when entering manager thread
500 */
501 continue;
502 }
503 futex_down(&async_futex);
504 if (!list_empty(&timeout_list)) {
505 waiter = list_get_instance(timeout_list.next,awaiter_t,link);
506 gettimeofday(&tv,NULL);
507 if (tv_gteq(&tv, &waiter->expires)) {
508 futex_up(&async_futex);
509 handle_expired_timeouts();
510 continue;
511 } else
512 timeout = tv_sub(&waiter->expires, &tv);
513 } else
514 timeout = SYNCH_NO_TIMEOUT;
515 futex_up(&async_futex);
516
517 callid = ipc_wait_cycle(&call, timeout, SYNCH_FLAGS_NONE);
518
519 if (!callid) {
520 handle_expired_timeouts();
521 continue;
522 }
523
524 if (callid & IPC_CALLID_ANSWERED)
525 continue;
526
527 handle_call(callid, &call);
528 }
529}
530
531/** Function to start async_manager as a standalone thread
532 *
533 * When more kernel threads are used, one async manager should
534 * exist per thread. The particular implementation may change,
535 * currently one async_manager is started automatically per kernel
536 * thread except main thread.
537 */
538static int async_manager_thread(void *arg)
539{
540 futex_up(&async_futex); /* async_futex is always locked when entering
541 * manager */
542 async_manager();
543}
544
545/** Add one manager to manager list */
546void async_create_manager(void)
547{
548 pstid_t ptid;
549
550 ptid = psthread_create(async_manager_thread, NULL);
551 psthread_add_manager(ptid);
552}
553
554/** Remove one manager from manager list */
555void async_destroy_manager(void)
556{
557 psthread_remove_manager();
558}
559
560/** Initialize internal structures needed for async manager */
561int _async_init(void)
562{
563 if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1, &conn_hash_table_ops)) {
564 printf("%s: cannot create hash table\n", "async");
565 return ENOMEM;
566 }
567
568}
569
570/** IPC handler for messages in async framework
571 *
572 * Notify thread that is waiting for this message, that it arrived
573 */
574static void reply_received(void *private, int retval,
575 ipc_call_t *data)
576{
577 amsg_t *msg = (amsg_t *) private;
578
579 msg->retval = retval;
580
581 futex_down(&async_futex);
582 /* Copy data after futex_down, just in case the
583 * call was detached
584 */
585 if (msg->dataptr)
586 *msg->dataptr = *data;
587
588 write_barrier();
589 /* Remove message from timeout list */
590 if (msg->wdata.inlist)
591 list_remove(&msg->wdata.link);
592 msg->done = 1;
593 if (! msg->wdata.active) {
594 msg->wdata.active = 1;
595 psthread_add_ready(msg->wdata.ptid);
596 }
597 futex_up(&async_futex);
598}
599
600/** Send message and return id of the sent message
601 *
602 * The return value can be used as input for async_wait() to wait
603 * for completion.
604 */
605aid_t async_send_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
606 ipc_call_t *dataptr)
607{
608 amsg_t *msg;
609
610 msg = malloc(sizeof(*msg));
611 msg->done = 0;
612 msg->dataptr = dataptr;
613
614 msg->wdata.active = 1; /* We may sleep in next method, but it
615 * will use it's own mechanism */
616 ipc_call_async_2(phoneid,method,arg1,arg2,msg,reply_received,1);
617
618 return (aid_t) msg;
619}
620
621/** Send message and return id of the sent message
622 *
623 * The return value can be used as input for async_wait() to wait
624 * for completion.
625 */
626aid_t async_send_3(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
627 ipcarg_t arg3, ipc_call_t *dataptr)
628{
629 amsg_t *msg;
630
631 msg = malloc(sizeof(*msg));
632 msg->done = 0;
633 msg->dataptr = dataptr;
634
635 msg->wdata.active = 1; /* We may sleep in next method, but it
636 * will use it's own mechanism */
637 ipc_call_async_3(phoneid,method,arg1,arg2,arg3, msg,reply_received,1);
638
639 return (aid_t) msg;
640}
641
642/** Wait for a message sent by async framework
643 *
644 * @param amsgid Message ID to wait for
645 * @param retval Pointer to variable where will be stored retval
646 * of the answered message. If NULL, it is ignored.
647 *
648 */
649void async_wait_for(aid_t amsgid, ipcarg_t *retval)
650{
651 amsg_t *msg = (amsg_t *) amsgid;
652 connection_t *conn;
653
654 futex_down(&async_futex);
655 if (msg->done) {
656 futex_up(&async_futex);
657 goto done;
658 }
659
660 msg->wdata.ptid = psthread_get_id();
661 msg->wdata.active = 0;
662 msg->wdata.inlist = 0;
663 /* Leave locked async_futex when entering this function */
664 psthread_schedule_next_adv(PS_TO_MANAGER);
665 /* futex is up automatically after psthread_schedule_next...*/
666done:
667 if (retval)
668 *retval = msg->retval;
669 free(msg);
670}
671
672/** Wait for a message sent by async framework with timeout
673 *
674 * @param amsgid Message ID to wait for
675 * @param retval Pointer to variable where will be stored retval
676 * of the answered message. If NULL, it is ignored.
677 * @param timeout Timeout in usecs
678 * @return 0 on success, ETIMEOUT if timeout expired
679 *
680 */
681int async_wait_timeout(aid_t amsgid, ipcarg_t *retval, suseconds_t timeout)
682{
683 amsg_t *msg = (amsg_t *) amsgid;
684 connection_t *conn;
685
686 /* TODO: Let it go through the event read at least once */
687 if (timeout < 0)
688 return ETIMEOUT;
689
690 futex_down(&async_futex);
691 if (msg->done) {
692 futex_up(&async_futex);
693 goto done;
694 }
695
696 gettimeofday(&msg->wdata.expires, NULL);
697 tv_add(&msg->wdata.expires, timeout);
698
699 msg->wdata.ptid = psthread_get_id();
700 msg->wdata.active = 0;
701 msg->wdata.inlist = 1;
702
703 insert_timeout(&msg->wdata);
704
705 /* Leave locked async_futex when entering this function */
706 psthread_schedule_next_adv(PS_TO_MANAGER);
707 /* futex is up automatically after psthread_schedule_next...*/
708
709 if (!msg->done)
710 return ETIMEOUT;
711
712done:
713 if (retval)
714 *retval = msg->retval;
715 free(msg);
716
717 return 0;
718}
719
720/** Wait specified time, but in the meantime handle incoming events
721 *
722 * @param timeout Time in microseconds to wait
723 */
724void async_usleep(suseconds_t timeout)
725{
726 amsg_t *msg;
727
728 msg = malloc(sizeof(*msg));
729 if (!msg)
730 return;
731
732 msg->wdata.ptid = psthread_get_id();
733 msg->wdata.inlist = 1;
734 msg->wdata.active = 0;
735
736 gettimeofday(&msg->wdata.expires, NULL);
737 tv_add(&msg->wdata.expires, timeout);
738
739 futex_down(&async_futex);
740 insert_timeout(&msg->wdata);
741 /* Leave locked async_futex when entering this function */
742 psthread_schedule_next_adv(PS_TO_MANAGER);
743 /* futex is up automatically after psthread_schedule_next...*/
744 free(msg);
745}
746
747/** Set function that is called, IPC_M_CONNECT_ME_TO is received
748 *
749 * @param conn Function that will form new psthread.
750 */
751void async_set_client_connection(async_client_conn_t conn)
752{
753 client_connection = conn;
754}
755void async_set_interrupt_received(async_client_conn_t conn)
756{
757 interrupt_received = conn;
758}
Note: See TracBrowser for help on using the repository browser.