]> git.saurik.com Git - apple/xnu.git/blob - osfmk/ipc/ipc_mqueue.c
4bec21084b2c88a77d49f5edc94f72c2342e4a4c
[apple/xnu.git] / osfmk / ipc / ipc_mqueue.c
1 /*
2 * Copyright (c) 2000-2007 Apple Inc. All rights reserved.
3 *
4 * @APPLE_OSREFERENCE_LICENSE_HEADER_START@
5 *
6 * This file contains Original Code and/or Modifications of Original Code
7 * as defined in and that are subject to the Apple Public Source License
8 * Version 2.0 (the 'License'). You may not use this file except in
9 * compliance with the License. The rights granted to you under the License
10 * may not be used to create, or enable the creation or redistribution of,
11 * unlawful or unlicensed copies of an Apple operating system, or to
12 * circumvent, violate, or enable the circumvention or violation of, any
13 * terms of an Apple operating system software license agreement.
14 *
15 * Please obtain a copy of the License at
16 * http://www.opensource.apple.com/apsl/ and read it before using this file.
17 *
18 * The Original Code and all software distributed under the License are
19 * distributed on an 'AS IS' basis, WITHOUT WARRANTY OF ANY KIND, EITHER
20 * EXPRESS OR IMPLIED, AND APPLE HEREBY DISCLAIMS ALL SUCH WARRANTIES,
21 * INCLUDING WITHOUT LIMITATION, ANY WARRANTIES OF MERCHANTABILITY,
22 * FITNESS FOR A PARTICULAR PURPOSE, QUIET ENJOYMENT OR NON-INFRINGEMENT.
23 * Please see the License for the specific language governing rights and
24 * limitations under the License.
25 *
26 * @APPLE_OSREFERENCE_LICENSE_HEADER_END@
27 */
28 /*
29 * @OSF_FREE_COPYRIGHT@
30 */
31 /*
32 * Mach Operating System
33 * Copyright (c) 1991,1990,1989 Carnegie Mellon University
34 * All Rights Reserved.
35 *
36 * Permission to use, copy, modify and distribute this software and its
37 * documentation is hereby granted, provided that both the copyright
38 * notice and this permission notice appear in all copies of the
39 * software, derivative works or modified versions, and any portions
40 * thereof, and that both notices appear in supporting documentation.
41 *
42 * CARNEGIE MELLON ALLOWS FREE USE OF THIS SOFTWARE IN ITS "AS IS"
43 * CONDITION. CARNEGIE MELLON DISCLAIMS ANY LIABILITY OF ANY KIND FOR
44 * ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF THIS SOFTWARE.
45 *
46 * Carnegie Mellon requests users of this software to return to
47 *
48 * Software Distribution Coordinator or Software.Distribution@CS.CMU.EDU
49 * School of Computer Science
50 * Carnegie Mellon University
51 * Pittsburgh PA 15213-3890
52 *
53 * any improvements or extensions that they make and grant Carnegie Mellon
54 * the rights to redistribute these changes.
55 */
56 /*
57 */
58 /*
59 * File: ipc/ipc_mqueue.c
60 * Author: Rich Draves
61 * Date: 1989
62 *
63 * Functions to manipulate IPC message queues.
64 */
65 /*
66 * NOTICE: This file was modified by SPARTA, Inc. in 2006 to introduce
67 * support for mandatory and extensible security protections. This notice
68 * is included in support of clause 2.2 (b) of the Apple Public License,
69 * Version 2.0.
70 */
71
72
73 #include <mach/port.h>
74 #include <mach/message.h>
75 #include <mach/sync_policy.h>
76
77 #include <kern/assert.h>
78 #include <kern/counters.h>
79 #include <kern/sched_prim.h>
80 #include <kern/ipc_kobject.h>
81 #include <kern/ipc_mig.h> /* XXX - for mach_msg_receive_continue */
82 #include <kern/misc_protos.h>
83 #include <kern/task.h>
84 #include <kern/thread.h>
85 #include <kern/waitq.h>
86
87 #include <ipc/port.h>
88 #include <ipc/ipc_mqueue.h>
89 #include <ipc/ipc_kmsg.h>
90 #include <ipc/ipc_port.h>
91 #include <ipc/ipc_pset.h>
92 #include <ipc/ipc_space.h>
93
94 #if MACH_FLIPC
95 #include <ipc/flipc.h>
96 #endif
97
98 #ifdef __LP64__
99 #include <vm/vm_map.h>
100 #endif
101
102 #include <sys/event.h>
103
104 extern char *proc_name_address(void *p);
105
106 int ipc_mqueue_full; /* address is event for queue space */
107 int ipc_mqueue_rcv; /* address is event for message arrival */
108
109 /* forward declarations */
110 static void ipc_mqueue_receive_results(wait_result_t result);
111 static void ipc_mqueue_peek_on_thread(
112 ipc_mqueue_t port_mq,
113 mach_msg_option_t option,
114 thread_t thread);
115
116 /*
117 * Routine: ipc_mqueue_init
118 * Purpose:
119 * Initialize a newly-allocated message queue.
120 */
121 void
122 ipc_mqueue_init(
123 ipc_mqueue_t mqueue,
124 boolean_t is_set)
125 {
126 if (is_set) {
127 waitq_set_init(&mqueue->imq_set_queue,
128 SYNC_POLICY_FIFO | SYNC_POLICY_PREPOST,
129 NULL, NULL);
130 } else {
131 waitq_init(&mqueue->imq_wait_queue, SYNC_POLICY_FIFO | SYNC_POLICY_PORT);
132 ipc_kmsg_queue_init(&mqueue->imq_messages);
133 mqueue->imq_seqno = 0;
134 mqueue->imq_msgcount = 0;
135 mqueue->imq_qlimit = MACH_PORT_QLIMIT_DEFAULT;
136 mqueue->imq_context = 0;
137 mqueue->imq_fullwaiters = FALSE;
138 #if MACH_FLIPC
139 mqueue->imq_fport = FPORT_NULL;
140 #endif
141 }
142 klist_init(&mqueue->imq_klist);
143 }
144
145 void
146 ipc_mqueue_deinit(
147 ipc_mqueue_t mqueue)
148 {
149 boolean_t is_set = imq_is_set(mqueue);
150
151 if (is_set) {
152 waitq_set_deinit(&mqueue->imq_set_queue);
153 } else {
154 waitq_deinit(&mqueue->imq_wait_queue);
155 }
156 }
157
158 /*
159 * Routine: imq_reserve_and_lock
160 * Purpose:
161 * Atomically lock an ipc_mqueue_t object and reserve
162 * an appropriate number of prepost linkage objects for
163 * use in wakeup operations.
164 * Conditions:
165 * mq is unlocked
166 */
167 void
168 imq_reserve_and_lock(ipc_mqueue_t mq, uint64_t *reserved_prepost)
169 {
170 *reserved_prepost = waitq_prepost_reserve(&mq->imq_wait_queue, 0,
171 WAITQ_KEEP_LOCKED);
172 }
173
174
175 /*
176 * Routine: imq_release_and_unlock
177 * Purpose:
178 * Unlock an ipc_mqueue_t object, re-enable interrupts,
179 * and release any unused prepost object reservations.
180 * Conditions:
181 * mq is locked
182 */
183 void
184 imq_release_and_unlock(ipc_mqueue_t mq, uint64_t reserved_prepost)
185 {
186 assert(imq_held(mq));
187 waitq_unlock(&mq->imq_wait_queue);
188 waitq_prepost_release_reserve(reserved_prepost);
189 }
190
191
192 /*
193 * Routine: ipc_mqueue_member
194 * Purpose:
195 * Indicate whether the (port) mqueue is a member of
196 * this portset's mqueue. We do this by checking
197 * whether the portset mqueue's waitq is an member of
198 * the port's mqueue waitq.
199 * Conditions:
200 * the portset's mqueue is not already a member
201 * this may block while allocating linkage structures.
202 */
203
204 boolean_t
205 ipc_mqueue_member(
206 ipc_mqueue_t port_mqueue,
207 ipc_mqueue_t set_mqueue)
208 {
209 struct waitq *port_waitq = &port_mqueue->imq_wait_queue;
210 struct waitq_set *set_waitq = &set_mqueue->imq_set_queue;
211
212 return waitq_member(port_waitq, set_waitq);
213 }
214
215 /*
216 * Routine: ipc_mqueue_remove
217 * Purpose:
218 * Remove the association between the queue and the specified
219 * set message queue.
220 */
221
222 kern_return_t
223 ipc_mqueue_remove(
224 ipc_mqueue_t mqueue,
225 ipc_mqueue_t set_mqueue)
226 {
227 struct waitq *mq_waitq = &mqueue->imq_wait_queue;
228 struct waitq_set *set_waitq = &set_mqueue->imq_set_queue;
229
230 return waitq_unlink(mq_waitq, set_waitq);
231 }
232
233 /*
234 * Routine: ipc_mqueue_remove_from_all
235 * Purpose:
236 * Remove the mqueue from all the sets it is a member of
237 * Conditions:
238 * Nothing locked.
239 * Returns:
240 * mqueue unlocked and set links deallocated
241 */
242 void
243 ipc_mqueue_remove_from_all(ipc_mqueue_t mqueue)
244 {
245 struct waitq *mq_waitq = &mqueue->imq_wait_queue;
246 kern_return_t kr;
247
248 imq_lock(mqueue);
249
250 assert(waitq_valid(mq_waitq));
251 kr = waitq_unlink_all_unlock(mq_waitq);
252 /* mqueue unlocked and set links deallocated */
253 }
254
255 /*
256 * Routine: ipc_mqueue_remove_all
257 * Purpose:
258 * Remove all the member queues from the specified set.
259 * Also removes the queue from any containing sets.
260 * Conditions:
261 * Nothing locked.
262 * Returns:
263 * mqueue unlocked all set links deallocated
264 */
265 void
266 ipc_mqueue_remove_all(ipc_mqueue_t mqueue)
267 {
268 struct waitq_set *mq_setq = &mqueue->imq_set_queue;
269
270 imq_lock(mqueue);
271 assert(waitqs_is_set(mq_setq));
272 waitq_set_unlink_all_unlock(mq_setq);
273 /* mqueue unlocked set links deallocated */
274 }
275
276
277 /*
278 * Routine: ipc_mqueue_add
279 * Purpose:
280 * Associate the portset's mqueue with the port's mqueue.
281 * This has to be done so that posting the port will wakeup
282 * a portset waiter. If there are waiters on the portset
283 * mqueue and messages on the port mqueue, try to match them
284 * up now.
285 * Conditions:
286 * May block.
287 */
288 kern_return_t
289 ipc_mqueue_add(
290 ipc_mqueue_t port_mqueue,
291 ipc_mqueue_t set_mqueue,
292 uint64_t *reserved_link,
293 uint64_t *reserved_prepost)
294 {
295 struct waitq *port_waitq = &port_mqueue->imq_wait_queue;
296 struct waitq_set *set_waitq = &set_mqueue->imq_set_queue;
297 ipc_kmsg_queue_t kmsgq;
298 ipc_kmsg_t kmsg, next;
299 kern_return_t kr;
300
301 assert(reserved_link && *reserved_link != 0);
302 assert(waitqs_is_linked(set_waitq));
303
304 imq_lock(port_mqueue);
305
306 /*
307 * The link operation is now under the same lock-hold as
308 * message iteration and thread wakeup, but doesn't have to be...
309 */
310 kr = waitq_link(port_waitq, set_waitq, WAITQ_ALREADY_LOCKED, reserved_link);
311 if (kr != KERN_SUCCESS) {
312 imq_unlock(port_mqueue);
313 return kr;
314 }
315
316 /*
317 * Now that the set has been added to the port, there may be
318 * messages queued on the port and threads waiting on the set
319 * waitq. Lets get them together.
320 */
321 kmsgq = &port_mqueue->imq_messages;
322 for (kmsg = ipc_kmsg_queue_first(kmsgq);
323 kmsg != IKM_NULL;
324 kmsg = next) {
325 next = ipc_kmsg_queue_next(kmsgq, kmsg);
326
327 for (;;) {
328 thread_t th;
329 mach_msg_size_t msize;
330 spl_t th_spl;
331
332 th = waitq_wakeup64_identify_locked(
333 port_waitq,
334 IPC_MQUEUE_RECEIVE,
335 THREAD_AWAKENED, &th_spl,
336 reserved_prepost, WAITQ_ALL_PRIORITIES,
337 WAITQ_KEEP_LOCKED);
338 /* waitq/mqueue still locked, thread locked */
339
340 if (th == THREAD_NULL) {
341 goto leave;
342 }
343
344 /*
345 * If the receiver waited with a facility not directly
346 * related to Mach messaging, then it isn't prepared to get
347 * handed the message directly. Just set it running, and
348 * go look for another thread that can.
349 */
350 if (th->ith_state != MACH_RCV_IN_PROGRESS) {
351 if (th->ith_state == MACH_PEEK_IN_PROGRESS) {
352 /*
353 * wakeup the peeking thread, but
354 * continue to loop over the threads
355 * waiting on the port's mqueue to see
356 * if there are any actual receivers
357 */
358 ipc_mqueue_peek_on_thread(port_mqueue,
359 th->ith_option,
360 th);
361 }
362 thread_unlock(th);
363 splx(th_spl);
364 continue;
365 }
366
367 /*
368 * Found a receiver. see if they can handle the message
369 * correctly (the message is not too large for them, or
370 * they didn't care to be informed that the message was
371 * too large). If they can't handle it, take them off
372 * the list and let them go back and figure it out and
373 * just move onto the next.
374 */
375 msize = ipc_kmsg_copyout_size(kmsg, th->map);
376 if (th->ith_rsize <
377 (msize + REQUESTED_TRAILER_SIZE(thread_is_64bit_addr(th), th->ith_option))) {
378 th->ith_state = MACH_RCV_TOO_LARGE;
379 th->ith_msize = msize;
380 if (th->ith_option & MACH_RCV_LARGE) {
381 /*
382 * let him go without message
383 */
384 th->ith_receiver_name = port_mqueue->imq_receiver_name;
385 th->ith_kmsg = IKM_NULL;
386 th->ith_seqno = 0;
387 thread_unlock(th);
388 splx(th_spl);
389 continue; /* find another thread */
390 }
391 } else {
392 th->ith_state = MACH_MSG_SUCCESS;
393 }
394
395 /*
396 * This thread is going to take this message,
397 * so give it to him.
398 */
399 ipc_kmsg_rmqueue(kmsgq, kmsg);
400 #if MACH_FLIPC
401 mach_node_t node = kmsg->ikm_node;
402 #endif
403 ipc_mqueue_release_msgcount(port_mqueue, IMQ_NULL);
404
405 th->ith_kmsg = kmsg;
406 th->ith_seqno = port_mqueue->imq_seqno++;
407 thread_unlock(th);
408 splx(th_spl);
409 #if MACH_FLIPC
410 if (MACH_NODE_VALID(node) && FPORT_VALID(port_mqueue->imq_fport)) {
411 flipc_msg_ack(node, port_mqueue, TRUE);
412 }
413 #endif
414 break; /* go to next message */
415 }
416 }
417 leave:
418 imq_unlock(port_mqueue);
419 return KERN_SUCCESS;
420 }
421
422
423 /*
424 * Routine: ipc_mqueue_has_klist
425 * Purpose:
426 * Returns whether the given mqueue imq_klist field can be used as a klist.
427 */
428 static inline bool
429 ipc_mqueue_has_klist(ipc_mqueue_t mqueue)
430 {
431 ipc_object_t object = imq_to_object(mqueue);
432 if (io_otype(object) != IOT_PORT) {
433 return true;
434 }
435 ipc_port_t port = ip_from_mq(mqueue);
436 if (port->ip_specialreply) {
437 return false;
438 }
439 return port->ip_sync_link_state == PORT_SYNC_LINK_ANY;
440 }
441
442 /*
443 * Routine: ipc_mqueue_changed
444 * Purpose:
445 * Wake up receivers waiting in a message queue.
446 * Conditions:
447 * The message queue is locked.
448 */
449 void
450 ipc_mqueue_changed(
451 ipc_space_t space,
452 ipc_mqueue_t mqueue)
453 {
454 if (ipc_mqueue_has_klist(mqueue) && SLIST_FIRST(&mqueue->imq_klist)) {
455 /*
456 * Indicate that this message queue is vanishing
457 *
458 * When this is called, the associated receive right may be in flight
459 * between two tasks: the one it used to live in, and the one that armed
460 * a port destroyed notification for it.
461 *
462 * The new process may want to register the port it gets back with an
463 * EVFILT_MACHPORT filter again, and may have pending sync IPC on this
464 * port pending already, in which case we want the imq_klist field to be
465 * reusable for nefarious purposes.
466 *
467 * Fortunately, we really don't need this linkage anymore after this
468 * point as EV_VANISHED / EV_EOF will be the last thing delivered ever.
469 *
470 * Note: we don't have the space lock here, however, this covers the
471 * case of when a task is terminating the space, triggering
472 * several knote_vanish() calls.
473 *
474 * We don't need the lock to observe that the space is inactive as
475 * we just deactivated it on the same thread.
476 *
477 * We still need to call knote_vanish() so that the knote is
478 * marked with EV_VANISHED or EV_EOF so that the detach step
479 * in filt_machportdetach is skipped correctly.
480 */
481 assert(space);
482 knote_vanish(&mqueue->imq_klist, is_active(space));
483 }
484
485 if (io_otype(imq_to_object(mqueue)) == IOT_PORT) {
486 ipc_port_adjust_sync_link_state_locked(ip_from_mq(mqueue), PORT_SYNC_LINK_ANY, NULL);
487 } else {
488 klist_init(&mqueue->imq_klist);
489 }
490
491 waitq_wakeup64_all_locked(&mqueue->imq_wait_queue,
492 IPC_MQUEUE_RECEIVE,
493 THREAD_RESTART,
494 NULL,
495 WAITQ_ALL_PRIORITIES,
496 WAITQ_KEEP_LOCKED);
497 }
498
499
500
501
502 /*
503 * Routine: ipc_mqueue_send
504 * Purpose:
505 * Send a message to a message queue. The message holds a reference
506 * for the destination port for this message queue in the
507 * msgh_remote_port field.
508 *
509 * If unsuccessful, the caller still has possession of
510 * the message and must do something with it. If successful,
511 * the message is queued, given to a receiver, or destroyed.
512 * Conditions:
513 * mqueue is locked.
514 * Returns:
515 * MACH_MSG_SUCCESS The message was accepted.
516 * MACH_SEND_TIMED_OUT Caller still has message.
517 * MACH_SEND_INTERRUPTED Caller still has message.
518 */
519 mach_msg_return_t
520 ipc_mqueue_send(
521 ipc_mqueue_t mqueue,
522 ipc_kmsg_t kmsg,
523 mach_msg_option_t option,
524 mach_msg_timeout_t send_timeout)
525 {
526 int wresult;
527
528 /*
529 * Don't block if:
530 * 1) We're under the queue limit.
531 * 2) Caller used the MACH_SEND_ALWAYS internal option.
532 * 3) Message is sent to a send-once right.
533 */
534 if (!imq_full(mqueue) ||
535 (!imq_full_kernel(mqueue) &&
536 ((option & MACH_SEND_ALWAYS) ||
537 (MACH_MSGH_BITS_REMOTE(kmsg->ikm_header->msgh_bits) ==
538 MACH_MSG_TYPE_PORT_SEND_ONCE)))) {
539 mqueue->imq_msgcount++;
540 assert(mqueue->imq_msgcount > 0);
541 imq_unlock(mqueue);
542 } else {
543 thread_t cur_thread = current_thread();
544 ipc_port_t port = ip_from_mq(mqueue);
545 struct turnstile *send_turnstile = TURNSTILE_NULL;
546 uint64_t deadline;
547
548 /*
549 * We have to wait for space to be granted to us.
550 */
551 if ((option & MACH_SEND_TIMEOUT) && (send_timeout == 0)) {
552 imq_unlock(mqueue);
553 return MACH_SEND_TIMED_OUT;
554 }
555 if (imq_full_kernel(mqueue)) {
556 imq_unlock(mqueue);
557 return MACH_SEND_NO_BUFFER;
558 }
559 mqueue->imq_fullwaiters = TRUE;
560
561 if (option & MACH_SEND_TIMEOUT) {
562 clock_interval_to_deadline(send_timeout, 1000 * NSEC_PER_USEC, &deadline);
563 } else {
564 deadline = 0;
565 }
566
567 thread_set_pending_block_hint(cur_thread, kThreadWaitPortSend);
568
569 send_turnstile = turnstile_prepare((uintptr_t)port,
570 port_send_turnstile_address(port),
571 TURNSTILE_NULL, TURNSTILE_SYNC_IPC);
572
573 ipc_port_send_update_inheritor(port, send_turnstile,
574 TURNSTILE_DELAYED_UPDATE);
575
576 wresult = waitq_assert_wait64_leeway(
577 &send_turnstile->ts_waitq,
578 IPC_MQUEUE_FULL,
579 THREAD_ABORTSAFE,
580 TIMEOUT_URGENCY_USER_NORMAL,
581 deadline,
582 TIMEOUT_NO_LEEWAY);
583
584 imq_unlock(mqueue);
585 turnstile_update_inheritor_complete(send_turnstile,
586 TURNSTILE_INTERLOCK_NOT_HELD);
587
588 if (wresult == THREAD_WAITING) {
589 wresult = thread_block(THREAD_CONTINUE_NULL);
590 counter(c_ipc_mqueue_send_block++);
591 }
592
593 /* Call turnstile complete with interlock held */
594 imq_lock(mqueue);
595 turnstile_complete((uintptr_t)port, port_send_turnstile_address(port), NULL, TURNSTILE_SYNC_IPC);
596 imq_unlock(mqueue);
597
598 /* Call cleanup after dropping the interlock */
599 turnstile_cleanup();
600
601 switch (wresult) {
602 case THREAD_AWAKENED:
603 /*
604 * we can proceed - inherited msgcount from waker
605 * or the message queue has been destroyed and the msgcount
606 * has been reset to zero (will detect in ipc_mqueue_post()).
607 */
608 break;
609
610 case THREAD_TIMED_OUT:
611 assert(option & MACH_SEND_TIMEOUT);
612 return MACH_SEND_TIMED_OUT;
613
614 case THREAD_INTERRUPTED:
615 return MACH_SEND_INTERRUPTED;
616
617 case THREAD_RESTART:
618 /* mqueue is being destroyed */
619 return MACH_SEND_INVALID_DEST;
620 default:
621 panic("ipc_mqueue_send");
622 }
623 }
624
625 ipc_mqueue_post(mqueue, kmsg, option);
626 return MACH_MSG_SUCCESS;
627 }
628
629 /*
630 * Routine: ipc_mqueue_override_send
631 * Purpose:
632 * Set an override qos on the first message in the queue
633 * (if the queue is full). This is a send-possible override
634 * that will go away as soon as we drain a message from the
635 * queue.
636 *
637 * Conditions:
638 * The message queue is not locked.
639 * The caller holds a reference on the message queue.
640 */
641 extern void
642 ipc_mqueue_override_send(
643 ipc_mqueue_t mqueue,
644 mach_msg_priority_t override)
645 {
646 boolean_t __unused full_queue_empty = FALSE;
647
648 imq_lock(mqueue);
649 assert(imq_valid(mqueue));
650 assert(!imq_is_set(mqueue));
651
652 if (imq_full(mqueue)) {
653 ipc_kmsg_t first = ipc_kmsg_queue_first(&mqueue->imq_messages);
654
655 if (first && ipc_kmsg_override_qos(&mqueue->imq_messages, first, override)) {
656 ipc_object_t object = imq_to_object(mqueue);
657 assert(io_otype(object) == IOT_PORT);
658 ipc_port_t port = ip_object_to_port(object);
659 if (ip_active(port) &&
660 port->ip_receiver_name != MACH_PORT_NULL &&
661 is_active(port->ip_receiver) &&
662 ipc_mqueue_has_klist(mqueue)) {
663 KNOTE(&mqueue->imq_klist, 0);
664 }
665 }
666 if (!first) {
667 full_queue_empty = TRUE;
668 }
669 }
670 imq_unlock(mqueue);
671
672 #if DEVELOPMENT || DEBUG
673 if (full_queue_empty) {
674 ipc_port_t port = ip_from_mq(mqueue);
675 int dst_pid = 0;
676 if (ip_active(port) && !port->ip_tempowner &&
677 port->ip_receiver_name && port->ip_receiver &&
678 port->ip_receiver != ipc_space_kernel) {
679 dst_pid = task_pid(port->ip_receiver->is_task);
680 }
681 }
682 #endif
683 }
684
685 /*
686 * Routine: ipc_mqueue_release_msgcount
687 * Purpose:
688 * Release a message queue reference in the case where we
689 * found a waiter.
690 *
691 * Conditions:
692 * The message queue is locked.
693 * The message corresponding to this reference is off the queue.
694 * There is no need to pass reserved preposts because this will
695 * never prepost to anyone
696 */
697 void
698 ipc_mqueue_release_msgcount(ipc_mqueue_t port_mq, ipc_mqueue_t set_mq)
699 {
700 struct turnstile *send_turnstile = port_send_turnstile(ip_from_mq(port_mq));
701 (void)set_mq;
702 assert(imq_held(port_mq));
703 assert(port_mq->imq_msgcount > 1 || ipc_kmsg_queue_empty(&port_mq->imq_messages));
704
705 port_mq->imq_msgcount--;
706
707 if (!imq_full(port_mq) && port_mq->imq_fullwaiters &&
708 send_turnstile != TURNSTILE_NULL) {
709 /*
710 * boost the priority of the awoken thread
711 * (WAITQ_PROMOTE_PRIORITY) to ensure it uses
712 * the message queue slot we've just reserved.
713 *
714 * NOTE: this will never prepost
715 *
716 * The wakeup happens on a turnstile waitq
717 * which will wakeup the highest priority waiter.
718 * A potential downside of this would be starving low
719 * priority senders if there is a constant churn of
720 * high priority threads trying to send to this port.
721 */
722 if (waitq_wakeup64_one(&send_turnstile->ts_waitq,
723 IPC_MQUEUE_FULL,
724 THREAD_AWAKENED,
725 WAITQ_PROMOTE_PRIORITY) != KERN_SUCCESS) {
726 port_mq->imq_fullwaiters = FALSE;
727 } else {
728 /* gave away our slot - add reference back */
729 port_mq->imq_msgcount++;
730 }
731 }
732
733 if (ipc_kmsg_queue_empty(&port_mq->imq_messages)) {
734 /* no more msgs: invalidate the port's prepost object */
735 waitq_clear_prepost_locked(&port_mq->imq_wait_queue);
736 }
737 }
738
739 /*
740 * Routine: ipc_mqueue_post
741 * Purpose:
742 * Post a message to a waiting receiver or enqueue it. If a
743 * receiver is waiting, we can release our reserved space in
744 * the message queue.
745 *
746 * Conditions:
747 * mqueue is unlocked
748 * If we need to queue, our space in the message queue is reserved.
749 */
750 void
751 ipc_mqueue_post(
752 ipc_mqueue_t mqueue,
753 ipc_kmsg_t kmsg,
754 mach_msg_option_t __unused option)
755 {
756 uint64_t reserved_prepost = 0;
757 boolean_t destroy_msg = FALSE;
758
759 ipc_kmsg_trace_send(kmsg, option);
760
761 /*
762 * While the msg queue is locked, we have control of the
763 * kmsg, so the ref in it for the port is still good.
764 *
765 * Check for a receiver for the message.
766 */
767 imq_reserve_and_lock(mqueue, &reserved_prepost);
768
769 /* we may have raced with port destruction! */
770 if (!imq_valid(mqueue)) {
771 destroy_msg = TRUE;
772 goto out_unlock;
773 }
774
775 for (;;) {
776 struct waitq *waitq = &mqueue->imq_wait_queue;
777 spl_t th_spl;
778 thread_t receiver;
779 mach_msg_size_t msize;
780
781 receiver = waitq_wakeup64_identify_locked(waitq,
782 IPC_MQUEUE_RECEIVE,
783 THREAD_AWAKENED,
784 &th_spl,
785 &reserved_prepost,
786 WAITQ_ALL_PRIORITIES,
787 WAITQ_KEEP_LOCKED);
788 /* waitq still locked, thread locked */
789
790 if (receiver == THREAD_NULL) {
791 /*
792 * no receivers; queue kmsg if space still reserved
793 * Reservations are cancelled when the port goes inactive.
794 * note that this will enqueue the message for any
795 * "peeking" receivers.
796 *
797 * Also, post the knote to wake up any threads waiting
798 * on that style of interface if this insertion is of
799 * note (first insertion, or adjusted override qos all
800 * the way to the head of the queue).
801 *
802 * This is just for ports. portset knotes are stay-active,
803 * and their threads get awakened through the !MACH_RCV_IN_PROGRESS
804 * logic below).
805 */
806 if (mqueue->imq_msgcount > 0) {
807 if (ipc_kmsg_enqueue_qos(&mqueue->imq_messages, kmsg)) {
808 /* if the space is dead there is no point calling KNOTE */
809 ipc_object_t object = imq_to_object(mqueue);
810 assert(io_otype(object) == IOT_PORT);
811 ipc_port_t port = ip_object_to_port(object);
812 if (ip_active(port) &&
813 port->ip_receiver_name != MACH_PORT_NULL &&
814 is_active(port->ip_receiver) &&
815 ipc_mqueue_has_klist(mqueue)) {
816 KNOTE(&mqueue->imq_klist, 0);
817 }
818 }
819 break;
820 }
821
822 /*
823 * Otherwise, the message queue must belong to an inactive
824 * port, so just destroy the message and pretend it was posted.
825 */
826 destroy_msg = TRUE;
827 goto out_unlock;
828 }
829
830 /*
831 * If a thread is attempting a "peek" into the message queue
832 * (MACH_PEEK_IN_PROGRESS), then we enqueue the message and set the
833 * thread running. A successful peek is essentially the same as
834 * message delivery since the peeking thread takes responsibility
835 * for delivering the message and (eventually) removing it from
836 * the mqueue. Only one thread can successfully use the peek
837 * facility on any given port, so we exit the waitq loop after
838 * encountering such a thread.
839 */
840 if (receiver->ith_state == MACH_PEEK_IN_PROGRESS && mqueue->imq_msgcount > 0) {
841 ipc_kmsg_enqueue_qos(&mqueue->imq_messages, kmsg);
842 ipc_mqueue_peek_on_thread(mqueue, receiver->ith_option, receiver);
843 thread_unlock(receiver);
844 splx(th_spl);
845 break; /* Message was posted, so break out of loop */
846 }
847
848 /*
849 * If the receiver waited with a facility not directly related
850 * to Mach messaging, then it isn't prepared to get handed the
851 * message directly. Just set it running, and go look for
852 * another thread that can.
853 */
854 if (receiver->ith_state != MACH_RCV_IN_PROGRESS) {
855 thread_unlock(receiver);
856 splx(th_spl);
857 continue;
858 }
859
860
861 /*
862 * We found a waiting thread.
863 * If the message is too large or the scatter list is too small
864 * the thread we wake up will get that as its status.
865 */
866 msize = ipc_kmsg_copyout_size(kmsg, receiver->map);
867 if (receiver->ith_rsize <
868 (msize + REQUESTED_TRAILER_SIZE(thread_is_64bit_addr(receiver), receiver->ith_option))) {
869 receiver->ith_msize = msize;
870 receiver->ith_state = MACH_RCV_TOO_LARGE;
871 } else {
872 receiver->ith_state = MACH_MSG_SUCCESS;
873 }
874
875 /*
876 * If there is no problem with the upcoming receive, or the
877 * receiver thread didn't specifically ask for special too
878 * large error condition, go ahead and select it anyway.
879 */
880 if ((receiver->ith_state == MACH_MSG_SUCCESS) ||
881 !(receiver->ith_option & MACH_RCV_LARGE)) {
882 receiver->ith_kmsg = kmsg;
883 receiver->ith_seqno = mqueue->imq_seqno++;
884 #if MACH_FLIPC
885 mach_node_t node = kmsg->ikm_node;
886 #endif
887 thread_unlock(receiver);
888 splx(th_spl);
889
890 /* we didn't need our reserved spot in the queue */
891 ipc_mqueue_release_msgcount(mqueue, IMQ_NULL);
892
893 #if MACH_FLIPC
894 if (MACH_NODE_VALID(node) && FPORT_VALID(mqueue->imq_fport)) {
895 flipc_msg_ack(node, mqueue, TRUE);
896 }
897 #endif
898 break;
899 }
900
901 /*
902 * Otherwise, this thread needs to be released to run
903 * and handle its error without getting the message. We
904 * need to go back and pick another one.
905 */
906 receiver->ith_receiver_name = mqueue->imq_receiver_name;
907 receiver->ith_kmsg = IKM_NULL;
908 receiver->ith_seqno = 0;
909 thread_unlock(receiver);
910 splx(th_spl);
911 }
912
913 out_unlock:
914 /* clear the waitq boost we may have been given */
915 waitq_clear_promotion_locked(&mqueue->imq_wait_queue, current_thread());
916 imq_release_and_unlock(mqueue, reserved_prepost);
917 if (destroy_msg) {
918 ipc_kmsg_destroy(kmsg);
919 }
920
921 current_task()->messages_sent++;
922 return;
923 }
924
925
926 static void
927 ipc_mqueue_receive_results(wait_result_t saved_wait_result)
928 {
929 thread_t self = current_thread();
930 mach_msg_option_t option = self->ith_option;
931
932 /*
933 * why did we wake up?
934 */
935 switch (saved_wait_result) {
936 case THREAD_TIMED_OUT:
937 self->ith_state = MACH_RCV_TIMED_OUT;
938 return;
939
940 case THREAD_INTERRUPTED:
941 self->ith_state = MACH_RCV_INTERRUPTED;
942 return;
943
944 case THREAD_RESTART:
945 /* something bad happened to the port/set */
946 self->ith_state = MACH_RCV_PORT_CHANGED;
947 return;
948
949 case THREAD_AWAKENED:
950 /*
951 * We do not need to go select a message, somebody
952 * handed us one (or a too-large indication).
953 */
954 switch (self->ith_state) {
955 case MACH_RCV_SCATTER_SMALL:
956 case MACH_RCV_TOO_LARGE:
957 /*
958 * Somebody tried to give us a too large
959 * message. If we indicated that we cared,
960 * then they only gave us the indication,
961 * otherwise they gave us the indication
962 * AND the message anyway.
963 */
964 if (option & MACH_RCV_LARGE) {
965 return;
966 }
967
968 case MACH_MSG_SUCCESS:
969 case MACH_PEEK_READY:
970 return;
971
972 default:
973 panic("ipc_mqueue_receive_results: strange ith_state");
974 }
975
976 default:
977 panic("ipc_mqueue_receive_results: strange wait_result");
978 }
979 }
980
981 void
982 ipc_mqueue_receive_continue(
983 __unused void *param,
984 wait_result_t wresult)
985 {
986 ipc_mqueue_receive_results(wresult);
987 mach_msg_receive_continue(); /* hard-coded for now */
988 }
989
990 /*
991 * Routine: ipc_mqueue_receive
992 * Purpose:
993 * Receive a message from a message queue.
994 *
995 * Conditions:
996 * Our caller must hold a reference for the port or port set
997 * to which this queue belongs, to keep the queue
998 * from being deallocated.
999 *
1000 * The kmsg is returned with clean header fields
1001 * and with the circular bit turned off through the ith_kmsg
1002 * field of the thread's receive continuation state.
1003 * Returns:
1004 * MACH_MSG_SUCCESS Message returned in ith_kmsg.
1005 * MACH_RCV_TOO_LARGE Message size returned in ith_msize.
1006 * MACH_RCV_TIMED_OUT No message obtained.
1007 * MACH_RCV_INTERRUPTED No message obtained.
1008 * MACH_RCV_PORT_DIED Port/set died; no message.
1009 * MACH_RCV_PORT_CHANGED Port moved into set; no msg.
1010 *
1011 */
1012
1013 void
1014 ipc_mqueue_receive(
1015 ipc_mqueue_t mqueue,
1016 mach_msg_option_t option,
1017 mach_msg_size_t max_size,
1018 mach_msg_timeout_t rcv_timeout,
1019 int interruptible)
1020 {
1021 wait_result_t wresult;
1022 thread_t self = current_thread();
1023
1024 imq_lock(mqueue);
1025 wresult = ipc_mqueue_receive_on_thread(mqueue, option, max_size,
1026 rcv_timeout, interruptible,
1027 self);
1028 /* mqueue unlocked */
1029 if (wresult == THREAD_NOT_WAITING) {
1030 return;
1031 }
1032
1033 if (wresult == THREAD_WAITING) {
1034 counter((interruptible == THREAD_ABORTSAFE) ?
1035 c_ipc_mqueue_receive_block_user++ :
1036 c_ipc_mqueue_receive_block_kernel++);
1037
1038 if (self->ith_continuation) {
1039 thread_block(ipc_mqueue_receive_continue);
1040 }
1041 /* NOTREACHED */
1042
1043 wresult = thread_block(THREAD_CONTINUE_NULL);
1044 }
1045 ipc_mqueue_receive_results(wresult);
1046 }
1047
1048 static int
1049 mqueue_process_prepost_receive(void *ctx, struct waitq *waitq,
1050 struct waitq_set *wqset)
1051 {
1052 ipc_mqueue_t port_mq, *pmq_ptr;
1053
1054 (void)wqset;
1055 port_mq = (ipc_mqueue_t)waitq;
1056
1057 /*
1058 * If there are no messages on this queue, skip it and remove
1059 * it from the prepost list
1060 */
1061 if (ipc_kmsg_queue_empty(&port_mq->imq_messages)) {
1062 return WQ_ITERATE_INVALIDATE_CONTINUE;
1063 }
1064
1065 /*
1066 * There are messages waiting on this port.
1067 * Instruct the prepost iteration logic to break, but keep the
1068 * waitq locked.
1069 */
1070 pmq_ptr = (ipc_mqueue_t *)ctx;
1071 if (pmq_ptr) {
1072 *pmq_ptr = port_mq;
1073 }
1074 return WQ_ITERATE_BREAK_KEEP_LOCKED;
1075 }
1076
1077 /*
1078 * Routine: ipc_mqueue_receive_on_thread
1079 * Purpose:
1080 * Receive a message from a message queue using a specified thread.
1081 * If no message available, assert_wait on the appropriate waitq.
1082 *
1083 * Conditions:
1084 * Assumes thread is self.
1085 * Called with mqueue locked.
1086 * Returns with mqueue unlocked.
1087 * May have assert-waited. Caller must block in those cases.
1088 */
1089 wait_result_t
1090 ipc_mqueue_receive_on_thread(
1091 ipc_mqueue_t mqueue,
1092 mach_msg_option_t option,
1093 mach_msg_size_t max_size,
1094 mach_msg_timeout_t rcv_timeout,
1095 int interruptible,
1096 thread_t thread)
1097 {
1098 wait_result_t wresult;
1099 uint64_t deadline;
1100 struct turnstile *rcv_turnstile = TURNSTILE_NULL;
1101
1102 /* called with mqueue locked */
1103
1104 /* no need to reserve anything: we never prepost to anyone */
1105
1106 if (!imq_valid(mqueue)) {
1107 /* someone raced us to destroy this mqueue/port! */
1108 imq_unlock(mqueue);
1109 /*
1110 * ipc_mqueue_receive_results updates the thread's ith_state
1111 * TODO: differentiate between rights being moved and
1112 * rights/ports being destroyed (21885327)
1113 */
1114 return THREAD_RESTART;
1115 }
1116
1117 if (imq_is_set(mqueue)) {
1118 ipc_mqueue_t port_mq = IMQ_NULL;
1119
1120 (void)waitq_set_iterate_preposts(&mqueue->imq_set_queue,
1121 &port_mq,
1122 mqueue_process_prepost_receive);
1123
1124 if (port_mq != IMQ_NULL) {
1125 /*
1126 * We get here if there is at least one message
1127 * waiting on port_mq. We have instructed the prepost
1128 * iteration logic to leave both the port_mq and the
1129 * set mqueue locked.
1130 *
1131 * TODO: previously, we would place this port at the
1132 * back of the prepost list...
1133 */
1134 imq_unlock(mqueue);
1135
1136 /*
1137 * Continue on to handling the message with just
1138 * the port mqueue locked.
1139 */
1140 if (option & MACH_PEEK_MSG) {
1141 ipc_mqueue_peek_on_thread(port_mq, option, thread);
1142 } else {
1143 ipc_mqueue_select_on_thread(port_mq, mqueue, option,
1144 max_size, thread);
1145 }
1146
1147 imq_unlock(port_mq);
1148 return THREAD_NOT_WAITING;
1149 }
1150 } else if (imq_is_queue(mqueue)) {
1151 ipc_kmsg_queue_t kmsgs;
1152
1153 /*
1154 * Receive on a single port. Just try to get the messages.
1155 */
1156 kmsgs = &mqueue->imq_messages;
1157 if (ipc_kmsg_queue_first(kmsgs) != IKM_NULL) {
1158 if (option & MACH_PEEK_MSG) {
1159 ipc_mqueue_peek_on_thread(mqueue, option, thread);
1160 } else {
1161 ipc_mqueue_select_on_thread(mqueue, IMQ_NULL, option,
1162 max_size, thread);
1163 }
1164 imq_unlock(mqueue);
1165 return THREAD_NOT_WAITING;
1166 }
1167 } else {
1168 panic("Unknown mqueue type 0x%x: likely memory corruption!\n",
1169 mqueue->imq_wait_queue.waitq_type);
1170 }
1171
1172 /*
1173 * Looks like we'll have to block. The mqueue we will
1174 * block on (whether the set's or the local port's) is
1175 * still locked.
1176 */
1177 if (option & MACH_RCV_TIMEOUT) {
1178 if (rcv_timeout == 0) {
1179 imq_unlock(mqueue);
1180 thread->ith_state = MACH_RCV_TIMED_OUT;
1181 return THREAD_NOT_WAITING;
1182 }
1183 }
1184
1185 thread->ith_option = option;
1186 thread->ith_rsize = max_size;
1187 thread->ith_msize = 0;
1188
1189 if (option & MACH_PEEK_MSG) {
1190 thread->ith_state = MACH_PEEK_IN_PROGRESS;
1191 } else {
1192 thread->ith_state = MACH_RCV_IN_PROGRESS;
1193 }
1194
1195 if (option & MACH_RCV_TIMEOUT) {
1196 clock_interval_to_deadline(rcv_timeout, 1000 * NSEC_PER_USEC, &deadline);
1197 } else {
1198 deadline = 0;
1199 }
1200
1201 /*
1202 * Threads waiting on a special reply port
1203 * (not portset or regular ports)
1204 * will wait on its receive turnstile.
1205 *
1206 * Donate waiting thread's turnstile and
1207 * setup inheritor for special reply port.
1208 * Based on the state of the special reply
1209 * port, the inheritor would be the send
1210 * turnstile of the connection port on which
1211 * the send of sync ipc would happen or
1212 * workloop's turnstile who would reply to
1213 * the sync ipc message.
1214 *
1215 * Pass in mqueue wait in waitq_assert_wait to
1216 * support port set wakeup. The mqueue waitq of port
1217 * will be converted to to turnstile waitq
1218 * in waitq_assert_wait instead of global waitqs.
1219 */
1220 if (imq_is_queue(mqueue) && ip_from_mq(mqueue)->ip_specialreply) {
1221 ipc_port_t port = ip_from_mq(mqueue);
1222 rcv_turnstile = turnstile_prepare((uintptr_t)port,
1223 port_rcv_turnstile_address(port),
1224 TURNSTILE_NULL, TURNSTILE_SYNC_IPC);
1225
1226 ipc_port_recv_update_inheritor(port, rcv_turnstile,
1227 TURNSTILE_DELAYED_UPDATE);
1228 }
1229
1230 thread_set_pending_block_hint(thread, kThreadWaitPortReceive);
1231 wresult = waitq_assert_wait64_locked(&mqueue->imq_wait_queue,
1232 IPC_MQUEUE_RECEIVE,
1233 interruptible,
1234 TIMEOUT_URGENCY_USER_NORMAL,
1235 deadline,
1236 TIMEOUT_NO_LEEWAY,
1237 thread);
1238 /* preposts should be detected above, not here */
1239 if (wresult == THREAD_AWAKENED) {
1240 panic("ipc_mqueue_receive_on_thread: sleep walking");
1241 }
1242
1243 imq_unlock(mqueue);
1244
1245 /* Check if its a port mqueue and if it needs to call turnstile_update_inheritor_complete */
1246 if (rcv_turnstile != TURNSTILE_NULL) {
1247 turnstile_update_inheritor_complete(rcv_turnstile, TURNSTILE_INTERLOCK_NOT_HELD);
1248 }
1249 /* Its callers responsibility to call turnstile_complete to get the turnstile back */
1250
1251 return wresult;
1252 }
1253
1254
1255 /*
1256 * Routine: ipc_mqueue_peek_on_thread
1257 * Purpose:
1258 * A receiver discovered that there was a message on the queue
1259 * before he had to block. Tell a thread about the message queue,
1260 * but don't pick off any messages.
1261 * Conditions:
1262 * port_mq locked
1263 * at least one message on port_mq's message queue
1264 *
1265 * Returns: (on thread->ith_state)
1266 * MACH_PEEK_READY ith_peekq contains a message queue
1267 */
1268 void
1269 ipc_mqueue_peek_on_thread(
1270 ipc_mqueue_t port_mq,
1271 mach_msg_option_t option,
1272 thread_t thread)
1273 {
1274 (void)option;
1275 assert(option & MACH_PEEK_MSG);
1276 assert(ipc_kmsg_queue_first(&port_mq->imq_messages) != IKM_NULL);
1277
1278 /*
1279 * Take a reference on the mqueue's associated port:
1280 * the peeking thread will be responsible to release this reference
1281 * using ip_release_mq()
1282 */
1283 ip_reference_mq(port_mq);
1284 thread->ith_peekq = port_mq;
1285 thread->ith_state = MACH_PEEK_READY;
1286 }
1287
1288 /*
1289 * Routine: ipc_mqueue_select_on_thread
1290 * Purpose:
1291 * A receiver discovered that there was a message on the queue
1292 * before he had to block. Pick the message off the queue and
1293 * "post" it to thread.
1294 * Conditions:
1295 * mqueue locked.
1296 * thread not locked.
1297 * There is a message.
1298 * No need to reserve prepost objects - it will never prepost
1299 *
1300 * Returns:
1301 * MACH_MSG_SUCCESS Actually selected a message for ourselves.
1302 * MACH_RCV_TOO_LARGE May or may not have pull it, but it is large
1303 */
1304 void
1305 ipc_mqueue_select_on_thread(
1306 ipc_mqueue_t port_mq,
1307 ipc_mqueue_t set_mq,
1308 mach_msg_option_t option,
1309 mach_msg_size_t max_size,
1310 thread_t thread)
1311 {
1312 ipc_kmsg_t kmsg;
1313 mach_msg_return_t mr = MACH_MSG_SUCCESS;
1314 mach_msg_size_t msize;
1315
1316 /*
1317 * Do some sanity checking of our ability to receive
1318 * before pulling the message off the queue.
1319 */
1320 kmsg = ipc_kmsg_queue_first(&port_mq->imq_messages);
1321 assert(kmsg != IKM_NULL);
1322
1323 /*
1324 * If we really can't receive it, but we had the
1325 * MACH_RCV_LARGE option set, then don't take it off
1326 * the queue, instead return the appropriate error
1327 * (and size needed).
1328 */
1329 msize = ipc_kmsg_copyout_size(kmsg, thread->map);
1330 if (msize + REQUESTED_TRAILER_SIZE(thread_is_64bit_addr(thread), option) > max_size) {
1331 mr = MACH_RCV_TOO_LARGE;
1332 if (option & MACH_RCV_LARGE) {
1333 thread->ith_receiver_name = port_mq->imq_receiver_name;
1334 thread->ith_kmsg = IKM_NULL;
1335 thread->ith_msize = msize;
1336 thread->ith_seqno = 0;
1337 thread->ith_state = mr;
1338 return;
1339 }
1340 }
1341
1342 ipc_kmsg_rmqueue(&port_mq->imq_messages, kmsg);
1343 #if MACH_FLIPC
1344 if (MACH_NODE_VALID(kmsg->ikm_node) && FPORT_VALID(port_mq->imq_fport)) {
1345 flipc_msg_ack(kmsg->ikm_node, port_mq, TRUE);
1346 }
1347 #endif
1348 ipc_mqueue_release_msgcount(port_mq, set_mq);
1349 thread->ith_seqno = port_mq->imq_seqno++;
1350 thread->ith_kmsg = kmsg;
1351 thread->ith_state = mr;
1352
1353 current_task()->messages_received++;
1354 return;
1355 }
1356
1357 /*
1358 * Routine: ipc_mqueue_peek_locked
1359 * Purpose:
1360 * Peek at a (non-set) message queue to see if it has a message
1361 * matching the sequence number provided (if zero, then the
1362 * first message in the queue) and return vital info about the
1363 * message.
1364 *
1365 * Conditions:
1366 * The ipc_mqueue_t is locked by callers.
1367 * Other locks may be held by callers, so this routine cannot block.
1368 * Caller holds reference on the message queue.
1369 */
1370 unsigned
1371 ipc_mqueue_peek_locked(ipc_mqueue_t mq,
1372 mach_port_seqno_t * seqnop,
1373 mach_msg_size_t * msg_sizep,
1374 mach_msg_id_t * msg_idp,
1375 mach_msg_max_trailer_t * msg_trailerp,
1376 ipc_kmsg_t *kmsgp)
1377 {
1378 ipc_kmsg_queue_t kmsgq;
1379 ipc_kmsg_t kmsg;
1380 mach_port_seqno_t seqno, msgoff;
1381 unsigned res = 0;
1382
1383 assert(!imq_is_set(mq));
1384
1385 seqno = 0;
1386 if (seqnop != NULL) {
1387 seqno = *seqnop;
1388 }
1389
1390 if (seqno == 0) {
1391 seqno = mq->imq_seqno;
1392 msgoff = 0;
1393 } else if (seqno >= mq->imq_seqno &&
1394 seqno < mq->imq_seqno + mq->imq_msgcount) {
1395 msgoff = seqno - mq->imq_seqno;
1396 } else {
1397 goto out;
1398 }
1399
1400 /* look for the message that would match that seqno */
1401 kmsgq = &mq->imq_messages;
1402 kmsg = ipc_kmsg_queue_first(kmsgq);
1403 while (msgoff-- && kmsg != IKM_NULL) {
1404 kmsg = ipc_kmsg_queue_next(kmsgq, kmsg);
1405 }
1406 if (kmsg == IKM_NULL) {
1407 goto out;
1408 }
1409
1410 /* found one - return the requested info */
1411 if (seqnop != NULL) {
1412 *seqnop = seqno;
1413 }
1414 if (msg_sizep != NULL) {
1415 *msg_sizep = kmsg->ikm_header->msgh_size;
1416 }
1417 if (msg_idp != NULL) {
1418 *msg_idp = kmsg->ikm_header->msgh_id;
1419 }
1420 if (msg_trailerp != NULL) {
1421 memcpy(msg_trailerp,
1422 (mach_msg_max_trailer_t *)((vm_offset_t)kmsg->ikm_header +
1423 round_msg(kmsg->ikm_header->msgh_size)),
1424 sizeof(mach_msg_max_trailer_t));
1425 }
1426 if (kmsgp != NULL) {
1427 *kmsgp = kmsg;
1428 }
1429
1430 res = 1;
1431
1432 out:
1433 return res;
1434 }
1435
1436
1437 /*
1438 * Routine: ipc_mqueue_peek
1439 * Purpose:
1440 * Peek at a (non-set) message queue to see if it has a message
1441 * matching the sequence number provided (if zero, then the
1442 * first message in the queue) and return vital info about the
1443 * message.
1444 *
1445 * Conditions:
1446 * The ipc_mqueue_t is unlocked.
1447 * Locks may be held by callers, so this routine cannot block.
1448 * Caller holds reference on the message queue.
1449 */
1450 unsigned
1451 ipc_mqueue_peek(ipc_mqueue_t mq,
1452 mach_port_seqno_t * seqnop,
1453 mach_msg_size_t * msg_sizep,
1454 mach_msg_id_t * msg_idp,
1455 mach_msg_max_trailer_t * msg_trailerp,
1456 ipc_kmsg_t *kmsgp)
1457 {
1458 unsigned res;
1459
1460 imq_lock(mq);
1461
1462 res = ipc_mqueue_peek_locked(mq, seqnop, msg_sizep, msg_idp,
1463 msg_trailerp, kmsgp);
1464
1465 imq_unlock(mq);
1466 return res;
1467 }
1468
1469 /*
1470 * Routine: ipc_mqueue_release_peek_ref
1471 * Purpose:
1472 * Release the reference on an mqueue's associated port which was
1473 * granted to a thread in ipc_mqueue_peek_on_thread (on the
1474 * MACH_PEEK_MSG thread wakeup path).
1475 *
1476 * Conditions:
1477 * The ipc_mqueue_t should be locked on entry.
1478 * The ipc_mqueue_t will be _unlocked_ on return
1479 * (and potentially invalid!)
1480 *
1481 */
1482 void
1483 ipc_mqueue_release_peek_ref(ipc_mqueue_t mq)
1484 {
1485 assert(!imq_is_set(mq));
1486 assert(imq_held(mq));
1487
1488 /*
1489 * clear any preposts this mq may have generated
1490 * (which would cause subsequent immediate wakeups)
1491 */
1492 waitq_clear_prepost_locked(&mq->imq_wait_queue);
1493
1494 imq_unlock(mq);
1495
1496 /*
1497 * release the port reference: we need to do this outside the lock
1498 * because we might be holding the last port reference!
1499 **/
1500 ip_release_mq(mq);
1501 }
1502
1503 /*
1504 * peek at the contained port message queues, break prepost iteration as soon
1505 * as we spot a message on one of the message queues referenced by the set's
1506 * prepost list. No need to lock each message queue, as only the head of each
1507 * queue is checked. If a message wasn't there before we entered here, no need
1508 * to find it (if we do, great).
1509 */
1510 static int
1511 mqueue_peek_iterator(void *ctx, struct waitq *waitq,
1512 struct waitq_set *wqset)
1513 {
1514 ipc_mqueue_t port_mq = (ipc_mqueue_t)waitq;
1515 ipc_kmsg_queue_t kmsgs = &port_mq->imq_messages;
1516
1517 (void)ctx;
1518 (void)wqset;
1519
1520 if (ipc_kmsg_queue_first(kmsgs) != IKM_NULL) {
1521 return WQ_ITERATE_BREAK; /* break out of the prepost iteration */
1522 }
1523 return WQ_ITERATE_CONTINUE;
1524 }
1525
1526 /*
1527 * Routine: ipc_mqueue_set_peek
1528 * Purpose:
1529 * Peek at a message queue set to see if it has any ports
1530 * with messages.
1531 *
1532 * Conditions:
1533 * Locks may be held by callers, so this routine cannot block.
1534 * Caller holds reference on the message queue.
1535 */
1536 unsigned
1537 ipc_mqueue_set_peek(ipc_mqueue_t mq)
1538 {
1539 int ret;
1540
1541 imq_lock(mq);
1542
1543 /*
1544 * We may have raced with port destruction where the mqueue is marked
1545 * as invalid. In that case, even though we don't have messages, we
1546 * have an end-of-life event to deliver.
1547 */
1548 if (!imq_is_valid(mq)) {
1549 return 1;
1550 }
1551
1552 ret = waitq_set_iterate_preposts(&mq->imq_set_queue, NULL,
1553 mqueue_peek_iterator);
1554
1555 imq_unlock(mq);
1556
1557 return ret == WQ_ITERATE_BREAK;
1558 }
1559
1560 /*
1561 * Routine: ipc_mqueue_set_gather_member_names
1562 * Purpose:
1563 * Discover all ports which are members of a given port set.
1564 * Because the waitq linkage mechanism was redesigned to save
1565 * significan amounts of memory, it no longer keeps back-pointers
1566 * from a port set to a port. Therefore, we must iterate over all
1567 * ports within a given IPC space and individually query them to
1568 * see if they are members of the given set. Port names of ports
1569 * found to be members of the given set will be gathered into the
1570 * provided 'names' array. Actual returned names are limited to
1571 * maxnames entries, but we keep counting the actual number of
1572 * members to let the caller decide to retry if necessary.
1573 *
1574 * Conditions:
1575 * Locks may be held by callers, so this routine cannot block.
1576 * Caller holds reference on the message queue (via port set).
1577 */
1578 void
1579 ipc_mqueue_set_gather_member_names(
1580 ipc_space_t space,
1581 ipc_mqueue_t set_mq,
1582 ipc_entry_num_t maxnames,
1583 mach_port_name_t *names,
1584 ipc_entry_num_t *actualp)
1585 {
1586 ipc_entry_t table;
1587 ipc_entry_num_t tsize;
1588 struct waitq_set *wqset;
1589 ipc_entry_num_t actual = 0;
1590
1591 assert(set_mq != IMQ_NULL);
1592 wqset = &set_mq->imq_set_queue;
1593
1594 assert(space != IS_NULL);
1595 is_read_lock(space);
1596 if (!is_active(space)) {
1597 is_read_unlock(space);
1598 goto out;
1599 }
1600
1601 if (!waitq_set_is_valid(wqset)) {
1602 is_read_unlock(space);
1603 goto out;
1604 }
1605
1606 table = space->is_table;
1607 tsize = space->is_table_size;
1608 for (ipc_entry_num_t idx = 0; idx < tsize; idx++) {
1609 ipc_entry_t entry = &table[idx];
1610
1611 /* only receive rights can be members of port sets */
1612 if ((entry->ie_bits & MACH_PORT_TYPE_RECEIVE) != MACH_PORT_TYPE_NONE) {
1613 ipc_port_t port = ip_object_to_port(entry->ie_object);
1614 ipc_mqueue_t mq = &port->ip_messages;
1615
1616 assert(IP_VALID(port));
1617 if (ip_active(port) &&
1618 waitq_member(&mq->imq_wait_queue, wqset)) {
1619 if (actual < maxnames) {
1620 names[actual] = mq->imq_receiver_name;
1621 }
1622 actual++;
1623 }
1624 }
1625 }
1626
1627 is_read_unlock(space);
1628
1629 out:
1630 *actualp = actual;
1631 }
1632
1633
1634 /*
1635 * Routine: ipc_mqueue_destroy_locked
1636 * Purpose:
1637 * Destroy a (non-set) message queue.
1638 * Set any blocked senders running.
1639 * Destroy the kmsgs in the queue.
1640 * Conditions:
1641 * mqueue locked
1642 * Receivers were removed when the receive right was "changed"
1643 */
1644 boolean_t
1645 ipc_mqueue_destroy_locked(ipc_mqueue_t mqueue)
1646 {
1647 ipc_kmsg_queue_t kmqueue;
1648 ipc_kmsg_t kmsg;
1649 boolean_t reap = FALSE;
1650 struct turnstile *send_turnstile = port_send_turnstile(ip_from_mq(mqueue));
1651
1652 assert(!imq_is_set(mqueue));
1653
1654 /*
1655 * rouse all blocked senders
1656 * (don't boost anyone - we're tearing this queue down)
1657 * (never preposts)
1658 */
1659 mqueue->imq_fullwaiters = FALSE;
1660
1661 if (send_turnstile != TURNSTILE_NULL) {
1662 waitq_wakeup64_all(&send_turnstile->ts_waitq,
1663 IPC_MQUEUE_FULL,
1664 THREAD_RESTART,
1665 WAITQ_ALL_PRIORITIES);
1666 }
1667
1668 /*
1669 * Move messages from the specified queue to the per-thread
1670 * clean/drain queue while we have the mqueue lock.
1671 */
1672 kmqueue = &mqueue->imq_messages;
1673 while ((kmsg = ipc_kmsg_dequeue(kmqueue)) != IKM_NULL) {
1674 #if MACH_FLIPC
1675 if (MACH_NODE_VALID(kmsg->ikm_node) && FPORT_VALID(mqueue->imq_fport)) {
1676 flipc_msg_ack(kmsg->ikm_node, mqueue, TRUE);
1677 }
1678 #endif
1679 boolean_t first;
1680 first = ipc_kmsg_delayed_destroy(kmsg);
1681 if (first) {
1682 reap = first;
1683 }
1684 }
1685
1686 /*
1687 * Wipe out message count, both for messages about to be
1688 * reaped and for reserved space for (previously) woken senders.
1689 * This is the indication to them that their reserved space is gone
1690 * (the mqueue was destroyed).
1691 */
1692 mqueue->imq_msgcount = 0;
1693
1694 /* invalidate the waitq for subsequent mqueue operations */
1695 waitq_invalidate_locked(&mqueue->imq_wait_queue);
1696
1697 /* clear out any preposting we may have done */
1698 waitq_clear_prepost_locked(&mqueue->imq_wait_queue);
1699
1700 /*
1701 * assert that we are destroying / invalidating a queue that's
1702 * not a member of any other queue.
1703 */
1704 assert(mqueue->imq_preposts == 0);
1705 assert(mqueue->imq_in_pset == 0);
1706
1707 return reap;
1708 }
1709
1710 /*
1711 * Routine: ipc_mqueue_set_qlimit
1712 * Purpose:
1713 * Changes a message queue limit; the maximum number
1714 * of messages which may be queued.
1715 * Conditions:
1716 * Nothing locked.
1717 */
1718
1719 void
1720 ipc_mqueue_set_qlimit(
1721 ipc_mqueue_t mqueue,
1722 mach_port_msgcount_t qlimit)
1723 {
1724 assert(qlimit <= MACH_PORT_QLIMIT_MAX);
1725
1726 /* wake up senders allowed by the new qlimit */
1727 imq_lock(mqueue);
1728 if (qlimit > mqueue->imq_qlimit) {
1729 mach_port_msgcount_t i, wakeup;
1730 struct turnstile *send_turnstile = port_send_turnstile(ip_from_mq(mqueue));
1731
1732 /* caution: wakeup, qlimit are unsigned */
1733 wakeup = qlimit - mqueue->imq_qlimit;
1734
1735 for (i = 0; i < wakeup; i++) {
1736 /*
1737 * boost the priority of the awoken thread
1738 * (WAITQ_PROMOTE_PRIORITY) to ensure it uses
1739 * the message queue slot we've just reserved.
1740 *
1741 * NOTE: this will never prepost
1742 */
1743 if (send_turnstile == TURNSTILE_NULL ||
1744 waitq_wakeup64_one(&send_turnstile->ts_waitq,
1745 IPC_MQUEUE_FULL,
1746 THREAD_AWAKENED,
1747 WAITQ_PROMOTE_PRIORITY) == KERN_NOT_WAITING) {
1748 mqueue->imq_fullwaiters = FALSE;
1749 break;
1750 }
1751 mqueue->imq_msgcount++; /* give it to the awakened thread */
1752 }
1753 }
1754 mqueue->imq_qlimit = qlimit;
1755 imq_unlock(mqueue);
1756 }
1757
1758 /*
1759 * Routine: ipc_mqueue_set_seqno
1760 * Purpose:
1761 * Changes an mqueue's sequence number.
1762 * Conditions:
1763 * Caller holds a reference to the queue's containing object.
1764 */
1765 void
1766 ipc_mqueue_set_seqno(
1767 ipc_mqueue_t mqueue,
1768 mach_port_seqno_t seqno)
1769 {
1770 imq_lock(mqueue);
1771 mqueue->imq_seqno = seqno;
1772 imq_unlock(mqueue);
1773 }
1774
1775
1776 /*
1777 * Routine: ipc_mqueue_copyin
1778 * Purpose:
1779 * Convert a name in a space to a message queue.
1780 * Conditions:
1781 * Nothing locked. If successful, the caller gets a ref for
1782 * for the object. This ref ensures the continued existence of
1783 * the queue.
1784 * Returns:
1785 * MACH_MSG_SUCCESS Found a message queue.
1786 * MACH_RCV_INVALID_NAME The space is dead.
1787 * MACH_RCV_INVALID_NAME The name doesn't denote a right.
1788 * MACH_RCV_INVALID_NAME
1789 * The denoted right is not receive or port set.
1790 * MACH_RCV_IN_SET Receive right is a member of a set.
1791 */
1792
1793 mach_msg_return_t
1794 ipc_mqueue_copyin(
1795 ipc_space_t space,
1796 mach_port_name_t name,
1797 ipc_mqueue_t *mqueuep,
1798 ipc_object_t *objectp)
1799 {
1800 ipc_entry_t entry;
1801 ipc_entry_bits_t bits;
1802 ipc_object_t object;
1803 ipc_mqueue_t mqueue;
1804
1805 is_read_lock(space);
1806 if (!is_active(space)) {
1807 is_read_unlock(space);
1808 return MACH_RCV_INVALID_NAME;
1809 }
1810
1811 entry = ipc_entry_lookup(space, name);
1812 if (entry == IE_NULL) {
1813 is_read_unlock(space);
1814 return MACH_RCV_INVALID_NAME;
1815 }
1816
1817 bits = entry->ie_bits;
1818 object = entry->ie_object;
1819
1820 if (bits & MACH_PORT_TYPE_RECEIVE) {
1821 ipc_port_t port = ip_object_to_port(object);
1822
1823 assert(port != IP_NULL);
1824
1825 ip_lock(port);
1826 require_ip_active(port);
1827 assert(port->ip_receiver_name == name);
1828 assert(port->ip_receiver == space);
1829 is_read_unlock(space);
1830 mqueue = &port->ip_messages;
1831 } else if (bits & MACH_PORT_TYPE_PORT_SET) {
1832 ipc_pset_t pset = ips_object_to_pset(object);
1833
1834 assert(pset != IPS_NULL);
1835
1836 ips_lock(pset);
1837 assert(ips_active(pset));
1838 is_read_unlock(space);
1839
1840 mqueue = &pset->ips_messages;
1841 } else {
1842 is_read_unlock(space);
1843 /* guard exception if we never held the receive right in this entry */
1844 if ((bits & MACH_PORT_TYPE_EX_RECEIVE) == 0) {
1845 mach_port_guard_exception(name, 0, 0, kGUARD_EXC_RCV_INVALID_NAME);
1846 }
1847 return MACH_RCV_INVALID_NAME;
1848 }
1849
1850 /*
1851 * At this point, the object is locked and active,
1852 * the space is unlocked, and mqueue is initialized.
1853 */
1854
1855 io_reference(object);
1856 io_unlock(object);
1857
1858 *objectp = object;
1859 *mqueuep = mqueue;
1860 return MACH_MSG_SUCCESS;
1861 }
1862
1863 void
1864 imq_lock(ipc_mqueue_t mq)
1865 {
1866 ipc_object_t object = imq_to_object(mq);
1867 ipc_object_validate(object);
1868 waitq_lock(&(mq)->imq_wait_queue);
1869 }
1870
1871 unsigned int
1872 imq_lock_try(ipc_mqueue_t mq)
1873 {
1874 ipc_object_t object = imq_to_object(mq);
1875 ipc_object_validate(object);
1876 return waitq_lock_try(&(mq)->imq_wait_queue);
1877 }