![[BACK]](/cvsweb/icons/back.gif){kind=icon}
Return to ipc_mqueue.c CVS log ![[TXT]](/cvsweb/icons/text.gif){kind=icon}
![[DIR]](/cvsweb/icons/dir.gif){kind=icon}
Up to [Apple XNU] / XNU / osfmk / ipc|
Return to ipc_mqueue.c CVS log | Up to [Apple XNU] / XNU / osfmk / ipc |
1.1 ! root 1: /* ! 2: * Copyright (c) 2000 Apple Computer, Inc. All rights reserved. ! 3: * ! 4: * @APPLE_LICENSE_HEADER_START@ ! 5: * ! 6: * The contents of this file constitute Original Code as defined in and ! 7: * are subject to the Apple Public Source License Version 1.1 (the ! 8: * "License"). You may not use this file except in compliance with the ! 9: * License. Please obtain a copy of the License at ! 10: * http://www.apple.com/publicsource and read it before using this file. ! 11: * ! 12: * This Original Code and all software distributed under the License are ! 13: * distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, EITHER ! 14: * EXPRESS OR IMPLIED, AND APPLE HEREBY DISCLAIMS ALL SUCH WARRANTIES, ! 15: * INCLUDING WITHOUT LIMITATION, ANY WARRANTIES OF MERCHANTABILITY, ! 16: * FITNESS FOR A PARTICULAR PURPOSE OR NON-INFRINGEMENT. Please see the ! 17: * License for the specific language governing rights and limitations ! 18: * under the License. ! 19: * ! 20: * @APPLE_LICENSE_HEADER_END@ ! 21: */ ! 22: /* ! 23: * @OSF_FREE_COPYRIGHT@ ! 24: */ ! 25: /* ! 26: * Mach Operating System ! 27: * Copyright (c) 1991,1990,1989 Carnegie Mellon University ! 28: * All Rights Reserved. ! 29: * ! 30: * Permission to use, copy, modify and distribute this software and its ! 31: * documentation is hereby granted, provided that both the copyright ! 32: * notice and this permission notice appear in all copies of the ! 33: * software, derivative works or modified versions, and any portions ! 34: * thereof, and that both notices appear in supporting documentation. ! 35: * ! 36: * CARNEGIE MELLON ALLOWS FREE USE OF THIS SOFTWARE IN ITS "AS IS" ! 37: * CONDITION. CARNEGIE MELLON DISCLAIMS ANY LIABILITY OF ANY KIND FOR ! 38: * ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF THIS SOFTWARE. ! 39: * ! 40: * Carnegie Mellon requests users of this software to return to ! 41: * ! 42: * Software Distribution Coordinator or [email protected] ! 43: * School of Computer Science ! 44: * Carnegie Mellon University ! 45: * Pittsburgh PA 15213-3890 ! 46: * ! 47: * any improvements or extensions that they make and grant Carnegie Mellon ! 48: * the rights to redistribute these changes. ! 49: */ ! 50: /* ! 51: */ ! 52: /* ! 53: * File: ipc/ipc_mqueue.c ! 54: * Author: Rich Draves ! 55: * Date: 1989 ! 56: * ! 57: * Functions to manipulate IPC message queues. ! 58: */ ! 59: ! 60: #include <dipc.h> ! 61: ! 62: #include <mach/port.h> ! 63: #include <mach/message.h> ! 64: #include <mach/sync_policy.h> ! 65: ! 66: #include <kern/assert.h> ! 67: #include <kern/counters.h> ! 68: #include <kern/sched_prim.h> ! 69: #include <kern/ipc_kobject.h> ! 70: #include <kern/misc_protos.h> ! 71: #include <kern/task.h> ! 72: #include <kern/thread.h> ! 73: #include <kern/wait_queue.h> ! 74: ! 75: #include <ipc/ipc_mqueue.h> ! 76: #include <ipc/ipc_kmsg.h> ! 77: #include <ipc/ipc_port.h> ! 78: #include <ipc/ipc_pset.h> ! 79: #include <ipc/ipc_space.h> ! 80: ! 81: #include <ddb/tr.h> ! 82: ! 83: int ipc_mqueue_full; /* address is event for queue space */ ! 84: int ipc_mqueue_rcv; /* address is event for message arrival */ ! 85: ! 86: #define TR_ENABLE 0 ! 87: ! 88: /* ! 89: * Routine: ipc_mqueue_init ! 90: * Purpose: ! 91: * Initialize a newly-allocated message queue. ! 92: */ ! 93: void ! 94: ipc_mqueue_init( ! 95: ipc_mqueue_t mqueue, ! 96: boolean_t is_set) ! 97: { ! 98: if (is_set) { ! 99: wait_queue_sub_init(&mqueue->imq_set_queue, SYNC_POLICY_FIFO); ! 100: } else { ! 101: wait_queue_init(&mqueue->imq_wait_queue, SYNC_POLICY_FIFO); ! 102: ipc_kmsg_queue_init(&mqueue->imq_messages); ! 103: mqueue->imq_seqno = 0; ! 104: mqueue->imq_msgcount = 0; ! 105: mqueue->imq_qlimit = MACH_PORT_QLIMIT_DEFAULT; ! 106: mqueue->imq_fullwaiters = FALSE; ! 107: } ! 108: } ! 109: ! 110: /* ! 111: * Routine: ipc_mqueue_member ! 112: * Purpose: ! 113: * Indicate whether the (port) mqueue is a member of ! 114: * this portset's mqueue. We do this by checking ! 115: * whether the portset mqueue's waitq is an member of ! 116: * the port's mqueue waitq. ! 117: * Conditions: ! 118: * the portset's mqueue is not already a member ! 119: * this may block while allocating linkage structures. ! 120: */ ! 121: ! 122: boolean_t ! 123: ipc_mqueue_member( ! 124: ipc_mqueue_t port_mqueue, ! 125: ipc_mqueue_t set_mqueue) ! 126: { ! 127: wait_queue_t port_waitq = &port_mqueue->imq_wait_queue; ! 128: wait_queue_t set_waitq = &set_mqueue->imq_wait_queue; ! 129: ! 130: return (wait_queue_member(port_waitq, set_waitq)); ! 131: ! 132: } ! 133: ! 134: /* ! 135: * Routine: ipc_mqueue_remove ! 136: * Purpose: ! 137: * Remove the association between the queue and the specified ! 138: * subordinate message queue. ! 139: */ ! 140: ! 141: kern_return_t ! 142: ipc_mqueue_remove( ! 143: ipc_mqueue_t mqueue, ! 144: ipc_mqueue_t sub_mqueue) ! 145: { ! 146: wait_queue_t mq_waitq = &mqueue->imq_wait_queue; ! 147: wait_queue_sub_t sub_waitq = &sub_mqueue->imq_set_queue; ! 148: ! 149: if (wait_queue_member(mq_waitq, sub_waitq)) { ! 150: wait_queue_unlink(mq_waitq, sub_waitq); ! 151: return KERN_SUCCESS; ! 152: } ! 153: return KERN_NOT_IN_SET; ! 154: } ! 155: ! 156: /* ! 157: * Routine: ipc_mqueue_remove_one ! 158: * Purpose: ! 159: * Find and remove one subqueue from the queue. ! 160: * Conditions: ! 161: * Will return the set mqueue that was removed ! 162: */ ! 163: void ! 164: ipc_mqueue_remove_one( ! 165: ipc_mqueue_t mqueue, ! 166: ipc_mqueue_t *sub_queuep) ! 167: { ! 168: wait_queue_t mq_waitq = &mqueue->imq_wait_queue; ! 169: ! 170: wait_queue_unlink_one(mq_waitq, (wait_queue_sub_t *)sub_queuep); ! 171: return; ! 172: } ! 173: ! 174: ! 175: /* ! 176: * Routine: ipc_mqueue_add ! 177: * Purpose: ! 178: * Associate the portset's mqueue with the port's mqueue. ! 179: * This has to be done so that posting the port will wakeup ! 180: * a portset waiter. If there are waiters on the portset ! 181: * mqueue and messages on the port mqueue, try to match them ! 182: * up now. ! 183: * Conditions: ! 184: * May block. ! 185: */ ! 186: kern_return_t ! 187: ipc_mqueue_add( ! 188: ipc_mqueue_t port_mqueue, ! 189: ipc_mqueue_t set_mqueue) ! 190: { ! 191: wait_queue_t port_waitq = &port_mqueue->imq_wait_queue; ! 192: wait_queue_sub_t set_waitq = &set_mqueue->imq_set_queue; ! 193: ipc_kmsg_queue_t kmsgq; ! 194: ipc_kmsg_t kmsg, next; ! 195: kern_return_t kr; ! 196: ! 197: kr = wait_queue_link(port_waitq, set_waitq); ! 198: if (kr != KERN_SUCCESS) ! 199: return kr; ! 200: ! 201: /* ! 202: * Now that the set has been added to the port, there may be ! 203: * messages queued on the port and threads waiting on the set ! 204: * waitq. Lets get them together. ! 205: */ ! 206: imq_lock(port_mqueue); ! 207: kmsgq = &port_mqueue->imq_messages; ! 208: for (kmsg = ipc_kmsg_queue_first(kmsgq); ! 209: kmsg != IKM_NULL; ! 210: kmsg = next) { ! 211: next = ipc_kmsg_queue_next(kmsgq, kmsg); ! 212: ! 213: for (;;) { ! 214: thread_t th; ! 215: spl_t s; ! 216: ! 217: s = splsched(); ! 218: th = wait_queue_wakeup_identity_locked(port_waitq, ! 219: IPC_MQUEUE_RECEIVE, ! 220: THREAD_AWAKENED, ! 221: FALSE); ! 222: /* waitq/mqueue still locked, thread locked */ ! 223: ! 224: if (th == THREAD_NULL) { ! 225: splx(s); ! 226: goto leave; ! 227: } ! 228: ! 229: /* ! 230: * Found a receiver. see if they can handle the message ! 231: * correctly (the message is not too large for them, or ! 232: * they didn't care to be informed that the message was ! 233: * too large). If they can't handle it, take them off ! 234: * the list and let them go back and figure it out and ! 235: * just move onto the next. ! 236: */ ! 237: if (th->ith_msize < ! 238: kmsg->ikm_header.msgh_size + ! 239: REQUESTED_TRAILER_SIZE(th->ith_option)) { ! 240: th->ith_state = MACH_RCV_TOO_LARGE; ! 241: th->ith_msize = kmsg->ikm_header.msgh_size; ! 242: if (th->ith_option & MACH_RCV_LARGE) { ! 243: /* ! 244: * let him go without message ! 245: */ ! 246: th->ith_kmsg = IKM_NULL; ! 247: th->ith_seqno = 0; ! 248: thread_unlock(th); ! 249: splx(s); ! 250: continue; /* find another thread */ ! 251: } ! 252: } else { ! 253: th->ith_state = MACH_MSG_SUCCESS; ! 254: } ! 255: ! 256: /* ! 257: * This thread is going to take this message, ! 258: * so give it to him. ! 259: */ ! 260: ipc_mqueue_release_msgcount(port_mqueue); ! 261: ipc_kmsg_rmqueue(kmsgq, kmsg); ! 262: th->ith_kmsg = kmsg; ! 263: th->ith_seqno = port_mqueue->imq_seqno++; ! 264: thread_unlock(th); ! 265: splx(s); ! 266: break; /* go to next message */ ! 267: } ! 268: ! 269: } ! 270: leave: ! 271: imq_unlock(port_mqueue); ! 272: return KERN_SUCCESS; ! 273: } ! 274: ! 275: /* ! 276: * Routine: ipc_mqueue_changed ! 277: * Purpose: ! 278: * Wake up receivers waiting in a message queue. ! 279: * Conditions: ! 280: * The message queue is locked. ! 281: */ ! 282: ! 283: void ! 284: ipc_mqueue_changed( ! 285: ipc_mqueue_t mqueue) ! 286: { ! 287: wait_queue_wakeup_all_locked(&mqueue->imq_wait_queue, ! 288: IPC_MQUEUE_RECEIVE, ! 289: THREAD_RESTART, ! 290: FALSE); /* unlock waitq? */ ! 291: } ! 292: ! 293: ! 294: ! 295: ! 296: /* ! 297: * Routine: ipc_mqueue_send ! 298: * Purpose: ! 299: * Send a message to a message queue. The message holds a reference ! 300: * for the destination port for this message queue in the ! 301: * msgh_remote_port field. ! 302: * ! 303: * If unsuccessful, the caller still has possession of ! 304: * the message and must do something with it. If successful, ! 305: * the message is queued, given to a receiver, or destroyed. ! 306: * Conditions: ! 307: * Nothing locked. ! 308: * Returns: ! 309: * MACH_MSG_SUCCESS The message was accepted. ! 310: * MACH_SEND_TIMED_OUT Caller still has message. ! 311: * MACH_SEND_INTERRUPTED Caller still has message. ! 312: */ ! 313: mach_msg_return_t ! 314: ipc_mqueue_send( ! 315: ipc_mqueue_t mqueue, ! 316: ipc_kmsg_t kmsg, ! 317: mach_msg_option_t option, ! 318: mach_msg_timeout_t timeout) ! 319: { ! 320: int save_wait_result; ! 321: ! 322: /* ! 323: * Don't block if: ! 324: * 1) We're under the queue limit. ! 325: * 2) Caller used the MACH_SEND_ALWAYS internal option. ! 326: * 3) Message is sent to a send-once right. ! 327: */ ! 328: imq_lock(mqueue); ! 329: ! 330: if (!imq_full(mqueue) || ! 331: (option & MACH_SEND_ALWAYS) || ! 332: (MACH_MSGH_BITS_REMOTE(kmsg->ikm_header.msgh_bits) == ! 333: MACH_MSG_TYPE_PORT_SEND_ONCE)) { ! 334: mqueue->imq_msgcount++; ! 335: imq_unlock(mqueue); ! 336: ! 337: } else { ! 338: ! 339: /* ! 340: * We have to wait for space to be granted to us. ! 341: */ ! 342: if ((option & MACH_SEND_TIMEOUT) && (timeout == 0)) { ! 343: imq_unlock(mqueue); ! 344: return MACH_SEND_TIMED_OUT; ! 345: } ! 346: mqueue->imq_fullwaiters = TRUE; ! 347: wait_queue_assert_wait_locked(&mqueue->imq_wait_queue, ! 348: IPC_MQUEUE_FULL, ! 349: THREAD_ABORTSAFE, ! 350: TRUE); /* unlock? */ ! 351: /* wait/mqueue is unlocked */ ! 352: ! 353: if (option & MACH_SEND_TIMEOUT) ! 354: thread_set_timer(timeout, 1000*NSEC_PER_USEC); ! 355: ! 356: counter(c_ipc_mqueue_send_block++); ! 357: save_wait_result = thread_block((void (*)(void)) 0); ! 358: ! 359: if (option & MACH_SEND_TIMEOUT) ! 360: thread_cancel_timer(); ! 361: ! 362: ! 363: switch (save_wait_result) { ! 364: case THREAD_TIMED_OUT: ! 365: assert(option & MACH_SEND_TIMEOUT); ! 366: return MACH_SEND_TIMED_OUT; ! 367: ! 368: case THREAD_AWAKENED: ! 369: break; /* we can proceed - inherited msgcount from waker */ ! 370: ! 371: case THREAD_INTERRUPTED: ! 372: return MACH_SEND_INTERRUPTED; ! 373: ! 374: case THREAD_RESTART: ! 375: default: ! 376: panic("ipc_mqueue_send"); ! 377: } ! 378: } ! 379: ! 380: ipc_mqueue_post(mqueue, kmsg); ! 381: return MACH_MSG_SUCCESS; ! 382: } ! 383: ! 384: /* ! 385: * Routine: ipc_mqueue_release_msgcount ! 386: * Purpose: ! 387: * Release a message queue reference in the case where we ! 388: * found a waiter. ! 389: * ! 390: * Conditions: ! 391: * The message queue is locked ! 392: */ ! 393: void ! 394: ipc_mqueue_release_msgcount( ! 395: ipc_mqueue_t mqueue) ! 396: { ! 397: assert(imq_held(mqueue)); ! 398: assert(mqueue->imq_msgcount > 0); ! 399: ! 400: mqueue->imq_msgcount--; ! 401: if (!imq_full(mqueue) && mqueue->imq_fullwaiters) { ! 402: if (wait_queue_wakeup_one_locked(&mqueue->imq_wait_queue, ! 403: IPC_MQUEUE_FULL, ! 404: THREAD_AWAKENED, ! 405: FALSE) != KERN_SUCCESS) { ! 406: mqueue->imq_fullwaiters = FALSE; ! 407: } else { ! 408: mqueue->imq_msgcount++; /* gave it away */ ! 409: } ! 410: } ! 411: } ! 412: ! 413: /* ! 414: * Routine: ipc_mqueue_post ! 415: * Purpose: ! 416: * Post a message to a waiting receiver or enqueue it. If a ! 417: * receiver is waiting, we can release our reserved space in ! 418: * the message queue. ! 419: * ! 420: * Conditions: ! 421: * If we need to queue, our space in the message queue is reserved. ! 422: */ ! 423: void ! 424: ipc_mqueue_post( ! 425: register ipc_mqueue_t mqueue, ! 426: register ipc_kmsg_t kmsg) ! 427: { ! 428: ! 429: /* ! 430: * While the msg queue is locked, we have control of the ! 431: * kmsg, so the ref in it for the port is still good. ! 432: * ! 433: * Check for a receiver for the message. ! 434: */ ! 435: imq_lock(mqueue); ! 436: for (;;) { ! 437: wait_queue_t waitq = &mqueue->imq_wait_queue; ! 438: thread_t receiver; ! 439: spl_t s; ! 440: ! 441: s = splsched(); ! 442: receiver = wait_queue_wakeup_identity_locked(waitq, ! 443: IPC_MQUEUE_RECEIVE, ! 444: THREAD_AWAKENED, ! 445: FALSE); ! 446: /* waitq still locked, thread locked */ ! 447: ! 448: if (receiver == THREAD_NULL) { ! 449: /* ! 450: * no receivers; queue kmsg ! 451: */ ! 452: assert(mqueue->imq_msgcount > 0); ! 453: ipc_kmsg_enqueue_macro(&mqueue->imq_messages, kmsg); ! 454: splx(s); ! 455: break; ! 456: } ! 457: ! 458: /* ! 459: * We found a waiting thread. ! 460: * If the message is too large or the scatter list is too small ! 461: * the thread we wake up will get that as its status. ! 462: */ ! 463: if (receiver->ith_msize < ! 464: (kmsg->ikm_header.msgh_size) + ! 465: REQUESTED_TRAILER_SIZE(receiver->ith_option)) { ! 466: receiver->ith_msize = kmsg->ikm_header.msgh_size; ! 467: receiver->ith_state = MACH_RCV_TOO_LARGE; ! 468: } else { ! 469: receiver->ith_state = ! 470: #ifdef DOITFORSCATTER ! 471: (receiver->ith_scatter_list != MACH_MSG_BODY_NULL) ? ! 472: ipc_kmsg_check_scatter(kmsg, ! 473: receiver->ith_option, ! 474: &receiver->ith_scatter_list, ! 475: &receiver->ith_scatter_list_size) : ! 476: #endif /* DOITFORSCATTER */ ! 477: MACH_MSG_SUCCESS; ! 478: } ! 479: ! 480: /* ! 481: * If there is no problem with the upcoming receive, or the ! 482: * receiver thread didn't specifically ask for special too ! 483: * large error condition, go ahead and select it anyway. ! 484: */ ! 485: if ((receiver->ith_state == MACH_MSG_SUCCESS) || ! 486: !(receiver->ith_option & MACH_RCV_LARGE)) { ! 487: ! 488: receiver->ith_kmsg = kmsg; ! 489: receiver->ith_seqno = mqueue->imq_seqno++; ! 490: thread_unlock(receiver); ! 491: splx(s); ! 492: ! 493: /* we didn't need our reserved spot in the queue */ ! 494: ipc_mqueue_release_msgcount(mqueue); ! 495: break; ! 496: } ! 497: ! 498: /* ! 499: * Otherwise, this thread needs to be released to run ! 500: * and handle its error without getting the message. We ! 501: * need to go back and pick another one. ! 502: */ ! 503: receiver->ith_kmsg = IKM_NULL; ! 504: receiver->ith_seqno = 0; ! 505: thread_unlock(receiver); ! 506: splx(s); ! 507: } ! 508: ! 509: imq_unlock(mqueue); ! 510: current_task()->messages_sent++; ! 511: return; ! 512: } ! 513: ! 514: ! 515: /* ! 516: * Routine: ipc_mqueue_receive ! 517: * Purpose: ! 518: * Receive a message from a message queue. ! 519: * ! 520: * If continuation is non-zero, then we might discard ! 521: * our kernel stack when we block. We will continue ! 522: * after unblocking by executing continuation. ! 523: * ! 524: * If resume is true, then we are resuming a receive ! 525: * operation after a blocked receive discarded our stack. ! 526: * Conditions: ! 527: * Our caller must hold a reference for the port or port set ! 528: * to which this queue belongs, to keep the queue ! 529: * from being deallocated. ! 530: * ! 531: * The kmsg is returned with clean header fields ! 532: * and with the circular bit turned off. ! 533: * Returns: ! 534: * MACH_MSG_SUCCESS Message returned in kmsgp. ! 535: * MACH_RCV_TOO_LARGE Message size returned in kmsgp. ! 536: * MACH_RCV_TIMED_OUT No message obtained. ! 537: * MACH_RCV_INTERRUPTED No message obtained. ! 538: * MACH_RCV_PORT_DIED Port/set died; no message. ! 539: * MACH_RCV_PORT_CHANGED Port moved into set; no msg. ! 540: * ! 541: */ ! 542: ! 543: mach_msg_return_t ! 544: ipc_mqueue_receive( ! 545: ipc_mqueue_t mqueue, ! 546: mach_msg_option_t option, ! 547: mach_msg_size_t max_size, ! 548: mach_msg_timeout_t timeout, ! 549: int interruptible, ! 550: ipc_kmsg_t *kmsgp, ! 551: mach_port_seqno_t *seqnop) ! 552: { ! 553: ipc_port_t port; ! 554: mach_msg_return_t mr, mr2; ! 555: ipc_kmsg_queue_t kmsgs; ! 556: kern_return_t save_wait_result; ! 557: thread_t self; ! 558: ! 559: imq_lock(mqueue); ! 560: ! 561: if (imq_is_set(mqueue)) { ! 562: wait_queue_link_t wql; ! 563: ipc_mqueue_t port_mq; ! 564: queue_t q; ! 565: ! 566: q = &mqueue->imq_setlinks; ! 567: ! 568: /* ! 569: * If we are waiting on a portset mqueue, we need to see if ! 570: * any of the member ports have work for us. If so, try to ! 571: * deliver one of those messages. By holding the portset's ! 572: * mqueue lock during the search, we tie up any attempts by ! 573: * mqueue_deliver or portset membership changes that may ! 574: * cross our path. But this is a lock order violation, so we ! 575: * have to do it "softly." If we don't find a message waiting ! 576: * for us, we will assert our intention to wait while still ! 577: * holding that lock. When we release the lock, the deliver/ ! 578: * change will succeed and find us. ! 579: */ ! 580: search_set: ! 581: queue_iterate(q, wql, wait_queue_link_t, wql_sublinks) { ! 582: port_mq = (ipc_mqueue_t)wql->wql_queue; ! 583: kmsgs = &port_mq->imq_messages; ! 584: ! 585: if (!ipc_kmsg_queue_first(kmsgs)) /* peek first */ ! 586: continue; ! 587: ! 588: if (!imq_lock_try(port_mq)) { ! 589: imq_unlock(mqueue); ! 590: imq_lock(mqueue); ! 591: goto search_set; /* start again at beginning - SMP */ ! 592: } ! 593: ! 594: /* ! 595: * If there is still a message to be had, we will ! 596: * try to select it (may not succeed because of size ! 597: * and options). In any case, we deliver those ! 598: * results back to the user. ! 599: * ! 600: * We also move the port's linkage to the tail of the ! 601: * list for this set (fairness). Future versions will ! 602: * sort by timestamp or priority. ! 603: */ ! 604: if (ipc_kmsg_queue_first(kmsgs) == IKM_NULL) { ! 605: imq_unlock(port_mq); ! 606: continue; ! 607: } ! 608: queue_remove(q, wql, wait_queue_link_t, wql_sublinks); ! 609: queue_enter(q, wql, wait_queue_link_t, wql_sublinks); ! 610: imq_unlock(mqueue); ! 611: ! 612: mr = ipc_mqueue_select(port_mq, option, max_size, kmsgp, seqnop); ! 613: imq_unlock(port_mq); ! 614: return mr; ! 615: ! 616: } ! 617: ! 618: } else { ! 619: ! 620: /* ! 621: * Receive on a single port. Just try to get the messages. ! 622: */ ! 623: kmsgs = &mqueue->imq_messages; ! 624: if (ipc_kmsg_queue_first(kmsgs) != IKM_NULL) { ! 625: mr = ipc_mqueue_select(mqueue, option, max_size, kmsgp, seqnop); ! 626: imq_unlock(mqueue); ! 627: return mr; ! 628: } ! 629: } ! 630: ! 631: /* ! 632: * Looks like we'll have to block. The mqueue we will ! 633: * block on (whether the set's or the local port's) is ! 634: * still locked. ! 635: */ ! 636: if (option & MACH_RCV_TIMEOUT) { ! 637: if (timeout == 0) { ! 638: imq_unlock(mqueue); ! 639: return MACH_RCV_TIMED_OUT; ! 640: } ! 641: } ! 642: self = current_thread(); ! 643: ! 644: self->ith_state = MACH_RCV_IN_PROGRESS; ! 645: self->ith_option = option; ! 646: self->ith_msize = max_size; ! 647: ! 648: wait_queue_assert_wait_locked(&mqueue->imq_wait_queue, ! 649: IPC_MQUEUE_RECEIVE, ! 650: interruptible, ! 651: TRUE); /* unlock? */ ! 652: /* mqueue/waitq is unlocked */ ! 653: ! 654: if (option & MACH_RCV_TIMEOUT) { ! 655: thread_set_timer(timeout, 1000*NSEC_PER_USEC); ! 656: } ! 657: ! 658: if (interruptible == THREAD_ABORTSAFE) { ! 659: counter(c_ipc_mqueue_receive_block_user++); ! 660: } else { ! 661: counter(c_ipc_mqueue_receive_block_kernel++); ! 662: } ! 663: ! 664: save_wait_result = thread_block((void (*)(void))0); ! 665: ! 666: if (option & MACH_RCV_TIMEOUT) ! 667: thread_cancel_timer(); ! 668: ! 669: /* ! 670: * why did we wake up? ! 671: */ ! 672: switch (save_wait_result) { ! 673: case THREAD_TIMED_OUT: ! 674: return MACH_RCV_TIMED_OUT; ! 675: ! 676: case THREAD_INTERRUPTED: ! 677: return MACH_RCV_INTERRUPTED; ! 678: ! 679: case THREAD_RESTART: ! 680: /* something bad happened to the port/set */ ! 681: return MACH_RCV_PORT_CHANGED; ! 682: ! 683: case THREAD_AWAKENED: ! 684: /* ! 685: * We do not need to go select a message, somebody ! 686: * handed us one (or a too-large indication). ! 687: */ ! 688: ! 689: mr = MACH_MSG_SUCCESS; ! 690: ! 691: switch (self->ith_state) { ! 692: case MACH_RCV_SCATTER_SMALL: ! 693: case MACH_RCV_TOO_LARGE: ! 694: /* ! 695: * Somebody tried to give us a too large ! 696: * message. If we indicated that we cared, ! 697: * then they only gave us the indication, ! 698: * otherwise they gave us the indication ! 699: * AND the message anyway. ! 700: */ ! 701: if (option & MACH_RCV_LARGE) { ! 702: if (self->ith_state == MACH_RCV_TOO_LARGE) ! 703: *kmsgp = (ipc_kmsg_t)self->ith_msize; ! 704: return(self->ith_state); ! 705: } else { ! 706: mr = self->ith_state; ! 707: /* fall thru to SUCCESS case */ ! 708: } ! 709: ! 710: case MACH_MSG_SUCCESS: ! 711: /* pick up the message that was handed to us */ ! 712: *kmsgp = self->ith_kmsg; ! 713: *seqnop = self->ith_seqno; ! 714: return mr; ! 715: ! 716: default: ! 717: panic("ipc_mqueue_receive: strange ith_state"); ! 718: } ! 719: ! 720: default: ! 721: panic("ipc_mqueue_receive: strange wait_result"); ! 722: } ! 723: } ! 724: ! 725: ! 726: /* ! 727: * Routine: ipc_mqueue_select ! 728: * Purpose: ! 729: * A receiver discovered that there was a message on the queue ! 730: * before he had to block. Pick the message off the queue and ! 731: * "post" it to himself. ! 732: * Conditions: ! 733: * mqueue locked. ! 734: * There is a message. ! 735: * Returns: ! 736: * MACH_MSG_SUCCESS Actually selected a message for ourselves. ! 737: * MACH_RCV_TOO_LARGE May or may not have pull it, but it is large ! 738: */ ! 739: kern_return_t ! 740: ipc_mqueue_select( ! 741: ipc_mqueue_t mqueue, ! 742: mach_msg_option_t option, ! 743: mach_msg_size_t max_size, ! 744: ipc_kmsg_t *kmsgp, ! 745: mach_port_seqno_t *seqnop) ! 746: { ! 747: ipc_kmsg_t kmsg; ! 748: mach_port_seqno_t seqno; ! 749: mach_msg_return_t mr; ! 750: ! 751: mr = MACH_MSG_SUCCESS; ! 752: ! 753: ! 754: /* ! 755: * Do some sanity checking of our ability to receive ! 756: * before pulling the message off the queue. ! 757: */ ! 758: kmsg = ipc_kmsg_queue_first(&mqueue->imq_messages); ! 759: ! 760: assert(kmsg != IKM_NULL); ! 761: ! 762: if (kmsg->ikm_header.msgh_size + ! 763: REQUESTED_TRAILER_SIZE(option) > max_size) { ! 764: mr = MACH_RCV_TOO_LARGE; ! 765: } ! 766: #ifdef DOITFORSCATTER ! 767: else if (self->ith_scatter_list != MACH_MSG_BODY_NULL) { ! 768: mr = ipc_kmsg_check_scatter( ! 769: kmsg, ! 770: self->ith_option, ! 771: &self->ith_scatter_list, ! 772: &self->ith_scatter_list_size); ! 773: } ! 774: #endif /* DOITFORSCATTER */ ! 775: ! 776: /* ! 777: * If we really can't receive it, but we had the ! 778: * MACH_RCV_LARGE option set, then don't take it off ! 779: * the queue, instead return the appropriate error ! 780: * (and size needed). ! 781: */ ! 782: if (((mr == MACH_RCV_TOO_LARGE) || ! 783: (mr == MACH_RCV_SCATTER_SMALL)) && ! 784: (option & MACH_RCV_LARGE)) { ! 785: *seqnop = 0; ! 786: *kmsgp = (ipc_kmsg_t)kmsg->ikm_header.msgh_size; ! 787: return mr; ! 788: } ! 789: ! 790: ipc_kmsg_rmqueue_first_macro(&mqueue->imq_messages, kmsg); ! 791: ipc_mqueue_release_msgcount(mqueue); ! 792: *seqnop = mqueue->imq_seqno++; ! 793: *kmsgp = kmsg; ! 794: ! 795: current_task()->messages_received++; ! 796: return mr; ! 797: } ! 798: ! 799: /* ! 800: * Routine: ipc_mqueue_destroy ! 801: * Purpose: ! 802: * Destroy a message queue. Set any blocked senders running. ! 803: * Destroy the kmsgs in the queue. ! 804: * Conditions: ! 805: * Nothing locked. ! 806: * Receivers were removed when the receive right was "changed" ! 807: */ ! 808: void ! 809: ipc_mqueue_destroy( ! 810: ipc_mqueue_t mqueue) ! 811: { ! 812: ipc_kmsg_queue_t kmqueue; ! 813: ipc_kmsg_t kmsg; ! 814: ! 815: imq_lock(mqueue); ! 816: /* ! 817: * rouse all blocked senders ! 818: */ ! 819: mqueue->imq_fullwaiters = FALSE; ! 820: wait_queue_wakeup_all_locked(&mqueue->imq_wait_queue, ! 821: IPC_MQUEUE_FULL, ! 822: THREAD_AWAKENED, ! 823: FALSE); ! 824: ! 825: kmqueue = &mqueue->imq_messages; ! 826: ! 827: while ((kmsg = ipc_kmsg_dequeue(kmqueue)) != IKM_NULL) { ! 828: imq_unlock(mqueue); ! 829: ipc_kmsg_destroy_dest(kmsg); ! 830: imq_lock(mqueue); ! 831: } ! 832: imq_unlock(mqueue); ! 833: } ! 834: ! 835: /* ! 836: * Routine: ipc_mqueue_set_qlimit ! 837: * Purpose: ! 838: * Changes a message queue limit; the maximum number ! 839: * of messages which may be queued. ! 840: * Conditions: ! 841: * Nothing locked. ! 842: */ ! 843: ! 844: void ! 845: ipc_mqueue_set_qlimit( ! 846: ipc_mqueue_t mqueue, ! 847: mach_port_msgcount_t qlimit) ! 848: { ! 849: /* wake up senders allowed by the new qlimit */ ! 850: imq_lock(mqueue); ! 851: if (qlimit > mqueue->imq_qlimit) { ! 852: mach_port_msgcount_t i, wakeup; ! 853: ! 854: /* caution: wakeup, qlimit are unsigned */ ! 855: wakeup = qlimit - mqueue->imq_qlimit; ! 856: ! 857: for (i = 0; i < wakeup; i++) { ! 858: if (wait_queue_wakeup_one_locked(&mqueue->imq_wait_queue, ! 859: IPC_MQUEUE_FULL, ! 860: THREAD_AWAKENED, ! 861: FALSE) == KERN_NOT_WAITING) { ! 862: mqueue->imq_fullwaiters = FALSE; ! 863: break; ! 864: } ! 865: } ! 866: } ! 867: mqueue->imq_qlimit = qlimit; ! 868: imq_unlock(mqueue); ! 869: } ! 870: ! 871: /* ! 872: * Routine: ipc_mqueue_set_seqno ! 873: * Purpose: ! 874: * Changes an mqueue's sequence number. ! 875: * Conditions: ! 876: * Caller holds a reference to the queue's containing object. ! 877: */ ! 878: void ! 879: ipc_mqueue_set_seqno( ! 880: ipc_mqueue_t mqueue, ! 881: mach_port_seqno_t seqno) ! 882: { ! 883: imq_lock(mqueue); ! 884: mqueue->imq_seqno = seqno; ! 885: imq_unlock(mqueue); ! 886: } ! 887: ! 888: ! 889: /* ! 890: * Routine: ipc_mqueue_copyin ! 891: * Purpose: ! 892: * Convert a name in a space to a message queue. ! 893: * Conditions: ! 894: * Nothing locked. If successful, the caller gets a ref for ! 895: * for the object. This ref ensures the continued existence of ! 896: * the queue. ! 897: * Returns: ! 898: * MACH_MSG_SUCCESS Found a message queue. ! 899: * MACH_RCV_INVALID_NAME The space is dead. ! 900: * MACH_RCV_INVALID_NAME The name doesn't denote a right. ! 901: * MACH_RCV_INVALID_NAME ! 902: * The denoted right is not receive or port set. ! 903: * MACH_RCV_IN_SET Receive right is a member of a set. ! 904: */ ! 905: ! 906: mach_msg_return_t ! 907: ipc_mqueue_copyin( ! 908: ipc_space_t space, ! 909: mach_port_name_t name, ! 910: ipc_mqueue_t *mqueuep, ! 911: ipc_object_t *objectp) ! 912: { ! 913: ipc_entry_t entry; ! 914: ipc_object_t object; ! 915: ipc_mqueue_t mqueue; ! 916: ! 917: is_read_lock(space); ! 918: if (!space->is_active) { ! 919: is_read_unlock(space); ! 920: return MACH_RCV_INVALID_NAME; ! 921: } ! 922: ! 923: entry = ipc_entry_lookup(space, name); ! 924: if (entry == IE_NULL) { ! 925: is_read_unlock(space); ! 926: return MACH_RCV_INVALID_NAME; ! 927: } ! 928: ! 929: object = entry->ie_object; ! 930: ! 931: if (entry->ie_bits & MACH_PORT_TYPE_RECEIVE) { ! 932: ipc_port_t port; ! 933: ipc_pset_t pset; ! 934: ! 935: port = (ipc_port_t) object; ! 936: assert(port != IP_NULL); ! 937: ! 938: ip_lock(port); ! 939: assert(ip_active(port)); ! 940: assert(port->ip_receiver_name == name); ! 941: assert(port->ip_receiver == space); ! 942: is_read_unlock(space); ! 943: mqueue = &port->ip_messages; ! 944: ! 945: } else if (entry->ie_bits & MACH_PORT_TYPE_PORT_SET) { ! 946: ipc_pset_t pset; ! 947: ! 948: pset = (ipc_pset_t) object; ! 949: assert(pset != IPS_NULL); ! 950: ! 951: ips_lock(pset); ! 952: assert(ips_active(pset)); ! 953: assert(pset->ips_local_name == name); ! 954: is_read_unlock(space); ! 955: ! 956: mqueue = &pset->ips_messages; ! 957: } else { ! 958: is_read_unlock(space); ! 959: return MACH_RCV_INVALID_NAME; ! 960: } ! 961: ! 962: /* ! 963: * At this point, the object is locked and active, ! 964: * the space is unlocked, and mqueue is initialized. ! 965: */ ! 966: ! 967: io_reference(object); ! 968: io_unlock(object); ! 969: ! 970: *objectp = object; ! 971: *mqueuep = mqueue; ! 972: return MACH_MSG_SUCCESS; ! 973: } ! 974:
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.