![[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 Darwin 0.x] / kernel / ipc|
Return to ipc_mqueue.c CVS log | Up to [Apple Darwin 0.x] / kernel / ipc |
1.1 root 1: /*
2: * Copyright (c) 1999 Apple Computer, Inc. All rights reserved.
3: *
4: * @APPLE_LICENSE_HEADER_START@
5: *
6: * Portions Copyright (c) 1999 Apple Computer, Inc. All Rights
7: * Reserved. This file contains Original Code and/or Modifications of
8: * Original Code as defined in and that are subject to the Apple Public
9: * Source License Version 1.1 (the "License"). You may not use this file
10: * except in compliance with the License. Please obtain a copy of the
11: * License at http://www.apple.com/publicsource and read it before using
12: * this file.
13: *
14: * The Original Code and all software distributed under the License are
15: * distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, EITHER
16: * EXPRESS OR IMPLIED, AND APPLE HEREBY DISCLAIMS ALL SUCH WARRANTIES,
17: * INCLUDING WITHOUT LIMITATION, ANY WARRANTIES OF MERCHANTABILITY,
18: * FITNESS FOR A PARTICULAR PURPOSE OR NON-INFRINGEMENT. Please see the
19: * License for the specific language governing rights and limitations
20: * under the License.
21: *
22: * @APPLE_LICENSE_HEADER_END@
23: */
24:
25: /*
26: * Copyright (c) 1995, 1994, 1993, 1992, 1991, 1990
27: * Open Software Foundation, Inc.
28: *
29: * Permission to use, copy, modify, and distribute this software and
30: * its documentation for any purpose and without fee is hereby granted,
31: * provided that the above copyright notice appears in all copies and
32: * that both the copyright notice and this permission notice appear in
33: * supporting documentation, and that the name of ("OSF") or Open Software
34: * Foundation not be used in advertising or publicity pertaining to
35: * distribution of the software without specific, written prior permission.
36: *
37: * OSF DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE
38: * INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
39: * FOR A PARTICULAR PURPOSE. IN NO EVENT SHALL OSF BE LIABLE FOR ANY
40: * SPECIAL, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
41: * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN
42: * ACTION OF CONTRACT, NEGLIGENCE, OR OTHER TORTIOUS ACTION, ARISING
43: * OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE
44: */
45: /*
46: * OSF Research Institute MK6.1 (unencumbered) 1/31/1995
47: */
48: /*
49: * Mach Operating System
50: * Copyright (c) 1991,1990,1989 Carnegie Mellon University
51: * All Rights Reserved.
52: *
53: * Permission to use, copy, modify and distribute this software and its
54: * documentation is hereby granted, provided that both the copyright
55: * notice and this permission notice appear in all copies of the
56: * software, derivative works or modified versions, and any portions
57: * thereof, and that both notices appear in supporting documentation.
58: *
59: * CARNEGIE MELLON ALLOWS FREE USE OF THIS SOFTWARE IN ITS "AS IS"
60: * CONDITION. CARNEGIE MELLON DISCLAIMS ANY LIABILITY OF ANY KIND FOR
61: * ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF THIS SOFTWARE.
62: *
63: * Carnegie Mellon requests users of this software to return to
64: *
65: * Software Distribution Coordinator or [email protected]
66: * School of Computer Science
67: * Carnegie Mellon University
68: * Pittsburgh PA 15213-3890
69: *
70: * any improvements or extensions that they make and grant Carnegie Mellon
71: * the rights to redistribute these changes.
72: */
73: /*
74: * File: ipc/ipc_mqueue.c
75: * Author: Rich Draves
76: * Date: 1989
77: *
78: * Functions to manipulate IPC message queues.
79: */
80:
81: #include <mach/port.h>
82: #include <mach/message.h>
83: #include <kern/assert.h>
84: #include <kern/counters.h>
85: #include <kern/sched_prim.h>
86: #include <kern/ipc_sched.h>
87: #include <kern/ipc_kobject.h>
88: #include <ipc/ipc_mqueue.h>
89: #include <ipc/ipc_thread.h>
90: #include <ipc/ipc_kmsg.h>
91: #include <ipc/ipc_port.h>
92: #include <ipc/ipc_pset.h>
93: #include <ipc/ipc_space.h>
94: #include <ipc/ipc_marequest.h>
95:
96: /*
97: * Routine: ipc_mqueue_init
98: * Purpose:
99: * Initialize a newly-allocated message queue.
100: */
101:
102: void
103: ipc_mqueue_init(
104: ipc_mqueue_t mqueue)
105: {
106: imq_lock_init(mqueue);
107: ipc_kmsg_queue_init(&mqueue->imq_messages);
108: ipc_thread_queue_init(&mqueue->imq_threads);
109: }
110:
111: /*
112: * Routine: ipc_mqueue_move
113: * Purpose:
114: * Move messages from one queue (source) to another (dest).
115: * Only moves messages sent to the specified port.
116: * Conditions:
117: * Both queues must be locked.
118: * (This is sufficient to manipulate port->ip_seqno.)
119: */
120:
121: void
122: ipc_mqueue_move(
123: ipc_mqueue_t dest,
124: ipc_mqueue_t source,
125: ipc_port_t port)
126: {
127: ipc_kmsg_queue_t oldq, newq;
128: ipc_thread_queue_t blockedq;
129: ipc_kmsg_t kmsg, next;
130: ipc_thread_t th;
131:
132: oldq = &source->imq_messages;
133: newq = &dest->imq_messages;
134: blockedq = &dest->imq_threads;
135:
136: for (kmsg = ipc_kmsg_queue_first(oldq);
137: kmsg != IKM_NULL; kmsg = next) {
138: next = ipc_kmsg_queue_next(oldq, kmsg);
139:
140: /* only move messages sent to port */
141:
142: if (kmsg->ikm_header.msgh_remote_port != (mach_port_t) port)
143: continue;
144:
145: ipc_kmsg_rmqueue(oldq, kmsg);
146:
147: /* before adding kmsg to newq, check for a blocked receiver */
148:
149: if (kmsg->ikm_header.msgh_bits & MACH_MSGH_BITS_OLD_FORMAT) {
150: while ((th = ipc_thread_dequeue(blockedq)) != ITH_NULL) {
151: assert(ipc_kmsg_queue_empty(newq));
152:
153: /* check if the receiver can handle the message */
154:
155: if (!(th->ith_rcv_option & MACH_RCV_OLD_FORMAT)) {
156: th->ith_state = MACH_RCV_HEADER_ERROR;
157: th->ith_kmsg = kmsg;
158: th->ith_seqno = port->ip_seqno++;
159: thread_go(th);
160:
161: goto next_kmsg;
162: }
163: else
164: if (kmsg->ikm_header.msgh_size <= th->ith_msize) {
165: th->ith_state = MACH_MSG_SUCCESS;
166: th->ith_kmsg = kmsg;
167: th->ith_seqno = port->ip_seqno++;
168: thread_go(th);
169:
170: goto next_kmsg;
171: }
172: else {
173: th->ith_state = MACH_RCV_TOO_LARGE;
174: th->ith_msize = kmsg->ikm_header.msgh_size;
175: }
176:
177: thread_go(th);
178: }
179: }
180: else {
181: boolean_t msg_is_complex =
182: (kmsg->ikm_header.msgh_bits &
183: MACH_MSGH_BITS_COMPLEX) != 0;
184:
185: while ((th = ipc_thread_dequeue(blockedq)) != ITH_NULL) {
186: assert(ipc_kmsg_queue_empty(newq));
187:
188: /*
189: * Check if the receiver can handle the message
190: */
191: if (th->ith_rcv_option & MACH_RCV_OLD_FORMAT) {
192: if (msg_is_complex ||
193: kmsg->ikm_header.msgh_size >
194: sizeof (mach_msg_header_t) ||
195: th->ith_msize <
196: sizeof (mach_msg_header_t)) {
197: th->ith_state = MACH_RCV_HEADER_ERROR;
198: th->ith_kmsg = kmsg;
199: th->ith_seqno = port->ip_seqno++;
200: thread_go(th);
201:
202: goto next_kmsg;
203: }
204:
205: /*
206: * As a special case, allow simple
207: * messages which consist of only a
208: * header to be received with the old
209: * messaging primatives.
210: */
211: th->ith_state = MACH_MSG_SUCCESS;
212: }
213: else
214: if (kmsg->ikm_header.msgh_size > (th->ith_msize -
215: REQUESTED_TRAILER_SIZE(th->ith_rcv_option))) {
216: /*
217: * The receive buffer isn't large enough
218: */
219: th->ith_state = MACH_RCV_TOO_LARGE;
220:
221: if (th->ith_rcv_option & MACH_RCV_LARGE) {
222: th->ith_msize = kmsg->ikm_header.msgh_size;
223: thread_go(th);
224: continue;
225: }
226: }
227: else
228: if (msg_is_complex && th->ith_list != IKM_NULL &&
229: (th->ith_state =
230: ipc_kmsg_check_scatter(
231: kmsg, th->ith_rcv_option,
232: &th->ith_list))
233: != MACH_MSG_SUCCESS) {
234: /*
235: * Check for type/VM errors.
236: */
237: if (th->ith_state == MACH_RCV_INVALID_TYPE ||
238: (th->ith_state & MACH_MSG_VM_KERNEL) != 0) {
239: thread_go(th);
240: continue;
241: }
242:
243: /*
244: * The scatter list isn't large enough.
245: */
246: if (th->ith_rcv_option & MACH_RCV_LARGE) {
247: th->ith_msize = kmsg->ikm_header.msgh_size;
248: thread_go(th);
249: continue;
250: }
251: }
252: else
253: th->ith_state = MACH_MSG_SUCCESS;
254:
255: /*
256: * Got a receiver. Possibly with a size error.
257: * Hand the receiver the message and process the
258: * next one.
259: */
260: th->ith_kmsg = kmsg;
261: th->ith_seqno = port->ip_seqno++;
262: thread_go(th);
263: goto next_kmsg;
264: }
265: }
266:
267: /* didn't find a receiver to handle the message */
268:
269: ipc_kmsg_enqueue(newq, kmsg);
270: next_kmsg:;
271: }
272: }
273:
274: /*
275: * Routine: ipc_mqueue_changed
276: * Purpose:
277: * Wake up receivers waiting in a message queue.
278: * Conditions:
279: * The message queue is locked.
280: */
281:
282: void
283: ipc_mqueue_changed(
284: ipc_mqueue_t mqueue,
285: mach_msg_return_t mr)
286: {
287: ipc_thread_t th;
288:
289: while ((th = ipc_thread_dequeue(&mqueue->imq_threads)) != ITH_NULL) {
290: th->ith_state = mr;
291: thread_go(th);
292: }
293: }
294:
295: /*
296: * Routine: ipc_mqueue_send
297: * Purpose:
298: * Send a message to a port. The message holds a reference
299: * for the destination port in the msgh_remote_port field.
300: *
301: * If unsuccessful, the caller still has possession of
302: * the message and must do something with it. If successful,
303: * the message is queued, given to a receiver, destroyed,
304: * or handled directly by the kernel via mach_msg.
305: * Conditions:
306: * Nothing locked.
307: * Returns:
308: * MACH_MSG_SUCCESS The message was accepted.
309: * MACH_SEND_TIMED_OUT Caller still has message.
310: * MACH_SEND_INTERRUPTED Caller still has message.
311: */
312:
313: mach_msg_return_t
314: ipc_mqueue_send(
315: ipc_kmsg_t kmsg,
316: mach_msg_option_t option,
317: mach_msg_timeout_t timeout,
318: void (*continuation)(void))
319: {
320: ipc_port_t port;
321:
322: port = (ipc_port_t) kmsg->ikm_header.msgh_remote_port;
323: assert(IP_VALID(port));
324:
325: ip_lock(port);
326:
327: if (port->ip_receiver == ipc_space_kernel) {
328: ipc_kmsg_t reply;
329:
330: /*
331: * We can check ip_receiver == ipc_space_kernel
332: * before checking that the port is active because
333: * ipc_port_dealloc_kernel clears ip_receiver
334: * before destroying a kernel port.
335: */
336:
337: assert(ip_active(port));
338: ip_unlock(port);
339:
340: reply = ipc_kobject_server(kmsg);
341: if (reply != IKM_NULL)
342: ipc_mqueue_send_always(reply);
343:
344: return MACH_MSG_SUCCESS;
345: }
346:
347: for (;;) {
348: ipc_thread_t self;
349:
350: /*
351: * Can't deliver to a dead port.
352: * However, we can pretend it got sent
353: * and was then immediately destroyed.
354: */
355:
356: if (!ip_active(port)) {
357: /*
358: * We can't let ipc_kmsg_destroy deallocate
359: * the port right, because we might end up
360: * in an infinite loop trying to deliver
361: * a send-once notification.
362: */
363:
364: ip_release(port);
365: ip_check_unlock(port);
366: kmsg->ikm_header.msgh_remote_port = MACH_PORT_NULL;
367: ipc_kmsg_destroy(kmsg);
368: return MACH_MSG_SUCCESS;
369: }
370:
371: /*
372: * Don't block if:
373: * 1) We're under the queue limit.
374: * 2) Caller used the MACH_SEND_ALWAYS internal option.
375: * 3) Message is sent to a send-once right.
376: */
377:
378: if ((port->ip_msgcount < port->ip_qlimit) ||
379: (option & MACH_SEND_ALWAYS) ||
380: (MACH_MSGH_BITS_REMOTE(kmsg->ikm_header.msgh_bits) ==
381: MACH_MSG_TYPE_PORT_SEND_ONCE))
382: break;
383:
384: /* must block waiting for queue to clear */
385:
386: self = current_thread();
387:
388: if (option & MACH_SEND_TIMEOUT) {
389: if (timeout == 0) {
390: ip_unlock(port);
391: return MACH_SEND_TIMED_OUT;
392: }
393:
394: thread_will_wait_with_timeout(self, timeout);
395: } else
396: thread_will_wait(self);
397:
398: ipc_thread_enqueue(&port->ip_blocked, self);
399: self->ith_state = MACH_SEND_IN_PROGRESS;
400:
401: ip_unlock(port);
402: counter(c_ipc_mqueue_send_block++);
403: thread_block_with_continuation((void (*)(void)) 0);
404: ip_lock(port);
405:
406: /* why did we wake up? */
407:
408: if (self->ith_state == MACH_MSG_SUCCESS)
409: continue;
410: assert(self->ith_state == MACH_SEND_IN_PROGRESS);
411:
412: /* take ourselves off blocked queue */
413:
414: ipc_thread_rmqueue(&port->ip_blocked, self);
415:
416: /*
417: * Thread wakeup-reason field tells us why
418: * the wait was interrupted.
419: */
420:
421: switch (self->ith_wait_result) {
422: case THREAD_INTERRUPTED:
423: case THREAD_SHOULD_TERMINATE:
424: /* send was interrupted - give up */
425:
426: ip_unlock(port);
427: return MACH_SEND_INTERRUPTED;
428:
429: case THREAD_TIMED_OUT:
430: /* timeout expired */
431:
432: assert(option & MACH_SEND_TIMEOUT);
433: timeout = 0;
434: break;
435:
436: case THREAD_RESTART:
437: default:
438: panic("ipc_mqueue_send");
439: }
440: }
441:
442: if (kmsg->ikm_header.msgh_bits & MACH_MSGH_BITS_CIRCULAR) {
443: ip_unlock(port);
444:
445: /* don't allow the creation of a circular loop */
446:
447: ipc_kmsg_destroy(kmsg);
448: return MACH_MSG_SUCCESS;
449: }
450:
451: {
452: ipc_mqueue_t mqueue;
453: ipc_pset_t pset;
454: ipc_thread_t receiver;
455: ipc_thread_queue_t receivers;
456:
457: port->ip_msgcount++;
458: assert(port->ip_msgcount > 0);
459:
460: pset = port->ip_pset;
461: if (pset == IPS_NULL)
462: mqueue = &port->ip_messages;
463: else
464: mqueue = &pset->ips_messages;
465:
466: imq_lock(mqueue);
467: receivers = &mqueue->imq_threads;
468:
469: /*
470: * Can unlock the port now that the msg queue is locked
471: * and we know the port is active. While the msg queue
472: * is locked, we have control of the kmsg, so the ref in
473: * it for the port is still good. If the msg queue is in
474: * a set (dead or alive), then we're OK because the port
475: * is still a member of the set and the set won't go away
476: * until the port is taken out, which tries to lock the
477: * set's msg queue to remove the port's msgs.
478: */
479:
480: ip_unlock(port);
481:
482: /* check for a receiver for the message */
483:
484: if (kmsg->ikm_header.msgh_bits & MACH_MSGH_BITS_OLD_FORMAT) {
485: for (;;) {
486: receiver = ipc_thread_queue_first(receivers);
487: if (receiver == ITH_NULL) {
488: /* no receivers; queue kmsg */
489:
490: ipc_kmsg_enqueue_macro(&mqueue->imq_messages, kmsg);
491: imq_unlock(mqueue);
492: break;
493: }
494:
495: ipc_thread_rmqueue_first_macro(receivers, receiver);
496: assert(ipc_kmsg_queue_empty(&mqueue->imq_messages));
497:
498: if (!(receiver->ith_rcv_option & MACH_RCV_OLD_FORMAT)) {
499: receiver->ith_state = MACH_RCV_HEADER_ERROR;
500: receiver->ith_kmsg = kmsg;
501: receiver->ith_seqno = port->ip_seqno++;
502: imq_unlock(mqueue);
503:
504: thread_go(receiver);
505: break;
506: }
507: else
508: if (kmsg->ikm_header.msgh_size <= receiver->ith_msize) {
509: /* got a successful receiver */
510:
511: receiver->ith_state = MACH_MSG_SUCCESS;
512: receiver->ith_kmsg = kmsg;
513: receiver->ith_seqno = port->ip_seqno++;
514: imq_unlock(mqueue);
515:
516: if (option & MACH_SEND_SWITCH)
517: thread_go_and_switch(continuation, receiver);
518: else
519: thread_go(receiver);
520: break;
521: }
522: else {
523: receiver->ith_state = MACH_RCV_TOO_LARGE;
524: receiver->ith_msize = kmsg->ikm_header.msgh_size;
525: }
526:
527: thread_go(receiver);
528: }
529: }
530: else {
531: boolean_t msg_is_complex =
532: (kmsg->ikm_header.msgh_bits &
533: MACH_MSGH_BITS_COMPLEX) != 0;
534:
535: for (;;) {
536: receiver = ipc_thread_queue_first(receivers);
537: if (receiver == ITH_NULL) {
538: /* no receivers; queue kmsg */
539:
540: ipc_kmsg_enqueue_macro(&mqueue->imq_messages, kmsg);
541: imq_unlock(mqueue);
542: break;
543: }
544:
545: ipc_thread_rmqueue_first_macro(receivers, receiver);
546: assert(ipc_kmsg_queue_empty(&mqueue->imq_messages));
547:
548: /*
549: * Got a potential receiver, verify sizes
550: */
551: if (receiver->ith_rcv_option & MACH_RCV_OLD_FORMAT) {
552: if (msg_is_complex ||
553: kmsg->ikm_header.msgh_size >
554: sizeof (mach_msg_header_t) ||
555: receiver->ith_msize <
556: sizeof (mach_msg_header_t)) {
557: receiver->ith_state = MACH_RCV_HEADER_ERROR;
558: receiver->ith_kmsg = kmsg;
559: receiver->ith_seqno = port->ip_seqno++;
560: imq_unlock(mqueue);
561: thread_go(receiver);
562: break;
563: }
564:
565: receiver->ith_state = MACH_MSG_SUCCESS;
566: }
567: else
568: if (kmsg->ikm_header.msgh_size > (receiver->ith_msize -
569: REQUESTED_TRAILER_SIZE(receiver->ith_rcv_option))) {
570: /*
571: * The receive buffer isn't large enough
572: */
573: receiver->ith_state = MACH_RCV_TOO_LARGE;
574:
575: if (receiver->ith_rcv_option & MACH_RCV_LARGE) {
576: receiver->ith_msize = kmsg->ikm_header.msgh_size;
577: thread_go(receiver);
578: continue;
579: }
580: }
581: else
582: if (msg_is_complex && receiver->ith_list != IKM_NULL &&
583: (receiver->ith_state =
584: ipc_kmsg_check_scatter(
585: kmsg, receiver->ith_rcv_option,
586: &receiver->ith_list))
587: != MACH_MSG_SUCCESS) {
588: /*
589: * Check for type/VM errors.
590: */
591: if (receiver->ith_state == MACH_RCV_INVALID_TYPE ||
592: (receiver->ith_state & MACH_MSG_VM_KERNEL) != 0) {
593: thread_go(receiver);
594: continue;
595: }
596:
597: /*
598: * The scatter list isn't large enough.
599: */
600: if (receiver->ith_rcv_option & MACH_RCV_LARGE) {
601: receiver->ith_msize = kmsg->ikm_header.msgh_size;
602: thread_go(receiver);
603: continue;
604: }
605: }
606: else
607: receiver->ith_state = MACH_MSG_SUCCESS;
608:
609: /*
610: * Got a valid receiver. Possibly with an error.
611: * Hand the receiver the message.
612: */
613: receiver->ith_kmsg = kmsg;
614: receiver->ith_seqno = port->ip_seqno++;
615: imq_unlock(mqueue);
616: thread_go(receiver);
617: break;
618: }
619: }
620: }
621:
622: return MACH_MSG_SUCCESS;
623: }
624:
625: mach_msg_return_t
626: ipc_mqueue_send_interrupt(kmsg)
627: ipc_kmsg_t kmsg;
628: {
629: ipc_port_t port;
630:
631: port = (ipc_port_t) kmsg->ikm_header.msgh_remote_port;
632:
633: if (!ip_lock_try(port))
634: return MACH_MSG_IPC_KERNEL;
635:
636: if (!ip_active(port)) {
637: ip_unlock(port);
638: return MACH_SEND_INVALID_DEST;
639: }
640:
641: assert(!(kmsg->ikm_header.msgh_bits &
642: (MACH_MSGH_BITS_COMPLEX|MACH_MSGH_BITS_OLD_FORMAT)));
643:
644: {
645: ipc_mqueue_t mqueue;
646: ipc_pset_t pset;
647: ipc_thread_t receiver;
648: ipc_thread_queue_t receivers;
649:
650: pset = port->ip_pset;
651: if (pset == IPS_NULL)
652: mqueue = &port->ip_messages;
653: else
654: mqueue = &pset->ips_messages;
655:
656: if (!imq_lock_try(mqueue)) {
657: ip_unlock(port);
658: return MACH_MSG_IPC_KERNEL;
659: }
660: receivers = &mqueue->imq_threads;
661:
662: port->ip_msgcount++;
663:
664: /*
665: * Can unlock the port now that the msg queue is locked
666: * and we know the port is active. While the msg queue
667: * is locked, we have control of the kmsg, so the ref in
668: * it for the port is still good. If the msg queue is in
669: * a set (dead or alive), then we're OK because the port
670: * is still a member of the set and the set won't go away
671: * until the port is taken out, which tries to lock the
672: * set's msg queue to remove the port's msgs.
673: */
674:
675: ip_unlock(port);
676:
677: /* check for a receiver for the message */
678:
679: for (;;) {
680: receiver = ipc_thread_queue_first(receivers);
681: if (receiver == ITH_NULL) {
682: /* no receivers; queue kmsg */
683:
684: ipc_kmsg_enqueue_macro(&mqueue->imq_messages, kmsg);
685: imq_unlock(mqueue);
686: break;
687: }
688:
689: ipc_thread_rmqueue_first_macro(receivers, receiver);
690: assert(ipc_kmsg_queue_empty(&mqueue->imq_messages));
691:
692: /*
693: * Got a potential receiver, verify sizes
694: */
695: if (receiver->ith_rcv_option & MACH_RCV_OLD_FORMAT) {
696: if ( kmsg->ikm_header.msgh_size >
697: sizeof (mach_msg_header_t) ||
698: receiver->ith_msize <
699: sizeof (mach_msg_header_t)) {
700: receiver->ith_state = MACH_RCV_HEADER_ERROR;
701: receiver->ith_kmsg = kmsg;
702: receiver->ith_seqno = port->ip_seqno++;
703: imq_unlock(mqueue);
704: thread_go(receiver);
705: break;
706: }
707:
708: receiver->ith_state = MACH_MSG_SUCCESS;
709: }
710: else
711: if (kmsg->ikm_header.msgh_size > (receiver->ith_msize -
712: REQUESTED_TRAILER_SIZE(receiver->ith_rcv_option))) {
713: /*
714: * The receive buffer isn't large enough
715: */
716: receiver->ith_state = MACH_RCV_TOO_LARGE;
717:
718: if (receiver->ith_rcv_option & MACH_RCV_LARGE) {
719: receiver->ith_msize = kmsg->ikm_header.msgh_size;
720: thread_go(receiver);
721: continue;
722: }
723: }
724: else
725: receiver->ith_state = MACH_MSG_SUCCESS;
726:
727: /*
728: * Got a valid receiver. Possibly with an error.
729: * Hand the receiver the message.
730: */
731: receiver->ith_kmsg = kmsg;
732: receiver->ith_seqno = port->ip_seqno++;
733: imq_unlock(mqueue);
734: thread_go(receiver);
735: break;
736: }
737: }
738:
739: return MACH_MSG_SUCCESS;
740: }
741:
742: /*
743: * Routine: ipc_mqueue_copyin
744: * Purpose:
745: * Convert a name in a space to a message queue.
746: * Conditions:
747: * Nothing locked. If successful, the message queue
748: * is returned locked and caller gets a ref for the object.
749: * This ref ensures the continued existence of the queue.
750: * Returns:
751: * MACH_MSG_SUCCESS Found a message queue.
752: * MACH_RCV_INVALID_NAME The space is dead.
753: * MACH_RCV_INVALID_NAME The name doesn't denote a right.
754: * MACH_RCV_INVALID_NAME
755: * The denoted right is not receive or port set.
756: * MACH_RCV_IN_SET Receive right is a member of a set.
757: */
758:
759: mach_msg_return_t
760: ipc_mqueue_copyin(
761: ipc_space_t space,
762: mach_port_t name,
763: ipc_mqueue_t *mqueuep,
764: ipc_object_t *objectp)
765: {
766: ipc_entry_t entry;
767: ipc_entry_bits_t bits;
768: ipc_object_t object;
769: ipc_mqueue_t mqueue;
770:
771: is_read_lock(space);
772: if (!space->is_active) {
773: is_read_unlock(space);
774: return MACH_RCV_INVALID_NAME;
775: }
776:
777: entry = ipc_entry_lookup(space, name);
778: if (entry == IE_NULL) {
779: is_read_unlock(space);
780: return MACH_RCV_INVALID_NAME;
781: }
782:
783: bits = entry->ie_bits;
784: object = entry->ie_object;
785:
786: if (bits & MACH_PORT_TYPE_RECEIVE) {
787: ipc_port_t port;
788: ipc_pset_t pset;
789:
790: port = (ipc_port_t) object;
791: assert(port != IP_NULL);
792:
793: ip_lock(port);
794: assert(ip_active(port));
795: assert(port->ip_receiver_name == name);
796: assert(port->ip_receiver == space);
797: is_read_unlock(space);
798:
799: pset = port->ip_pset;
800: if (pset != IPS_NULL) {
801: ips_lock(pset);
802: if (ips_active(pset)) {
803: ips_unlock(pset);
804: ip_unlock(port);
805: return MACH_RCV_IN_SET;
806: }
807:
808: ipc_pset_remove(pset, port);
809: ips_check_unlock(pset);
810: assert(port->ip_pset == IPS_NULL);
811: }
812:
813: mqueue = &port->ip_messages;
814: } else if (bits & MACH_PORT_TYPE_PORT_SET) {
815: ipc_pset_t pset;
816:
817: pset = (ipc_pset_t) object;
818: assert(pset != IPS_NULL);
819:
820: ips_lock(pset);
821: assert(ips_active(pset));
822: assert(pset->ips_local_name == name);
823: is_read_unlock(space);
824:
825: mqueue = &pset->ips_messages;
826: } else {
827: is_read_unlock(space);
828: return MACH_RCV_INVALID_NAME;
829: }
830:
831: /*
832: * At this point, the object is locked and active,
833: * the space is unlocked, and mqueue is initialized.
834: */
835:
836: io_reference(object);
837: imq_lock(mqueue);
838: io_unlock(object);
839:
840: *objectp = object;
841: *mqueuep = mqueue;
842: return MACH_MSG_SUCCESS;
843: }
844:
845: /*
846: * Routine: ipc_mqueue_receive
847: * Purpose:
848: * Receive a message from a message queue.
849: *
850: * If continuation is non-zero, then we might discard
851: * our kernel stack when we block. We will continue
852: * after unblocking by executing continuation.
853: *
854: * If resume is true, then we are resuming a receive
855: * operation after a blocked receive discarded our stack.
856: * Conditions:
857: * The message queue is locked; it is unlocked.
858: *
859: * Our caller must hold a reference for the port or port set
860: * to which this queue belongs, to keep the queue
861: * from being deallocated. Furthermore, the port or set
862: * must have been active when the queue was locked.
863: *
864: * The kmsg is returned with clean header fields
865: * and with the circular bit turned off.
866: * Returns:
867: * MACH_MSG_SUCCESS Message returned in kmsgp.
868: * MACH_RCV_TOO_LARGE Message size returned in kmsgp.
869: * MACH_RCV_TIMED_OUT No message obtained.
870: * MACH_RCV_INTERRUPTED No message obtained.
871: * MACH_RCV_PORT_DIED Port/set died; no message.
872: * MACH_RCV_PORT_CHANGED Port moved into set; no msg.
873: */
874:
875: mach_msg_return_t
876: ipc_mqueue_receive(
877: ipc_mqueue_t mqueue,
878: mach_msg_option_t option,
879: mach_msg_size_t max_size,
880: mach_msg_timeout_t timeout,
881: boolean_t resume,
882: void (*continuation)(void),
883: ipc_kmsg_t *kmsgp,
884: mach_port_seqno_t *seqnop,
885: ipc_kmsg_t *list)
886: {
887: ipc_port_t port;
888: ipc_kmsg_t kmsg;
889: mach_port_seqno_t seqno;
890: mach_msg_return_t mr;
891:
892: {
893: ipc_kmsg_queue_t kmsgs = &mqueue->imq_messages;
894: ipc_thread_t self = current_thread();
895:
896: if (resume)
897: goto after_thread_block;
898:
899: for (;;) {
900: kmsg = ipc_kmsg_queue_first(kmsgs);
901: if (kmsg != IKM_NULL) {
902: if (kmsg->ikm_header.msgh_bits &
903: MACH_MSGH_BITS_OLD_FORMAT) {
904: /*
905: * Old format message. (typed data) Verify
906: * that one of the old messaging primatives
907: * is being used for the receive operation.
908: */
909: if (!(option & MACH_RCV_OLD_FORMAT)) {
910: self->ith_state = MACH_RCV_HEADER_ERROR;
911: ipc_kmsg_rmqueue_first_macro(kmsgs, kmsg);
912: port = (ipc_port_t)
913: kmsg->ikm_header.msgh_remote_port;
914: seqno = port->ip_seqno++;
915: break;
916: }
917:
918: /* check space requirements */
919:
920: if (kmsg->ikm_header.msgh_size > max_size) {
921: * (mach_msg_size_t *) kmsgp =
922: kmsg->ikm_header.msgh_size;
923: imq_unlock(mqueue);
924: return MACH_RCV_TOO_LARGE;
925: }
926:
927: self->ith_state = MACH_MSG_SUCCESS;
928: }
929: else {
930: /*
931: * New format message. (untyped data) Verify
932: * that one of the new messaging primatives
933: * is being used for the receive operation.
934: */
935: if (option & MACH_RCV_OLD_FORMAT) {
936: if ((kmsg->ikm_header.msgh_bits &
937: MACH_MSGH_BITS_COMPLEX) ||
938: kmsg->ikm_header.msgh_size >
939: sizeof (mach_msg_header_t) ||
940: max_size <
941: sizeof (mach_msg_header_t)) {
942: self->ith_state = MACH_RCV_HEADER_ERROR;
943: ipc_kmsg_rmqueue_first_macro(kmsgs, kmsg);
944: port = (ipc_port_t)
945: kmsg->ikm_header.msgh_remote_port;
946: seqno = port->ip_seqno++;
947: break;
948: }
949:
950: self->ith_state = MACH_MSG_SUCCESS;
951: }
952: else
953: /*
954: * Verify sizes.
955: */
956: if (kmsg->ikm_header.msgh_size > (max_size -
957: REQUESTED_TRAILER_SIZE(option))) {
958: /*
959: * The receive buffer isn't large enough.
960: */
961: if (option & MACH_RCV_LARGE) {
962: *(mach_msg_size_t *)kmsgp =
963: kmsg->ikm_header.msgh_size;
964: imq_unlock(mqueue);
965: return MACH_RCV_TOO_LARGE;
966: }
967:
968: self->ith_state = MACH_RCV_TOO_LARGE;
969: }
970: else
971: if ((kmsg->ikm_header.msgh_bits &
972: MACH_MSGH_BITS_COMPLEX) &&
973: list && (*list) != IKM_NULL &&
974: (self->ith_state =
975: ipc_kmsg_check_scatter(
976: kmsg, option, list))
977: != MACH_MSG_SUCCESS) {
978: /*
979: * Check for type/VM errors.
980: */
981: if (self->ith_state == MACH_RCV_INVALID_TYPE ||
982: (self->ith_state & MACH_MSG_VM_KERNEL) != 0) {
983: imq_unlock(mqueue);
984: return self->ith_state;
985: }
986:
987: /*
988: * The scatter list isn't large enough.
989: */
990: if (option & MACH_RCV_LARGE) {
991: *(mach_msg_size_t *)kmsgp =
992: kmsg->ikm_header.msgh_size;
993: imq_unlock(mqueue);
994: return self->ith_state;
995: }
996: }
997: else
998: self->ith_state = MACH_MSG_SUCCESS;
999: }
1000:
1001: ipc_kmsg_rmqueue_first_macro(kmsgs, kmsg);
1002: port = (ipc_port_t) kmsg->ikm_header.msgh_remote_port;
1003: seqno = port->ip_seqno++;
1004: break;
1005: }
1006:
1007: /* must block waiting for a message */
1008:
1009: if (option & MACH_RCV_TIMEOUT) {
1010: if (timeout == 0) {
1011: imq_unlock(mqueue);
1012: return MACH_RCV_TIMED_OUT;
1013: }
1014:
1015: thread_will_wait_with_timeout(self, timeout);
1016: } else
1017: thread_will_wait(self);
1018:
1019: ipc_thread_enqueue_macro(&mqueue->imq_threads, self);
1020: self->ith_state = MACH_RCV_IN_PROGRESS;
1021: self->ith_msize = max_size;
1022: self->ith_rcv_option = option;
1023: self->ith_list = list ? *list : IKM_NULL;
1024:
1025: imq_unlock(mqueue);
1026: if (continuation != (void (*)(void)) 0) {
1027: counter(c_ipc_mqueue_receive_block_user++);
1028: thread_block_with_continuation(continuation);
1029: } else {
1030: counter(c_ipc_mqueue_receive_block_kernel++);
1031: thread_block_with_continuation((void (*)(void)) 0);
1032: }
1033: after_thread_block:
1034: imq_lock(mqueue);
1035: if (list)
1036: (*list) = self->ith_list;
1037:
1038: /* why did we wake up? */
1039:
1040: if (self->ith_state == MACH_MSG_SUCCESS) {
1041: return_kmsg:
1042: /* pick up the message that was handed to us */
1043:
1044: kmsg = self->ith_kmsg;
1045: seqno = self->ith_seqno;
1046: port = (ipc_port_t) kmsg->ikm_header.msgh_remote_port;
1047: break;
1048: }
1049:
1050: switch (self->ith_state) {
1051: case MACH_RCV_INVALID_TYPE:
1052: /* bad invalid scatter list */
1053:
1054: imq_unlock(mqueue);
1055: return self->ith_state;
1056:
1057: case MACH_RCV_TOO_LARGE:
1058: case MACH_RCV_SCATTER_SMALL:
1059: /*
1060: * The receive buffer isn't large enough or
1061: * there is a scatter list and it doesn't match
1062: * the gather list.
1063: */
1064: if (!(self->ith_rcv_option &
1065: (MACH_RCV_LARGE|MACH_RCV_OLD_FORMAT)))
1066: goto return_kmsg;
1067:
1068: *(mach_msg_size_t *) kmsgp = self->ith_msize;
1069:
1070: /* fall-through */
1071:
1072: case MACH_RCV_PORT_DIED:
1073: case MACH_RCV_PORT_CHANGED:
1074: /* something bad happened to the port/set */
1075:
1076: imq_unlock(mqueue);
1077: return self->ith_state;
1078:
1079: case MACH_RCV_IN_PROGRESS:
1080: /*
1081: * Awakened for other than IPC completion.
1082: * Remove ourselves from the waiting queue,
1083: * then check the wakeup cause.
1084: */
1085:
1086: ipc_thread_rmqueue(&mqueue->imq_threads, self);
1087:
1088: switch (self->ith_wait_result) {
1089: case THREAD_INTERRUPTED:
1090: case THREAD_SHOULD_TERMINATE:
1091: /* receive was interrupted - give up */
1092:
1093: imq_unlock(mqueue);
1094: return MACH_RCV_INTERRUPTED;
1095:
1096: case THREAD_TIMED_OUT:
1097: /* timeout expired */
1098:
1099: assert(option & MACH_RCV_TIMEOUT);
1100: timeout = 0;
1101: break;
1102:
1103: case THREAD_RESTART:
1104: default:
1105: panic("ipc_mqueue_receive");
1106: }
1107: break;
1108:
1109: case MACH_RCV_HEADER_ERROR:
1110: /*
1111: * There was a message was queued on the
1112: * port, but its format was incompatible
1113: * with the receive operation we were
1114: * performing.
1115: */
1116: goto return_kmsg;
1117:
1118: default:
1119: panic("ipc_mqueue_receive: strange ith_state");
1120: }
1121: }
1122:
1123: mr = self->ith_state;
1124:
1125: /* we have a kmsg; unlock the msg queue */
1126:
1127: imq_unlock(mqueue);
1128: }
1129:
1130: {
1131: ipc_marequest_t marequest;
1132:
1133: marequest = kmsg->ikm_marequest;
1134: if (marequest != IMAR_NULL) {
1135: ipc_marequest_destroy(marequest);
1136: kmsg->ikm_marequest = IMAR_NULL;
1137: }
1138: assert((kmsg->ikm_header.msgh_bits & MACH_MSGH_BITS_CIRCULAR) == 0);
1139:
1140: assert(port == (ipc_port_t) kmsg->ikm_header.msgh_remote_port);
1141: ip_lock(port);
1142:
1143: if (ip_active(port)) {
1144: ipc_thread_queue_t senders;
1145: ipc_thread_t sender;
1146:
1147: assert(port->ip_msgcount > 0);
1148: port->ip_msgcount--;
1149:
1150: senders = &port->ip_blocked;
1151: sender = ipc_thread_queue_first(senders);
1152:
1153: if ((sender != ITH_NULL) &&
1154: (port->ip_msgcount < port->ip_qlimit)) {
1155: ipc_thread_rmqueue(senders, sender);
1156: sender->ith_state = MACH_MSG_SUCCESS;
1157: thread_go(sender);
1158: }
1159: }
1160:
1161: ip_unlock(port);
1162: }
1163:
1164: *kmsgp = kmsg;
1165: *seqnop = seqno;
1166: return mr;
1167: }
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.