Changes in uspace/lib/c/generic/async.c [9c31643:58cbf8d5] in mainline
- File:
-
- 1 edited
-
uspace/lib/c/generic/async.c (modified) (75 diffs)
Legend:
- Unmodified
- Added
- Removed
-
uspace/lib/c/generic/async.c
r9c31643 r58cbf8d5 40 40 * programming. 41 41 * 42 * You should be able to write very simple multithreaded programs, the async 43 * framework will automatically take care of most synchronization problems. 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. 44 45 * 45 46 * Example of use (pseudo C): … … 53 54 * int fibril1(void *arg) 54 55 * { 55 * conn = async_connect_me_to(); 56 * c1 = async_send(conn); 57 * c2 = async_send(conn); 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 * 58 66 * async_wait_for(c1); 59 67 * async_wait_for(c2); … … 90 98 #include <ipc/ipc.h> 91 99 #include <async.h> 100 #include "private/async.h" 92 101 #undef LIBC_ASYNC_C_ 93 102 94 103 #include <futex.h> 95 104 #include <fibril.h> 96 #include <stdio.h>97 105 #include <adt/hash_table.h> 98 106 #include <adt/list.h> … … 100 108 #include <errno.h> 101 109 #include <sys/time.h> 102 #include < arch/barrier.h>110 #include <libarch/barrier.h> 103 111 #include <bool.h> 104 #include "private/async.h" 105 112 #include <malloc.h> 113 #include <mem.h> 114 #include <stdlib.h> 115 #include <macros.h> 116 117 #define CLIENT_HASH_TABLE_BUCKETS 32 118 #define CONN_HASH_TABLE_BUCKETS 32 119 120 /** Async framework global futex */ 106 121 atomic_t async_futex = FUTEX_INITIALIZER; 107 122 … … 109 124 atomic_t threads_in_ipc_wait = { 0 }; 110 125 111 typedef struct { 112 awaiter_t wdata; 113 114 /** If reply was received. */ 115 bool done; 116 117 /** Pointer to where the answer data is stored. */ 118 ipc_call_t *dataptr; 119 120 sysarg_t retval; 121 } amsg_t; 122 123 /** 124 * Structures of this type are used to group information about 125 * a call and about a message queue link. 126 */ 126 /** Naming service session */ 127 async_sess_t *session_ns; 128 129 /** Call data */ 127 130 typedef struct { 128 131 link_t link; 132 129 133 ipc_callid_t callid; 130 134 ipc_call_t call; 131 135 } msg_t; 132 136 137 /* Client connection data */ 133 138 typedef struct { 134 sysarg_t in_task_hash;135 139 link_t link; 136 int refcnt; 140 141 task_id_t in_task_id; 142 atomic_t refcnt; 137 143 void *data; 138 144 } client_t; 139 145 146 /* Server connection data */ 140 147 typedef struct { 141 148 awaiter_t wdata; … … 144 151 link_t link; 145 152 146 /** Incoming client task hash. */ 147 sysarg_t in_task_hash; 153 /** Incoming client task ID. */ 154 task_id_t in_task_id; 155 148 156 /** Incoming phone hash. */ 149 157 sysarg_t in_phone_hash; … … 153 161 154 162 /** Messages that should be delivered to this fibril. */ 155 li nk_t msg_queue;163 list_t msg_queue; 156 164 157 165 /** Identification of the opening call. */ … … 159 167 /** Call data of the opening call. */ 160 168 ipc_call_t call; 169 /** Local argument or NULL if none. */ 170 void *carg; 161 171 162 172 /** Identification of the closing call. */ … … 164 174 165 175 /** Fibril function that will be used to handle the connection. */ 166 void (*cfibril)(ipc_callid_t, ipc_call_t *);176 async_client_conn_t cfibril; 167 177 } connection_t; 168 178 169 179 /** Identifier of the incoming connection handled by the current fibril. */ 170 static fibril_local connection_t * FIBRIL_connection;180 static fibril_local connection_t *fibril_connection; 171 181 172 182 static void *default_client_data_constructor(void) … … 194 204 } 195 205 196 void *async_client_data_get(void)197 {198 assert(FIBRIL_connection);199 return FIBRIL_connection->client->data;200 }201 202 206 /** Default fibril function that gets called to handle new connection. 203 207 * … … 206 210 * @param callid Hash of the incoming call. 207 211 * @param call Data of the incoming call. 208 * 209 */ 210 static void default_client_connection(ipc_callid_t callid, ipc_call_t *call) 212 * @param arg Local argument 213 * 214 */ 215 static void default_client_connection(ipc_callid_t callid, ipc_call_t *call, 216 void *arg) 211 217 { 212 218 ipc_answer_0(callid, ENOENT); 213 219 } 214 215 /**216 * Pointer to a fibril function that will be used to handle connections.217 */218 static async_client_conn_t client_connection = default_client_connection;219 220 220 221 /** Default fibril function that gets called to handle interrupt notifications. … … 224 225 * @param callid Hash of the incoming call. 225 226 * @param call Data of the incoming call. 227 * @param arg Local argument. 226 228 * 227 229 */ … … 230 232 } 231 233 232 /** 233 * Pointer to a fibril function that will be used to handle interrupt 234 * notifications. 235 */ 236 static async_client_conn_t interrupt_received = default_interrupt_received; 234 static async_client_conn_t client_connection = default_client_connection; 235 static async_interrupt_handler_t interrupt_received = default_interrupt_received; 236 237 /** Setter for client_connection function pointer. 238 * 239 * @param conn Function that will implement a new connection fibril. 240 * 241 */ 242 void async_set_client_connection(async_client_conn_t conn) 243 { 244 client_connection = conn; 245 } 246 247 /** Setter for interrupt_received function pointer. 248 * 249 * @param intr Function that will implement a new interrupt 250 * notification fibril. 251 */ 252 void async_set_interrupt_received(async_interrupt_handler_t intr) 253 { 254 interrupt_received = intr; 255 } 256 257 /** Mutex protecting inactive_exch_list and avail_phone_cv. 258 * 259 */ 260 static FIBRIL_MUTEX_INITIALIZE(async_sess_mutex); 261 262 /** List of all currently inactive exchanges. 263 * 264 */ 265 static LIST_INITIALIZE(inactive_exch_list); 266 267 /** Condition variable to wait for a phone to become available. 268 * 269 */ 270 static FIBRIL_CONDVAR_INITIALIZE(avail_phone_cv); 237 271 238 272 static hash_table_t client_hash_table; … … 240 274 static LIST_INITIALIZE(timeout_list); 241 275 242 #define CLIENT_HASH_TABLE_BUCKETS 32243 #define CONN_HASH_TABLE_BUCKETS 32244 245 276 static hash_index_t client_hash(unsigned long key[]) 246 277 { 247 278 assert(key); 279 248 280 return (((key[0]) >> 4) % CLIENT_HASH_TABLE_BUCKETS); 249 281 } … … 251 283 static int client_compare(unsigned long key[], hash_count_t keys, link_t *item) 252 284 { 285 assert(key); 286 assert(keys == 2); 287 assert(item); 288 253 289 client_t *client = hash_table_get_instance(item, client_t, link); 254 return (key[0] == client->in_task_hash); 290 return (key[0] == LOWER32(client->in_task_id) && 291 (key[1] == UPPER32(client->in_task_id))); 255 292 } 256 293 … … 276 313 { 277 314 assert(key); 315 278 316 return (((key[0]) >> 4) % CONN_HASH_TABLE_BUCKETS); 279 317 } … … 290 328 static int conn_compare(unsigned long key[], hash_count_t keys, link_t *item) 291 329 { 330 assert(key); 331 assert(item); 332 292 333 connection_t *conn = hash_table_get_instance(item, connection_t, link); 293 334 return (key[0] == conn->in_phone_hash); … … 312 353 void async_insert_timeout(awaiter_t *wd) 313 354 { 355 assert(wd); 356 314 357 wd->to_event.occurred = false; 315 358 wd->to_event.inlist = true; 316 359 317 link_t *tmp = timeout_list. next;318 while (tmp != &timeout_list ) {360 link_t *tmp = timeout_list.head.next; 361 while (tmp != &timeout_list.head) { 319 362 awaiter_t *cur 320 363 = list_get_instance(tmp, awaiter_t, to_event.link); … … 326 369 } 327 370 328 list_ append(&wd->to_event.link, tmp);371 list_insert_before(&wd->to_event.link, tmp); 329 372 } 330 373 … … 344 387 static bool route_call(ipc_callid_t callid, ipc_call_t *call) 345 388 { 389 assert(call); 390 346 391 futex_down(&async_futex); 347 392 … … 398 443 static int notification_fibril(void *arg) 399 444 { 445 assert(arg); 446 400 447 msg_t *msg = (msg_t *) arg; 401 448 interrupt_received(msg->callid, &msg->call); … … 418 465 static bool process_notification(ipc_callid_t callid, ipc_call_t *call) 419 466 { 467 assert(call); 468 420 469 futex_down(&async_futex); 421 470 … … 430 479 431 480 fid_t fid = fibril_create(notification_fibril, msg); 481 if (fid == 0) { 482 free(msg); 483 futex_up(&async_futex); 484 return false; 485 } 486 432 487 fibril_add_ready(fid); 433 488 … … 450 505 ipc_callid_t async_get_call_timeout(ipc_call_t *call, suseconds_t usecs) 451 506 { 452 assert(FIBRIL_connection); 507 assert(call); 508 assert(fibril_connection); 453 509 454 510 /* Why doing this? 455 * GCC 4.1.0 coughs on FIBRIL_connection-> dereference.511 * GCC 4.1.0 coughs on fibril_connection-> dereference. 456 512 * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot. 457 513 * I would never expect to find so many errors in 458 514 * a compiler. 459 515 */ 460 connection_t *conn = FIBRIL_connection;516 connection_t *conn = fibril_connection; 461 517 462 518 futex_down(&async_futex); … … 510 566 } 511 567 512 msg_t *msg = list_get_instance( conn->msg_queue.next, msg_t, link);568 msg_t *msg = list_get_instance(list_first(&conn->msg_queue), msg_t, link); 513 569 list_remove(&msg->link); 514 570 … … 521 577 } 522 578 579 static client_t *async_client_get(task_id_t client_id, bool create) 580 { 581 unsigned long key[2] = { 582 LOWER32(client_id), 583 UPPER32(client_id), 584 }; 585 client_t *client = NULL; 586 587 futex_down(&async_futex); 588 link_t *lnk = hash_table_find(&client_hash_table, key); 589 if (lnk) { 590 client = hash_table_get_instance(lnk, client_t, link); 591 atomic_inc(&client->refcnt); 592 } else if (create) { 593 client = malloc(sizeof(client_t)); 594 if (client) { 595 client->in_task_id = client_id; 596 client->data = async_client_data_create(); 597 598 atomic_set(&client->refcnt, 1); 599 hash_table_insert(&client_hash_table, key, &client->link); 600 } 601 } 602 603 futex_up(&async_futex); 604 return client; 605 } 606 607 static void async_client_put(client_t *client) 608 { 609 bool destroy; 610 unsigned long key[2] = { 611 LOWER32(client->in_task_id), 612 UPPER32(client->in_task_id) 613 }; 614 615 futex_down(&async_futex); 616 617 if (atomic_predec(&client->refcnt) == 0) { 618 hash_table_remove(&client_hash_table, key, 2); 619 destroy = true; 620 } else 621 destroy = false; 622 623 futex_up(&async_futex); 624 625 if (destroy) { 626 if (client->data) 627 async_client_data_destroy(client->data); 628 629 free(client); 630 } 631 } 632 633 void *async_get_client_data(void) 634 { 635 assert(fibril_connection); 636 return fibril_connection->client->data; 637 } 638 639 void *async_get_client_data_by_id(task_id_t client_id) 640 { 641 client_t *client = async_client_get(client_id, false); 642 if (!client) 643 return NULL; 644 if (!client->data) { 645 async_client_put(client); 646 return NULL; 647 } 648 649 return client->data; 650 } 651 652 void async_put_client_data_by_id(task_id_t client_id) 653 { 654 client_t *client = async_client_get(client_id, false); 655 656 assert(client); 657 assert(client->data); 658 659 /* Drop the reference we got in async_get_client_data_by_hash(). */ 660 async_client_put(client); 661 662 /* Drop our own reference we got at the beginning of this function. */ 663 async_client_put(client); 664 } 665 523 666 /** Wrapper for client connection fibril. 524 667 * … … 533 676 static int connection_fibril(void *arg) 534 677 { 678 assert(arg); 679 535 680 /* 536 681 * Setup fibril-local connection pointer. 537 682 */ 538 FIBRIL_connection = (connection_t *) arg; 539 540 futex_down(&async_futex); 683 fibril_connection = (connection_t *) arg; 541 684 542 685 /* … … 545 688 * hash in a new tracking structure. 546 689 */ 547 548 unsigned long key = FIBRIL_connection->in_task_hash; 549 link_t *lnk = hash_table_find(&client_hash_table, &key); 550 551 client_t *client; 552 553 if (lnk) { 554 client = hash_table_get_instance(lnk, client_t, link); 555 client->refcnt++; 556 } else { 557 client = malloc(sizeof(client_t)); 558 if (!client) { 559 ipc_answer_0(FIBRIL_connection->callid, ENOMEM); 560 futex_up(&async_futex); 561 return 0; 562 } 563 564 client->in_task_hash = FIBRIL_connection->in_task_hash; 565 566 async_serialize_start(); 567 client->data = async_client_data_create(); 568 async_serialize_end(); 569 570 client->refcnt = 1; 571 hash_table_insert(&client_hash_table, &key, &client->link); 572 } 573 574 futex_up(&async_futex); 575 576 FIBRIL_connection->client = client; 690 691 client_t *client = async_client_get(fibril_connection->in_task_id, true); 692 if (!client) { 693 ipc_answer_0(fibril_connection->callid, ENOMEM); 694 return 0; 695 } 696 697 fibril_connection->client = client; 577 698 578 699 /* 579 700 * Call the connection handler function. 580 701 */ 581 FIBRIL_connection->cfibril(FIBRIL_connection->callid,582 & FIBRIL_connection->call);702 fibril_connection->cfibril(fibril_connection->callid, 703 &fibril_connection->call, fibril_connection->carg); 583 704 584 705 /* 585 706 * Remove the reference for this client task connection. 586 707 */ 587 bool destroy; 588 589 futex_down(&async_futex); 590 591 if (--client->refcnt == 0) { 592 hash_table_remove(&client_hash_table, &key, 1); 593 destroy = true; 594 } else 595 destroy = false; 596 597 futex_up(&async_futex); 598 599 if (destroy) { 600 if (client->data) 601 async_client_data_destroy(client->data); 602 603 free(client); 604 } 708 async_client_put(client); 605 709 606 710 /* … … 608 712 */ 609 713 futex_down(&async_futex); 610 key = FIBRIL_connection->in_phone_hash;714 unsigned long key = fibril_connection->in_phone_hash; 611 715 hash_table_remove(&conn_hash_table, &key, 1); 612 716 futex_up(&async_futex); … … 615 719 * Answer all remaining messages with EHANGUP. 616 720 */ 617 while (!list_empty(& FIBRIL_connection->msg_queue)) {721 while (!list_empty(&fibril_connection->msg_queue)) { 618 722 msg_t *msg = 619 list_get_instance( FIBRIL_connection->msg_queue.next, msg_t,620 link);723 list_get_instance(list_first(&fibril_connection->msg_queue), 724 msg_t, link); 621 725 622 726 list_remove(&msg->link); … … 629 733 * i.e. IPC_M_PHONE_HUNGUP. 630 734 */ 631 if ( FIBRIL_connection->close_callid)632 ipc_answer_0( FIBRIL_connection->close_callid, EOK);633 634 free( FIBRIL_connection);735 if (fibril_connection->close_callid) 736 ipc_answer_0(fibril_connection->close_callid, EOK); 737 738 free(fibril_connection); 635 739 return 0; 636 740 } … … 638 742 /** Create a new fibril for a new connection. 639 743 * 640 * Create new fibril for connection, fill in connection structures and insert s744 * Create new fibril for connection, fill in connection structures and insert 641 745 * it into the hash table, so that later we can easily do routing of messages to 642 746 * particular fibrils. 643 747 * 644 * @param in_task_ hashIdentification of the incoming connection.748 * @param in_task_id Identification of the incoming connection. 645 749 * @param in_phone_hash Identification of the incoming connection. 646 750 * @param callid Hash of the opening IPC_M_CONNECT_ME_TO call. … … 651 755 * @param cfibril Fibril function that should be called upon opening the 652 756 * connection. 757 * @param carg Extra argument to pass to the connection fibril 653 758 * 654 759 * @return New fibril id or NULL on failure. 655 760 * 656 761 */ 657 fid_t async_new_connection( sysarg_t in_task_hash, sysarg_t in_phone_hash,762 fid_t async_new_connection(task_id_t in_task_id, sysarg_t in_phone_hash, 658 763 ipc_callid_t callid, ipc_call_t *call, 659 void (*cfibril)(ipc_callid_t, ipc_call_t *))764 async_client_conn_t cfibril, void *carg) 660 765 { 661 766 connection_t *conn = malloc(sizeof(*conn)); … … 667 772 } 668 773 669 conn->in_task_ hash = in_task_hash;774 conn->in_task_id = in_task_id; 670 775 conn->in_phone_hash = in_phone_hash; 671 776 list_initialize(&conn->msg_queue); 672 777 conn->callid = callid; 673 778 conn->close_callid = 0; 779 conn->carg = carg; 674 780 675 781 if (call) … … 681 787 conn->wdata.fid = fibril_create(connection_fibril, conn); 682 788 683 if ( !conn->wdata.fid) {789 if (conn->wdata.fid == 0) { 684 790 free(conn); 791 685 792 if (callid) 686 793 ipc_answer_0(callid, ENOMEM); 794 687 795 return (uintptr_t) NULL; 688 796 } … … 711 819 static void handle_call(ipc_callid_t callid, ipc_call_t *call) 712 820 { 821 assert(call); 822 713 823 /* Unrouted call - take some default action */ 714 824 if ((callid & IPC_CALLID_NOTIFICATION)) { … … 721 831 case IPC_M_CONNECT_ME_TO: 722 832 /* Open new connection with fibril, etc. */ 723 async_new_connection(call->in_task_ hash, IPC_GET_ARG5(*call),724 callid, call, client_connection );833 async_new_connection(call->in_task_id, IPC_GET_ARG5(*call), 834 callid, call, client_connection, NULL); 725 835 return; 726 836 } … … 742 852 futex_down(&async_futex); 743 853 744 link_t *cur = timeout_list.next;745 while (cur != &timeout_list) {854 link_t *cur = list_first(&timeout_list); 855 while (cur != NULL) { 746 856 awaiter_t *waiter = 747 857 list_get_instance(cur, awaiter_t, to_event.link); … … 749 859 if (tv_gt(&waiter->to_event.expires, &tv)) 750 860 break; 751 752 cur = cur->next;753 861 754 862 list_remove(&waiter->to_event.link); … … 764 872 fibril_add_ready(waiter->fid); 765 873 } 874 875 cur = list_first(&timeout_list); 766 876 } 767 877 … … 790 900 suseconds_t timeout; 791 901 if (!list_empty(&timeout_list)) { 792 awaiter_t *waiter = list_get_instance( timeout_list.next,793 awaiter_t, to_event.link);902 awaiter_t *waiter = list_get_instance( 903 list_first(&timeout_list), awaiter_t, to_event.link); 794 904 795 905 struct timeval tv; … … 853 963 { 854 964 fid_t fid = fibril_create(async_manager_fibril, NULL); 855 fibril_add_manager(fid); 965 if (fid != 0) 966 fibril_add_manager(fid); 856 967 } 857 968 … … 867 978 void __async_init(void) 868 979 { 869 if (!hash_table_create(&client_hash_table, CLIENT_HASH_TABLE_BUCKETS, 1,870 &client_hash_table_ops))980 if (!hash_table_create(&client_hash_table, CLIENT_HASH_TABLE_BUCKETS, 981 2, &client_hash_table_ops)) 871 982 abort(); 872 983 873 if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_BUCKETS, 1,874 &conn_hash_table_ops))984 if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_BUCKETS, 985 1, &conn_hash_table_ops)) 875 986 abort(); 987 988 session_ns = (async_sess_t *) malloc(sizeof(async_sess_t)); 989 if (session_ns == NULL) 990 abort(); 991 992 session_ns->mgmt = EXCHANGE_ATOMIC; 993 session_ns->phone = PHONE_NS; 994 session_ns->arg1 = 0; 995 session_ns->arg2 = 0; 996 session_ns->arg3 = 0; 997 998 fibril_mutex_initialize(&session_ns->remote_state_mtx); 999 session_ns->remote_state_data = NULL; 1000 1001 list_initialize(&session_ns->exch_list); 1002 fibril_mutex_initialize(&session_ns->mutex); 1003 atomic_set(&session_ns->refcnt, 0); 876 1004 } 877 1005 … … 888 1016 * 889 1017 */ 890 static void reply_received(void *arg, int retval, ipc_call_t *data) 891 { 1018 void reply_received(void *arg, int retval, ipc_call_t *data) 1019 { 1020 assert(arg); 1021 892 1022 futex_down(&async_futex); 893 1023 … … 919 1049 * completion. 920 1050 * 921 * @param phoneid Handle of the phone that will be used for the send.922 * @param method Service-defined method.1051 * @param exch Exchange for sending the message. 1052 * @param imethod Service-defined interface and method. 923 1053 * @param arg1 Service-defined payload argument. 924 1054 * @param arg2 Service-defined payload argument. … … 931 1061 * 932 1062 */ 933 aid_t async_send_fast( int phoneid, sysarg_tmethod, sysarg_t arg1,1063 aid_t async_send_fast(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1, 934 1064 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, ipc_call_t *dataptr) 935 1065 { 1066 if (exch == NULL) 1067 return 0; 1068 936 1069 amsg_t *msg = malloc(sizeof(amsg_t)); 937 938 if (!msg) 1070 if (msg == NULL) 939 1071 return 0; 940 1072 … … 950 1082 msg->wdata.active = true; 951 1083 952 ipc_call_async_4( phoneid,method, arg1, arg2, arg3, arg4, msg,1084 ipc_call_async_4(exch->phone, imethod, arg1, arg2, arg3, arg4, msg, 953 1085 reply_received, true); 954 1086 … … 961 1093 * completion. 962 1094 * 963 * @param phoneid Handle of the phone that will be used for the send.964 * @param method Service-defined method.1095 * @param exch Exchange for sending the message. 1096 * @param imethod Service-defined interface and method. 965 1097 * @param arg1 Service-defined payload argument. 966 1098 * @param arg2 Service-defined payload argument. … … 974 1106 * 975 1107 */ 976 aid_t async_send_slow( int phoneid, sysarg_tmethod, sysarg_t arg1,1108 aid_t async_send_slow(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1, 977 1109 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5, 978 1110 ipc_call_t *dataptr) 979 1111 { 1112 if (exch == NULL) 1113 return 0; 1114 980 1115 amsg_t *msg = malloc(sizeof(amsg_t)); 981 1116 982 if ( !msg)1117 if (msg == NULL) 983 1118 return 0; 984 1119 … … 994 1129 msg->wdata.active = true; 995 1130 996 ipc_call_async_5( phoneid, method, arg1, arg2, arg3, arg4, arg5, msg,997 reply_received, true);1131 ipc_call_async_5(exch->phone, imethod, arg1, arg2, arg3, arg4, arg5, 1132 msg, reply_received, true); 998 1133 999 1134 return (aid_t) msg; … … 1009 1144 void async_wait_for(aid_t amsgid, sysarg_t *retval) 1010 1145 { 1146 assert(amsgid); 1147 1011 1148 amsg_t *msg = (amsg_t *) amsgid; 1012 1149 … … 1045 1182 int async_wait_timeout(aid_t amsgid, sysarg_t *retval, suseconds_t timeout) 1046 1183 { 1184 assert(amsgid); 1185 1047 1186 amsg_t *msg = (amsg_t *) amsgid; 1048 1187 … … 1113 1252 } 1114 1253 1115 /** Setter for client_connection function pointer.1116 *1117 * @param conn Function that will implement a new connection fibril.1118 *1119 */1120 void async_set_client_connection(async_client_conn_t conn)1121 {1122 client_connection = conn;1123 }1124 1125 /** Setter for interrupt_received function pointer.1126 *1127 * @param intr Function that will implement a new interrupt1128 * notification fibril.1129 */1130 void async_set_interrupt_received(async_client_conn_t intr)1131 {1132 interrupt_received = intr;1133 }1134 1135 1254 /** Pseudo-synchronous message sending - fast version. 1136 1255 * … … 1140 1259 * transferring more arguments, see the slower async_req_slow(). 1141 1260 * 1142 * @param phoneid Hash of the phone through which to make the call.1143 * @param method Method of the call.1261 * @param exch Exchange for sending the message. 1262 * @param imethod Interface and method of the call. 1144 1263 * @param arg1 Service-defined payload argument. 1145 1264 * @param arg2 Service-defined payload argument. … … 1155 1274 * 1156 1275 */ 1157 sysarg_t async_req_fast( int phoneid, sysarg_tmethod, sysarg_t arg1,1276 sysarg_t async_req_fast(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1, 1158 1277 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t *r1, sysarg_t *r2, 1159 1278 sysarg_t *r3, sysarg_t *r4, sysarg_t *r5) 1160 1279 { 1280 if (exch == NULL) 1281 return ENOENT; 1282 1161 1283 ipc_call_t result; 1162 aid_t eid = async_send_4(phoneid,method, arg1, arg2, arg3, arg4,1284 aid_t aid = async_send_4(exch, imethod, arg1, arg2, arg3, arg4, 1163 1285 &result); 1164 1286 1165 1287 sysarg_t rc; 1166 async_wait_for( eid, &rc);1288 async_wait_for(aid, &rc); 1167 1289 1168 1290 if (r1) … … 1188 1310 * Send message asynchronously and return only after the reply arrives. 1189 1311 * 1190 * @param phoneid Hash of the phone through which to make the call.1191 * @param method Method of the call.1312 * @param exch Exchange for sending the message. 1313 * @param imethod Interface and method of the call. 1192 1314 * @param arg1 Service-defined payload argument. 1193 1315 * @param arg2 Service-defined payload argument. … … 1204 1326 * 1205 1327 */ 1206 sysarg_t async_req_slow( int phoneid, sysarg_tmethod, sysarg_t arg1,1328 sysarg_t async_req_slow(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1, 1207 1329 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5, sysarg_t *r1, 1208 1330 sysarg_t *r2, sysarg_t *r3, sysarg_t *r4, sysarg_t *r5) 1209 1331 { 1332 if (exch == NULL) 1333 return ENOENT; 1334 1210 1335 ipc_call_t result; 1211 aid_t eid = async_send_5(phoneid,method, arg1, arg2, arg3, arg4, arg5,1336 aid_t aid = async_send_5(exch, imethod, arg1, arg2, arg3, arg4, arg5, 1212 1337 &result); 1213 1338 1214 1339 sysarg_t rc; 1215 async_wait_for( eid, &rc);1340 async_wait_for(aid, &rc); 1216 1341 1217 1342 if (r1) … … 1233 1358 } 1234 1359 1235 void async_msg_0(int phone, sysarg_t imethod) 1236 { 1237 ipc_call_async_0(phone, imethod, NULL, NULL, true); 1238 } 1239 1240 void async_msg_1(int phone, sysarg_t imethod, sysarg_t arg1) 1241 { 1242 ipc_call_async_1(phone, imethod, arg1, NULL, NULL, true); 1243 } 1244 1245 void async_msg_2(int phone, sysarg_t imethod, sysarg_t arg1, sysarg_t arg2) 1246 { 1247 ipc_call_async_2(phone, imethod, arg1, arg2, NULL, NULL, true); 1248 } 1249 1250 void async_msg_3(int phone, sysarg_t imethod, sysarg_t arg1, sysarg_t arg2, 1251 sysarg_t arg3) 1252 { 1253 ipc_call_async_3(phone, imethod, arg1, arg2, arg3, NULL, NULL, true); 1254 } 1255 1256 void async_msg_4(int phone, sysarg_t imethod, sysarg_t arg1, sysarg_t arg2, 1257 sysarg_t arg3, sysarg_t arg4) 1258 { 1259 ipc_call_async_4(phone, imethod, arg1, arg2, arg3, arg4, NULL, NULL, 1260 true); 1261 } 1262 1263 void async_msg_5(int phone, sysarg_t imethod, sysarg_t arg1, sysarg_t arg2, 1264 sysarg_t arg3, sysarg_t arg4, sysarg_t arg5) 1265 { 1266 ipc_call_async_5(phone, imethod, arg1, arg2, arg3, arg4, arg5, NULL, 1267 NULL, true); 1360 void async_msg_0(async_exch_t *exch, sysarg_t imethod) 1361 { 1362 if (exch != NULL) 1363 ipc_call_async_0(exch->phone, imethod, NULL, NULL, true); 1364 } 1365 1366 void async_msg_1(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1) 1367 { 1368 if (exch != NULL) 1369 ipc_call_async_1(exch->phone, imethod, arg1, NULL, NULL, true); 1370 } 1371 1372 void async_msg_2(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1, 1373 sysarg_t arg2) 1374 { 1375 if (exch != NULL) 1376 ipc_call_async_2(exch->phone, imethod, arg1, arg2, NULL, NULL, 1377 true); 1378 } 1379 1380 void async_msg_3(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1, 1381 sysarg_t arg2, sysarg_t arg3) 1382 { 1383 if (exch != NULL) 1384 ipc_call_async_3(exch->phone, imethod, arg1, arg2, arg3, NULL, 1385 NULL, true); 1386 } 1387 1388 void async_msg_4(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1, 1389 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4) 1390 { 1391 if (exch != NULL) 1392 ipc_call_async_4(exch->phone, imethod, arg1, arg2, arg3, arg4, 1393 NULL, NULL, true); 1394 } 1395 1396 void async_msg_5(async_exch_t *exch, sysarg_t imethod, sysarg_t arg1, 1397 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5) 1398 { 1399 if (exch != NULL) 1400 ipc_call_async_5(exch->phone, imethod, arg1, arg2, arg3, arg4, 1401 arg5, NULL, NULL, true); 1268 1402 } 1269 1403 … … 1302 1436 } 1303 1437 1304 int async_forward_fast(ipc_callid_t callid, int phoneid, sysarg_t imethod, 1305 sysarg_t arg1, sysarg_t arg2, unsigned int mode) 1306 { 1307 return ipc_forward_fast(callid, phoneid, imethod, arg1, arg2, mode); 1308 } 1309 1310 int async_forward_slow(ipc_callid_t callid, int phoneid, sysarg_t imethod, 1311 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, sysarg_t arg5, 1312 unsigned int mode) 1313 { 1314 return ipc_forward_slow(callid, phoneid, imethod, arg1, arg2, arg3, arg4, 1315 arg5, mode); 1438 int async_forward_fast(ipc_callid_t callid, async_exch_t *exch, 1439 sysarg_t imethod, sysarg_t arg1, sysarg_t arg2, unsigned int mode) 1440 { 1441 if (exch == NULL) 1442 return ENOENT; 1443 1444 return ipc_forward_fast(callid, exch->phone, imethod, arg1, arg2, mode); 1445 } 1446 1447 int async_forward_slow(ipc_callid_t callid, async_exch_t *exch, 1448 sysarg_t imethod, sysarg_t arg1, sysarg_t arg2, sysarg_t arg3, 1449 sysarg_t arg4, sysarg_t arg5, unsigned int mode) 1450 { 1451 if (exch == NULL) 1452 return ENOENT; 1453 1454 return ipc_forward_slow(callid, exch->phone, imethod, arg1, arg2, arg3, 1455 arg4, arg5, mode); 1316 1456 } 1317 1457 … … 1320 1460 * Ask through phone for a new connection to some service. 1321 1461 * 1322 * @param phone Phone handle used for contacting the other side.1462 * @param exch Exchange for sending the message. 1323 1463 * @param arg1 User defined argument. 1324 1464 * @param arg2 User defined argument. … … 1326 1466 * @param client_receiver Connection handing routine. 1327 1467 * 1328 * @return New phone handle on success or a negative error code. 1329 * 1330 */ 1331 int async_connect_to_me(int phone, sysarg_t arg1, sysarg_t arg2, 1332 sysarg_t arg3, async_client_conn_t client_receiver) 1333 { 1334 sysarg_t task_hash; 1468 * @return Zero on success or a negative error code. 1469 * 1470 */ 1471 int async_connect_to_me(async_exch_t *exch, sysarg_t arg1, sysarg_t arg2, 1472 sysarg_t arg3, async_client_conn_t client_receiver, void *carg) 1473 { 1474 if (exch == NULL) 1475 return ENOENT; 1476 1335 1477 sysarg_t phone_hash; 1336 int rc = async_req_3_5(phone, IPC_M_CONNECT_TO_ME, arg1, arg2, arg3, 1337 NULL, NULL, NULL, &task_hash, &phone_hash); 1478 sysarg_t rc; 1479 1480 aid_t req; 1481 ipc_call_t answer; 1482 req = async_send_3(exch, IPC_M_CONNECT_TO_ME, arg1, arg2, arg3, 1483 &answer); 1484 async_wait_for(req, &rc); 1485 if (rc != EOK) 1486 return (int) rc; 1487 1488 phone_hash = IPC_GET_ARG5(answer); 1489 1490 if (client_receiver != NULL) 1491 async_new_connection(answer.in_task_id, phone_hash, 0, NULL, 1492 client_receiver, carg); 1493 1494 return EOK; 1495 } 1496 1497 /** Wrapper for making IPC_M_CONNECT_ME calls using the async framework. 1498 * 1499 * Ask through for a cloned connection to some service. 1500 * 1501 * @param mgmt Exchange management style. 1502 * @param exch Exchange for sending the message. 1503 * 1504 * @return New session on success or NULL on error. 1505 * 1506 */ 1507 async_sess_t *async_connect_me(exch_mgmt_t mgmt, async_exch_t *exch) 1508 { 1509 if (exch == NULL) { 1510 errno = ENOENT; 1511 return NULL; 1512 } 1513 1514 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t)); 1515 if (sess == NULL) { 1516 errno = ENOMEM; 1517 return NULL; 1518 } 1519 1520 ipc_call_t result; 1521 1522 amsg_t *msg = malloc(sizeof(amsg_t)); 1523 if (msg == NULL) { 1524 free(sess); 1525 errno = ENOMEM; 1526 return NULL; 1527 } 1528 1529 msg->done = false; 1530 msg->dataptr = &result; 1531 1532 msg->wdata.to_event.inlist = false; 1533 1534 /* 1535 * We may sleep in the next method, 1536 * but it will use its own means 1537 */ 1538 msg->wdata.active = true; 1539 1540 ipc_call_async_0(exch->phone, IPC_M_CONNECT_ME, msg, 1541 reply_received, true); 1542 1543 sysarg_t rc; 1544 async_wait_for((aid_t) msg, &rc); 1545 1546 if (rc != EOK) { 1547 errno = rc; 1548 free(sess); 1549 return NULL; 1550 } 1551 1552 int phone = (int) IPC_GET_ARG5(result); 1553 1554 if (phone < 0) { 1555 errno = phone; 1556 free(sess); 1557 return NULL; 1558 } 1559 1560 sess->mgmt = mgmt; 1561 sess->phone = phone; 1562 sess->arg1 = 0; 1563 sess->arg2 = 0; 1564 sess->arg3 = 0; 1565 1566 fibril_mutex_initialize(&sess->remote_state_mtx); 1567 sess->remote_state_data = NULL; 1568 1569 list_initialize(&sess->exch_list); 1570 fibril_mutex_initialize(&sess->mutex); 1571 atomic_set(&sess->refcnt, 0); 1572 1573 return sess; 1574 } 1575 1576 static int async_connect_me_to_internal(int phone, sysarg_t arg1, sysarg_t arg2, 1577 sysarg_t arg3, sysarg_t arg4) 1578 { 1579 ipc_call_t result; 1580 1581 amsg_t *msg = malloc(sizeof(amsg_t)); 1582 if (msg == NULL) 1583 return ENOENT; 1584 1585 msg->done = false; 1586 msg->dataptr = &result; 1587 1588 msg->wdata.to_event.inlist = false; 1589 1590 /* 1591 * We may sleep in the next method, 1592 * but it will use its own means 1593 */ 1594 msg->wdata.active = true; 1595 1596 ipc_call_async_4(phone, IPC_M_CONNECT_ME_TO, arg1, arg2, arg3, arg4, 1597 msg, reply_received, true); 1598 1599 sysarg_t rc; 1600 async_wait_for((aid_t) msg, &rc); 1601 1338 1602 if (rc != EOK) 1339 1603 return rc; 1340 1604 1341 if (client_receiver != NULL) 1342 async_new_connection(task_hash, phone_hash, 0, NULL, 1343 client_receiver); 1344 1345 return EOK; 1605 return (int) IPC_GET_ARG5(result); 1346 1606 } 1347 1607 1348 1608 /** Wrapper for making IPC_M_CONNECT_ME_TO calls using the async framework. 1349 1609 * 1350 * Ask through phone for a new connection to some service. 1351 * 1352 * @param phone Phone handle used for contacting the other side. 1353 * @param arg1 User defined argument. 1354 * @param arg2 User defined argument. 1355 * @param arg3 User defined argument. 1356 * 1357 * @return New phone handle on success or a negative error code. 1358 * 1359 */ 1360 int async_connect_me_to(int phone, sysarg_t arg1, sysarg_t arg2, 1361 sysarg_t arg3) 1362 { 1363 sysarg_t newphid; 1364 int rc = async_req_3_5(phone, IPC_M_CONNECT_ME_TO, arg1, arg2, arg3, 1365 NULL, NULL, NULL, NULL, &newphid); 1366 1367 if (rc != EOK) 1368 return rc; 1369 1370 return newphid; 1610 * Ask through for a new connection to some service. 1611 * 1612 * @param mgmt Exchange management style. 1613 * @param exch Exchange for sending the message. 1614 * @param arg1 User defined argument. 1615 * @param arg2 User defined argument. 1616 * @param arg3 User defined argument. 1617 * 1618 * @return New session on success or NULL on error. 1619 * 1620 */ 1621 async_sess_t *async_connect_me_to(exch_mgmt_t mgmt, async_exch_t *exch, 1622 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3) 1623 { 1624 if (exch == NULL) { 1625 errno = ENOENT; 1626 return NULL; 1627 } 1628 1629 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t)); 1630 if (sess == NULL) { 1631 errno = ENOMEM; 1632 return NULL; 1633 } 1634 1635 int phone = async_connect_me_to_internal(exch->phone, arg1, arg2, arg3, 1636 0); 1637 1638 if (phone < 0) { 1639 errno = phone; 1640 free(sess); 1641 return NULL; 1642 } 1643 1644 sess->mgmt = mgmt; 1645 sess->phone = phone; 1646 sess->arg1 = arg1; 1647 sess->arg2 = arg2; 1648 sess->arg3 = arg3; 1649 1650 fibril_mutex_initialize(&sess->remote_state_mtx); 1651 sess->remote_state_data = NULL; 1652 1653 list_initialize(&sess->exch_list); 1654 fibril_mutex_initialize(&sess->mutex); 1655 atomic_set(&sess->refcnt, 0); 1656 1657 return sess; 1371 1658 } 1372 1659 … … 1376 1663 * success. 1377 1664 * 1378 * @param phoneid Phone handle used for contacting the other side. 1379 * @param arg1 User defined argument. 1380 * @param arg2 User defined argument. 1381 * @param arg3 User defined argument. 1382 * 1383 * @return New phone handle on success or a negative error code. 1384 * 1385 */ 1386 int async_connect_me_to_blocking(int phoneid, sysarg_t arg1, sysarg_t arg2, 1387 sysarg_t arg3) 1388 { 1389 sysarg_t newphid; 1390 int rc = async_req_4_5(phoneid, IPC_M_CONNECT_ME_TO, arg1, arg2, arg3, 1391 IPC_FLAG_BLOCKING, NULL, NULL, NULL, NULL, &newphid); 1392 1393 if (rc != EOK) 1394 return rc; 1395 1396 return newphid; 1665 * @param mgmt Exchange management style. 1666 * @param exch Exchange for sending the message. 1667 * @param arg1 User defined argument. 1668 * @param arg2 User defined argument. 1669 * @param arg3 User defined argument. 1670 * 1671 * @return New session on success or NULL on error. 1672 * 1673 */ 1674 async_sess_t *async_connect_me_to_blocking(exch_mgmt_t mgmt, async_exch_t *exch, 1675 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3) 1676 { 1677 if (exch == NULL) { 1678 errno = ENOENT; 1679 return NULL; 1680 } 1681 1682 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t)); 1683 if (sess == NULL) { 1684 errno = ENOMEM; 1685 return NULL; 1686 } 1687 1688 int phone = async_connect_me_to_internal(exch->phone, arg1, arg2, arg3, 1689 IPC_FLAG_BLOCKING); 1690 1691 if (phone < 0) { 1692 errno = phone; 1693 free(sess); 1694 return NULL; 1695 } 1696 1697 sess->mgmt = mgmt; 1698 sess->phone = phone; 1699 sess->arg1 = arg1; 1700 sess->arg2 = arg2; 1701 sess->arg3 = arg3; 1702 1703 fibril_mutex_initialize(&sess->remote_state_mtx); 1704 sess->remote_state_data = NULL; 1705 1706 list_initialize(&sess->exch_list); 1707 fibril_mutex_initialize(&sess->mutex); 1708 atomic_set(&sess->refcnt, 0); 1709 1710 return sess; 1397 1711 } 1398 1712 … … 1400 1714 * 1401 1715 */ 1402 int async_connect_kbox(task_id_t id) 1403 { 1404 return ipc_connect_kbox(id); 1716 async_sess_t *async_connect_kbox(task_id_t id) 1717 { 1718 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t)); 1719 if (sess == NULL) { 1720 errno = ENOMEM; 1721 return NULL; 1722 } 1723 1724 int phone = ipc_connect_kbox(id); 1725 if (phone < 0) { 1726 errno = phone; 1727 free(sess); 1728 return NULL; 1729 } 1730 1731 sess->mgmt = EXCHANGE_ATOMIC; 1732 sess->phone = phone; 1733 sess->arg1 = 0; 1734 sess->arg2 = 0; 1735 sess->arg3 = 0; 1736 1737 fibril_mutex_initialize(&sess->remote_state_mtx); 1738 sess->remote_state_data = NULL; 1739 1740 list_initialize(&sess->exch_list); 1741 fibril_mutex_initialize(&sess->mutex); 1742 atomic_set(&sess->refcnt, 0); 1743 1744 return sess; 1745 } 1746 1747 static int async_hangup_internal(int phone) 1748 { 1749 return ipc_hangup(phone); 1405 1750 } 1406 1751 1407 1752 /** Wrapper for ipc_hangup. 1408 1753 * 1409 * @param phone Phone handleto hung up.1754 * @param sess Session to hung up. 1410 1755 * 1411 1756 * @return Zero on success or a negative error code. 1412 1757 * 1413 1758 */ 1414 int async_hangup(int phone) 1415 { 1416 return ipc_hangup(phone); 1759 int async_hangup(async_sess_t *sess) 1760 { 1761 assert(sess); 1762 1763 if (atomic_get(&sess->refcnt) > 0) 1764 return EBUSY; 1765 1766 int rc = async_hangup_internal(sess->phone); 1767 if (rc == EOK) 1768 free(sess); 1769 1770 return rc; 1417 1771 } 1418 1772 … … 1423 1777 } 1424 1778 1779 /** Start new exchange in a session. 1780 * 1781 * @param session Session. 1782 * 1783 * @return New exchange or NULL on error. 1784 * 1785 */ 1786 async_exch_t *async_exchange_begin(async_sess_t *sess) 1787 { 1788 if (sess == NULL) 1789 return NULL; 1790 1791 async_exch_t *exch; 1792 1793 fibril_mutex_lock(&async_sess_mutex); 1794 1795 if (!list_empty(&sess->exch_list)) { 1796 /* 1797 * There are inactive exchanges in the session. 1798 */ 1799 exch = (async_exch_t *) 1800 list_get_instance(list_first(&sess->exch_list), 1801 async_exch_t, sess_link); 1802 1803 list_remove(&exch->sess_link); 1804 list_remove(&exch->global_link); 1805 } else { 1806 /* 1807 * There are no available exchanges in the session. 1808 */ 1809 1810 if ((sess->mgmt == EXCHANGE_ATOMIC) || 1811 (sess->mgmt == EXCHANGE_SERIALIZE)) { 1812 exch = (async_exch_t *) malloc(sizeof(async_exch_t)); 1813 if (exch != NULL) { 1814 link_initialize(&exch->sess_link); 1815 link_initialize(&exch->global_link); 1816 exch->sess = sess; 1817 exch->phone = sess->phone; 1818 } 1819 } else { /* EXCHANGE_PARALLEL */ 1820 /* 1821 * Make a one-time attempt to connect a new data phone. 1822 */ 1823 1824 int phone; 1825 1826 retry: 1827 phone = async_connect_me_to_internal(sess->phone, sess->arg1, 1828 sess->arg2, sess->arg3, 0); 1829 if (phone >= 0) { 1830 exch = (async_exch_t *) malloc(sizeof(async_exch_t)); 1831 if (exch != NULL) { 1832 link_initialize(&exch->sess_link); 1833 link_initialize(&exch->global_link); 1834 exch->sess = sess; 1835 exch->phone = phone; 1836 } else 1837 async_hangup_internal(phone); 1838 } else if (!list_empty(&inactive_exch_list)) { 1839 /* 1840 * We did not manage to connect a new phone. But we 1841 * can try to close some of the currently inactive 1842 * connections in other sessions and try again. 1843 */ 1844 exch = (async_exch_t *) 1845 list_get_instance(list_first(&inactive_exch_list), 1846 async_exch_t, global_link); 1847 1848 list_remove(&exch->sess_link); 1849 list_remove(&exch->global_link); 1850 async_hangup_internal(exch->phone); 1851 free(exch); 1852 goto retry; 1853 } else { 1854 /* 1855 * Wait for a phone to become available. 1856 */ 1857 fibril_condvar_wait(&avail_phone_cv, &async_sess_mutex); 1858 goto retry; 1859 } 1860 } 1861 } 1862 1863 fibril_mutex_unlock(&async_sess_mutex); 1864 1865 if (exch != NULL) { 1866 atomic_inc(&sess->refcnt); 1867 1868 if (sess->mgmt == EXCHANGE_SERIALIZE) 1869 fibril_mutex_lock(&sess->mutex); 1870 } 1871 1872 return exch; 1873 } 1874 1875 /** Finish an exchange. 1876 * 1877 * @param exch Exchange to finish. 1878 * 1879 */ 1880 void async_exchange_end(async_exch_t *exch) 1881 { 1882 if (exch == NULL) 1883 return; 1884 1885 async_sess_t *sess = exch->sess; 1886 1887 atomic_dec(&sess->refcnt); 1888 1889 if (sess->mgmt == EXCHANGE_SERIALIZE) 1890 fibril_mutex_unlock(&sess->mutex); 1891 1892 fibril_mutex_lock(&async_sess_mutex); 1893 1894 list_append(&exch->sess_link, &sess->exch_list); 1895 list_append(&exch->global_link, &inactive_exch_list); 1896 fibril_condvar_signal(&avail_phone_cv); 1897 1898 fibril_mutex_unlock(&async_sess_mutex); 1899 } 1900 1425 1901 /** Wrapper for IPC_M_SHARE_IN calls using the async framework. 1426 1902 * 1427 * @param phoneid Phone that will be used to contact the receiving side.1428 * @param dst 1429 * @param size 1430 * @param arg 1431 * @param flags 1903 * @param exch Exchange for sending the message. 1904 * @param dst Destination address space area base. 1905 * @param size Size of the destination address space area. 1906 * @param arg User defined argument. 1907 * @param flags Storage for the received flags. Can be NULL. 1432 1908 * 1433 1909 * @return Zero on success or a negative error code from errno.h. 1434 1910 * 1435 1911 */ 1436 int async_share_in_start(int phoneid, void *dst, size_t size, sysarg_t arg, 1437 unsigned int *flags) 1438 { 1912 int async_share_in_start(async_exch_t *exch, void *dst, size_t size, 1913 sysarg_t arg, unsigned int *flags) 1914 { 1915 if (exch == NULL) 1916 return ENOENT; 1917 1439 1918 sysarg_t tmp_flags; 1440 int res = async_req_3_2( phoneid, IPC_M_SHARE_IN, (sysarg_t) dst,1919 int res = async_req_3_2(exch, IPC_M_SHARE_IN, (sysarg_t) dst, 1441 1920 (sysarg_t) size, arg, NULL, &tmp_flags); 1442 1921 … … 1496 1975 /** Wrapper for IPC_M_SHARE_OUT calls using the async framework. 1497 1976 * 1498 * @param phoneid Phone that will be used to contact the receiving side.1499 * @param src 1500 * @param flags 1977 * @param exch Exchange for sending the message. 1978 * @param src Source address space area base address. 1979 * @param flags Flags to be used for sharing. Bits can be only cleared. 1501 1980 * 1502 1981 * @return Zero on success or a negative error code from errno.h. 1503 1982 * 1504 1983 */ 1505 int async_share_out_start(int phoneid, void *src, unsigned int flags) 1506 { 1507 return async_req_3_0(phoneid, IPC_M_SHARE_OUT, (sysarg_t) src, 0, 1984 int async_share_out_start(async_exch_t *exch, void *src, unsigned int flags) 1985 { 1986 if (exch == NULL) 1987 return ENOENT; 1988 1989 return async_req_3_0(exch, IPC_M_SHARE_OUT, (sysarg_t) src, 0, 1508 1990 (sysarg_t) flags); 1509 1991 } … … 1558 2040 } 1559 2041 2042 /** Start IPC_M_DATA_READ using the async framework. 2043 * 2044 * @param exch Exchange for sending the message. 2045 * @param dst Address of the beginning of the destination buffer. 2046 * @param size Size of the destination buffer (in bytes). 2047 * @param dataptr Storage of call data (arg 2 holds actual data size). 2048 * 2049 * @return Hash of the sent message or 0 on error. 2050 * 2051 */ 2052 aid_t async_data_read(async_exch_t *exch, void *dst, size_t size, 2053 ipc_call_t *dataptr) 2054 { 2055 return async_send_2(exch, IPC_M_DATA_READ, (sysarg_t) dst, 2056 (sysarg_t) size, dataptr); 2057 } 2058 1560 2059 /** Wrapper for IPC_M_DATA_READ calls using the async framework. 1561 2060 * 1562 * @param phoneid Phone that will be used to contact the receiving side.1563 * @param dst 1564 * @param size 2061 * @param exch Exchange for sending the message. 2062 * @param dst Address of the beginning of the destination buffer. 2063 * @param size Size of the destination buffer. 1565 2064 * 1566 2065 * @return Zero on success or a negative error code from errno.h. 1567 2066 * 1568 2067 */ 1569 int async_data_read_start(int phoneid, void *dst, size_t size) 1570 { 1571 return async_req_2_0(phoneid, IPC_M_DATA_READ, (sysarg_t) dst, 2068 int async_data_read_start(async_exch_t *exch, void *dst, size_t size) 2069 { 2070 if (exch == NULL) 2071 return ENOENT; 2072 2073 return async_req_2_0(exch, IPC_M_DATA_READ, (sysarg_t) dst, 1572 2074 (sysarg_t) size); 1573 2075 } … … 1625 2127 * 1626 2128 */ 1627 int async_data_read_forward_fast(int phoneid, sysarg_t method, sysarg_t arg1, 1628 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, ipc_call_t *dataptr) 1629 { 2129 int async_data_read_forward_fast(async_exch_t *exch, sysarg_t imethod, 2130 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, 2131 ipc_call_t *dataptr) 2132 { 2133 if (exch == NULL) 2134 return ENOENT; 2135 1630 2136 ipc_callid_t callid; 1631 2137 if (!async_data_read_receive(&callid, NULL)) { … … 1634 2140 } 1635 2141 1636 aid_t msg = async_send_fast( phoneid,method, arg1, arg2, arg3, arg4,2142 aid_t msg = async_send_fast(exch, imethod, arg1, arg2, arg3, arg4, 1637 2143 dataptr); 1638 2144 if (msg == 0) { … … 1641 2147 } 1642 2148 1643 int retval = ipc_forward_fast(callid, phoneid, 0, 0, 0,2149 int retval = ipc_forward_fast(callid, exch->phone, 0, 0, 0, 1644 2150 IPC_FF_ROUTE_FROM_ME); 1645 2151 if (retval != EOK) { … … 1657 2163 /** Wrapper for IPC_M_DATA_WRITE calls using the async framework. 1658 2164 * 1659 * @param phoneid Phone that will be used to contact the receiving side.1660 * @param src 1661 * @param size 2165 * @param exch Exchange for sending the message. 2166 * @param src Address of the beginning of the source buffer. 2167 * @param size Size of the source buffer. 1662 2168 * 1663 2169 * @return Zero on success or a negative error code from errno.h. 1664 2170 * 1665 2171 */ 1666 int async_data_write_start(int phoneid, const void *src, size_t size) 1667 { 1668 return async_req_2_0(phoneid, IPC_M_DATA_WRITE, (sysarg_t) src, 2172 int async_data_write_start(async_exch_t *exch, const void *src, size_t size) 2173 { 2174 if (exch == NULL) 2175 return ENOENT; 2176 2177 return async_req_2_0(exch, IPC_M_DATA_WRITE, (sysarg_t) src, 1669 2178 (sysarg_t) size); 1670 2179 } … … 1743 2252 size_t *received) 1744 2253 { 2254 assert(data); 2255 1745 2256 ipc_callid_t callid; 1746 2257 size_t size; … … 1810 2321 * 1811 2322 */ 1812 int async_data_write_forward_fast(int phoneid, sysarg_t method, sysarg_t arg1, 1813 sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, ipc_call_t *dataptr) 1814 { 2323 int async_data_write_forward_fast(async_exch_t *exch, sysarg_t imethod, 2324 sysarg_t arg1, sysarg_t arg2, sysarg_t arg3, sysarg_t arg4, 2325 ipc_call_t *dataptr) 2326 { 2327 if (exch == NULL) 2328 return ENOENT; 2329 1815 2330 ipc_callid_t callid; 1816 2331 if (!async_data_write_receive(&callid, NULL)) { … … 1819 2334 } 1820 2335 1821 aid_t msg = async_send_fast( phoneid,method, arg1, arg2, arg3, arg4,2336 aid_t msg = async_send_fast(exch, imethod, arg1, arg2, arg3, arg4, 1822 2337 dataptr); 1823 2338 if (msg == 0) { … … 1826 2341 } 1827 2342 1828 int retval = ipc_forward_fast(callid, phoneid, 0, 0, 0,2343 int retval = ipc_forward_fast(callid, exch->phone, 0, 0, 0, 1829 2344 IPC_FF_ROUTE_FROM_ME); 1830 2345 if (retval != EOK) { … … 1840 2355 } 1841 2356 2357 /** Wrapper for sending an exchange over different exchange for cloning 2358 * 2359 * @param exch Exchange to be used for sending. 2360 * @param clone_exch Exchange to be cloned. 2361 * 2362 */ 2363 int async_exchange_clone(async_exch_t *exch, async_exch_t *clone_exch) 2364 { 2365 return async_req_1_0(exch, IPC_M_CONNECTION_CLONE, clone_exch->phone); 2366 } 2367 2368 /** Wrapper for receiving the IPC_M_CONNECTION_CLONE calls. 2369 * 2370 * If the current call is IPC_M_CONNECTION_CLONE then a new 2371 * async session is created for the accepted phone. 2372 * 2373 * @param mgmt Exchange management style. 2374 * 2375 * @return New async session or NULL on failure. 2376 * 2377 */ 2378 async_sess_t *async_clone_receive(exch_mgmt_t mgmt) 2379 { 2380 /* Accept the phone */ 2381 ipc_call_t call; 2382 ipc_callid_t callid = async_get_call(&call); 2383 int phone = (int) IPC_GET_ARG1(call); 2384 2385 if ((IPC_GET_IMETHOD(call) != IPC_M_CONNECTION_CLONE) || 2386 (phone < 0)) { 2387 async_answer_0(callid, EINVAL); 2388 return NULL; 2389 } 2390 2391 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t)); 2392 if (sess == NULL) { 2393 async_answer_0(callid, ENOMEM); 2394 return NULL; 2395 } 2396 2397 sess->mgmt = mgmt; 2398 sess->phone = phone; 2399 sess->arg1 = 0; 2400 sess->arg2 = 0; 2401 sess->arg3 = 0; 2402 2403 fibril_mutex_initialize(&sess->remote_state_mtx); 2404 sess->remote_state_data = NULL; 2405 2406 list_initialize(&sess->exch_list); 2407 fibril_mutex_initialize(&sess->mutex); 2408 atomic_set(&sess->refcnt, 0); 2409 2410 /* Acknowledge the cloned phone */ 2411 async_answer_0(callid, EOK); 2412 2413 return sess; 2414 } 2415 2416 /** Wrapper for receiving the IPC_M_CONNECT_TO_ME calls. 2417 * 2418 * If the current call is IPC_M_CONNECT_TO_ME then a new 2419 * async session is created for the accepted phone. 2420 * 2421 * @param mgmt Exchange management style. 2422 * 2423 * @return New async session. 2424 * @return NULL on failure. 2425 * 2426 */ 2427 async_sess_t *async_callback_receive(exch_mgmt_t mgmt) 2428 { 2429 /* Accept the phone */ 2430 ipc_call_t call; 2431 ipc_callid_t callid = async_get_call(&call); 2432 int phone = (int) IPC_GET_ARG5(call); 2433 2434 if ((IPC_GET_IMETHOD(call) != IPC_M_CONNECT_TO_ME) || 2435 (phone < 0)) { 2436 async_answer_0(callid, EINVAL); 2437 return NULL; 2438 } 2439 2440 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t)); 2441 if (sess == NULL) { 2442 async_answer_0(callid, ENOMEM); 2443 return NULL; 2444 } 2445 2446 sess->mgmt = mgmt; 2447 sess->phone = phone; 2448 sess->arg1 = 0; 2449 sess->arg2 = 0; 2450 sess->arg3 = 0; 2451 2452 fibril_mutex_initialize(&sess->remote_state_mtx); 2453 sess->remote_state_data = NULL; 2454 2455 list_initialize(&sess->exch_list); 2456 fibril_mutex_initialize(&sess->mutex); 2457 atomic_set(&sess->refcnt, 0); 2458 2459 /* Acknowledge the connected phone */ 2460 async_answer_0(callid, EOK); 2461 2462 return sess; 2463 } 2464 2465 /** Wrapper for receiving the IPC_M_CONNECT_TO_ME calls. 2466 * 2467 * If the call is IPC_M_CONNECT_TO_ME then a new 2468 * async session is created. However, the phone is 2469 * not accepted automatically. 2470 * 2471 * @param mgmt Exchange management style. 2472 * @param call Call data. 2473 * 2474 * @return New async session. 2475 * @return NULL on failure. 2476 * @return NULL if the call is not IPC_M_CONNECT_TO_ME. 2477 * 2478 */ 2479 async_sess_t *async_callback_receive_start(exch_mgmt_t mgmt, ipc_call_t *call) 2480 { 2481 int phone = (int) IPC_GET_ARG5(*call); 2482 2483 if ((IPC_GET_IMETHOD(*call) != IPC_M_CONNECT_TO_ME) || 2484 (phone < 0)) 2485 return NULL; 2486 2487 async_sess_t *sess = (async_sess_t *) malloc(sizeof(async_sess_t)); 2488 if (sess == NULL) 2489 return NULL; 2490 2491 sess->mgmt = mgmt; 2492 sess->phone = phone; 2493 sess->arg1 = 0; 2494 sess->arg2 = 0; 2495 sess->arg3 = 0; 2496 2497 fibril_mutex_initialize(&sess->remote_state_mtx); 2498 sess->remote_state_data = NULL; 2499 2500 list_initialize(&sess->exch_list); 2501 fibril_mutex_initialize(&sess->mutex); 2502 atomic_set(&sess->refcnt, 0); 2503 2504 return sess; 2505 } 2506 2507 int async_state_change_start(async_exch_t *exch, sysarg_t arg1, sysarg_t arg2, 2508 sysarg_t arg3, async_exch_t *other_exch) 2509 { 2510 return async_req_5_0(exch, IPC_M_STATE_CHANGE_AUTHORIZE, 2511 arg1, arg2, arg3, 0, other_exch->phone); 2512 } 2513 2514 bool async_state_change_receive(ipc_callid_t *callid, sysarg_t *arg1, 2515 sysarg_t *arg2, sysarg_t *arg3) 2516 { 2517 assert(callid); 2518 2519 ipc_call_t call; 2520 *callid = async_get_call(&call); 2521 2522 if (IPC_GET_IMETHOD(call) != IPC_M_STATE_CHANGE_AUTHORIZE) 2523 return false; 2524 2525 if (arg1) 2526 *arg1 = IPC_GET_ARG1(call); 2527 if (arg2) 2528 *arg2 = IPC_GET_ARG2(call); 2529 if (arg3) 2530 *arg3 = IPC_GET_ARG3(call); 2531 2532 return true; 2533 } 2534 2535 int async_state_change_finalize(ipc_callid_t callid, async_exch_t *other_exch) 2536 { 2537 return ipc_answer_1(callid, EOK, other_exch->phone); 2538 } 2539 2540 /** Lock and get session remote state 2541 * 2542 * Lock and get the local replica of the remote state 2543 * in stateful sessions. The call should be paired 2544 * with async_remote_state_release*(). 2545 * 2546 * @param[in] sess Stateful session. 2547 * 2548 * @return Local replica of the remote state. 2549 * 2550 */ 2551 void *async_remote_state_acquire(async_sess_t *sess) 2552 { 2553 fibril_mutex_lock(&sess->remote_state_mtx); 2554 return sess->remote_state_data; 2555 } 2556 2557 /** Update the session remote state 2558 * 2559 * Update the local replica of the remote state 2560 * in stateful sessions. The remote state must 2561 * be already locked. 2562 * 2563 * @param[in] sess Stateful session. 2564 * @param[in] state New local replica of the remote state. 2565 * 2566 */ 2567 void async_remote_state_update(async_sess_t *sess, void *state) 2568 { 2569 assert(fibril_mutex_is_locked(&sess->remote_state_mtx)); 2570 sess->remote_state_data = state; 2571 } 2572 2573 /** Release the session remote state 2574 * 2575 * Unlock the local replica of the remote state 2576 * in stateful sessions. 2577 * 2578 * @param[in] sess Stateful session. 2579 * 2580 */ 2581 void async_remote_state_release(async_sess_t *sess) 2582 { 2583 assert(fibril_mutex_is_locked(&sess->remote_state_mtx)); 2584 2585 fibril_mutex_unlock(&sess->remote_state_mtx); 2586 } 2587 2588 /** Release the session remote state and end an exchange 2589 * 2590 * Unlock the local replica of the remote state 2591 * in stateful sessions. This is convenience function 2592 * which gets the session pointer from the exchange 2593 * and also ends the exchange. 2594 * 2595 * @param[in] exch Stateful session's exchange. 2596 * 2597 */ 2598 void async_remote_state_release_exchange(async_exch_t *exch) 2599 { 2600 if (exch == NULL) 2601 return; 2602 2603 async_sess_t *sess = exch->sess; 2604 assert(fibril_mutex_is_locked(&sess->remote_state_mtx)); 2605 2606 async_exchange_end(exch); 2607 fibril_mutex_unlock(&sess->remote_state_mtx); 2608 } 2609 1842 2610 /** @} 1843 2611 */
Note:
See TracChangeset
for help on using the changeset viewer.
