osfmk/ipc/ipc_mqueue.c

110 	ipc_mqueue_t	mqueue,
114 		wait_queue_set_init(&mqueue->imq_set_queue, SYNC_POLICY_FIFO|SYNC_POLICY_PREPOST);
116 		wait_queue_init(&mqueue->imq_wait_queue, SYNC_POLICY_FIFO);
117 		ipc_kmsg_queue_init(&mqueue->imq_messages);
118 		mqueue->imq_seqno = 0;
119 		mqueue->imq_msgcount = 0;
120 		mqueue->imq_qlimit = MACH_PORT_QLIMIT_DEFAULT;
121 		mqueue->imq_fullwaiters = FALSE;
128  *		Indicate whether the (port) mqueue is a member of
129  *		this portset's mqueue.  We do this by checking
130  *		whether the portset mqueue's waitq is an member of
131  *		the port's mqueue waitq.
133  *		the portset's mqueue is not already a member
158 	ipc_mqueue_t	  mqueue,
162 	wait_queue_t	 mq_waitq = &mqueue->imq_wait_queue;
171  *		Remove the mqueue from all the sets it is a member of
177 	ipc_mqueue_t	mqueue,
180 	wait_queue_t	mq_waitq = &mqueue->imq_wait_queue;
195 	ipc_mqueue_t	mqueue,
198 	wait_queue_set_t	mq_setq = &mqueue->imq_set_queue;
208  *		Associate the portset's mqueue with the port's mqueue.
211  *		mqueue and messages on the port mqueue, try to match them
255 			/* waitq/mqueue still locked, thread locked */
328 	ipc_mqueue_t		mqueue)
331 				&mqueue->imq_wait_queue,
351  *		mqueue is locked.
359 	ipc_mqueue_t		mqueue,
373 	if (!imq_full(mqueue) ||
374 	    (!imq_full_kernel(mqueue) &&
378 		mqueue->imq_msgcount++;
379 		assert(mqueue->imq_msgcount > 0);
380 		imq_unlock(mqueue);
390 			imq_unlock(mqueue);
394 		if (imq_full_kernel(mqueue)) {
395 			imq_unlock(mqueue);
399 		mqueue->imq_fullwaiters = TRUE;
406 						&mqueue->imq_wait_queue,
413 		imq_unlock(mqueue);
428 			assert(mqueue->imq_msgcount > 0);
435 			/* mqueue is being destroyed */
442 	ipc_mqueue_post(mqueue, kmsg);
459 	ipc_mqueue_t mqueue)
461 	assert(imq_held(mqueue));
462 	assert(mqueue->imq_msgcount > 1 || ipc_kmsg_queue_empty(&mqueue->imq_messages));
464 	mqueue->imq_msgcount--;
466 	if (!imq_full(mqueue) && mqueue->imq_fullwaiters) {
468 						&mqueue->imq_wait_queue,
472 			mqueue->imq_fullwaiters = FALSE;
475 			mqueue->imq_msgcount++;
492 	register ipc_mqueue_t 	mqueue,
504 	imq_lock(mqueue);
506 		wait_queue_t waitq = &mqueue->imq_wait_queue;
521 			assert(mqueue->imq_msgcount > 0);
522 			ipc_kmsg_enqueue_macro(&mqueue->imq_messages, kmsg);
561 			receiver->ith_seqno = mqueue->imq_seqno++;
565 			ipc_mqueue_release_msgcount(mqueue);
574 		receiver->ith_receiver_name = mqueue->imq_receiver_name;
580 	imq_unlock(mqueue);
681 	ipc_mqueue_t            mqueue,
690         wresult = ipc_mqueue_receive_on_thread(mqueue, option, max_size,
712         ipc_mqueue_t            mqueue,
725 	imq_lock(mqueue);
727 	if (imq_is_set(mqueue)) {
730 		q = &mqueue->imq_preposts;
733 		 * If we are waiting on a portset mqueue, we need to see if
737 		 * mqueue lock during the search, we tie up any attempts by
760 				imq_unlock(mqueue);
764 				imq_lock(mqueue);
781 			 * while we still have the portset mqueue locked.
784 			imq_unlock(mqueue);
788 			 * the port mqueue locked.
802 	  	kmsgs = &mqueue->imq_messages;
804 			ipc_mqueue_select_on_thread(mqueue, option, max_size, thread);
805 			imq_unlock(mqueue);
812 	 * Looks like we'll have to block.  The mqueue we will
818 			imq_unlock(mqueue);
835 	wresult = wait_queue_assert_wait64_locked(&mqueue->imq_wait_queue,
846 	imq_unlock(mqueue);
859  *		mqueue locked.
868 	ipc_mqueue_t		mqueue,
881 	kmsg = ipc_kmsg_queue_first(&mqueue->imq_messages);
894 			thread->ith_receiver_name = mqueue->imq_receiver_name;
903 	ipc_kmsg_rmqueue_first_macro(&mqueue->imq_messages, kmsg);
904 	ipc_mqueue_release_msgcount(mqueue);
905 	thread->ith_seqno = mqueue->imq_seqno++;
1059 	 * Iterate over the member ports through the mqueue set links
1089 	ipc_mqueue_t	mqueue)
1097 	imq_lock(mqueue);
1101 	mqueue->imq_fullwaiters = FALSE;
1103 				&mqueue->imq_wait_queue,
1110 	 * clean/drain queue while we have the mqueue lock.
1112 	kmqueue = &mqueue->imq_messages;
1120 	imq_unlock(mqueue);
1142 	 ipc_mqueue_t			mqueue,
1151 	 imq_lock(mqueue);
1152 	 if (qlimit > mqueue->imq_qlimit) {
1156 		 wakeup = qlimit - mqueue->imq_qlimit;
1160 							&mqueue->imq_wait_queue,
1164 					 mqueue->imq_fullwaiters = FALSE;
1167 			 mqueue->imq_msgcount++;  /* give it to the awakened thread */
1170 	mqueue->imq_qlimit = qlimit;
1171 	imq_unlock(mqueue);
1178  *		Changes an mqueue's sequence number.
1184 	ipc_mqueue_t		mqueue,
1190 	imq_lock(mqueue);
1191 	mqueue->imq_seqno = seqno;
1192 	imq_unlock(mqueue);
1223 	ipc_mqueue_t mqueue;
1250 		mqueue = &port->ip_messages;
1263 		mqueue = &pset->ips_messages;
1271 	 *	the space is unlocked, and mqueue is initialized.
1278 	*mqueuep = mqueue;