![[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 [GnuMach] / Gnu-Mach / ipc|
Return to ipc_mqueue.c CVS log | Up to [GnuMach] / Gnu-Mach / ipc |
1.1 root 1: /*
2: * Mach Operating System
3: * Copyright (c) 1991,1990,1989 Carnegie Mellon University.
4: * Copyright (c) 1993,1994 The University of Utah and
5: * the Computer Systems Laboratory (CSL).
6: * All rights reserved.
7: *
8: * Permission to use, copy, modify and distribute this software and its
9: * documentation is hereby granted, provided that both the copyright
10: * notice and this permission notice appear in all copies of the
11: * software, derivative works or modified versions, and any portions
12: * thereof, and that both notices appear in supporting documentation.
13: *
14: * CARNEGIE MELLON, THE UNIVERSITY OF UTAH AND CSL ALLOW FREE USE OF
15: * THIS SOFTWARE IN ITS "AS IS" CONDITION, AND DISCLAIM ANY LIABILITY
16: * OF ANY KIND FOR ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF
17: * THIS SOFTWARE.
18: *
19: * Carnegie Mellon requests users of this software to return to
20: *
21: * Software Distribution Coordinator or [email protected]
22: * School of Computer Science
23: * Carnegie Mellon University
24: * Pittsburgh PA 15213-3890
25: *
26: * any improvements or extensions that they make and grant Carnegie Mellon
27: * the rights to redistribute these changes.
28: */
29: /*
30: * File: ipc/ipc_mqueue.c
31: * Author: Rich Draves
32: * Date: 1989
33: *
34: * Functions to manipulate IPC message queues.
35: */
36:
37: #include <mach/port.h>
38: #include <mach/message.h>
39: #include <kern/assert.h>
40: #include <kern/counters.h>
1.1.1.3 ! root 41: #include <kern/debug.h>
1.1 root 42: #include <kern/sched_prim.h>
43: #include <kern/ipc_sched.h>
44: #include <kern/ipc_kobject.h>
45: #include <ipc/ipc_mqueue.h>
46: #include <ipc/ipc_thread.h>
47: #include <ipc/ipc_kmsg.h>
48: #include <ipc/ipc_port.h>
49: #include <ipc/ipc_pset.h>
50: #include <ipc/ipc_space.h>
51: #include <ipc/ipc_marequest.h>
52:
53:
54:
55: /*
56: * Routine: ipc_mqueue_init
57: * Purpose:
58: * Initialize a newly-allocated message queue.
59: */
60:
61: void
62: ipc_mqueue_init(
63: ipc_mqueue_t mqueue)
64: {
65: imq_lock_init(mqueue);
66: ipc_kmsg_queue_init(&mqueue->imq_messages);
67: ipc_thread_queue_init(&mqueue->imq_threads);
68: }
69:
70: /*
71: * Routine: ipc_mqueue_move
72: * Purpose:
73: * Move messages from one queue (source) to another (dest).
74: * Only moves messages sent to the specified port.
75: * Conditions:
76: * Both queues must be locked.
77: * (This is sufficient to manipulate port->ip_seqno.)
78: */
79:
80: void
81: ipc_mqueue_move(
82: ipc_mqueue_t dest,
83: ipc_mqueue_t source,
84: ipc_port_t port)
85: {
86: ipc_kmsg_queue_t oldq, newq;
87: ipc_thread_queue_t blockedq;
88: ipc_kmsg_t kmsg, next;
89: ipc_thread_t th;
90:
91: oldq = &source->imq_messages;
92: newq = &dest->imq_messages;
93: blockedq = &dest->imq_threads;
94:
95: for (kmsg = ipc_kmsg_queue_first(oldq);
96: kmsg != IKM_NULL; kmsg = next) {
97: next = ipc_kmsg_queue_next(oldq, kmsg);
98:
99: /* only move messages sent to port */
100:
101: if (kmsg->ikm_header.msgh_remote_port != (mach_port_t) port)
102: continue;
103:
104: ipc_kmsg_rmqueue(oldq, kmsg);
105:
106: /* before adding kmsg to newq, check for a blocked receiver */
107:
108: while ((th = ipc_thread_dequeue(blockedq)) != ITH_NULL) {
109: assert(ipc_kmsg_queue_empty(newq));
110:
111: thread_go(th);
112:
113: /* check if the receiver can handle the message */
114:
115: if (kmsg->ikm_header.msgh_size <= th->ith_msize) {
116: th->ith_state = MACH_MSG_SUCCESS;
117: th->ith_kmsg = kmsg;
118: th->ith_seqno = port->ip_seqno++;
119:
120: goto next_kmsg;
121: }
122:
123: th->ith_state = MACH_RCV_TOO_LARGE;
124: th->ith_msize = kmsg->ikm_header.msgh_size;
125: }
126:
127: /* didn't find a receiver to handle the message */
128:
129: ipc_kmsg_enqueue(newq, kmsg);
130: next_kmsg:;
131: }
132: }
133:
134: /*
135: * Routine: ipc_mqueue_changed
136: * Purpose:
137: * Wake up receivers waiting in a message queue.
138: * Conditions:
139: * The message queue is locked.
140: */
141:
142: void
143: ipc_mqueue_changed(
144: ipc_mqueue_t mqueue,
145: mach_msg_return_t mr)
146: {
147: ipc_thread_t th;
148:
149: while ((th = ipc_thread_dequeue(&mqueue->imq_threads)) != ITH_NULL) {
150: th->ith_state = mr;
151: thread_go(th);
152: }
153: }
154:
155: /*
156: * Routine: ipc_mqueue_send
157: * Purpose:
158: * Send a message to a port. The message holds a reference
159: * for the destination port in the msgh_remote_port field.
160: *
161: * If unsuccessful, the caller still has possession of
162: * the message and must do something with it. If successful,
163: * the message is queued, given to a receiver, destroyed,
164: * or handled directly by the kernel via mach_msg.
165: * Conditions:
166: * Nothing locked.
167: * Returns:
168: * MACH_MSG_SUCCESS The message was accepted.
169: * MACH_SEND_TIMED_OUT Caller still has message.
170: * MACH_SEND_INTERRUPTED Caller still has message.
171: */
172:
173: mach_msg_return_t
174: ipc_mqueue_send(kmsg, option, time_out)
175: ipc_kmsg_t kmsg;
176: mach_msg_option_t option;
177: mach_msg_timeout_t time_out;
178: {
179: ipc_port_t port;
180:
181: port = (ipc_port_t) kmsg->ikm_header.msgh_remote_port;
182: assert(IP_VALID(port));
183:
184: ip_lock(port);
185:
186: if (port->ip_receiver == ipc_space_kernel) {
187: ipc_kmsg_t reply;
188:
189: /*
190: * We can check ip_receiver == ipc_space_kernel
191: * before checking that the port is active because
192: * ipc_port_dealloc_kernel clears ip_receiver
193: * before destroying a kernel port.
194: */
195:
196: assert(ip_active(port));
197: ip_unlock(port);
198:
199: reply = ipc_kobject_server(kmsg);
200: if (reply != IKM_NULL)
201: ipc_mqueue_send_always(reply);
202:
203: return MACH_MSG_SUCCESS;
204: }
205:
206: for (;;) {
207: ipc_thread_t self;
208:
209: /*
210: * Can't deliver to a dead port.
211: * However, we can pretend it got sent
212: * and was then immediately destroyed.
213: */
214:
215: if (!ip_active(port)) {
216: /*
217: * We can't let ipc_kmsg_destroy deallocate
218: * the port right, because we might end up
219: * in an infinite loop trying to deliver
220: * a send-once notification.
221: */
222:
223: ip_release(port);
224: ip_check_unlock(port);
225: kmsg->ikm_header.msgh_remote_port = MACH_PORT_NULL;
226: ipc_kmsg_destroy(kmsg);
227: return MACH_MSG_SUCCESS;
228: }
229:
230: /*
231: * Don't block if:
232: * 1) We're under the queue limit.
233: * 2) Caller used the MACH_SEND_ALWAYS internal option.
234: * 3) Message is sent to a send-once right.
235: */
236:
237: if ((port->ip_msgcount < port->ip_qlimit) ||
238: (option & MACH_SEND_ALWAYS) ||
239: (MACH_MSGH_BITS_REMOTE(kmsg->ikm_header.msgh_bits) ==
240: MACH_MSG_TYPE_PORT_SEND_ONCE))
241: break;
242:
243: /* must block waiting for queue to clear */
244:
245: self = current_thread();
246:
247: if (option & MACH_SEND_TIMEOUT) {
248: if (time_out == 0) {
249: ip_unlock(port);
250: return MACH_SEND_TIMED_OUT;
251: }
252:
253: thread_will_wait_with_timeout(self, time_out);
254: } else
255: thread_will_wait(self);
256:
257: ipc_thread_enqueue(&port->ip_blocked, self);
258: self->ith_state = MACH_SEND_IN_PROGRESS;
259:
260: ip_unlock(port);
261: counter(c_ipc_mqueue_send_block++);
262: thread_block((void (*)(void)) 0);
263: ip_lock(port);
264:
265: /* why did we wake up? */
266:
267: if (self->ith_state == MACH_MSG_SUCCESS)
268: continue;
269: assert(self->ith_state == MACH_SEND_IN_PROGRESS);
270:
271: /* take ourselves off blocked queue */
272:
273: ipc_thread_rmqueue(&port->ip_blocked, self);
274:
275: /*
276: * Thread wakeup-reason field tells us why
277: * the wait was interrupted.
278: */
279:
280: switch (self->ith_wait_result) {
281: case THREAD_INTERRUPTED:
282: /* send was interrupted - give up */
283:
284: ip_unlock(port);
285: return MACH_SEND_INTERRUPTED;
286:
287: case THREAD_TIMED_OUT:
288: /* timeout expired */
289:
290: assert(option & MACH_SEND_TIMEOUT);
291: time_out = 0;
292: break;
293:
294: case THREAD_RESTART:
295: default:
296: #if MACH_ASSERT
297: assert(!"ipc_mqueue_send");
298: #else
299: panic("ipc_mqueue_send");
300: #endif
301: }
302: }
303:
304: if (kmsg->ikm_header.msgh_bits & MACH_MSGH_BITS_CIRCULAR) {
305: ip_unlock(port);
306:
307: /* don't allow the creation of a circular loop */
308:
309: ipc_kmsg_destroy(kmsg);
310: return MACH_MSG_SUCCESS;
311: }
312:
313: {
314: ipc_mqueue_t mqueue;
315: ipc_pset_t pset;
316: ipc_thread_t receiver;
317: ipc_thread_queue_t receivers;
318:
319: port->ip_msgcount++;
320: assert(port->ip_msgcount > 0);
321:
322: pset = port->ip_pset;
323: if (pset == IPS_NULL)
324: mqueue = &port->ip_messages;
325: else
326: mqueue = &pset->ips_messages;
327:
328: imq_lock(mqueue);
329: receivers = &mqueue->imq_threads;
330:
331: /*
332: * Can unlock the port now that the msg queue is locked
333: * and we know the port is active. While the msg queue
334: * is locked, we have control of the kmsg, so the ref in
335: * it for the port is still good. If the msg queue is in
336: * a set (dead or alive), then we're OK because the port
337: * is still a member of the set and the set won't go away
338: * until the port is taken out, which tries to lock the
339: * set's msg queue to remove the port's msgs.
340: */
341:
342: ip_unlock(port);
343:
344: /* check for a receiver for the message */
345:
346: for (;;) {
347: receiver = ipc_thread_queue_first(receivers);
348: if (receiver == ITH_NULL) {
349: /* no receivers; queue kmsg */
350:
351: ipc_kmsg_enqueue_macro(&mqueue->imq_messages, kmsg);
352: imq_unlock(mqueue);
353: break;
354: }
355:
356: ipc_thread_rmqueue_first_macro(receivers, receiver);
357: assert(ipc_kmsg_queue_empty(&mqueue->imq_messages));
358:
359: if (kmsg->ikm_header.msgh_size <= receiver->ith_msize) {
360: /* got a successful receiver */
361:
362: receiver->ith_state = MACH_MSG_SUCCESS;
363: receiver->ith_kmsg = kmsg;
364: receiver->ith_seqno = port->ip_seqno++;
365: imq_unlock(mqueue);
366:
367: thread_go(receiver);
368: break;
369: }
370:
371: receiver->ith_state = MACH_RCV_TOO_LARGE;
372: receiver->ith_msize = kmsg->ikm_header.msgh_size;
373: thread_go(receiver);
374: }
375: }
376:
1.1.1.3 ! root 377: current_task()->messages_sent++;
! 378:
1.1 root 379: return MACH_MSG_SUCCESS;
380: }
381:
382: /*
383: * Routine: ipc_mqueue_copyin
384: * Purpose:
385: * Convert a name in a space to a message queue.
386: * Conditions:
387: * Nothing locked. If successful, the message queue
388: * is returned locked and caller gets a ref for the object.
389: * This ref ensures the continued existence of the queue.
390: * Returns:
391: * MACH_MSG_SUCCESS Found a message queue.
392: * MACH_RCV_INVALID_NAME The space is dead.
393: * MACH_RCV_INVALID_NAME The name doesn't denote a right.
394: * MACH_RCV_INVALID_NAME
395: * The denoted right is not receive or port set.
396: * MACH_RCV_IN_SET Receive right is a member of a set.
397: */
398:
399: mach_msg_return_t
400: ipc_mqueue_copyin(
401: ipc_space_t space,
402: mach_port_t name,
403: ipc_mqueue_t *mqueuep,
404: ipc_object_t *objectp)
405: {
406: ipc_entry_t entry;
407: ipc_entry_bits_t bits;
408: ipc_object_t object;
409: ipc_mqueue_t mqueue;
410:
411: is_read_lock(space);
412: if (!space->is_active) {
413: is_read_unlock(space);
414: return MACH_RCV_INVALID_NAME;
415: }
416:
417: entry = ipc_entry_lookup(space, name);
418: if (entry == IE_NULL) {
419: is_read_unlock(space);
420: return MACH_RCV_INVALID_NAME;
421: }
422:
423: bits = entry->ie_bits;
424: object = entry->ie_object;
425:
426: if (bits & MACH_PORT_TYPE_RECEIVE) {
427: ipc_port_t port;
428: ipc_pset_t pset;
429:
430: port = (ipc_port_t) object;
431: assert(port != IP_NULL);
432:
433: ip_lock(port);
434: assert(ip_active(port));
435: assert(port->ip_receiver_name == name);
436: assert(port->ip_receiver == space);
437: is_read_unlock(space);
438:
439: pset = port->ip_pset;
440: if (pset != IPS_NULL) {
441: ips_lock(pset);
442: if (ips_active(pset)) {
443: ips_unlock(pset);
444: ip_unlock(port);
445: return MACH_RCV_IN_SET;
446: }
447:
448: ipc_pset_remove(pset, port);
449: ips_check_unlock(pset);
450: assert(port->ip_pset == IPS_NULL);
451: }
452:
453: mqueue = &port->ip_messages;
454: } else if (bits & MACH_PORT_TYPE_PORT_SET) {
455: ipc_pset_t pset;
456:
457: pset = (ipc_pset_t) object;
458: assert(pset != IPS_NULL);
459:
460: ips_lock(pset);
461: assert(ips_active(pset));
462: assert(pset->ips_local_name == name);
463: is_read_unlock(space);
464:
465: mqueue = &pset->ips_messages;
466: } else {
467: is_read_unlock(space);
468: return MACH_RCV_INVALID_NAME;
469: }
470:
471: /*
472: * At this point, the object is locked and active,
473: * the space is unlocked, and mqueue is initialized.
474: */
475:
476: io_reference(object);
477: imq_lock(mqueue);
478: io_unlock(object);
479:
480: *objectp = object;
481: *mqueuep = mqueue;
482: return MACH_MSG_SUCCESS;
483: }
484:
485: /*
486: * Routine: ipc_mqueue_receive
487: * Purpose:
488: * Receive a message from a message queue.
489: *
490: * If continuation is non-zero, then we might discard
491: * our kernel stack when we block. We will continue
492: * after unblocking by executing continuation.
493: *
494: * If resume is true, then we are resuming a receive
495: * operation after a blocked receive discarded our stack.
496: * Conditions:
497: * The message queue is locked; it will be returned unlocked.
498: *
499: * Our caller must hold a reference for the port or port set
500: * to which this queue belongs, to keep the queue
501: * from being deallocated. Furthermore, the port or set
502: * must have been active when the queue was locked.
503: *
504: * The kmsg is returned with clean header fields
505: * and with the circular bit turned off.
506: * Returns:
507: * MACH_MSG_SUCCESS Message returned in kmsgp.
508: * MACH_RCV_TOO_LARGE Message size returned in kmsgp.
509: * MACH_RCV_TIMED_OUT No message obtained.
510: * MACH_RCV_INTERRUPTED No message obtained.
511: * MACH_RCV_PORT_DIED Port/set died; no message.
512: * MACH_RCV_PORT_CHANGED Port moved into set; no msg.
513: *
514: */
515:
516: mach_msg_return_t
517: ipc_mqueue_receive(
518: ipc_mqueue_t mqueue,
519: mach_msg_option_t option,
520: mach_msg_size_t max_size,
521: mach_msg_timeout_t time_out,
522: boolean_t resume,
523: void (*continuation)(void),
524: ipc_kmsg_t *kmsgp,
525: mach_port_seqno_t *seqnop)
526: {
527: ipc_port_t port;
528: ipc_kmsg_t kmsg;
529: mach_port_seqno_t seqno;
530:
531: {
532: ipc_kmsg_queue_t kmsgs = &mqueue->imq_messages;
533: ipc_thread_t self = current_thread();
534:
535: if (resume)
536: goto after_thread_block;
537:
538: for (;;) {
539: kmsg = ipc_kmsg_queue_first(kmsgs);
540: if (kmsg != IKM_NULL) {
541: /* check space requirements */
542:
543: if (kmsg->ikm_header.msgh_size > max_size) {
544: * (mach_msg_size_t *) kmsgp =
545: kmsg->ikm_header.msgh_size;
546: imq_unlock(mqueue);
547: return MACH_RCV_TOO_LARGE;
548: }
549:
550: ipc_kmsg_rmqueue_first_macro(kmsgs, kmsg);
551: port = (ipc_port_t) kmsg->ikm_header.msgh_remote_port;
552: seqno = port->ip_seqno++;
553: break;
554: }
555:
556: /* must block waiting for a message */
557:
558: if (option & MACH_RCV_TIMEOUT) {
559: if (time_out == 0) {
560: imq_unlock(mqueue);
561: return MACH_RCV_TIMED_OUT;
562: }
563:
564: thread_will_wait_with_timeout(self, time_out);
565: } else
566: thread_will_wait(self);
567:
568: ipc_thread_enqueue_macro(&mqueue->imq_threads, self);
569: self->ith_state = MACH_RCV_IN_PROGRESS;
570: self->ith_msize = max_size;
571:
572: imq_unlock(mqueue);
573: if (continuation != (void (*)(void)) 0) {
574: counter(c_ipc_mqueue_receive_block_user++);
575: } else {
576: counter(c_ipc_mqueue_receive_block_kernel++);
577: }
578: thread_block(continuation);
579: after_thread_block:
580: imq_lock(mqueue);
581:
582: /* why did we wake up? */
583:
584: if (self->ith_state == MACH_MSG_SUCCESS) {
585: /* pick up the message that was handed to us */
586:
587: kmsg = self->ith_kmsg;
588: seqno = self->ith_seqno;
589: port = (ipc_port_t) kmsg->ikm_header.msgh_remote_port;
590: break;
591: }
592:
593: switch (self->ith_state) {
594: case MACH_RCV_TOO_LARGE:
595: /* pick up size of the too-large message */
596:
597: * (mach_msg_size_t *) kmsgp = self->ith_msize;
598: /* fall-through */
599:
600: case MACH_RCV_PORT_DIED:
601: case MACH_RCV_PORT_CHANGED:
602: /* something bad happened to the port/set */
603:
604: imq_unlock(mqueue);
605: return self->ith_state;
606:
607: case MACH_RCV_IN_PROGRESS:
608: /*
609: * Awakened for other than IPC completion.
610: * Remove ourselves from the waiting queue,
611: * then check the wakeup cause.
612: */
613:
614: ipc_thread_rmqueue(&mqueue->imq_threads, self);
615:
616: switch (self->ith_wait_result) {
617: case THREAD_INTERRUPTED:
618: /* receive was interrupted - give up */
619:
620: imq_unlock(mqueue);
621: return MACH_RCV_INTERRUPTED;
622:
623: case THREAD_TIMED_OUT:
624: /* timeout expired */
625:
626: assert(option & MACH_RCV_TIMEOUT);
627: time_out = 0;
628: break;
629:
630: case THREAD_RESTART:
631: default:
632: #if MACH_ASSERT
633: assert(!"ipc_mqueue_receive");
634: #else
635: panic("ipc_mqueue_receive");
636: #endif
637: }
638: break;
639:
640: default:
641: #if MACH_ASSERT
642: assert(!"ipc_mqueue_receive: strange ith_state");
643: #else
644: panic("ipc_mqueue_receive: strange ith_state");
645: #endif
646: }
647: }
648:
649: /* we have a kmsg; unlock the msg queue */
650:
651: imq_unlock(mqueue);
652: assert(kmsg->ikm_header.msgh_size <= max_size);
653: }
654:
655: {
656: ipc_marequest_t marequest;
657:
658: marequest = kmsg->ikm_marequest;
659: if (marequest != IMAR_NULL) {
660: ipc_marequest_destroy(marequest);
661: kmsg->ikm_marequest = IMAR_NULL;
662: }
663: assert((kmsg->ikm_header.msgh_bits & MACH_MSGH_BITS_CIRCULAR) == 0);
664:
665: assert(port == (ipc_port_t) kmsg->ikm_header.msgh_remote_port);
666: ip_lock(port);
667:
668: if (ip_active(port)) {
669: ipc_thread_queue_t senders;
670: ipc_thread_t sender;
671:
672: assert(port->ip_msgcount > 0);
673: port->ip_msgcount--;
674:
675: senders = &port->ip_blocked;
676: sender = ipc_thread_queue_first(senders);
677:
678: if ((sender != ITH_NULL) &&
679: (port->ip_msgcount < port->ip_qlimit)) {
680: ipc_thread_rmqueue(senders, sender);
681: sender->ith_state = MACH_MSG_SUCCESS;
682: thread_go(sender);
683: }
684: }
685:
686: ip_unlock(port);
687: }
688:
1.1.1.3 ! root 689: current_task()->messages_received++;
! 690:
1.1 root 691: *kmsgp = kmsg;
692: *seqnop = seqno;
693: return MACH_MSG_SUCCESS;
694: }
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.