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 [email protected]
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/counter.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_right.h>
91 #include <ipc/ipc_port.h>
92 #include <ipc/ipc_pset.h>
93 #include <ipc/ipc_space.h>
94
95 #if MACH_FLIPC
96 #include <ipc/flipc.h>
97 #endif
98
99 #ifdef __LP64__
100 #include <vm/vm_map.h>
101 #endif
102
103 #include <sys/event.h>
104
105 extern char *proc_name_address(void *p);
106
107 int ipc_mqueue_full; /* address is event for queue space */
108 int ipc_mqueue_rcv; /* address is event for message arrival */
109
110 /* forward declarations */
111 static void ipc_mqueue_receive_results(wait_result_t result);
112 static void ipc_mqueue_peek_on_thread_locked(
113 ipc_mqueue_t port_mq,
114 mach_msg_option64_t option,
115 thread_t thread);
116
117 /* Deliver message to message queue or waiting receiver */
118 static void ipc_mqueue_post(
119 ipc_mqueue_t mqueue,
120 ipc_kmsg_t kmsg,
121 mach_msg_option_t option);
122
123 /*
124 * Routine: ipc_mqueue_init
125 * Purpose:
126 * Initialize a newly-allocated message queue.
127 */
128 void
ipc_mqueue_init(ipc_mqueue_t mqueue)129 ipc_mqueue_init(
130 ipc_mqueue_t mqueue)
131 {
132 ipc_kmsg_queue_init(&mqueue->imq_messages);
133 mqueue->imq_qlimit = MACH_PORT_QLIMIT_DEFAULT;
134 klist_init(&mqueue->imq_klist);
135 }
136
137 /*
138 * Routine: ipc_mqueue_add_locked.
139 * Purpose:
140 * Associate the portset's mqueue with the port's mqueue.
141 * This has to be done so that posting the port will wakeup
142 * a portset waiter. If there are waiters on the portset
143 * mqueue and messages on the port mqueue, try to match them
144 * up now.
145 * Conditions:
146 * Port and Pset both locked.
147 */
148 kern_return_t
ipc_mqueue_add_locked(ipc_mqueue_t port_mqueue,ipc_pset_t pset,waitq_link_t * linkp)149 ipc_mqueue_add_locked(
150 ipc_mqueue_t port_mqueue,
151 ipc_pset_t pset,
152 waitq_link_t *linkp)
153 {
154 ipc_port_t port = ip_from_mq(port_mqueue);
155 struct waitq_set *wqset = &pset->ips_wqset;
156 circle_queue_t kmsgq = &port_mqueue->imq_messages;
157 kern_return_t kr = KERN_SUCCESS;
158 ipc_kmsg_t kmsg;
159
160 kr = waitq_link_locked(&port->ip_waitq, wqset, linkp);
161 if (kr != KERN_SUCCESS) {
162 return kr;
163 }
164
165 /*
166 * Now that the set has been added to the port, there may be
167 * messages queued on the port and threads waiting on the set
168 * waitq. Lets get them together.
169 *
170 * Only consider this set however, as the other ones have been
171 * posted to already.
172 */
173 while ((kmsg = ipc_kmsg_queue_first(kmsgq)) != IKM_NULL) {
174 thread_t th;
175 mach_msg_size_t msize, asize;
176
177 th = waitq_wakeup64_identify_locked(wqset, IPC_MQUEUE_RECEIVE,
178 THREAD_AWAKENED, WAITQ_KEEP_LOCKED);
179 /* port and pset still locked, thread not runnable */
180
181 if (th == THREAD_NULL) {
182 /*
183 * Didn't find a thread to wake up but messages
184 * are enqueued, prepost the set instead,
185 * as calling waitq_wakeup64_identify_locked()
186 * on the set directly will not take care of it.
187 */
188 waitq_link_prepost_locked(&port->ip_waitq, wqset);
189 break;
190 }
191
192 /*
193 * Because we hold the thread off the runqueue at this point,
194 * it's safe to modify ith_ fields on the thread, as
195 * until it is resumed, it must be off core or in between
196 * the assert wait and returning from the continuation.
197 */
198
199 /*
200 * If the receiver waited with a facility not directly
201 * related to Mach messaging, then it isn't prepared to get
202 * handed the message directly. Just set it running, and
203 * go look for another thread that can.
204 */
205 if (th->ith_state != MACH_RCV_IN_PROGRESS) {
206 if (th->ith_state == MACH_PEEK_IN_PROGRESS) {
207 /*
208 * wakeup the peeking thread, but
209 * continue to loop over the threads
210 * waiting on the port's mqueue to see
211 * if there are any actual receivers
212 */
213 ipc_mqueue_peek_on_thread_locked(port_mqueue,
214 th->ith_option,
215 th);
216 }
217
218 waitq_resume_identified_thread(wqset, th,
219 THREAD_AWAKENED, WAITQ_WAKEUP_DEFAULT);
220 continue;
221 }
222
223 /*
224 * Found a receiver. see if they can handle the message
225 * correctly (the message is not too large for them, or
226 * they didn't care to be informed that the message was
227 * too large). If they can't handle it, take them off
228 * the list and let them go back and figure it out and
229 * just move onto the next.
230 */
231 msize = ipc_kmsg_copyout_size(kmsg, th->map);
232 asize = ipc_kmsg_aux_data_size(kmsg);
233
234 if (ipc_kmsg_too_large(msize, asize, th->ith_option,
235 th->ith_max_msize, th->ith_max_asize, th)) {
236 th->ith_state = MACH_RCV_TOO_LARGE;
237 th->ith_msize = msize;
238 th->ith_asize = asize;
239 if (th->ith_option & MACH_RCV_LARGE) {
240 /*
241 * let him go without message
242 */
243 th->ith_receiver_name = port_mqueue->imq_receiver_name;
244 th->ith_kmsg = IKM_NULL;
245 th->ith_seqno = 0;
246
247 waitq_resume_identified_thread(wqset, th,
248 THREAD_AWAKENED, WAITQ_WAKEUP_DEFAULT);
249
250 continue; /* find another thread */
251 }
252 } else {
253 th->ith_state = MACH_MSG_SUCCESS;
254 }
255
256 /*
257 * This thread is going to take this message,
258 * so give it the message.
259 */
260 ipc_kmsg_rmqueue(kmsgq, kmsg);
261
262 #if MACH_FLIPC
263 mach_node_t node = kmsg->ikm_node;
264 #endif
265
266 ipc_mqueue_release_msgcount(port_mqueue);
267
268 th->ith_kmsg = kmsg;
269 th->ith_seqno = port_mqueue->imq_seqno++;
270
271 waitq_resume_identified_thread(wqset, th,
272 THREAD_AWAKENED, WAITQ_WAKEUP_DEFAULT);
273
274 #if MACH_FLIPC
275 if (MACH_NODE_VALID(node) && FPORT_VALID(port_mqueue->imq_fport)) {
276 flipc_msg_ack(node, port_mqueue, TRUE);
277 }
278 #endif
279 }
280
281 return KERN_SUCCESS;
282 }
283
284
285 /*
286 * Routine: ipc_port_has_klist
287 * Purpose:
288 * Returns whether the given port imq_klist field can be used as a klist.
289 */
290 bool
ipc_port_has_klist(ipc_port_t port)291 ipc_port_has_klist(ipc_port_t port)
292 {
293 return !port->ip_specialreply &&
294 port->ip_sync_link_state == PORT_SYNC_LINK_ANY;
295 }
296
297 static inline struct klist *
ipc_object_klist(ipc_object_t object)298 ipc_object_klist(ipc_object_t object)
299 {
300 if (io_otype(object) == IOT_PORT) {
301 ipc_port_t port = ip_object_to_port(object);
302
303 return ipc_port_has_klist(port) ? &port->ip_klist : NULL;
304 }
305 return &ips_object_to_pset(object)->ips_klist;
306 }
307
308 /*
309 * Routine: ipc_mqueue_changed
310 * Purpose:
311 * Wake up receivers waiting in a message queue.
312 * Conditions:
313 * The object containing the message queue is locked.
314 */
315 void
ipc_mqueue_changed(ipc_space_t space,struct waitq * waitq)316 ipc_mqueue_changed(
317 ipc_space_t space,
318 struct waitq *waitq)
319 {
320 ipc_object_t object = io_from_waitq(waitq);
321 struct klist *klist = ipc_object_klist(object);
322
323 if (klist && SLIST_FIRST(klist)) {
324 /*
325 * Indicate that this message queue is vanishing
326 *
327 * When this is called, the associated receive right may be in flight
328 * between two tasks: the one it used to live in, and the one that armed
329 * a port destroyed notification for it.
330 *
331 * The new process may want to register the port it gets back with an
332 * EVFILT_MACHPORT filter again, and may have pending sync IPC on this
333 * port pending already, in which case we want the imq_klist field to be
334 * reusable for nefarious purposes.
335 *
336 * Fortunately, we really don't need this linkage anymore after this
337 * point as EV_VANISHED / EV_EOF will be the last thing delivered ever.
338 *
339 * Note: we don't have the space lock here, however, this covers the
340 * case of when a task is terminating the space, triggering
341 * several knote_vanish() calls.
342 *
343 * We don't need the lock to observe that the space is inactive as
344 * we just deactivated it on the same thread.
345 *
346 * We still need to call knote_vanish() so that the knote is
347 * marked with EV_VANISHED or EV_EOF so that the detach step
348 * in filt_machportdetach is skipped correctly.
349 */
350 assert(space);
351 knote_vanish(klist, is_active(space));
352 }
353
354 if (io_otype(object) == IOT_PORT) {
355 ipc_port_t port = ip_object_to_port(object);
356 if (!port->ip_specialreply) {
357 ipc_port_adjust_sync_link_state_locked(port,
358 PORT_SYNC_LINK_ANY, NULL);
359 }
360 } else {
361 klist_init(klist);
362 }
363
364 /*
365 * do not pass WAITQ_UPDATE_INHERITOR, ipc_port_destroy()
366 * needs to handle this manually, and the port lock
367 * is the waitq lock, so there's really no inefficiency there.
368 */
369 waitq_wakeup64_all_locked(waitq, IPC_MQUEUE_RECEIVE,
370 THREAD_RESTART, WAITQ_KEEP_LOCKED);
371 }
372
373
374
375
376 /*
377 * Routine: ipc_mqueue_send
378 * Purpose:
379 * Send a message to a message queue. The message holds a reference
380 * for the destination port for this message queue in the
381 * msgh_remote_port field.
382 *
383 * If unsuccessful, the caller still has possession of
384 * the message and must do something with it. If successful,
385 * the message is queued, given to a receiver, or destroyed.
386 * Conditions:
387 * port is locked.
388 * Returns:
389 * MACH_MSG_SUCCESS The message was accepted.
390 * MACH_SEND_TIMED_OUT Caller still has message.
391 * MACH_SEND_INTERRUPTED Caller still has message.
392 */
393 mach_msg_return_t
ipc_mqueue_send_locked(ipc_mqueue_t mqueue,ipc_kmsg_t kmsg,mach_msg_option_t option,mach_msg_timeout_t send_timeout)394 ipc_mqueue_send_locked(
395 ipc_mqueue_t mqueue,
396 ipc_kmsg_t kmsg,
397 mach_msg_option_t option,
398 mach_msg_timeout_t send_timeout)
399 {
400 ipc_port_t port = ip_from_mq(mqueue);
401 int wresult;
402
403 /*
404 * Don't block if:
405 * 1) We're under the queue limit.
406 * 2) Caller used the MACH_SEND_ALWAYS internal option.
407 * 3) Message is sent to a send-once right.
408 */
409 if (!imq_full(mqueue) ||
410 (!imq_full_kernel(mqueue) &&
411 ((option & MACH_SEND_ALWAYS) ||
412 (MACH_MSGH_BITS_REMOTE(ikm_header(kmsg)->msgh_bits) ==
413 MACH_MSG_TYPE_PORT_SEND_ONCE)))) {
414 mqueue->imq_msgcount++;
415 assert(mqueue->imq_msgcount > 0);
416 ip_mq_unlock(port);
417 } else {
418 thread_t cur_thread = current_thread();
419 struct turnstile *send_turnstile = TURNSTILE_NULL;
420 uint64_t deadline;
421
422 /*
423 * We have to wait for space to be granted to us.
424 */
425 if ((option & MACH_SEND_TIMEOUT) && (send_timeout == 0)) {
426 ip_mq_unlock(port);
427 return MACH_SEND_TIMED_OUT;
428 }
429 if (imq_full_kernel(mqueue)) {
430 ip_mq_unlock(port);
431 return MACH_SEND_NO_BUFFER;
432 }
433 port->ip_fullwaiters = true;
434
435 if (option & MACH_SEND_TIMEOUT) {
436 clock_interval_to_deadline(send_timeout, 1000 * NSEC_PER_USEC, &deadline);
437 } else {
438 deadline = 0;
439 }
440
441 thread_set_pending_block_hint(cur_thread, kThreadWaitPortSend);
442
443 send_turnstile = turnstile_prepare((uintptr_t)port,
444 port_send_turnstile_address(port),
445 TURNSTILE_NULL, TURNSTILE_SYNC_IPC);
446
447 ipc_port_send_update_inheritor(port, send_turnstile,
448 TURNSTILE_DELAYED_UPDATE);
449
450 wresult = waitq_assert_wait64_leeway(
451 &send_turnstile->ts_waitq,
452 IPC_MQUEUE_FULL,
453 THREAD_ABORTSAFE,
454 TIMEOUT_URGENCY_USER_NORMAL,
455 deadline,
456 TIMEOUT_NO_LEEWAY);
457
458 ip_mq_unlock(port);
459 turnstile_update_inheritor_complete(send_turnstile,
460 TURNSTILE_INTERLOCK_NOT_HELD);
461
462 if (wresult == THREAD_WAITING) {
463 wresult = thread_block(THREAD_CONTINUE_NULL);
464 }
465
466 /* Call turnstile complete with interlock held */
467 ip_mq_lock(port);
468 turnstile_complete((uintptr_t)port, port_send_turnstile_address(port), NULL, TURNSTILE_SYNC_IPC);
469 ip_mq_unlock(port);
470
471 /* Call cleanup after dropping the interlock */
472 turnstile_cleanup();
473
474 switch (wresult) {
475 case THREAD_AWAKENED:
476 /*
477 * we can proceed - inherited msgcount from waker
478 * or the message queue has been destroyed and the msgcount
479 * has been reset to zero (will detect in ipc_mqueue_post()).
480 */
481 break;
482
483 case THREAD_TIMED_OUT:
484 assert(option & MACH_SEND_TIMEOUT);
485 return MACH_SEND_TIMED_OUT;
486
487 case THREAD_INTERRUPTED:
488 return MACH_SEND_INTERRUPTED;
489
490 case THREAD_RESTART:
491 /* mqueue is being destroyed */
492 return MACH_SEND_INVALID_DEST;
493 default:
494 panic("ipc_mqueue_send");
495 }
496 }
497
498 ipc_mqueue_post(mqueue, kmsg, option);
499 return MACH_MSG_SUCCESS;
500 }
501
502 /*
503 * Routine: ipc_mqueue_override_send_locked
504 * Purpose:
505 * Set an override qos on the first message in the queue
506 * (if the queue is full). This is a send-possible override
507 * that will go away as soon as we drain a message from the
508 * queue.
509 *
510 * Conditions:
511 * The port corresponding to mqueue is locked.
512 * The caller holds a reference on the message queue.
513 */
514 void
ipc_mqueue_override_send_locked(ipc_mqueue_t mqueue,mach_msg_qos_t qos_ovr)515 ipc_mqueue_override_send_locked(
516 ipc_mqueue_t mqueue,
517 mach_msg_qos_t qos_ovr)
518 {
519 ipc_port_t port = ip_from_mq(mqueue);
520
521 assert(waitq_is_valid(&port->ip_waitq));
522
523 if (imq_full(mqueue)) {
524 ipc_kmsg_t first = ipc_kmsg_queue_first(&mqueue->imq_messages);
525
526 if (first && ipc_kmsg_override_qos(&mqueue->imq_messages, first, qos_ovr)) {
527 if (ip_in_a_space(port) &&
528 is_active(ip_get_receiver(port)) &&
529 ipc_port_has_klist(port)) {
530 KNOTE(&port->ip_klist, 0);
531 }
532 }
533 }
534 }
535
536 /*
537 * Routine: ipc_mqueue_release_msgcount
538 * Purpose:
539 * Release a message queue reference in the case where we
540 * found a waiter.
541 *
542 * Conditions:
543 * The port corresponding to message queue is locked.
544 * The message corresponding to this reference is off the queue.
545 * There is no need to pass reserved preposts because this will
546 * never prepost to anyone
547 */
548 void
ipc_mqueue_release_msgcount(ipc_mqueue_t port_mq)549 ipc_mqueue_release_msgcount(ipc_mqueue_t port_mq)
550 {
551 ipc_port_t port = ip_from_mq(port_mq);
552 struct turnstile *send_turnstile = port_send_turnstile(port);
553
554 ip_mq_lock_held(port);
555 assert(port_mq->imq_msgcount > 1 || ipc_kmsg_queue_empty(&port_mq->imq_messages));
556
557 port_mq->imq_msgcount--;
558
559 if (!imq_full(port_mq) && port->ip_fullwaiters &&
560 send_turnstile != TURNSTILE_NULL) {
561 /*
562 * boost the priority of the awoken thread
563 * (WAITQ_PROMOTE_PRIORITY) to ensure it uses
564 * the message queue slot we've just reserved.
565 *
566 * NOTE: this will never prepost
567 *
568 * The wakeup happens on a turnstile waitq
569 * which will wakeup the highest priority waiter.
570 * A potential downside of this would be starving low
571 * priority senders if there is a constant churn of
572 * high priority threads trying to send to this port.
573 */
574 if (waitq_wakeup64_one(&send_turnstile->ts_waitq,
575 IPC_MQUEUE_FULL,
576 THREAD_AWAKENED,
577 WAITQ_PROMOTE_PRIORITY) != KERN_SUCCESS) {
578 port->ip_fullwaiters = false;
579 } else {
580 /* gave away our slot - add reference back */
581 port_mq->imq_msgcount++;
582 }
583 }
584
585 if (ipc_kmsg_queue_empty(&port_mq->imq_messages)) {
586 waitq_clear_prepost_locked(&port->ip_waitq);
587 }
588 }
589
590 /*
591 * Routine: ipc_mqueue_post
592 * Purpose:
593 * Post a message to a waiting receiver or enqueue it. If a
594 * receiver is waiting, we can release our reserved space in
595 * the message queue.
596 *
597 * Conditions:
598 * port is unlocked
599 * If we need to queue, our space in the message queue is reserved.
600 */
601 static void
ipc_mqueue_post(ipc_mqueue_t mqueue,ipc_kmsg_t kmsg,mach_msg_option_t __unused option)602 ipc_mqueue_post(
603 ipc_mqueue_t mqueue,
604 ipc_kmsg_t kmsg,
605 mach_msg_option_t __unused option)
606 {
607 ipc_port_t port = ip_from_mq(mqueue);
608 struct waitq *waitq = &port->ip_waitq;
609 boolean_t destroy_msg = FALSE;
610
611 ipc_kmsg_trace_send(kmsg, option);
612
613 /*
614 * While the msg queue is locked, we have control of the
615 * kmsg, so the ref in it for the port is still good.
616 *
617 * Check for a receiver for the message.
618 */
619 ip_mq_lock(port);
620
621 /* we may have raced with port destruction! */
622 if (!waitq_is_valid(&port->ip_waitq)) {
623 destroy_msg = TRUE;
624 goto out_unlock;
625 }
626
627 for (;;) {
628 thread_t receiver;
629 mach_msg_size_t msize, asize;
630
631 receiver = waitq_wakeup64_identify_locked(waitq,
632 IPC_MQUEUE_RECEIVE, THREAD_AWAKENED, WAITQ_KEEP_LOCKED);
633 /* waitq still locked, thread not runnable */
634
635 if (receiver == THREAD_NULL) {
636 /*
637 * no receivers; queue kmsg if space still reserved
638 * Reservations are cancelled when the port goes inactive.
639 * note that this will enqueue the message for any
640 * "peeking" receivers.
641 *
642 * Also, post the knote to wake up any threads waiting
643 * on that style of interface if this insertion is of
644 * note (first insertion, or adjusted override qos all
645 * the way to the head of the queue).
646 *
647 * This is just for ports. port-sets knotes are being
648 * posted to by the waitq_wakeup64_identify_locked()
649 * above already.
650 */
651 if (mqueue->imq_msgcount == 0) {
652 /*
653 * The message queue must belong
654 * to an inactive port, so just destroy
655 * the message and pretend it was posted.
656 */
657 destroy_msg = TRUE;
658 } else if (!ipc_kmsg_enqueue_qos(&mqueue->imq_messages, kmsg)) {
659 /*
660 * queue was not empty and qos
661 * didn't change, nothing to do.
662 */
663 } else if (ip_in_a_space(port) &&
664 is_active(ip_get_receiver(port)) &&
665 ipc_port_has_klist(port)) {
666 /*
667 * queue was empty or qos changed
668 * we need to tell kqueue, unless
669 * the space is getting torn down
670 */
671 KNOTE(&port->ip_klist, 0);
672 }
673 break;
674 }
675
676 /*
677 * If a thread is attempting a "peek" into the message queue
678 * (MACH_PEEK_IN_PROGRESS), then we enqueue the message and set the
679 * thread running. A successful peek is essentially the same as
680 * message delivery since the peeking thread takes responsibility
681 * for delivering the message and (eventually) removing it from
682 * the mqueue. Only one thread can successfully use the peek
683 * facility on any given port, so we exit the waitq loop after
684 * encountering such a thread.
685 */
686 if (receiver->ith_state == MACH_PEEK_IN_PROGRESS && mqueue->imq_msgcount > 0) {
687 ipc_kmsg_enqueue_qos(&mqueue->imq_messages, kmsg);
688 ipc_mqueue_peek_on_thread_locked(mqueue, receiver->ith_option, receiver);
689 waitq_resume_identified_thread(waitq, receiver,
690 THREAD_AWAKENED, WAITQ_WAKEUP_DEFAULT);
691 break; /* Message was posted, so break out of loop */
692 }
693
694 /*
695 * If the receiver waited with a facility not directly related
696 * to Mach messaging, then it isn't prepared to get handed the
697 * message directly. Just set it running, and go look for
698 * another thread that can.
699 */
700 if (receiver->ith_state != MACH_RCV_IN_PROGRESS) {
701 waitq_resume_identified_thread(waitq, receiver,
702 THREAD_AWAKENED, WAITQ_WAKEUP_DEFAULT);
703
704 continue;
705 }
706
707
708 /*
709 * We found a waiting thread.
710 * If the message is too large or the scatter list is too small
711 * the thread we wake up will get that as its status.
712 */
713 msize = ipc_kmsg_copyout_size(kmsg, receiver->map);
714 asize = ipc_kmsg_aux_data_size(kmsg);
715
716 if (ipc_kmsg_too_large(msize, asize, receiver->ith_option,
717 receiver->ith_max_msize, receiver->ith_max_asize, receiver)) {
718 receiver->ith_msize = msize;
719 receiver->ith_asize = asize;
720 receiver->ith_state = MACH_RCV_TOO_LARGE;
721 } else {
722 receiver->ith_state = MACH_MSG_SUCCESS;
723 }
724
725 /*
726 * If there is no problem with the upcoming receive, or the
727 * receiver thread didn't specifically ask for special too
728 * large error condition, go ahead and select it anyway.
729 */
730 if ((receiver->ith_state == MACH_MSG_SUCCESS) ||
731 !(receiver->ith_option & MACH_RCV_LARGE)) {
732 receiver->ith_kmsg = kmsg;
733 receiver->ith_seqno = mqueue->imq_seqno++;
734 #if MACH_FLIPC
735 mach_node_t node = kmsg->ikm_node;
736 #endif
737 waitq_resume_identified_thread(waitq, receiver,
738 THREAD_AWAKENED, WAITQ_WAKEUP_DEFAULT);
739
740 /* we didn't need our reserved spot in the queue */
741 ipc_mqueue_release_msgcount(mqueue);
742
743 #if MACH_FLIPC
744 if (MACH_NODE_VALID(node) && FPORT_VALID(mqueue->imq_fport)) {
745 flipc_msg_ack(node, mqueue, TRUE);
746 }
747 #endif
748 break;
749 }
750
751 /*
752 * Otherwise, this thread needs to be released to run
753 * and handle its error without getting the message. We
754 * need to go back and pick another one.
755 */
756 receiver->ith_receiver_name = mqueue->imq_receiver_name;
757 receiver->ith_kmsg = IKM_NULL;
758 receiver->ith_seqno = 0;
759
760 waitq_resume_identified_thread(waitq, receiver,
761 THREAD_AWAKENED, WAITQ_WAKEUP_DEFAULT);
762 }
763
764 out_unlock:
765 /* clear the waitq boost we may have been given */
766 waitq_clear_promotion_locked(waitq, current_thread());
767 waitq_unlock(waitq);
768
769 if (destroy_msg) {
770 ipc_kmsg_destroy(kmsg, IPC_KMSG_DESTROY_ALL);
771 }
772
773 counter_inc(¤t_task()->messages_sent);
774 return;
775 }
776
777
778 static void
ipc_mqueue_receive_results(wait_result_t saved_wait_result)779 ipc_mqueue_receive_results(wait_result_t saved_wait_result)
780 {
781 thread_t self = current_thread();
782 mach_msg_option64_t option64 = self->ith_option;
783
784 /*
785 * why did we wake up?
786 */
787 switch (saved_wait_result) {
788 case THREAD_TIMED_OUT:
789 self->ith_state = MACH_RCV_TIMED_OUT;
790 return;
791
792 case THREAD_INTERRUPTED:
793 self->ith_state = MACH_RCV_INTERRUPTED;
794 return;
795
796 case THREAD_RESTART:
797 /* something bad happened to the port/set */
798 self->ith_state = MACH_RCV_PORT_CHANGED;
799 return;
800
801 case THREAD_AWAKENED:
802 /*
803 * We do not need to go select a message, somebody
804 * handed us one (or a too-large indication).
805 */
806 switch (self->ith_state) {
807 case MACH_RCV_SCATTER_SMALL:
808 case MACH_RCV_TOO_LARGE:
809 /*
810 * Somebody tried to give us a too large
811 * message. If we indicated that we cared,
812 * then they only gave us the indication,
813 * otherwise they gave us the indication
814 * AND the message anyway.
815 */
816 if (option64 & MACH_RCV_LARGE) {
817 return;
818 }
819 return;
820 case MACH_MSG_SUCCESS:
821 return;
822 case MACH_PEEK_READY:
823 return;
824
825 default:
826 panic("ipc_mqueue_receive_results: strange ith_state %d", self->ith_state);
827 }
828
829 default:
830 panic("ipc_mqueue_receive_results: strange wait_result %d", saved_wait_result);
831 }
832 }
833
834 void
ipc_mqueue_receive_continue(__unused void * param,wait_result_t wresult)835 ipc_mqueue_receive_continue(
836 __unused void *param,
837 wait_result_t wresult)
838 {
839 ipc_mqueue_receive_results(wresult);
840 mach_msg_receive_continue(); /* hard-coded for now */
841 }
842
843 /*
844 * Routine: ipc_mqueue_receive
845 * Purpose:
846 * Receive a message from a message queue.
847 *
848 * Conditions:
849 * Our caller must hold a reference for the port or port set
850 * to which this queue belongs, to keep the queue
851 * from being deallocated.
852 *
853 * The kmsg is returned with clean header fields
854 * and with the circular bit turned off through the ith_kmsg
855 * field of the thread's receive continuation state.
856 * Returns:
857 * MACH_MSG_SUCCESS Message returned in ith_kmsg.
858 * MACH_RCV_TOO_LARGE Message size returned in ith_msize,
859 * Auxiliary data size returned in ith_asize
860 * MACH_RCV_TIMED_OUT No message obtained.
861 * MACH_RCV_INTERRUPTED No message obtained.
862 * MACH_RCV_PORT_DIED Port/set died; no message.
863 * MACH_RCV_PORT_CHANGED Port moved into set; no msg.
864 *
865 */
866
867 void
ipc_mqueue_receive(struct waitq * waitq,mach_msg_option64_t option64,mach_msg_size_t max_size,mach_msg_size_t max_aux_size,mach_msg_timeout_t rcv_timeout,int interruptible,bool has_continuation)868 ipc_mqueue_receive(
869 struct waitq *waitq,
870 mach_msg_option64_t option64,
871 mach_msg_size_t max_size,
872 mach_msg_size_t max_aux_size, /* 0 if no aux buffer */
873 mach_msg_timeout_t rcv_timeout,
874 int interruptible,
875 bool has_continuation)
876 {
877 wait_result_t wresult;
878 thread_t self = current_thread();
879
880 waitq_lock(waitq);
881
882 wresult = ipc_mqueue_receive_on_thread_and_unlock(waitq, option64, max_size,
883 max_aux_size, rcv_timeout, interruptible, self);
884 /* object unlocked */
885 if (wresult == THREAD_NOT_WAITING) {
886 return;
887 }
888
889 if (wresult == THREAD_WAITING) {
890 if (has_continuation) {
891 wresult = thread_block(ipc_mqueue_receive_continue);
892 /* NOTREACHED */
893 }
894 wresult = thread_block(THREAD_CONTINUE_NULL);
895 }
896 ipc_mqueue_receive_results(wresult);
897 }
898
899 /*
900 * Routine: ipc_mqueue_receive_on_thread_and_unlock
901 * Purpose:
902 * Receive a message from a message queue using a specified thread.
903 * If no message available, assert_wait on the appropriate waitq.
904 *
905 * Conditions:
906 * Assumes thread is self.
907 * The port/port-set waitq is locked on entry, unlocked on return.
908 * May have assert-waited. Caller must block in those cases.
909 */
910 wait_result_t
ipc_mqueue_receive_on_thread_and_unlock(struct waitq * waitq,mach_msg_option64_t option64,mach_msg_size_t max_msg_size,mach_msg_size_t max_aux_size,mach_msg_timeout_t rcv_timeout,int interruptible,thread_t thread)911 ipc_mqueue_receive_on_thread_and_unlock(
912 struct waitq *waitq,
913 mach_msg_option64_t option64,
914 mach_msg_size_t max_msg_size,
915 mach_msg_size_t max_aux_size,
916 mach_msg_timeout_t rcv_timeout,
917 int interruptible,
918 thread_t thread)
919 {
920 ipc_object_t object = io_from_waitq(waitq);
921 ipc_port_t port = IP_NULL;
922 wait_result_t wresult;
923 uint64_t deadline;
924 struct turnstile *rcv_turnstile = TURNSTILE_NULL;
925
926 if (waitq_type(waitq) == WQT_PORT_SET) {
927 ipc_pset_t pset = ips_object_to_pset(object);
928 struct waitq *port_wq;
929
930 /*
931 * Put the message at the back of the prepost list
932 * if it's not a PEEK.
933 *
934 * Might drop the pset lock temporarily.
935 */
936 port_wq = waitq_set_first_prepost(&pset->ips_wqset, WQS_PREPOST_LOCK |
937 ((option64 & MACH64_PEEK_MSG) ? WQS_PREPOST_PEEK: 0));
938
939 /* Returns with port locked */
940
941 if (port_wq != NULL) {
942 /*
943 * We get here if there is at least one message
944 * waiting on port_wq. We have instructed the prepost
945 * iteration logic to leave both the port_wq and the
946 * set waitq locked.
947 *
948 * Continue on to handling the message with just
949 * the port waitq locked.
950 */
951 io_unlock(object);
952 port = ip_from_waitq(port_wq);
953 }
954 } else if (waitq_type(waitq) == WQT_PORT) {
955 port = ip_from_waitq(waitq);
956 if (ipc_kmsg_queue_empty(&port->ip_messages.imq_messages)) {
957 port = IP_NULL;
958 }
959 } else {
960 panic("Unknown waitq type (%p/0x%x)", waitq, waitq_type(waitq));
961 }
962
963 if (port) {
964 if (option64 & MACH64_PEEK_MSG) {
965 ipc_mqueue_peek_on_thread_locked(&port->ip_messages,
966 option64, thread);
967 } else {
968 ipc_mqueue_select_on_thread_locked(&port->ip_messages,
969 option64, max_msg_size, max_aux_size, thread);
970 }
971 ip_mq_unlock(port);
972 return THREAD_NOT_WAITING;
973 }
974
975 if (!waitq_is_valid(waitq)) {
976 /* someone raced us to destroy this mqueue/port! */
977 io_unlock(object);
978 /*
979 * ipc_mqueue_receive_results updates the thread's ith_state
980 * TODO: differentiate between rights being moved and
981 * rights/ports being destroyed (21885327)
982 */
983 return THREAD_RESTART;
984 }
985
986 /*
987 * Looks like we'll have to block. The waitq we will
988 * block on (whether the set's or the local port's) is
989 * still locked.
990 */
991 if ((option64 & MACH_RCV_TIMEOUT) && rcv_timeout == 0) {
992 io_unlock(object);
993 thread->ith_state = MACH_RCV_TIMED_OUT;
994 return THREAD_NOT_WAITING;
995 }
996
997 thread->ith_option = option64;
998 thread->ith_max_msize = max_msg_size;
999 thread->ith_msize = 0;
1000
1001 thread->ith_max_asize = max_aux_size;
1002 thread->ith_asize = 0;
1003
1004 if (option64 & MACH64_PEEK_MSG) {
1005 thread->ith_state = MACH_PEEK_IN_PROGRESS;
1006 } else {
1007 thread->ith_state = MACH_RCV_IN_PROGRESS;
1008 }
1009
1010 if (option64 & MACH_RCV_TIMEOUT) {
1011 clock_interval_to_deadline(rcv_timeout, 1000 * NSEC_PER_USEC, &deadline);
1012 } else {
1013 deadline = 0;
1014 }
1015
1016 /*
1017 * Threads waiting on a reply port (not portset)
1018 * will wait on its receive turnstile.
1019 *
1020 * Donate waiting thread's turnstile and
1021 * setup inheritor for special reply port.
1022 * Based on the state of the special reply
1023 * port, the inheritor would be the send
1024 * turnstile of the connection port on which
1025 * the send of sync ipc would happen or
1026 * workloop's turnstile who would reply to
1027 * the sync ipc message.
1028 *
1029 * Pass in mqueue wait in waitq_assert_wait to
1030 * support port set wakeup. The mqueue waitq of port
1031 * will be converted to to turnstile waitq
1032 * in waitq_assert_wait instead of global waitqs.
1033 */
1034 if (waitq_type(waitq) == WQT_PORT) {
1035 port = ip_from_waitq(waitq);
1036 rcv_turnstile = turnstile_prepare((uintptr_t)port,
1037 port_rcv_turnstile_address(port),
1038 TURNSTILE_NULL, TURNSTILE_SYNC_IPC);
1039
1040 ipc_port_recv_update_inheritor(port, rcv_turnstile,
1041 TURNSTILE_DELAYED_UPDATE);
1042 }
1043
1044 thread_set_pending_block_hint(thread, kThreadWaitPortReceive);
1045 wresult = waitq_assert_wait64_locked(waitq,
1046 IPC_MQUEUE_RECEIVE,
1047 interruptible,
1048 TIMEOUT_URGENCY_USER_NORMAL,
1049 deadline,
1050 TIMEOUT_NO_LEEWAY,
1051 thread);
1052 if (wresult == THREAD_AWAKENED) {
1053 /*
1054 * The first thing we did was to look for preposts
1055 * (using waitq_set_first_prepost() for sets, or looking
1056 * at the port's queue for ports).
1057 *
1058 * Since we found none, we kept the waitq locked.
1059 *
1060 * It ensures that waitq_assert_wait64_locked() can't
1061 * find pre-posts either, won't drop the waitq lock
1062 * either (even for a set), and can't return THREAD_AWAKENED.
1063 */
1064 panic("ipc_mqueue_receive_on_thread: sleep walking");
1065 }
1066
1067 io_unlock(object);
1068
1069 /*
1070 * After this point, a waiting thread could be found by the wakeup
1071 * identify path, and the other side now owns the ith_ fields until
1072 * this thread blocks and resumes in the continuation
1073 */
1074
1075 /* Check if its a port mqueue and if it needs to call turnstile_update_inheritor_complete */
1076 if (rcv_turnstile != TURNSTILE_NULL) {
1077 turnstile_update_inheritor_complete(rcv_turnstile, TURNSTILE_INTERLOCK_NOT_HELD);
1078 }
1079 /* Its callers responsibility to call turnstile_complete to get the turnstile back */
1080
1081 return wresult;
1082 }
1083
1084
1085 /*
1086 * Routine: ipc_mqueue_peek_on_thread_locked
1087 * Purpose:
1088 * A receiver discovered that there was a message on the queue
1089 * before he had to block. Tell a thread about the message queue,
1090 * but don't pick off any messages.
1091 * Conditions:
1092 * port_mq locked
1093 * at least one message on port_mq's message queue
1094 *
1095 * Returns: (on thread->ith_state)
1096 * MACH_PEEK_READY ith_peekq contains a message queue
1097 */
1098 void
ipc_mqueue_peek_on_thread_locked(ipc_mqueue_t port_mq,__assert_only mach_msg_option64_t option64,thread_t thread)1099 ipc_mqueue_peek_on_thread_locked(
1100 ipc_mqueue_t port_mq,
1101 __assert_only mach_msg_option64_t option64,
1102 thread_t thread)
1103 {
1104 assert(option64 & MACH64_PEEK_MSG);
1105 assert(ipc_kmsg_queue_first(&port_mq->imq_messages) != IKM_NULL);
1106
1107 /*
1108 * Take a reference on the mqueue's associated port:
1109 * the peeking thread will be responsible to release this reference
1110 */
1111 ip_validate(ip_from_mq(port_mq));
1112 ip_reference(ip_from_mq(port_mq));
1113 thread->ith_peekq = port_mq;
1114 thread->ith_state = MACH_PEEK_READY;
1115 }
1116
1117 /*
1118 * Routine: ipc_mqueue_select_on_thread_locked
1119 * Purpose:
1120 * A receiver discovered that there was a message on the queue
1121 * before he had to block. Pick the message off the queue and
1122 * "post" it to thread.
1123 * Conditions:
1124 * port locked.
1125 * thread not locked.
1126 * There is a message.
1127 * No need to reserve prepost objects - it will never prepost
1128 *
1129 * Returns:
1130 * MACH_MSG_SUCCESS Actually selected a message for ourselves.
1131 * MACH_RCV_TOO_LARGE May or may not have pull it, but it is large
1132 */
1133 void
ipc_mqueue_select_on_thread_locked(ipc_mqueue_t port_mq,mach_msg_option64_t option64,mach_msg_size_t max_msg_size,mach_msg_size_t max_aux_size,thread_t thread)1134 ipc_mqueue_select_on_thread_locked(
1135 ipc_mqueue_t port_mq,
1136 mach_msg_option64_t option64,
1137 mach_msg_size_t max_msg_size,
1138 mach_msg_size_t max_aux_size,
1139 thread_t thread)
1140 {
1141 ipc_kmsg_t kmsg;
1142 mach_msg_size_t msize, asize;
1143
1144 mach_msg_return_t mr = MACH_MSG_SUCCESS;
1145
1146 /*
1147 * Do some sanity checking of our ability to receive
1148 * before pulling the message off the queue.
1149 */
1150 kmsg = ipc_kmsg_queue_first(&port_mq->imq_messages);
1151 assert(kmsg != IKM_NULL);
1152
1153 /*
1154 * If we really can't receive it, but we had the
1155 * MACH_RCV_LARGE option set, then don't take it off
1156 * the queue, instead return the appropriate error
1157 * (and size needed).
1158 */
1159 msize = ipc_kmsg_copyout_size(kmsg, thread->map);
1160 asize = ipc_kmsg_aux_data_size(kmsg);
1161
1162 if (ipc_kmsg_too_large(msize, asize, option64,
1163 max_msg_size, max_aux_size, thread)) {
1164 mr = MACH_RCV_TOO_LARGE;
1165 if (option64 & MACH_RCV_LARGE) {
1166 thread->ith_receiver_name = port_mq->imq_receiver_name;
1167 thread->ith_kmsg = IKM_NULL;
1168 thread->ith_msize = msize;
1169 thread->ith_asize = asize;
1170 thread->ith_seqno = 0;
1171 thread->ith_state = mr;
1172 return;
1173 }
1174 }
1175
1176 ipc_kmsg_rmqueue(&port_mq->imq_messages, kmsg);
1177 #if MACH_FLIPC
1178 if (MACH_NODE_VALID(kmsg->ikm_node) && FPORT_VALID(port_mq->imq_fport)) {
1179 flipc_msg_ack(kmsg->ikm_node, port_mq, TRUE);
1180 }
1181 #endif
1182 ipc_mqueue_release_msgcount(port_mq);
1183 thread->ith_seqno = port_mq->imq_seqno++;
1184 thread->ith_kmsg = kmsg;
1185 thread->ith_state = mr;
1186
1187 counter_inc(¤t_task()->messages_received);
1188 return;
1189 }
1190
1191 /*
1192 * Routine: ipc_mqueue_peek_locked
1193 * Purpose:
1194 * Peek at a (non-set) message queue to see if it has a message
1195 * matching the sequence number provided (if zero, then the
1196 * first message in the queue) and return vital info about the
1197 * message.
1198 *
1199 * Conditions:
1200 * The io object corresponding to mq is locked by callers.
1201 * Other locks may be held by callers, so this routine cannot block.
1202 * Caller holds reference on the message queue.
1203 */
1204 unsigned
ipc_mqueue_peek_locked(ipc_mqueue_t mq,mach_port_seqno_t * seqnop,mach_msg_size_t * msg_sizep,mach_msg_id_t * msg_idp,mach_msg_max_trailer_t * msg_trailerp,ipc_kmsg_t * kmsgp)1205 ipc_mqueue_peek_locked(ipc_mqueue_t mq,
1206 mach_port_seqno_t * seqnop,
1207 mach_msg_size_t * msg_sizep,
1208 mach_msg_id_t * msg_idp,
1209 mach_msg_max_trailer_t * msg_trailerp,
1210 ipc_kmsg_t *kmsgp)
1211 {
1212 ipc_kmsg_queue_t kmsgq;
1213 ipc_kmsg_t kmsg;
1214 mach_port_seqno_t seqno, msgoff;
1215 unsigned res = 0;
1216 mach_msg_header_t *hdr;
1217
1218 seqno = 0;
1219 if (seqnop != NULL) {
1220 seqno = *seqnop;
1221 }
1222
1223 if (seqno == 0) {
1224 seqno = mq->imq_seqno;
1225 msgoff = 0;
1226 } else if (seqno >= mq->imq_seqno &&
1227 seqno < mq->imq_seqno + mq->imq_msgcount) {
1228 msgoff = seqno - mq->imq_seqno;
1229 } else {
1230 goto out;
1231 }
1232
1233 /* look for the message that would match that seqno */
1234 kmsgq = &mq->imq_messages;
1235 kmsg = ipc_kmsg_queue_first(kmsgq);
1236 while (msgoff-- && kmsg != IKM_NULL) {
1237 kmsg = ipc_kmsg_queue_next(kmsgq, kmsg);
1238 }
1239 if (kmsg == IKM_NULL) {
1240 goto out;
1241 }
1242
1243 #if __has_feature(ptrauth_calls)
1244 /*
1245 * Validate kmsg signature before doing anything with it. Since we are holding
1246 * the mqueue lock here, and only header + trailer will be peeked on, just
1247 * do a partial validation to finish quickly.
1248 *
1249 * Partial kmsg signature is only supported on PAC devices.
1250 */
1251 ipc_kmsg_validate_sig(kmsg, true);
1252 #endif
1253
1254 hdr = ikm_header(kmsg);
1255 /* found one - return the requested info */
1256 if (seqnop != NULL) {
1257 *seqnop = seqno;
1258 }
1259 if (msg_sizep != NULL) {
1260 *msg_sizep = hdr->msgh_size;
1261 }
1262 if (msg_idp != NULL) {
1263 *msg_idp = hdr->msgh_id;
1264 }
1265 if (msg_trailerp != NULL) {
1266 memcpy(msg_trailerp, ipc_kmsg_get_trailer(kmsg, false), sizeof(mach_msg_max_trailer_t));
1267 }
1268 if (kmsgp != NULL) {
1269 *kmsgp = kmsg;
1270 }
1271
1272 res = 1;
1273
1274 out:
1275 return res;
1276 }
1277
1278
1279 /*
1280 * Routine: ipc_mqueue_peek
1281 * Purpose:
1282 * Peek at a (non-set) message queue to see if it has a message
1283 * matching the sequence number provided (if zero, then the
1284 * first message in the queue) and return vital info about the
1285 * message.
1286 *
1287 * Conditions:
1288 * The ipc_mqueue_t is unlocked.
1289 * Locks may be held by callers, so this routine cannot block.
1290 * Caller holds reference on the message queue.
1291 */
1292 unsigned
ipc_mqueue_peek(ipc_mqueue_t mq,mach_port_seqno_t * seqnop,mach_msg_size_t * msg_sizep,mach_msg_id_t * msg_idp,mach_msg_max_trailer_t * msg_trailerp,ipc_kmsg_t * kmsgp)1293 ipc_mqueue_peek(ipc_mqueue_t mq,
1294 mach_port_seqno_t * seqnop,
1295 mach_msg_size_t * msg_sizep,
1296 mach_msg_id_t * msg_idp,
1297 mach_msg_max_trailer_t * msg_trailerp,
1298 ipc_kmsg_t *kmsgp)
1299 {
1300 ipc_port_t port = ip_from_mq(mq);
1301 unsigned res;
1302
1303 ip_mq_lock(port);
1304
1305 res = ipc_mqueue_peek_locked(mq, seqnop, msg_sizep, msg_idp,
1306 msg_trailerp, kmsgp);
1307
1308 ip_mq_unlock(port);
1309 return res;
1310 }
1311
1312 #if MACH_FLIPC
1313 /*
1314 * Routine: ipc_mqueue_release_peek_ref
1315 * Purpose:
1316 * Release the reference on an mqueue's associated port which was
1317 * granted to a thread in ipc_mqueue_peek_on_thread (on the
1318 * MACH64_PEEK_MSG thread wakeup path).
1319 *
1320 * Conditions:
1321 * The ipc_mqueue_t should be locked on entry.
1322 * The ipc_mqueue_t will be _unlocked_ on return
1323 * (and potentially invalid!)
1324 *
1325 */
1326 void
ipc_mqueue_release_peek_ref(ipc_mqueue_t mqueue)1327 ipc_mqueue_release_peek_ref(ipc_mqueue_t mqueue)
1328 {
1329 ipc_port_t port = ip_from_mq(mqueue);
1330
1331 ip_mq_lock_held(port);
1332
1333 /*
1334 * clear any preposts this mq may have generated
1335 * (which would cause subsequent immediate wakeups)
1336 */
1337 waitq_clear_prepost_locked(&port->ip_waitq);
1338
1339 ip_mq_unlock(port);
1340
1341 /*
1342 * release the port reference: we need to do this outside the lock
1343 * because we might be holding the last port reference!
1344 **/
1345 ip_release(port);
1346 }
1347 #endif /* MACH_FLIPC */
1348
1349 /*
1350 * Routine: ipc_mqueue_destroy_locked
1351 * Purpose:
1352 * Destroy a message queue.
1353 * Set any blocked senders running.
1354 * Destroy the kmsgs in the queue.
1355 * Conditions:
1356 * port locked
1357 * Receivers were removed when the receive right was "changed"
1358 */
1359 boolean_t
ipc_mqueue_destroy_locked(ipc_mqueue_t mqueue,waitq_link_list_t * free_l)1360 ipc_mqueue_destroy_locked(ipc_mqueue_t mqueue, waitq_link_list_t *free_l)
1361 {
1362 ipc_port_t port = ip_from_mq(mqueue);
1363 boolean_t reap = FALSE;
1364 struct turnstile *send_turnstile = port_send_turnstile(port);
1365
1366 /*
1367 * rouse all blocked senders
1368 * (don't boost anyone - we're tearing this queue down)
1369 * (never preposts)
1370 */
1371 port->ip_fullwaiters = false;
1372
1373 if (send_turnstile != TURNSTILE_NULL) {
1374 waitq_wakeup64_all(&send_turnstile->ts_waitq,
1375 IPC_MQUEUE_FULL,
1376 THREAD_RESTART, WAITQ_WAKEUP_DEFAULT);
1377 }
1378
1379 #if MACH_FLIPC
1380 ipc_kmsg_t kmsg;
1381
1382 cqe_foreach_element_safe(kmsg, &mqueue->imq_messages, ikm_link) {
1383 if (MACH_NODE_VALID(kmsg->ikm_node) &&
1384 FPORT_VALID(mqueue->imq_fport)) {
1385 flipc_msg_ack(kmsg->ikm_node, mqueue, TRUE);
1386 }
1387 }
1388 #endif
1389
1390 /*
1391 * Move messages from the specified queue to the per-thread
1392 * clean/drain queue while we have the mqueue lock.
1393 */
1394 reap = ipc_kmsg_delayed_destroy_queue(&mqueue->imq_messages);
1395
1396 /*
1397 * Wipe out message count, both for messages about to be
1398 * reaped and for reserved space for (previously) woken senders.
1399 * This is the indication to them that their reserved space is gone
1400 * (the mqueue was destroyed).
1401 */
1402 mqueue->imq_msgcount = 0;
1403
1404 /*
1405 * invalidate the waitq for subsequent mqueue operations,
1406 * the port lock could be dropped after invalidating the mqueue.
1407 */
1408
1409 waitq_invalidate(&port->ip_waitq);
1410
1411 waitq_unlink_all_locked(&port->ip_waitq, NULL, free_l);
1412
1413 return reap;
1414 }
1415
1416 /*
1417 * Routine: ipc_mqueue_set_qlimit_locked
1418 * Purpose:
1419 * Changes a message queue limit; the maximum number
1420 * of messages which may be queued.
1421 * Conditions:
1422 * Port locked.
1423 */
1424
1425 void
ipc_mqueue_set_qlimit_locked(ipc_mqueue_t mqueue,mach_port_msgcount_t qlimit)1426 ipc_mqueue_set_qlimit_locked(
1427 ipc_mqueue_t mqueue,
1428 mach_port_msgcount_t qlimit)
1429 {
1430 ipc_port_t port = ip_from_mq(mqueue);
1431
1432 assert(qlimit <= MACH_PORT_QLIMIT_MAX);
1433
1434 /* wake up senders allowed by the new qlimit */
1435 if (qlimit > mqueue->imq_qlimit) {
1436 mach_port_msgcount_t i, wakeup;
1437 struct turnstile *send_turnstile = port_send_turnstile(port);
1438
1439 /* caution: wakeup, qlimit are unsigned */
1440 wakeup = qlimit - mqueue->imq_qlimit;
1441
1442 for (i = 0; i < wakeup; i++) {
1443 /*
1444 * boost the priority of the awoken thread
1445 * (WAITQ_PROMOTE_PRIORITY) to ensure it uses
1446 * the message queue slot we've just reserved.
1447 *
1448 * NOTE: this will never prepost
1449 */
1450 if (send_turnstile == TURNSTILE_NULL ||
1451 waitq_wakeup64_one(&send_turnstile->ts_waitq,
1452 IPC_MQUEUE_FULL,
1453 THREAD_AWAKENED,
1454 WAITQ_PROMOTE_PRIORITY) == KERN_NOT_WAITING) {
1455 port->ip_fullwaiters = false;
1456 break;
1457 }
1458 mqueue->imq_msgcount++; /* give it to the awakened thread */
1459 }
1460 }
1461 mqueue->imq_qlimit = (uint16_t)qlimit;
1462 }
1463
1464 /*
1465 * Routine: ipc_mqueue_set_seqno_locked
1466 * Purpose:
1467 * Changes an mqueue's sequence number.
1468 * Conditions:
1469 * Caller holds a reference to the queue's containing object.
1470 */
1471 void
ipc_mqueue_set_seqno_locked(ipc_mqueue_t mqueue,mach_port_seqno_t seqno)1472 ipc_mqueue_set_seqno_locked(
1473 ipc_mqueue_t mqueue,
1474 mach_port_seqno_t seqno)
1475 {
1476 mqueue->imq_seqno = seqno;
1477 }
1478
1479
1480 /*
1481 * Routine: ipc_mqueue_copyin
1482 * Purpose:
1483 * Convert a name in a space to a message queue.
1484 * Conditions:
1485 * Nothing locked. If successful, the caller gets a ref for
1486 * for the object. This ref ensures the continued existence of
1487 * the queue.
1488 * Returns:
1489 * MACH_MSG_SUCCESS Found a message queue.
1490 * MACH_RCV_INVALID_NAME The space is dead.
1491 * MACH_RCV_INVALID_NAME The name doesn't denote a right.
1492 * MACH_RCV_INVALID_NAME
1493 * The denoted right is not receive or port set.
1494 * MACH_RCV_IN_SET Receive right is a member of a set.
1495 */
1496
1497 mach_msg_return_t
ipc_mqueue_copyin(ipc_space_t space,mach_port_name_t name,ipc_object_t * objectp)1498 ipc_mqueue_copyin(
1499 ipc_space_t space,
1500 mach_port_name_t name,
1501 ipc_object_t *objectp)
1502 {
1503 ipc_entry_bits_t bits;
1504 ipc_object_t object;
1505 kern_return_t kr;
1506
1507 kr = ipc_right_lookup_read(space, name, &bits, &object);
1508 if (kr != KERN_SUCCESS) {
1509 return MACH_RCV_INVALID_NAME;
1510 }
1511 /* object is locked and active */
1512
1513 if (bits & MACH_PORT_TYPE_RECEIVE) {
1514 __assert_only ipc_port_t port = ip_object_to_port(object);
1515 assert(ip_get_receiver_name(port) == name);
1516 assert(ip_in_space(port, space));
1517 }
1518 if (bits & (MACH_PORT_TYPE_RECEIVE | MACH_PORT_TYPE_PORT_SET)) {
1519 io_reference(object);
1520 io_unlock(object);
1521 } else {
1522 io_unlock(object);
1523 /* guard exception if we never held the receive right in this entry */
1524 if ((bits & MACH_PORT_TYPE_EX_RECEIVE) == 0) {
1525 mach_port_guard_exception(name, 0, 0, kGUARD_EXC_RCV_INVALID_NAME);
1526 }
1527 return MACH_RCV_INVALID_NAME;
1528 }
1529
1530 *objectp = object;
1531 return MACH_MSG_SUCCESS;
1532 }
1533