xnu-792.24.17.tar.gz

[apple/xnu.git] / osfmk / ipc / ipc_mqueue.c

diff --git a/osfmk/ipc/ipc_mqueue.c b/osfmk/ipc/ipc_mqueue.c
index 97f13b62c918d89ccfb7dcd6f45ba5e7be2c45d0..9150acbc069bf6f78e77dc6de319eaba726d39c2 100644 (file)
--- a/osfmk/ipc/ipc_mqueue.c
+++ b/osfmk/ipc/ipc_mqueue.c
@@ -1,5 +1,5 @@
 /*
 /*

- * Copyright (c) 2000 Apple Computer, Inc. All rights reserved.
+ * Copyright (c) 2000-2004 Apple Computer, Inc. All rights reserved.

  *
  * @APPLE_LICENSE_HEADER_START@
  * 
  *
  * @APPLE_LICENSE_HEADER_START@
  * 


@@ -65,6 +65,7 @@
 #include <kern/counters.h>
 #include <kern/sched_prim.h>
 #include <kern/ipc_kobject.h>
 #include <kern/counters.h>
 #include <kern/sched_prim.h>
 #include <kern/ipc_kobject.h>

+#include <kern/ipc_mig.h>      /* XXX - for mach_msg_receive_continue */

 #include <kern/misc_protos.h>
 #include <kern/task.h>
 #include <kern/thread.h>
 #include <kern/misc_protos.h>
 #include <kern/task.h>
 #include <kern/thread.h>


@@ -83,6 +84,9 @@ int ipc_mqueue_rcv;           /* address is event for message arrival */
 
 #define TR_ENABLE 0
 
 
 #define TR_ENABLE 0
 

+/* forward declarations */
+void ipc_mqueue_receive_results(wait_result_t result);
+

 /*
  *     Routine:        ipc_mqueue_init
  *     Purpose:
 /*
  *     Routine:        ipc_mqueue_init
  *     Purpose:


@@ -94,7 +98,7 @@ ipc_mqueue_init(
        boolean_t       is_set)
 {
        if (is_set) {
        boolean_t       is_set)
 {
        if (is_set) {

-               wait_queue_sub_init(&mqueue->imq_set_queue, SYNC_POLICY_FIFO);
+               wait_queue_set_init(&mqueue->imq_set_queue, SYNC_POLICY_FIFO);

        } else {
                wait_queue_init(&mqueue->imq_wait_queue, SYNC_POLICY_FIFO);
                ipc_kmsg_queue_init(&mqueue->imq_messages);
        } else {
                wait_queue_init(&mqueue->imq_wait_queue, SYNC_POLICY_FIFO);
                ipc_kmsg_queue_init(&mqueue->imq_messages);


@@ -119,11 +123,11 @@ ipc_mqueue_init(
 
 boolean_t
 ipc_mqueue_member(
 
 boolean_t
 ipc_mqueue_member(

-       ipc_mqueue_t    port_mqueue,
-       ipc_mqueue_t    set_mqueue)
+       ipc_mqueue_t            port_mqueue,
+       ipc_mqueue_t            set_mqueue)

 {
        wait_queue_t    port_waitq = &port_mqueue->imq_wait_queue;
 {
        wait_queue_t    port_waitq = &port_mqueue->imq_wait_queue;

-       wait_queue_t    set_waitq = &set_mqueue->imq_wait_queue;
+       wait_queue_set_t set_waitq = &set_mqueue->imq_set_queue;

 
        return (wait_queue_member(port_waitq, set_waitq));
 
 
        return (wait_queue_member(port_waitq, set_waitq));
 


@@ -133,39 +137,51 @@ ipc_mqueue_member(
  *     Routine:        ipc_mqueue_remove
  *     Purpose:
  *             Remove the association between the queue and the specified
  *     Routine:        ipc_mqueue_remove
  *     Purpose:
  *             Remove the association between the queue and the specified

- *             subordinate message queue.
+ *             set message queue.

  */
 
 kern_return_t
 ipc_mqueue_remove(
        ipc_mqueue_t     mqueue,
  */
 
 kern_return_t
 ipc_mqueue_remove(
        ipc_mqueue_t     mqueue,

-       ipc_mqueue_t     sub_mqueue)
+       ipc_mqueue_t     set_mqueue)

 {
        wait_queue_t     mq_waitq = &mqueue->imq_wait_queue;
 {
        wait_queue_t     mq_waitq = &mqueue->imq_wait_queue;

-       wait_queue_sub_t sub_waitq = &sub_mqueue->imq_set_queue;
+       wait_queue_set_t set_waitq = &set_mqueue->imq_set_queue;

 
 

-       if (wait_queue_member(mq_waitq, sub_waitq)) {
-           wait_queue_unlink(mq_waitq, sub_waitq);
-           return KERN_SUCCESS;
-       }
-       return KERN_NOT_IN_SET;    
+       return wait_queue_unlink(mq_waitq, set_waitq);

 }
 
 /*
 }
 
 /*

- *     Routine:        ipc_mqueue_remove_one
+ *     Routine:        ipc_mqueue_remove_from_all

  *     Purpose:
  *     Purpose:

- *             Find and remove one subqueue from the queue.
+ *             Remove the mqueue from all the sets it is a member of

  *     Conditions:
  *     Conditions:

- *             Will return the set mqueue that was removed
+ *             Nothing locked.

  */
 void
  */
 void

-ipc_mqueue_remove_one(
-       ipc_mqueue_t    mqueue,
-       ipc_mqueue_t    *sub_queuep)
+ipc_mqueue_remove_from_all(
+       ipc_mqueue_t    mqueue)

 {
        wait_queue_t    mq_waitq = &mqueue->imq_wait_queue;
 
 {
        wait_queue_t    mq_waitq = &mqueue->imq_wait_queue;
 

-       wait_queue_unlink_one(mq_waitq, (wait_queue_sub_t *)sub_queuep);
+       wait_queue_unlink_all(mq_waitq);
+       return;
+}
+
+/*
+ *     Routine:        ipc_mqueue_remove_all
+ *     Purpose:
+ *             Remove all the member queues from the specified set.
+ *     Conditions:
+ *             Nothing locked.
+ */
+void
+ipc_mqueue_remove_all(
+       ipc_mqueue_t    mqueue)
+{
+       wait_queue_set_t        mq_setq = &mqueue->imq_set_queue;
+
+       wait_queue_set_unlink_all(mq_setq);

        return;
 }
 
        return;
 }
 


@@ -187,7 +203,7 @@ ipc_mqueue_add(
        ipc_mqueue_t     set_mqueue)
 {
        wait_queue_t     port_waitq = &port_mqueue->imq_wait_queue;
        ipc_mqueue_t     set_mqueue)
 {
        wait_queue_t     port_waitq = &port_mqueue->imq_wait_queue;

-       wait_queue_sub_t set_waitq = &set_mqueue->imq_set_queue;
+       wait_queue_set_t set_waitq = &set_mqueue->imq_set_queue;

        ipc_kmsg_queue_t kmsgq;
        ipc_kmsg_t       kmsg, next;
        kern_return_t    kr;
        ipc_kmsg_queue_t kmsgq;
        ipc_kmsg_t       kmsg, next;
        kern_return_t    kr;


@@ -213,10 +229,11 @@ ipc_mqueue_add(
                for (;;) {
                        thread_t th;
 
                for (;;) {
                        thread_t th;
 

-                       th = wait_queue_wakeup_identity_locked(port_waitq,
-                                                                                                  IPC_MQUEUE_RECEIVE,
-                                                                                                  THREAD_AWAKENED,
-                                                                                                  FALSE);
+                       th = wait_queue_wakeup64_identity_locked(
+                                               port_waitq,
+                                               IPC_MQUEUE_RECEIVE,
+                                               THREAD_AWAKENED,
+                                               FALSE);

                        /* waitq/mqueue still locked, thread locked */
 
                        if (th == THREAD_NULL)
                        /* waitq/mqueue still locked, thread locked */
 
                        if (th == THREAD_NULL)


@@ -231,10 +248,10 @@ ipc_mqueue_add(
                         * just move onto the next.
                         */
                        if (th->ith_msize <
                         * just move onto the next.
                         */
                        if (th->ith_msize <

-                           kmsg->ikm_header.msgh_size +
+                           kmsg->ikm_header->msgh_size +

                            REQUESTED_TRAILER_SIZE(th->ith_option)) {
                                th->ith_state = MACH_RCV_TOO_LARGE;
                            REQUESTED_TRAILER_SIZE(th->ith_option)) {
                                th->ith_state = MACH_RCV_TOO_LARGE;

-                               th->ith_msize = kmsg->ikm_header.msgh_size;
+                               th->ith_msize = kmsg->ikm_header->msgh_size;

                                if (th->ith_option & MACH_RCV_LARGE) {
                                        /*
                                         * let him go without message
                                if (th->ith_option & MACH_RCV_LARGE) {
                                        /*
                                         * let him go without message


@@ -252,8 +269,9 @@ ipc_mqueue_add(
                         * This thread is going to take this message,
                         * so give it to him.
                         */
                         * This thread is going to take this message,
                         * so give it to him.
                         */

-                       ipc_mqueue_release_msgcount(port_mqueue);

                        ipc_kmsg_rmqueue(kmsgq, kmsg);
                        ipc_kmsg_rmqueue(kmsgq, kmsg);

+                       ipc_mqueue_release_msgcount(port_mqueue);
+

                        th->ith_kmsg = kmsg;
                        th->ith_seqno = port_mqueue->imq_seqno++;
                        thread_unlock(th);
                        th->ith_kmsg = kmsg;
                        th->ith_seqno = port_mqueue->imq_seqno++;
                        thread_unlock(th);


@@ -279,10 +297,11 @@ void
 ipc_mqueue_changed(
        ipc_mqueue_t            mqueue)
 {
 ipc_mqueue_changed(
        ipc_mqueue_t            mqueue)
 {

-       wait_queue_wakeup_all_locked(&mqueue->imq_wait_queue,
-                                                                IPC_MQUEUE_RECEIVE,
-                                                                THREAD_RESTART,
-                                                                FALSE);                /* unlock waitq? */
+       wait_queue_wakeup64_all_locked(
+                               &mqueue->imq_wait_queue,
+                               IPC_MQUEUE_RECEIVE,
+                               THREAD_RESTART,
+                               FALSE);         /* unlock waitq? */

 }
 
 
 }
 
 


@@ -310,9 +329,9 @@ ipc_mqueue_send(
        ipc_mqueue_t            mqueue,
        ipc_kmsg_t                      kmsg,
        mach_msg_option_t       option,
        ipc_mqueue_t            mqueue,
        ipc_kmsg_t                      kmsg,
        mach_msg_option_t       option,

-       mach_msg_timeout_t      timeout)
+       mach_msg_timeout_t      send_timeout)

 {
 {

-       int save_wait_result;
+       int wresult;

        spl_t s;
 
        /*
        spl_t s;
 
        /*


@@ -326,49 +345,55 @@ ipc_mqueue_send(
 
        if (!imq_full(mqueue) ||
                (option & MACH_SEND_ALWAYS) ||
 
        if (!imq_full(mqueue) ||
                (option & MACH_SEND_ALWAYS) ||

-               (MACH_MSGH_BITS_REMOTE(kmsg->ikm_header.msgh_bits) ==
+               (MACH_MSGH_BITS_REMOTE(kmsg->ikm_header->msgh_bits) ==

                 MACH_MSG_TYPE_PORT_SEND_ONCE)) {
                mqueue->imq_msgcount++;
                 MACH_MSG_TYPE_PORT_SEND_ONCE)) {
                mqueue->imq_msgcount++;

+               assert(mqueue->imq_msgcount > 0);

                imq_unlock(mqueue);
                splx(s);
        } else {
                imq_unlock(mqueue);
                splx(s);
        } else {

+               thread_t cur_thread = current_thread();
+               uint64_t deadline;

 
                /* 
                 * We have to wait for space to be granted to us.
                 */
 
                /* 
                 * We have to wait for space to be granted to us.
                 */

-               if ((option & MACH_SEND_TIMEOUT) && (timeout == 0)) {
+               if ((option & MACH_SEND_TIMEOUT) && (send_timeout == 0)) {

                        imq_unlock(mqueue);
                        splx(s);
                        return MACH_SEND_TIMED_OUT;
                }
                mqueue->imq_fullwaiters = TRUE;
                        imq_unlock(mqueue);
                        splx(s);
                        return MACH_SEND_TIMED_OUT;
                }
                mqueue->imq_fullwaiters = TRUE;

-               (void)wait_queue_assert_wait_locked(&mqueue->imq_wait_queue,
-                                                                         IPC_MQUEUE_FULL,
-                                                                         THREAD_ABORTSAFE,
-                                                                         TRUE); /* unlock? */
-               /* wait/mqueue is unlocked */
+               thread_lock(cur_thread);
+               if (option & MACH_SEND_TIMEOUT)
+                       clock_interval_to_deadline(send_timeout, 1000*NSEC_PER_USEC, &deadline);
+               else
+                       deadline = 0;
+               wresult = wait_queue_assert_wait64_locked(
+                                               &mqueue->imq_wait_queue,
+                                               IPC_MQUEUE_FULL,
+                                               THREAD_ABORTSAFE, deadline,
+                                               cur_thread);
+               thread_unlock(cur_thread);
+               imq_unlock(mqueue);

                splx(s);
                
                splx(s);
                

-               if (option & MACH_SEND_TIMEOUT)
-                       thread_set_timer(timeout, 1000*NSEC_PER_USEC);
-
-               counter(c_ipc_mqueue_send_block++);
-               save_wait_result = thread_block((void (*)(void)) 0);
+               if (wresult == THREAD_WAITING) {
+                       wresult = thread_block(THREAD_CONTINUE_NULL);
+                       counter(c_ipc_mqueue_send_block++);
+               }

                
                

-               switch (save_wait_result) {
+               switch (wresult) {

                case THREAD_TIMED_OUT:
                        assert(option & MACH_SEND_TIMEOUT);
                        return MACH_SEND_TIMED_OUT;
                        
                case THREAD_AWAKENED:
                        /* we can proceed - inherited msgcount from waker */
                case THREAD_TIMED_OUT:
                        assert(option & MACH_SEND_TIMEOUT);
                        return MACH_SEND_TIMED_OUT;
                        
                case THREAD_AWAKENED:
                        /* we can proceed - inherited msgcount from waker */

-                       if (option & MACH_SEND_TIMEOUT)
-                               thread_cancel_timer();
+                       assert(mqueue->imq_msgcount > 0);

                        break;
                        
                case THREAD_INTERRUPTED:
                        break;
                        
                case THREAD_INTERRUPTED:

-                       if (option & MACH_SEND_TIMEOUT)
-                               thread_cancel_timer();

                        return MACH_SEND_INTERRUPTED;
                        
                case THREAD_RESTART:
                        return MACH_SEND_INTERRUPTED;
                        
                case THREAD_RESTART:


@@ -388,24 +413,28 @@ ipc_mqueue_send(
  *             found a waiter.
  *
  *     Conditions:
  *             found a waiter.
  *
  *     Conditions:

- *             The message queue is locked
+ *             The message queue is locked.
+ *             The message corresponding to this reference is off the queue.

  */
 void
 ipc_mqueue_release_msgcount(
        ipc_mqueue_t mqueue)    
 {
        assert(imq_held(mqueue));
  */
 void
 ipc_mqueue_release_msgcount(
        ipc_mqueue_t mqueue)    
 {
        assert(imq_held(mqueue));

-       assert(mqueue->imq_msgcount > 0);
+       assert(mqueue->imq_msgcount > 1 || ipc_kmsg_queue_empty(&mqueue->imq_messages));

 
        mqueue->imq_msgcount--;
 
        mqueue->imq_msgcount--;

+

        if (!imq_full(mqueue) && mqueue->imq_fullwaiters) {
        if (!imq_full(mqueue) && mqueue->imq_fullwaiters) {

-               if (wait_queue_wakeup_one_locked(&mqueue->imq_wait_queue,
-                                                                                IPC_MQUEUE_FULL,
-                                                                                THREAD_AWAKENED,
-                                                                                FALSE) != KERN_SUCCESS) {
+               if (wait_queue_wakeup64_one_locked(
+                                               &mqueue->imq_wait_queue,
+                                               IPC_MQUEUE_FULL,
+                                               THREAD_AWAKENED,
+                                               FALSE) != KERN_SUCCESS) {

                        mqueue->imq_fullwaiters = FALSE;
                } else {
                        mqueue->imq_fullwaiters = FALSE;
                } else {

-                       mqueue->imq_msgcount++;  /* gave it away */
+                       /* gave away our slot - add reference back */
+                       mqueue->imq_msgcount++; 

                }
        }
 }
                }
        }
 }


@@ -440,10 +469,11 @@ ipc_mqueue_post(
                wait_queue_t waitq = &mqueue->imq_wait_queue;
                thread_t receiver;
 
                wait_queue_t waitq = &mqueue->imq_wait_queue;
                thread_t receiver;
 

-               receiver = wait_queue_wakeup_identity_locked(waitq,
-                                                                                                        IPC_MQUEUE_RECEIVE,
-                                                                                                        THREAD_AWAKENED,
-                                                                                                        FALSE);
+               receiver = wait_queue_wakeup64_identity_locked(
+                                                       waitq,
+                                                       IPC_MQUEUE_RECEIVE,
+                                                       THREAD_AWAKENED,
+                                                       FALSE);

                /* waitq still locked, thread locked */
 
                if (receiver == THREAD_NULL) {
                /* waitq still locked, thread locked */
 
                if (receiver == THREAD_NULL) {


@@ -461,9 +491,9 @@ ipc_mqueue_post(
                 * the thread we wake up will get that as its status.
                 */
                if (receiver->ith_msize <
                 * the thread we wake up will get that as its status.
                 */
                if (receiver->ith_msize <

-                   (kmsg->ikm_header.msgh_size) +
+                   (kmsg->ikm_header->msgh_size) +

                    REQUESTED_TRAILER_SIZE(receiver->ith_option)) {
                    REQUESTED_TRAILER_SIZE(receiver->ith_option)) {

-                       receiver->ith_msize = kmsg->ikm_header.msgh_size;
+                       receiver->ith_msize = kmsg->ikm_header->msgh_size;

                        receiver->ith_state = MACH_RCV_TOO_LARGE;
                } else {
                        receiver->ith_state = MACH_MSG_SUCCESS;
                        receiver->ith_state = MACH_RCV_TOO_LARGE;
                } else {
                        receiver->ith_state = MACH_MSG_SUCCESS;


@@ -504,12 +534,11 @@ ipc_mqueue_post(
 }
 
 
 }
 
 

-kern_return_t
-ipc_mqueue_receive_results(void)
+/* static */ void
+ipc_mqueue_receive_results(wait_result_t saved_wait_result)

 {
        thread_t                self = current_thread();
        mach_msg_option_t       option = self->ith_option;
 {
        thread_t                self = current_thread();
        mach_msg_option_t       option = self->ith_option;

-       kern_return_t           saved_wait_result = self->wait_result;

        kern_return_t           mr;
 
        /*
        kern_return_t           mr;
 
        /*


@@ -521,15 +550,11 @@ ipc_mqueue_receive_results(void)
                return;
 
        case THREAD_INTERRUPTED:
                return;
 
        case THREAD_INTERRUPTED:

-               if (option & MACH_RCV_TIMEOUT)
-                       thread_cancel_timer();

                self->ith_state = MACH_RCV_INTERRUPTED;
                return;
 
        case THREAD_RESTART:
                /* something bad happened to the port/set */
                self->ith_state = MACH_RCV_INTERRUPTED;
                return;
 
        case THREAD_RESTART:
                /* something bad happened to the port/set */

-               if (option & MACH_RCV_TIMEOUT)
-                       thread_cancel_timer();

                self->ith_state = MACH_RCV_PORT_CHANGED;
                return;
 
                self->ith_state = MACH_RCV_PORT_CHANGED;
                return;
 


@@ -538,9 +563,6 @@ ipc_mqueue_receive_results(void)
                 * We do not need to go select a message, somebody
                 * handed us one (or a too-large indication).
                 */
                 * We do not need to go select a message, somebody
                 * handed us one (or a too-large indication).
                 */

-               if (option & MACH_RCV_TIMEOUT)
-                       thread_cancel_timer();
-               

                mr = MACH_MSG_SUCCESS;
 
                switch (self->ith_state) {
                mr = MACH_MSG_SUCCESS;
 
                switch (self->ith_state) {


@@ -570,9 +592,11 @@ ipc_mqueue_receive_results(void)
 }
 
 void
 }
 
 void

-ipc_mqueue_receive_continue(void)
+ipc_mqueue_receive_continue(
+       __unused void *param,
+       wait_result_t wresult)

 {
 {

-       ipc_mqueue_receive_results();
+       ipc_mqueue_receive_results(wresult);

        mach_msg_receive_continue();  /* hard-coded for now */
 }
 
        mach_msg_receive_continue();  /* hard-coded for now */
 }
 


@@ -606,20 +630,17 @@ ipc_mqueue_receive_continue(void)
 
 void
 ipc_mqueue_receive(
 
 void
 ipc_mqueue_receive(

-       ipc_mqueue_t            mqueue,
-       mach_msg_option_t       option,
-       mach_msg_size_t         max_size,
-       mach_msg_timeout_t      timeout,
-       int                     interruptible)
+       ipc_mqueue_t         mqueue,
+       mach_msg_option_t    option,
+       mach_msg_size_t      max_size,
+       mach_msg_timeout_t   rcv_timeout,
+       int                  interruptible)

 {
 {

-       ipc_port_t port;
-       mach_msg_return_t mr, mr2;
-       ipc_kmsg_queue_t kmsgs;
-       kern_return_t   save_wait_result;
-       thread_t self;
-       ipc_kmsg_t              *kmsgp;
-       mach_port_seqno_t       *seqnop;
-       spl_t s;
+       ipc_kmsg_queue_t        kmsgs;
+       wait_result_t           wresult;
+       thread_t                self;
+       uint64_t                                deadline;
+       spl_t                   s;

 
        s = splsched();
        imq_lock(mqueue);
 
        s = splsched();
        imq_lock(mqueue);


@@ -644,7 +665,7 @@ ipc_mqueue_receive(
                 * change will succeed and find us.
                 */
        search_set:
                 * change will succeed and find us.
                 */
        search_set:

-               queue_iterate(q, wql, wait_queue_link_t, wql_sublinks) {
+               queue_iterate(q, wql, wait_queue_link_t, wql_setlinks) {

                        port_mq = (ipc_mqueue_t)wql->wql_queue;
                        kmsgs = &port_mq->imq_messages;
                        
                        port_mq = (ipc_mqueue_t)wql->wql_queue;
                        kmsgs = &port_mq->imq_messages;
                        


@@ -671,8 +692,8 @@ ipc_mqueue_receive(
                                imq_unlock(port_mq);
                                continue;
                        }
                                imq_unlock(port_mq);
                                continue;
                        }

-                       queue_remove(q, wql, wait_queue_link_t, wql_sublinks);
-                       queue_enter(q, wql, wait_queue_link_t, wql_sublinks);
+                       queue_remove(q, wql, wait_queue_link_t, wql_setlinks);
+                       queue_enter(q, wql, wait_queue_link_t, wql_setlinks);

                        imq_unlock(mqueue);
 
                        ipc_mqueue_select(port_mq, option, max_size);
                        imq_unlock(mqueue);
 
                        ipc_mqueue_select(port_mq, option, max_size);


@@ -703,7 +724,7 @@ ipc_mqueue_receive(
         */
        self = current_thread();
        if (option & MACH_RCV_TIMEOUT) {
         */
        self = current_thread();
        if (option & MACH_RCV_TIMEOUT) {

-               if (timeout == 0) {
+               if (rcv_timeout == 0) {

                        imq_unlock(mqueue);
                        splx(s);
                        self->ith_state = MACH_RCV_TIMED_OUT;
                        imq_unlock(mqueue);
                        splx(s);
                        self->ith_state = MACH_RCV_TIMED_OUT;


@@ -711,38 +732,36 @@ ipc_mqueue_receive(
                }
        }
 
                }
        }
 

+       thread_lock(self);

        self->ith_state = MACH_RCV_IN_PROGRESS;
        self->ith_option = option;
        self->ith_msize = max_size;
        self->ith_state = MACH_RCV_IN_PROGRESS;
        self->ith_option = option;
        self->ith_msize = max_size;

-               
-       (void)wait_queue_assert_wait_locked(&mqueue->imq_wait_queue,
-                                                                 IPC_MQUEUE_RECEIVE,
-                                                                 interruptible,
-                                                                 TRUE); /* unlock? */
-       /* mqueue/waitq is unlocked */
+
+       if (option & MACH_RCV_TIMEOUT)
+               clock_interval_to_deadline(rcv_timeout, 1000*NSEC_PER_USEC, &deadline);
+       else
+               deadline = 0;
+
+       wresult = wait_queue_assert_wait64_locked(&mqueue->imq_wait_queue,
+                                                                                               IPC_MQUEUE_RECEIVE,
+                                                                                               interruptible, deadline,
+                                                                                               self);
+       thread_unlock(self);
+       imq_unlock(mqueue);

        splx(s);
 
        splx(s);
 

-       if (option & MACH_RCV_TIMEOUT) {
-               thread_set_timer(timeout, 1000*NSEC_PER_USEC);
-       }
+       if (wresult == THREAD_WAITING) {
+               counter((interruptible == THREAD_ABORTSAFE) ? 
+                       c_ipc_mqueue_receive_block_user++ :
+                       c_ipc_mqueue_receive_block_kernel++);

 
 

-       if (interruptible == THREAD_ABORTSAFE) {
-               counter(c_ipc_mqueue_receive_block_user++);
-       } else {
-               counter(c_ipc_mqueue_receive_block_kernel++);
-       }
+               if (self->ith_continuation)
+                       thread_block(ipc_mqueue_receive_continue);
+                       /* NOTREACHED */

 
 

-#if defined (__i386__)
-       thread_block((void (*)(void))0);
-#else
-       if (self->ith_continuation) {
-               thread_block(ipc_mqueue_receive_continue);
-       } else {
-               thread_block((void (*)(void))0);
+               wresult = thread_block(THREAD_CONTINUE_NULL);

        }
        }

-#endif
-
-       ipc_mqueue_receive_results();  /* if we fell thru */
+       ipc_mqueue_receive_results(wresult);

 }
 
 
 }
 
 


@@ -767,8 +786,8 @@ ipc_mqueue_select(
 {
        thread_t self = current_thread();
        ipc_kmsg_t kmsg;
 {
        thread_t self = current_thread();
        ipc_kmsg_t kmsg;

-       mach_port_seqno_t seqno;

        mach_msg_return_t mr;
        mach_msg_return_t mr;

+       mach_msg_size_t rcv_size;

 
        mr = MACH_MSG_SUCCESS;
 
 
        mr = MACH_MSG_SUCCESS;
 


@@ -778,26 +797,24 @@ ipc_mqueue_select(
         * before pulling the message off the queue.
         */
        kmsg = ipc_kmsg_queue_first(&mqueue->imq_messages);
         * before pulling the message off the queue.
         */
        kmsg = ipc_kmsg_queue_first(&mqueue->imq_messages);

-

        assert(kmsg != IKM_NULL);
 
        assert(kmsg != IKM_NULL);
 

-       if (kmsg->ikm_header.msgh_size + 
-               REQUESTED_TRAILER_SIZE(option) > max_size) {
-               mr = MACH_RCV_TOO_LARGE;
-       } 
-

        /*
         * If we really can't receive it, but we had the
         * MACH_RCV_LARGE option set, then don't take it off
         * the queue, instead return the appropriate error
         * (and size needed).
         */
        /*
         * If we really can't receive it, but we had the
         * MACH_RCV_LARGE option set, then don't take it off
         * the queue, instead return the appropriate error
         * (and size needed).
         */

-       if ((mr == MACH_RCV_TOO_LARGE) && (option & MACH_RCV_LARGE)) {
-               self->ith_kmsg = IKM_NULL;
-               self->ith_msize = kmsg->ikm_header.msgh_size;
-               self->ith_seqno = 0;
-               self->ith_state = mr;
-               return;
+       rcv_size = ipc_kmsg_copyout_size(kmsg, self->map);
+       if (rcv_size + REQUESTED_TRAILER_SIZE(option) > max_size) {
+               mr = MACH_RCV_TOO_LARGE;
+               if (option & MACH_RCV_LARGE) {
+                       self->ith_kmsg = IKM_NULL;
+                       self->ith_msize = rcv_size;
+                       self->ith_seqno = 0;
+                       self->ith_state = mr;
+                       return;
+               }

        }
 
        ipc_kmsg_rmqueue_first_macro(&mqueue->imq_messages, kmsg);
        }
 
        ipc_kmsg_rmqueue_first_macro(&mqueue->imq_messages, kmsg);


@@ -834,10 +851,11 @@ ipc_mqueue_destroy(
         *      rouse all blocked senders
         */
        mqueue->imq_fullwaiters = FALSE;
         *      rouse all blocked senders
         */
        mqueue->imq_fullwaiters = FALSE;

-       wait_queue_wakeup_all_locked(&mqueue->imq_wait_queue,
-                                                                IPC_MQUEUE_FULL,
-                                                                THREAD_AWAKENED,
-                                                                FALSE);
+       wait_queue_wakeup64_all_locked(
+                               &mqueue->imq_wait_queue,
+                               IPC_MQUEUE_FULL,
+                               THREAD_AWAKENED,
+                               FALSE);

 
        kmqueue = &mqueue->imq_messages;
 
 
        kmqueue = &mqueue->imq_messages;
 


@@ -870,6 +888,8 @@ ipc_mqueue_set_qlimit(
 {
         spl_t s;
 
 {
         spl_t s;
 

+        assert(qlimit <= MACH_PORT_QLIMIT_MAX);
+

         /* wake up senders allowed by the new qlimit */
         s = splsched();
         imq_lock(mqueue);
         /* wake up senders allowed by the new qlimit */
         s = splsched();
         imq_lock(mqueue);


@@ -880,13 +900,15 @@ ipc_mqueue_set_qlimit(
                 wakeup = qlimit - mqueue->imq_qlimit;
 
                 for (i = 0; i < wakeup; i++) {
                 wakeup = qlimit - mqueue->imq_qlimit;
 
                 for (i = 0; i < wakeup; i++) {

-                        if (wait_queue_wakeup_one_locked(&mqueue->imq_wait_queue,
-                                                                                         IPC_MQUEUE_FULL,
-                                                                                         THREAD_AWAKENED,
-                                                                                         FALSE) == KERN_NOT_WAITING) {
+                        if (wait_queue_wakeup64_one_locked(
+                                                       &mqueue->imq_wait_queue,
+                                                       IPC_MQUEUE_FULL,
+                                                       THREAD_AWAKENED,
+                                                       FALSE) == KERN_NOT_WAITING) {

                                         mqueue->imq_fullwaiters = FALSE;
                                         break;
                         }
                                         mqueue->imq_fullwaiters = FALSE;
                                         break;
                         }

+                        mqueue->imq_msgcount++;  /* give it to the awakened thread */

                 }
         }
        mqueue->imq_qlimit = qlimit;
                 }
         }
        mqueue->imq_qlimit = qlimit;


@@ -960,7 +982,6 @@ ipc_mqueue_copyin(
 
        if (entry->ie_bits & MACH_PORT_TYPE_RECEIVE) {
                ipc_port_t port;
 
        if (entry->ie_bits & MACH_PORT_TYPE_RECEIVE) {
                ipc_port_t port;

-               ipc_pset_t pset;

 
                port = (ipc_port_t) object;
                assert(port != IP_NULL);
 
                port = (ipc_port_t) object;
                assert(port != IP_NULL);