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