xref: /xnu-10002.61.3/osfmk/ipc/ipc_mqueue.c (revision 0f4c859e951fba394238ab619495c4e1d54d0f34)
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(&current_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(&current_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