2 * Copyright (c) 2000-2007 Apple Inc. All rights reserved.
4 * @APPLE_OSREFERENCE_LICENSE_HEADER_START@
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.
15 * Please obtain a copy of the License at
16 * http://www.opensource.apple.com/apsl/ and read it before using this file.
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.
26 * @APPLE_OSREFERENCE_LICENSE_HEADER_END@
29 * @OSF_FREE_COPYRIGHT@
32 * Mach Operating System
33 * Copyright (c) 1991,1990,1989 Carnegie Mellon University
34 * All Rights Reserved.
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.
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.
46 * Carnegie Mellon requests users of this software to return to
48 * Software Distribution Coordinator or Software.Distribution@CS.CMU.EDU
49 * School of Computer Science
50 * Carnegie Mellon University
51 * Pittsburgh PA 15213-3890
53 * any improvements or extensions that they make and grant Carnegie Mellon
54 * the rights to redistribute these changes.
59 * File: ipc/ipc_mqueue.c
63 * Functions to manipulate IPC message queues.
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,
73 #include <mach/port.h>
74 #include <mach/message.h>
75 #include <mach/sync_policy.h>
77 #include <kern/assert.h>
78 #include <kern/counters.h>
79 #include <kern/sched_prim.h>
80 #include <kern/ipc_kobject.h>
81 #include <kern/ipc_mig.h> /* XXX - for mach_msg_receive_continue */
82 #include <kern/misc_protos.h>
83 #include <kern/task.h>
84 #include <kern/thread.h>
85 #include <kern/waitq.h>
88 #include <ipc/ipc_mqueue.h>
89 #include <ipc/ipc_kmsg.h>
90 #include <ipc/ipc_port.h>
91 #include <ipc/ipc_pset.h>
92 #include <ipc/ipc_space.h>
95 #include <ipc/flipc.h>
99 #include <vm/vm_map.h>
102 #include <sys/event.h>
104 extern char *proc_name_address(void *p
);
106 int ipc_mqueue_full
; /* address is event for queue space */
107 int ipc_mqueue_rcv
; /* address is event for message arrival */
109 /* forward declarations */
110 static void ipc_mqueue_receive_results(wait_result_t result
);
111 static void ipc_mqueue_peek_on_thread(
112 ipc_mqueue_t port_mq
,
113 mach_msg_option_t option
,
117 * Routine: ipc_mqueue_init
119 * Initialize a newly-allocated message queue.
127 waitq_set_init(&mqueue
->imq_set_queue
,
128 SYNC_POLICY_FIFO
| SYNC_POLICY_PREPOST
,
131 waitq_init(&mqueue
->imq_wait_queue
, SYNC_POLICY_FIFO
| SYNC_POLICY_PORT
);
132 ipc_kmsg_queue_init(&mqueue
->imq_messages
);
133 mqueue
->imq_seqno
= 0;
134 mqueue
->imq_msgcount
= 0;
135 mqueue
->imq_qlimit
= MACH_PORT_QLIMIT_DEFAULT
;
136 mqueue
->imq_context
= 0;
137 mqueue
->imq_fullwaiters
= FALSE
;
139 mqueue
->imq_fport
= FPORT_NULL
;
142 klist_init(&mqueue
->imq_klist
);
149 boolean_t is_set
= imq_is_set(mqueue
);
152 waitq_set_deinit(&mqueue
->imq_set_queue
);
154 waitq_deinit(&mqueue
->imq_wait_queue
);
159 * Routine: imq_reserve_and_lock
161 * Atomically lock an ipc_mqueue_t object and reserve
162 * an appropriate number of prepost linkage objects for
163 * use in wakeup operations.
168 imq_reserve_and_lock(ipc_mqueue_t mq
, uint64_t *reserved_prepost
)
170 *reserved_prepost
= waitq_prepost_reserve(&mq
->imq_wait_queue
, 0,
176 * Routine: imq_release_and_unlock
178 * Unlock an ipc_mqueue_t object, re-enable interrupts,
179 * and release any unused prepost object reservations.
184 imq_release_and_unlock(ipc_mqueue_t mq
, uint64_t reserved_prepost
)
186 assert(imq_held(mq
));
187 waitq_unlock(&mq
->imq_wait_queue
);
188 waitq_prepost_release_reserve(reserved_prepost
);
193 * Routine: ipc_mqueue_member
195 * Indicate whether the (port) mqueue is a member of
196 * this portset's mqueue. We do this by checking
197 * whether the portset mqueue's waitq is an member of
198 * the port's mqueue waitq.
200 * the portset's mqueue is not already a member
201 * this may block while allocating linkage structures.
206 ipc_mqueue_t port_mqueue
,
207 ipc_mqueue_t set_mqueue
)
209 struct waitq
*port_waitq
= &port_mqueue
->imq_wait_queue
;
210 struct waitq_set
*set_waitq
= &set_mqueue
->imq_set_queue
;
212 return waitq_member(port_waitq
, set_waitq
);
216 * Routine: ipc_mqueue_remove
218 * Remove the association between the queue and the specified
225 ipc_mqueue_t set_mqueue
)
227 struct waitq
*mq_waitq
= &mqueue
->imq_wait_queue
;
228 struct waitq_set
*set_waitq
= &set_mqueue
->imq_set_queue
;
230 return waitq_unlink(mq_waitq
, set_waitq
);
234 * Routine: ipc_mqueue_remove_from_all
236 * Remove the mqueue from all the sets it is a member of
240 * mqueue unlocked and set links deallocated
243 ipc_mqueue_remove_from_all(ipc_mqueue_t mqueue
)
245 struct waitq
*mq_waitq
= &mqueue
->imq_wait_queue
;
250 assert(waitq_valid(mq_waitq
));
251 kr
= waitq_unlink_all_unlock(mq_waitq
);
252 /* mqueue unlocked and set links deallocated */
256 * Routine: ipc_mqueue_remove_all
258 * Remove all the member queues from the specified set.
259 * Also removes the queue from any containing sets.
263 * mqueue unlocked all set links deallocated
266 ipc_mqueue_remove_all(ipc_mqueue_t mqueue
)
268 struct waitq_set
*mq_setq
= &mqueue
->imq_set_queue
;
271 assert(waitqs_is_set(mq_setq
));
272 waitq_set_unlink_all_unlock(mq_setq
);
273 /* mqueue unlocked set links deallocated */
278 * Routine: ipc_mqueue_add
280 * Associate the portset's mqueue with the port's mqueue.
281 * This has to be done so that posting the port will wakeup
282 * a portset waiter. If there are waiters on the portset
283 * mqueue and messages on the port mqueue, try to match them
290 ipc_mqueue_t port_mqueue
,
291 ipc_mqueue_t set_mqueue
,
292 uint64_t *reserved_link
,
293 uint64_t *reserved_prepost
)
295 struct waitq
*port_waitq
= &port_mqueue
->imq_wait_queue
;
296 struct waitq_set
*set_waitq
= &set_mqueue
->imq_set_queue
;
297 ipc_kmsg_queue_t kmsgq
;
298 ipc_kmsg_t kmsg
, next
;
301 assert(reserved_link
&& *reserved_link
!= 0);
302 assert(waitqs_is_linked(set_waitq
));
304 imq_lock(port_mqueue
);
307 * The link operation is now under the same lock-hold as
308 * message iteration and thread wakeup, but doesn't have to be...
310 kr
= waitq_link(port_waitq
, set_waitq
, WAITQ_ALREADY_LOCKED
, reserved_link
);
311 if (kr
!= KERN_SUCCESS
) {
312 imq_unlock(port_mqueue
);
317 * Now that the set has been added to the port, there may be
318 * messages queued on the port and threads waiting on the set
319 * waitq. Lets get them together.
321 kmsgq
= &port_mqueue
->imq_messages
;
322 for (kmsg
= ipc_kmsg_queue_first(kmsgq
);
325 next
= ipc_kmsg_queue_next(kmsgq
, kmsg
);
329 mach_msg_size_t msize
;
332 th
= waitq_wakeup64_identify_locked(
335 THREAD_AWAKENED
, &th_spl
,
336 reserved_prepost
, WAITQ_ALL_PRIORITIES
,
338 /* waitq/mqueue still locked, thread locked */
340 if (th
== THREAD_NULL
) {
345 * If the receiver waited with a facility not directly
346 * related to Mach messaging, then it isn't prepared to get
347 * handed the message directly. Just set it running, and
348 * go look for another thread that can.
350 if (th
->ith_state
!= MACH_RCV_IN_PROGRESS
) {
351 if (th
->ith_state
== MACH_PEEK_IN_PROGRESS
) {
353 * wakeup the peeking thread, but
354 * continue to loop over the threads
355 * waiting on the port's mqueue to see
356 * if there are any actual receivers
358 ipc_mqueue_peek_on_thread(port_mqueue
,
368 * Found a receiver. see if they can handle the message
369 * correctly (the message is not too large for them, or
370 * they didn't care to be informed that the message was
371 * too large). If they can't handle it, take them off
372 * the list and let them go back and figure it out and
373 * just move onto the next.
375 msize
= ipc_kmsg_copyout_size(kmsg
, th
->map
);
377 (msize
+ REQUESTED_TRAILER_SIZE(thread_is_64bit_addr(th
), th
->ith_option
))) {
378 th
->ith_state
= MACH_RCV_TOO_LARGE
;
379 th
->ith_msize
= msize
;
380 if (th
->ith_option
& MACH_RCV_LARGE
) {
382 * let him go without message
384 th
->ith_receiver_name
= port_mqueue
->imq_receiver_name
;
385 th
->ith_kmsg
= IKM_NULL
;
389 continue; /* find another thread */
392 th
->ith_state
= MACH_MSG_SUCCESS
;
396 * This thread is going to take this message,
399 ipc_kmsg_rmqueue(kmsgq
, kmsg
);
401 mach_node_t node
= kmsg
->ikm_node
;
403 ipc_mqueue_release_msgcount(port_mqueue
, IMQ_NULL
);
406 th
->ith_seqno
= port_mqueue
->imq_seqno
++;
410 if (MACH_NODE_VALID(node
) && FPORT_VALID(port_mqueue
->imq_fport
)) {
411 flipc_msg_ack(node
, port_mqueue
, TRUE
);
414 break; /* go to next message */
418 imq_unlock(port_mqueue
);
424 * Routine: ipc_mqueue_has_klist
426 * Returns whether the given mqueue imq_klist field can be used as a klist.
429 ipc_mqueue_has_klist(ipc_mqueue_t mqueue
)
431 ipc_object_t object
= imq_to_object(mqueue
);
432 if (io_otype(object
) != IOT_PORT
) {
435 ipc_port_t port
= ip_from_mq(mqueue
);
436 if (port
->ip_specialreply
) {
439 return port
->ip_sync_link_state
== PORT_SYNC_LINK_ANY
;
443 * Routine: ipc_mqueue_changed
445 * Wake up receivers waiting in a message queue.
447 * The message queue is locked.
454 if (ipc_mqueue_has_klist(mqueue
) && SLIST_FIRST(&mqueue
->imq_klist
)) {
456 * Indicate that this message queue is vanishing
458 * When this is called, the associated receive right may be in flight
459 * between two tasks: the one it used to live in, and the one that armed
460 * a port destroyed notification for it.
462 * The new process may want to register the port it gets back with an
463 * EVFILT_MACHPORT filter again, and may have pending sync IPC on this
464 * port pending already, in which case we want the imq_klist field to be
465 * reusable for nefarious purposes.
467 * Fortunately, we really don't need this linkage anymore after this
468 * point as EV_VANISHED / EV_EOF will be the last thing delivered ever.
470 * Note: we don't have the space lock here, however, this covers the
471 * case of when a task is terminating the space, triggering
472 * several knote_vanish() calls.
474 * We don't need the lock to observe that the space is inactive as
475 * we just deactivated it on the same thread.
477 * We still need to call knote_vanish() so that the knote is
478 * marked with EV_VANISHED or EV_EOF so that the detach step
479 * in filt_machportdetach is skipped correctly.
482 knote_vanish(&mqueue
->imq_klist
, is_active(space
));
485 if (io_otype(imq_to_object(mqueue
)) == IOT_PORT
) {
486 ipc_port_adjust_sync_link_state_locked(ip_from_mq(mqueue
), PORT_SYNC_LINK_ANY
, NULL
);
488 klist_init(&mqueue
->imq_klist
);
491 waitq_wakeup64_all_locked(&mqueue
->imq_wait_queue
,
495 WAITQ_ALL_PRIORITIES
,
503 * Routine: ipc_mqueue_send
505 * Send a message to a message queue. The message holds a reference
506 * for the destination port for this message queue in the
507 * msgh_remote_port field.
509 * If unsuccessful, the caller still has possession of
510 * the message and must do something with it. If successful,
511 * the message is queued, given to a receiver, or destroyed.
515 * MACH_MSG_SUCCESS The message was accepted.
516 * MACH_SEND_TIMED_OUT Caller still has message.
517 * MACH_SEND_INTERRUPTED Caller still has message.
523 mach_msg_option_t option
,
524 mach_msg_timeout_t send_timeout
)
530 * 1) We're under the queue limit.
531 * 2) Caller used the MACH_SEND_ALWAYS internal option.
532 * 3) Message is sent to a send-once right.
534 if (!imq_full(mqueue
) ||
535 (!imq_full_kernel(mqueue
) &&
536 ((option
& MACH_SEND_ALWAYS
) ||
537 (MACH_MSGH_BITS_REMOTE(kmsg
->ikm_header
->msgh_bits
) ==
538 MACH_MSG_TYPE_PORT_SEND_ONCE
)))) {
539 mqueue
->imq_msgcount
++;
540 assert(mqueue
->imq_msgcount
> 0);
543 thread_t cur_thread
= current_thread();
544 ipc_port_t port
= ip_from_mq(mqueue
);
545 struct turnstile
*send_turnstile
= TURNSTILE_NULL
;
549 * We have to wait for space to be granted to us.
551 if ((option
& MACH_SEND_TIMEOUT
) && (send_timeout
== 0)) {
553 return MACH_SEND_TIMED_OUT
;
555 if (imq_full_kernel(mqueue
)) {
557 return MACH_SEND_NO_BUFFER
;
559 mqueue
->imq_fullwaiters
= TRUE
;
561 if (option
& MACH_SEND_TIMEOUT
) {
562 clock_interval_to_deadline(send_timeout
, 1000 * NSEC_PER_USEC
, &deadline
);
567 thread_set_pending_block_hint(cur_thread
, kThreadWaitPortSend
);
569 send_turnstile
= turnstile_prepare((uintptr_t)port
,
570 port_send_turnstile_address(port
),
571 TURNSTILE_NULL
, TURNSTILE_SYNC_IPC
);
573 ipc_port_send_update_inheritor(port
, send_turnstile
,
574 TURNSTILE_DELAYED_UPDATE
);
576 wresult
= waitq_assert_wait64_leeway(
577 &send_turnstile
->ts_waitq
,
580 TIMEOUT_URGENCY_USER_NORMAL
,
585 turnstile_update_inheritor_complete(send_turnstile
,
586 TURNSTILE_INTERLOCK_NOT_HELD
);
588 if (wresult
== THREAD_WAITING
) {
589 wresult
= thread_block(THREAD_CONTINUE_NULL
);
590 counter(c_ipc_mqueue_send_block
++);
593 /* Call turnstile complete with interlock held */
595 turnstile_complete((uintptr_t)port
, port_send_turnstile_address(port
), NULL
, TURNSTILE_SYNC_IPC
);
598 /* Call cleanup after dropping the interlock */
602 case THREAD_AWAKENED
:
604 * we can proceed - inherited msgcount from waker
605 * or the message queue has been destroyed and the msgcount
606 * has been reset to zero (will detect in ipc_mqueue_post()).
610 case THREAD_TIMED_OUT
:
611 assert(option
& MACH_SEND_TIMEOUT
);
612 return MACH_SEND_TIMED_OUT
;
614 case THREAD_INTERRUPTED
:
615 return MACH_SEND_INTERRUPTED
;
618 /* mqueue is being destroyed */
619 return MACH_SEND_INVALID_DEST
;
621 panic("ipc_mqueue_send");
625 ipc_mqueue_post(mqueue
, kmsg
, option
);
626 return MACH_MSG_SUCCESS
;
630 * Routine: ipc_mqueue_override_send
632 * Set an override qos on the first message in the queue
633 * (if the queue is full). This is a send-possible override
634 * that will go away as soon as we drain a message from the
638 * The message queue is not locked.
639 * The caller holds a reference on the message queue.
642 ipc_mqueue_override_send(
644 mach_msg_priority_t override
)
646 boolean_t __unused full_queue_empty
= FALSE
;
649 assert(imq_valid(mqueue
));
650 assert(!imq_is_set(mqueue
));
652 if (imq_full(mqueue
)) {
653 ipc_kmsg_t first
= ipc_kmsg_queue_first(&mqueue
->imq_messages
);
655 if (first
&& ipc_kmsg_override_qos(&mqueue
->imq_messages
, first
, override
)) {
656 ipc_object_t object
= imq_to_object(mqueue
);
657 assert(io_otype(object
) == IOT_PORT
);
658 ipc_port_t port
= ip_object_to_port(object
);
659 if (ip_active(port
) &&
660 port
->ip_receiver_name
!= MACH_PORT_NULL
&&
661 is_active(port
->ip_receiver
) &&
662 ipc_mqueue_has_klist(mqueue
)) {
663 KNOTE(&mqueue
->imq_klist
, 0);
667 full_queue_empty
= TRUE
;
672 #if DEVELOPMENT || DEBUG
673 if (full_queue_empty
) {
674 ipc_port_t port
= ip_from_mq(mqueue
);
676 if (ip_active(port
) && !port
->ip_tempowner
&&
677 port
->ip_receiver_name
&& port
->ip_receiver
&&
678 port
->ip_receiver
!= ipc_space_kernel
) {
679 dst_pid
= task_pid(port
->ip_receiver
->is_task
);
686 * Routine: ipc_mqueue_release_msgcount
688 * Release a message queue reference in the case where we
692 * The message queue is locked.
693 * The message corresponding to this reference is off the queue.
694 * There is no need to pass reserved preposts because this will
695 * never prepost to anyone
698 ipc_mqueue_release_msgcount(ipc_mqueue_t port_mq
, ipc_mqueue_t set_mq
)
700 struct turnstile
*send_turnstile
= port_send_turnstile(ip_from_mq(port_mq
));
702 assert(imq_held(port_mq
));
703 assert(port_mq
->imq_msgcount
> 1 || ipc_kmsg_queue_empty(&port_mq
->imq_messages
));
705 port_mq
->imq_msgcount
--;
707 if (!imq_full(port_mq
) && port_mq
->imq_fullwaiters
&&
708 send_turnstile
!= TURNSTILE_NULL
) {
710 * boost the priority of the awoken thread
711 * (WAITQ_PROMOTE_PRIORITY) to ensure it uses
712 * the message queue slot we've just reserved.
714 * NOTE: this will never prepost
716 * The wakeup happens on a turnstile waitq
717 * which will wakeup the highest priority waiter.
718 * A potential downside of this would be starving low
719 * priority senders if there is a constant churn of
720 * high priority threads trying to send to this port.
722 if (waitq_wakeup64_one(&send_turnstile
->ts_waitq
,
725 WAITQ_PROMOTE_PRIORITY
) != KERN_SUCCESS
) {
726 port_mq
->imq_fullwaiters
= FALSE
;
728 /* gave away our slot - add reference back */
729 port_mq
->imq_msgcount
++;
733 if (ipc_kmsg_queue_empty(&port_mq
->imq_messages
)) {
734 /* no more msgs: invalidate the port's prepost object */
735 waitq_clear_prepost_locked(&port_mq
->imq_wait_queue
);
740 * Routine: ipc_mqueue_post
742 * Post a message to a waiting receiver or enqueue it. If a
743 * receiver is waiting, we can release our reserved space in
748 * If we need to queue, our space in the message queue is reserved.
754 mach_msg_option_t __unused option
)
756 uint64_t reserved_prepost
= 0;
757 boolean_t destroy_msg
= FALSE
;
759 ipc_kmsg_trace_send(kmsg
, option
);
762 * While the msg queue is locked, we have control of the
763 * kmsg, so the ref in it for the port is still good.
765 * Check for a receiver for the message.
767 imq_reserve_and_lock(mqueue
, &reserved_prepost
);
769 /* we may have raced with port destruction! */
770 if (!imq_valid(mqueue
)) {
776 struct waitq
*waitq
= &mqueue
->imq_wait_queue
;
779 mach_msg_size_t msize
;
781 receiver
= waitq_wakeup64_identify_locked(waitq
,
786 WAITQ_ALL_PRIORITIES
,
788 /* waitq still locked, thread locked */
790 if (receiver
== THREAD_NULL
) {
792 * no receivers; queue kmsg if space still reserved
793 * Reservations are cancelled when the port goes inactive.
794 * note that this will enqueue the message for any
795 * "peeking" receivers.
797 * Also, post the knote to wake up any threads waiting
798 * on that style of interface if this insertion is of
799 * note (first insertion, or adjusted override qos all
800 * the way to the head of the queue).
802 * This is just for ports. portset knotes are stay-active,
803 * and their threads get awakened through the !MACH_RCV_IN_PROGRESS
806 if (mqueue
->imq_msgcount
> 0) {
807 if (ipc_kmsg_enqueue_qos(&mqueue
->imq_messages
, kmsg
)) {
808 /* if the space is dead there is no point calling KNOTE */
809 ipc_object_t object
= imq_to_object(mqueue
);
810 assert(io_otype(object
) == IOT_PORT
);
811 ipc_port_t port
= ip_object_to_port(object
);
812 if (ip_active(port
) &&
813 port
->ip_receiver_name
!= MACH_PORT_NULL
&&
814 is_active(port
->ip_receiver
) &&
815 ipc_mqueue_has_klist(mqueue
)) {
816 KNOTE(&mqueue
->imq_klist
, 0);
823 * Otherwise, the message queue must belong to an inactive
824 * port, so just destroy the message and pretend it was posted.
831 * If a thread is attempting a "peek" into the message queue
832 * (MACH_PEEK_IN_PROGRESS), then we enqueue the message and set the
833 * thread running. A successful peek is essentially the same as
834 * message delivery since the peeking thread takes responsibility
835 * for delivering the message and (eventually) removing it from
836 * the mqueue. Only one thread can successfully use the peek
837 * facility on any given port, so we exit the waitq loop after
838 * encountering such a thread.
840 if (receiver
->ith_state
== MACH_PEEK_IN_PROGRESS
&& mqueue
->imq_msgcount
> 0) {
841 ipc_kmsg_enqueue_qos(&mqueue
->imq_messages
, kmsg
);
842 ipc_mqueue_peek_on_thread(mqueue
, receiver
->ith_option
, receiver
);
843 thread_unlock(receiver
);
845 break; /* Message was posted, so break out of loop */
849 * If the receiver waited with a facility not directly related
850 * to Mach messaging, then it isn't prepared to get handed the
851 * message directly. Just set it running, and go look for
852 * another thread that can.
854 if (receiver
->ith_state
!= MACH_RCV_IN_PROGRESS
) {
855 thread_unlock(receiver
);
862 * We found a waiting thread.
863 * If the message is too large or the scatter list is too small
864 * the thread we wake up will get that as its status.
866 msize
= ipc_kmsg_copyout_size(kmsg
, receiver
->map
);
867 if (receiver
->ith_rsize
<
868 (msize
+ REQUESTED_TRAILER_SIZE(thread_is_64bit_addr(receiver
), receiver
->ith_option
))) {
869 receiver
->ith_msize
= msize
;
870 receiver
->ith_state
= MACH_RCV_TOO_LARGE
;
872 receiver
->ith_state
= MACH_MSG_SUCCESS
;
876 * If there is no problem with the upcoming receive, or the
877 * receiver thread didn't specifically ask for special too
878 * large error condition, go ahead and select it anyway.
880 if ((receiver
->ith_state
== MACH_MSG_SUCCESS
) ||
881 !(receiver
->ith_option
& MACH_RCV_LARGE
)) {
882 receiver
->ith_kmsg
= kmsg
;
883 receiver
->ith_seqno
= mqueue
->imq_seqno
++;
885 mach_node_t node
= kmsg
->ikm_node
;
887 thread_unlock(receiver
);
890 /* we didn't need our reserved spot in the queue */
891 ipc_mqueue_release_msgcount(mqueue
, IMQ_NULL
);
894 if (MACH_NODE_VALID(node
) && FPORT_VALID(mqueue
->imq_fport
)) {
895 flipc_msg_ack(node
, mqueue
, TRUE
);
902 * Otherwise, this thread needs to be released to run
903 * and handle its error without getting the message. We
904 * need to go back and pick another one.
906 receiver
->ith_receiver_name
= mqueue
->imq_receiver_name
;
907 receiver
->ith_kmsg
= IKM_NULL
;
908 receiver
->ith_seqno
= 0;
909 thread_unlock(receiver
);
914 /* clear the waitq boost we may have been given */
915 waitq_clear_promotion_locked(&mqueue
->imq_wait_queue
, current_thread());
916 imq_release_and_unlock(mqueue
, reserved_prepost
);
918 ipc_kmsg_destroy(kmsg
);
921 current_task()->messages_sent
++;
927 ipc_mqueue_receive_results(wait_result_t saved_wait_result
)
929 thread_t self
= current_thread();
930 mach_msg_option_t option
= self
->ith_option
;
933 * why did we wake up?
935 switch (saved_wait_result
) {
936 case THREAD_TIMED_OUT
:
937 self
->ith_state
= MACH_RCV_TIMED_OUT
;
940 case THREAD_INTERRUPTED
:
941 self
->ith_state
= MACH_RCV_INTERRUPTED
;
945 /* something bad happened to the port/set */
946 self
->ith_state
= MACH_RCV_PORT_CHANGED
;
949 case THREAD_AWAKENED
:
951 * We do not need to go select a message, somebody
952 * handed us one (or a too-large indication).
954 switch (self
->ith_state
) {
955 case MACH_RCV_SCATTER_SMALL
:
956 case MACH_RCV_TOO_LARGE
:
958 * Somebody tried to give us a too large
959 * message. If we indicated that we cared,
960 * then they only gave us the indication,
961 * otherwise they gave us the indication
962 * AND the message anyway.
964 if (option
& MACH_RCV_LARGE
) {
968 case MACH_MSG_SUCCESS
:
969 case MACH_PEEK_READY
:
973 panic("ipc_mqueue_receive_results: strange ith_state");
977 panic("ipc_mqueue_receive_results: strange wait_result");
982 ipc_mqueue_receive_continue(
983 __unused
void *param
,
984 wait_result_t wresult
)
986 ipc_mqueue_receive_results(wresult
);
987 mach_msg_receive_continue(); /* hard-coded for now */
991 * Routine: ipc_mqueue_receive
993 * Receive a message from a message queue.
996 * Our caller must hold a reference for the port or port set
997 * to which this queue belongs, to keep the queue
998 * from being deallocated.
1000 * The kmsg is returned with clean header fields
1001 * and with the circular bit turned off through the ith_kmsg
1002 * field of the thread's receive continuation state.
1004 * MACH_MSG_SUCCESS Message returned in ith_kmsg.
1005 * MACH_RCV_TOO_LARGE Message size returned in ith_msize.
1006 * MACH_RCV_TIMED_OUT No message obtained.
1007 * MACH_RCV_INTERRUPTED No message obtained.
1008 * MACH_RCV_PORT_DIED Port/set died; no message.
1009 * MACH_RCV_PORT_CHANGED Port moved into set; no msg.
1015 ipc_mqueue_t mqueue
,
1016 mach_msg_option_t option
,
1017 mach_msg_size_t max_size
,
1018 mach_msg_timeout_t rcv_timeout
,
1021 wait_result_t wresult
;
1022 thread_t self
= current_thread();
1025 wresult
= ipc_mqueue_receive_on_thread(mqueue
, option
, max_size
,
1026 rcv_timeout
, interruptible
,
1028 /* mqueue unlocked */
1029 if (wresult
== THREAD_NOT_WAITING
) {
1033 if (wresult
== THREAD_WAITING
) {
1034 counter((interruptible
== THREAD_ABORTSAFE
) ?
1035 c_ipc_mqueue_receive_block_user
++ :
1036 c_ipc_mqueue_receive_block_kernel
++);
1038 if (self
->ith_continuation
) {
1039 thread_block(ipc_mqueue_receive_continue
);
1043 wresult
= thread_block(THREAD_CONTINUE_NULL
);
1045 ipc_mqueue_receive_results(wresult
);
1049 mqueue_process_prepost_receive(void *ctx
, struct waitq
*waitq
,
1050 struct waitq_set
*wqset
)
1052 ipc_mqueue_t port_mq
, *pmq_ptr
;
1055 port_mq
= (ipc_mqueue_t
)waitq
;
1058 * If there are no messages on this queue, skip it and remove
1059 * it from the prepost list
1061 if (ipc_kmsg_queue_empty(&port_mq
->imq_messages
)) {
1062 return WQ_ITERATE_INVALIDATE_CONTINUE
;
1066 * There are messages waiting on this port.
1067 * Instruct the prepost iteration logic to break, but keep the
1070 pmq_ptr
= (ipc_mqueue_t
*)ctx
;
1074 return WQ_ITERATE_BREAK_KEEP_LOCKED
;
1078 * Routine: ipc_mqueue_receive_on_thread
1080 * Receive a message from a message queue using a specified thread.
1081 * If no message available, assert_wait on the appropriate waitq.
1084 * Assumes thread is self.
1085 * Called with mqueue locked.
1086 * Returns with mqueue unlocked.
1087 * May have assert-waited. Caller must block in those cases.
1090 ipc_mqueue_receive_on_thread(
1091 ipc_mqueue_t mqueue
,
1092 mach_msg_option_t option
,
1093 mach_msg_size_t max_size
,
1094 mach_msg_timeout_t rcv_timeout
,
1098 wait_result_t wresult
;
1100 struct turnstile
*rcv_turnstile
= TURNSTILE_NULL
;
1102 /* called with mqueue locked */
1104 /* no need to reserve anything: we never prepost to anyone */
1106 if (!imq_valid(mqueue
)) {
1107 /* someone raced us to destroy this mqueue/port! */
1110 * ipc_mqueue_receive_results updates the thread's ith_state
1111 * TODO: differentiate between rights being moved and
1112 * rights/ports being destroyed (21885327)
1114 return THREAD_RESTART
;
1117 if (imq_is_set(mqueue
)) {
1118 ipc_mqueue_t port_mq
= IMQ_NULL
;
1120 (void)waitq_set_iterate_preposts(&mqueue
->imq_set_queue
,
1122 mqueue_process_prepost_receive
);
1124 if (port_mq
!= IMQ_NULL
) {
1126 * We get here if there is at least one message
1127 * waiting on port_mq. We have instructed the prepost
1128 * iteration logic to leave both the port_mq and the
1129 * set mqueue locked.
1131 * TODO: previously, we would place this port at the
1132 * back of the prepost list...
1137 * Continue on to handling the message with just
1138 * the port mqueue locked.
1140 if (option
& MACH_PEEK_MSG
) {
1141 ipc_mqueue_peek_on_thread(port_mq
, option
, thread
);
1143 ipc_mqueue_select_on_thread(port_mq
, mqueue
, option
,
1147 imq_unlock(port_mq
);
1148 return THREAD_NOT_WAITING
;
1150 } else if (imq_is_queue(mqueue
)) {
1151 ipc_kmsg_queue_t kmsgs
;
1154 * Receive on a single port. Just try to get the messages.
1156 kmsgs
= &mqueue
->imq_messages
;
1157 if (ipc_kmsg_queue_first(kmsgs
) != IKM_NULL
) {
1158 if (option
& MACH_PEEK_MSG
) {
1159 ipc_mqueue_peek_on_thread(mqueue
, option
, thread
);
1161 ipc_mqueue_select_on_thread(mqueue
, IMQ_NULL
, option
,
1165 return THREAD_NOT_WAITING
;
1168 panic("Unknown mqueue type 0x%x: likely memory corruption!\n",
1169 mqueue
->imq_wait_queue
.waitq_type
);
1173 * Looks like we'll have to block. The mqueue we will
1174 * block on (whether the set's or the local port's) is
1177 if (option
& MACH_RCV_TIMEOUT
) {
1178 if (rcv_timeout
== 0) {
1180 thread
->ith_state
= MACH_RCV_TIMED_OUT
;
1181 return THREAD_NOT_WAITING
;
1185 thread
->ith_option
= option
;
1186 thread
->ith_rsize
= max_size
;
1187 thread
->ith_msize
= 0;
1189 if (option
& MACH_PEEK_MSG
) {
1190 thread
->ith_state
= MACH_PEEK_IN_PROGRESS
;
1192 thread
->ith_state
= MACH_RCV_IN_PROGRESS
;
1195 if (option
& MACH_RCV_TIMEOUT
) {
1196 clock_interval_to_deadline(rcv_timeout
, 1000 * NSEC_PER_USEC
, &deadline
);
1202 * Threads waiting on a special reply port
1203 * (not portset or regular ports)
1204 * will wait on its receive turnstile.
1206 * Donate waiting thread's turnstile and
1207 * setup inheritor for special reply port.
1208 * Based on the state of the special reply
1209 * port, the inheritor would be the send
1210 * turnstile of the connection port on which
1211 * the send of sync ipc would happen or
1212 * workloop's turnstile who would reply to
1213 * the sync ipc message.
1215 * Pass in mqueue wait in waitq_assert_wait to
1216 * support port set wakeup. The mqueue waitq of port
1217 * will be converted to to turnstile waitq
1218 * in waitq_assert_wait instead of global waitqs.
1220 if (imq_is_queue(mqueue
) && ip_from_mq(mqueue
)->ip_specialreply
) {
1221 ipc_port_t port
= ip_from_mq(mqueue
);
1222 rcv_turnstile
= turnstile_prepare((uintptr_t)port
,
1223 port_rcv_turnstile_address(port
),
1224 TURNSTILE_NULL
, TURNSTILE_SYNC_IPC
);
1226 ipc_port_recv_update_inheritor(port
, rcv_turnstile
,
1227 TURNSTILE_DELAYED_UPDATE
);
1230 thread_set_pending_block_hint(thread
, kThreadWaitPortReceive
);
1231 wresult
= waitq_assert_wait64_locked(&mqueue
->imq_wait_queue
,
1234 TIMEOUT_URGENCY_USER_NORMAL
,
1238 /* preposts should be detected above, not here */
1239 if (wresult
== THREAD_AWAKENED
) {
1240 panic("ipc_mqueue_receive_on_thread: sleep walking");
1245 /* Check if its a port mqueue and if it needs to call turnstile_update_inheritor_complete */
1246 if (rcv_turnstile
!= TURNSTILE_NULL
) {
1247 turnstile_update_inheritor_complete(rcv_turnstile
, TURNSTILE_INTERLOCK_NOT_HELD
);
1249 /* Its callers responsibility to call turnstile_complete to get the turnstile back */
1256 * Routine: ipc_mqueue_peek_on_thread
1258 * A receiver discovered that there was a message on the queue
1259 * before he had to block. Tell a thread about the message queue,
1260 * but don't pick off any messages.
1263 * at least one message on port_mq's message queue
1265 * Returns: (on thread->ith_state)
1266 * MACH_PEEK_READY ith_peekq contains a message queue
1269 ipc_mqueue_peek_on_thread(
1270 ipc_mqueue_t port_mq
,
1271 mach_msg_option_t option
,
1275 assert(option
& MACH_PEEK_MSG
);
1276 assert(ipc_kmsg_queue_first(&port_mq
->imq_messages
) != IKM_NULL
);
1279 * Take a reference on the mqueue's associated port:
1280 * the peeking thread will be responsible to release this reference
1281 * using ip_release_mq()
1283 ip_reference_mq(port_mq
);
1284 thread
->ith_peekq
= port_mq
;
1285 thread
->ith_state
= MACH_PEEK_READY
;
1289 * Routine: ipc_mqueue_select_on_thread
1291 * A receiver discovered that there was a message on the queue
1292 * before he had to block. Pick the message off the queue and
1293 * "post" it to thread.
1296 * thread not locked.
1297 * There is a message.
1298 * No need to reserve prepost objects - it will never prepost
1301 * MACH_MSG_SUCCESS Actually selected a message for ourselves.
1302 * MACH_RCV_TOO_LARGE May or may not have pull it, but it is large
1305 ipc_mqueue_select_on_thread(
1306 ipc_mqueue_t port_mq
,
1307 ipc_mqueue_t set_mq
,
1308 mach_msg_option_t option
,
1309 mach_msg_size_t max_size
,
1313 mach_msg_return_t mr
= MACH_MSG_SUCCESS
;
1314 mach_msg_size_t msize
;
1317 * Do some sanity checking of our ability to receive
1318 * before pulling the message off the queue.
1320 kmsg
= ipc_kmsg_queue_first(&port_mq
->imq_messages
);
1321 assert(kmsg
!= IKM_NULL
);
1324 * If we really can't receive it, but we had the
1325 * MACH_RCV_LARGE option set, then don't take it off
1326 * the queue, instead return the appropriate error
1327 * (and size needed).
1329 msize
= ipc_kmsg_copyout_size(kmsg
, thread
->map
);
1330 if (msize
+ REQUESTED_TRAILER_SIZE(thread_is_64bit_addr(thread
), option
) > max_size
) {
1331 mr
= MACH_RCV_TOO_LARGE
;
1332 if (option
& MACH_RCV_LARGE
) {
1333 thread
->ith_receiver_name
= port_mq
->imq_receiver_name
;
1334 thread
->ith_kmsg
= IKM_NULL
;
1335 thread
->ith_msize
= msize
;
1336 thread
->ith_seqno
= 0;
1337 thread
->ith_state
= mr
;
1342 ipc_kmsg_rmqueue(&port_mq
->imq_messages
, kmsg
);
1344 if (MACH_NODE_VALID(kmsg
->ikm_node
) && FPORT_VALID(port_mq
->imq_fport
)) {
1345 flipc_msg_ack(kmsg
->ikm_node
, port_mq
, TRUE
);
1348 ipc_mqueue_release_msgcount(port_mq
, set_mq
);
1349 thread
->ith_seqno
= port_mq
->imq_seqno
++;
1350 thread
->ith_kmsg
= kmsg
;
1351 thread
->ith_state
= mr
;
1353 current_task()->messages_received
++;
1358 * Routine: ipc_mqueue_peek_locked
1360 * Peek at a (non-set) message queue to see if it has a message
1361 * matching the sequence number provided (if zero, then the
1362 * first message in the queue) and return vital info about the
1366 * The ipc_mqueue_t is locked by callers.
1367 * Other locks may be held by callers, so this routine cannot block.
1368 * Caller holds reference on the message queue.
1371 ipc_mqueue_peek_locked(ipc_mqueue_t mq
,
1372 mach_port_seqno_t
* seqnop
,
1373 mach_msg_size_t
* msg_sizep
,
1374 mach_msg_id_t
* msg_idp
,
1375 mach_msg_max_trailer_t
* msg_trailerp
,
1378 ipc_kmsg_queue_t kmsgq
;
1380 mach_port_seqno_t seqno
, msgoff
;
1383 assert(!imq_is_set(mq
));
1386 if (seqnop
!= NULL
) {
1391 seqno
= mq
->imq_seqno
;
1393 } else if (seqno
>= mq
->imq_seqno
&&
1394 seqno
< mq
->imq_seqno
+ mq
->imq_msgcount
) {
1395 msgoff
= seqno
- mq
->imq_seqno
;
1400 /* look for the message that would match that seqno */
1401 kmsgq
= &mq
->imq_messages
;
1402 kmsg
= ipc_kmsg_queue_first(kmsgq
);
1403 while (msgoff
-- && kmsg
!= IKM_NULL
) {
1404 kmsg
= ipc_kmsg_queue_next(kmsgq
, kmsg
);
1406 if (kmsg
== IKM_NULL
) {
1410 /* found one - return the requested info */
1411 if (seqnop
!= NULL
) {
1414 if (msg_sizep
!= NULL
) {
1415 *msg_sizep
= kmsg
->ikm_header
->msgh_size
;
1417 if (msg_idp
!= NULL
) {
1418 *msg_idp
= kmsg
->ikm_header
->msgh_id
;
1420 if (msg_trailerp
!= NULL
) {
1421 memcpy(msg_trailerp
,
1422 (mach_msg_max_trailer_t
*)((vm_offset_t
)kmsg
->ikm_header
+
1423 round_msg(kmsg
->ikm_header
->msgh_size
)),
1424 sizeof(mach_msg_max_trailer_t
));
1426 if (kmsgp
!= NULL
) {
1438 * Routine: ipc_mqueue_peek
1440 * Peek at a (non-set) message queue to see if it has a message
1441 * matching the sequence number provided (if zero, then the
1442 * first message in the queue) and return vital info about the
1446 * The ipc_mqueue_t is unlocked.
1447 * Locks may be held by callers, so this routine cannot block.
1448 * Caller holds reference on the message queue.
1451 ipc_mqueue_peek(ipc_mqueue_t mq
,
1452 mach_port_seqno_t
* seqnop
,
1453 mach_msg_size_t
* msg_sizep
,
1454 mach_msg_id_t
* msg_idp
,
1455 mach_msg_max_trailer_t
* msg_trailerp
,
1462 res
= ipc_mqueue_peek_locked(mq
, seqnop
, msg_sizep
, msg_idp
,
1463 msg_trailerp
, kmsgp
);
1470 * Routine: ipc_mqueue_release_peek_ref
1472 * Release the reference on an mqueue's associated port which was
1473 * granted to a thread in ipc_mqueue_peek_on_thread (on the
1474 * MACH_PEEK_MSG thread wakeup path).
1477 * The ipc_mqueue_t should be locked on entry.
1478 * The ipc_mqueue_t will be _unlocked_ on return
1479 * (and potentially invalid!)
1483 ipc_mqueue_release_peek_ref(ipc_mqueue_t mq
)
1485 assert(!imq_is_set(mq
));
1486 assert(imq_held(mq
));
1489 * clear any preposts this mq may have generated
1490 * (which would cause subsequent immediate wakeups)
1492 waitq_clear_prepost_locked(&mq
->imq_wait_queue
);
1497 * release the port reference: we need to do this outside the lock
1498 * because we might be holding the last port reference!
1504 * peek at the contained port message queues, break prepost iteration as soon
1505 * as we spot a message on one of the message queues referenced by the set's
1506 * prepost list. No need to lock each message queue, as only the head of each
1507 * queue is checked. If a message wasn't there before we entered here, no need
1508 * to find it (if we do, great).
1511 mqueue_peek_iterator(void *ctx
, struct waitq
*waitq
,
1512 struct waitq_set
*wqset
)
1514 ipc_mqueue_t port_mq
= (ipc_mqueue_t
)waitq
;
1515 ipc_kmsg_queue_t kmsgs
= &port_mq
->imq_messages
;
1520 if (ipc_kmsg_queue_first(kmsgs
) != IKM_NULL
) {
1521 return WQ_ITERATE_BREAK
; /* break out of the prepost iteration */
1523 return WQ_ITERATE_CONTINUE
;
1527 * Routine: ipc_mqueue_set_peek
1529 * Peek at a message queue set to see if it has any ports
1533 * Locks may be held by callers, so this routine cannot block.
1534 * Caller holds reference on the message queue.
1537 ipc_mqueue_set_peek(ipc_mqueue_t mq
)
1544 * We may have raced with port destruction where the mqueue is marked
1545 * as invalid. In that case, even though we don't have messages, we
1546 * have an end-of-life event to deliver.
1548 if (!imq_is_valid(mq
)) {
1552 ret
= waitq_set_iterate_preposts(&mq
->imq_set_queue
, NULL
,
1553 mqueue_peek_iterator
);
1557 return ret
== WQ_ITERATE_BREAK
;
1561 * Routine: ipc_mqueue_set_gather_member_names
1563 * Discover all ports which are members of a given port set.
1564 * Because the waitq linkage mechanism was redesigned to save
1565 * significan amounts of memory, it no longer keeps back-pointers
1566 * from a port set to a port. Therefore, we must iterate over all
1567 * ports within a given IPC space and individually query them to
1568 * see if they are members of the given set. Port names of ports
1569 * found to be members of the given set will be gathered into the
1570 * provided 'names' array. Actual returned names are limited to
1571 * maxnames entries, but we keep counting the actual number of
1572 * members to let the caller decide to retry if necessary.
1575 * Locks may be held by callers, so this routine cannot block.
1576 * Caller holds reference on the message queue (via port set).
1579 ipc_mqueue_set_gather_member_names(
1581 ipc_mqueue_t set_mq
,
1582 ipc_entry_num_t maxnames
,
1583 mach_port_name_t
*names
,
1584 ipc_entry_num_t
*actualp
)
1587 ipc_entry_num_t tsize
;
1588 struct waitq_set
*wqset
;
1589 ipc_entry_num_t actual
= 0;
1591 assert(set_mq
!= IMQ_NULL
);
1592 wqset
= &set_mq
->imq_set_queue
;
1594 assert(space
!= IS_NULL
);
1595 is_read_lock(space
);
1596 if (!is_active(space
)) {
1597 is_read_unlock(space
);
1601 if (!waitq_set_is_valid(wqset
)) {
1602 is_read_unlock(space
);
1606 table
= space
->is_table
;
1607 tsize
= space
->is_table_size
;
1608 for (ipc_entry_num_t idx
= 0; idx
< tsize
; idx
++) {
1609 ipc_entry_t entry
= &table
[idx
];
1611 /* only receive rights can be members of port sets */
1612 if ((entry
->ie_bits
& MACH_PORT_TYPE_RECEIVE
) != MACH_PORT_TYPE_NONE
) {
1613 ipc_port_t port
= ip_object_to_port(entry
->ie_object
);
1614 ipc_mqueue_t mq
= &port
->ip_messages
;
1616 assert(IP_VALID(port
));
1617 if (ip_active(port
) &&
1618 waitq_member(&mq
->imq_wait_queue
, wqset
)) {
1619 if (actual
< maxnames
) {
1620 names
[actual
] = mq
->imq_receiver_name
;
1627 is_read_unlock(space
);
1635 * Routine: ipc_mqueue_destroy_locked
1637 * Destroy a (non-set) message queue.
1638 * Set any blocked senders running.
1639 * Destroy the kmsgs in the queue.
1642 * Receivers were removed when the receive right was "changed"
1645 ipc_mqueue_destroy_locked(ipc_mqueue_t mqueue
)
1647 ipc_kmsg_queue_t kmqueue
;
1649 boolean_t reap
= FALSE
;
1650 struct turnstile
*send_turnstile
= port_send_turnstile(ip_from_mq(mqueue
));
1652 assert(!imq_is_set(mqueue
));
1655 * rouse all blocked senders
1656 * (don't boost anyone - we're tearing this queue down)
1659 mqueue
->imq_fullwaiters
= FALSE
;
1661 if (send_turnstile
!= TURNSTILE_NULL
) {
1662 waitq_wakeup64_all(&send_turnstile
->ts_waitq
,
1665 WAITQ_ALL_PRIORITIES
);
1669 * Move messages from the specified queue to the per-thread
1670 * clean/drain queue while we have the mqueue lock.
1672 kmqueue
= &mqueue
->imq_messages
;
1673 while ((kmsg
= ipc_kmsg_dequeue(kmqueue
)) != IKM_NULL
) {
1675 if (MACH_NODE_VALID(kmsg
->ikm_node
) && FPORT_VALID(mqueue
->imq_fport
)) {
1676 flipc_msg_ack(kmsg
->ikm_node
, mqueue
, TRUE
);
1680 first
= ipc_kmsg_delayed_destroy(kmsg
);
1687 * Wipe out message count, both for messages about to be
1688 * reaped and for reserved space for (previously) woken senders.
1689 * This is the indication to them that their reserved space is gone
1690 * (the mqueue was destroyed).
1692 mqueue
->imq_msgcount
= 0;
1694 /* invalidate the waitq for subsequent mqueue operations */
1695 waitq_invalidate_locked(&mqueue
->imq_wait_queue
);
1697 /* clear out any preposting we may have done */
1698 waitq_clear_prepost_locked(&mqueue
->imq_wait_queue
);
1701 * assert that we are destroying / invalidating a queue that's
1702 * not a member of any other queue.
1704 assert(mqueue
->imq_preposts
== 0);
1705 assert(mqueue
->imq_in_pset
== 0);
1711 * Routine: ipc_mqueue_set_qlimit
1713 * Changes a message queue limit; the maximum number
1714 * of messages which may be queued.
1720 ipc_mqueue_set_qlimit(
1721 ipc_mqueue_t mqueue
,
1722 mach_port_msgcount_t qlimit
)
1724 assert(qlimit
<= MACH_PORT_QLIMIT_MAX
);
1726 /* wake up senders allowed by the new qlimit */
1728 if (qlimit
> mqueue
->imq_qlimit
) {
1729 mach_port_msgcount_t i
, wakeup
;
1730 struct turnstile
*send_turnstile
= port_send_turnstile(ip_from_mq(mqueue
));
1732 /* caution: wakeup, qlimit are unsigned */
1733 wakeup
= qlimit
- mqueue
->imq_qlimit
;
1735 for (i
= 0; i
< wakeup
; i
++) {
1737 * boost the priority of the awoken thread
1738 * (WAITQ_PROMOTE_PRIORITY) to ensure it uses
1739 * the message queue slot we've just reserved.
1741 * NOTE: this will never prepost
1743 if (send_turnstile
== TURNSTILE_NULL
||
1744 waitq_wakeup64_one(&send_turnstile
->ts_waitq
,
1747 WAITQ_PROMOTE_PRIORITY
) == KERN_NOT_WAITING
) {
1748 mqueue
->imq_fullwaiters
= FALSE
;
1751 mqueue
->imq_msgcount
++; /* give it to the awakened thread */
1754 mqueue
->imq_qlimit
= qlimit
;
1759 * Routine: ipc_mqueue_set_seqno
1761 * Changes an mqueue's sequence number.
1763 * Caller holds a reference to the queue's containing object.
1766 ipc_mqueue_set_seqno(
1767 ipc_mqueue_t mqueue
,
1768 mach_port_seqno_t seqno
)
1771 mqueue
->imq_seqno
= seqno
;
1777 * Routine: ipc_mqueue_copyin
1779 * Convert a name in a space to a message queue.
1781 * Nothing locked. If successful, the caller gets a ref for
1782 * for the object. This ref ensures the continued existence of
1785 * MACH_MSG_SUCCESS Found a message queue.
1786 * MACH_RCV_INVALID_NAME The space is dead.
1787 * MACH_RCV_INVALID_NAME The name doesn't denote a right.
1788 * MACH_RCV_INVALID_NAME
1789 * The denoted right is not receive or port set.
1790 * MACH_RCV_IN_SET Receive right is a member of a set.
1796 mach_port_name_t name
,
1797 ipc_mqueue_t
*mqueuep
,
1798 ipc_object_t
*objectp
)
1801 ipc_entry_bits_t bits
;
1802 ipc_object_t object
;
1803 ipc_mqueue_t mqueue
;
1805 is_read_lock(space
);
1806 if (!is_active(space
)) {
1807 is_read_unlock(space
);
1808 return MACH_RCV_INVALID_NAME
;
1811 entry
= ipc_entry_lookup(space
, name
);
1812 if (entry
== IE_NULL
) {
1813 is_read_unlock(space
);
1814 return MACH_RCV_INVALID_NAME
;
1817 bits
= entry
->ie_bits
;
1818 object
= entry
->ie_object
;
1820 if (bits
& MACH_PORT_TYPE_RECEIVE
) {
1821 ipc_port_t port
= ip_object_to_port(object
);
1823 assert(port
!= IP_NULL
);
1826 require_ip_active(port
);
1827 assert(port
->ip_receiver_name
== name
);
1828 assert(port
->ip_receiver
== space
);
1829 is_read_unlock(space
);
1830 mqueue
= &port
->ip_messages
;
1831 } else if (bits
& MACH_PORT_TYPE_PORT_SET
) {
1832 ipc_pset_t pset
= ips_object_to_pset(object
);
1834 assert(pset
!= IPS_NULL
);
1837 assert(ips_active(pset
));
1838 is_read_unlock(space
);
1840 mqueue
= &pset
->ips_messages
;
1842 is_read_unlock(space
);
1843 /* guard exception if we never held the receive right in this entry */
1844 if ((bits
& MACH_PORT_TYPE_EX_RECEIVE
) == 0) {
1845 mach_port_guard_exception(name
, 0, 0, kGUARD_EXC_RCV_INVALID_NAME
);
1847 return MACH_RCV_INVALID_NAME
;
1851 * At this point, the object is locked and active,
1852 * the space is unlocked, and mqueue is initialized.
1855 io_reference(object
);
1860 return MACH_MSG_SUCCESS
;
1864 imq_lock(ipc_mqueue_t mq
)
1866 ipc_object_t object
= imq_to_object(mq
);
1867 ipc_object_validate(object
);
1868 waitq_lock(&(mq
)->imq_wait_queue
);
1872 imq_lock_try(ipc_mqueue_t mq
)
1874 ipc_object_t object
= imq_to_object(mq
);
1875 ipc_object_validate(object
);
1876 return waitq_lock_try(&(mq
)->imq_wait_queue
);