xref: /dragonfly/sys/kern/sys_mqueue.c (revision 858b3437)
1 /*	$NetBSD: sys_mqueue.c,v 1.16 2009/04/11 23:05:26 christos Exp $	*/
2 
3 /*
4  * Copyright (c) 2007, 2008 Mindaugas Rasiukevicius <rmind at NetBSD org>
5  * All rights reserved.
6  *
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted provided that the following conditions
9  * are met:
10  * 1. Redistributions of source code must retain the above copyright
11  *    notice, this list of conditions and the following disclaimer.
12  * 2. Redistributions in binary form must reproduce the above copyright
13  *    notice, this list of conditions and the following disclaimer in the
14  *    documentation and/or other materials provided with the distribution.
15  *
16  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
17  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
20  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
21  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
22  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
23  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
24  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
25  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
26  * SUCH DAMAGE.
27  */
28 
29 /*
30  * Implementation of POSIX message queues.
31  * Defined in the Base Definitions volume of IEEE Std 1003.1-2001.
32  *
33  * Locking
34  *
35  * Global list of message queues (mqueue_head) and proc_t::p_mqueue_cnt
36  * counter are protected by mqlist_mtx lock.  The very message queue and
37  * its members are protected by mqueue::mq_mtx.
38  *
39  * Lock order:
40  *	mqlist_mtx
41  *	  -> mqueue::mq_mtx
42  */
43 
44 #include <stdbool.h>
45 #include <sys/param.h>
46 #include <sys/types.h>
47 #include <sys/errno.h>
48 #include <sys/fcntl.h>
49 #include <sys/file.h>
50 #include <sys/filedesc.h>
51 #include <sys/ucred.h>
52 #include <sys/priv.h>
53 #include <sys/kernel.h>
54 #include <sys/malloc.h>
55 #include <sys/mplock2.h>
56 #include <sys/mqueue.h>
57 #include <sys/objcache.h>
58 #include <sys/proc.h>
59 #include <sys/queue.h>
60 #include <sys/select.h>
61 #include <sys/serialize.h>
62 #include <sys/signal.h>
63 #include <sys/signalvar.h>
64 #include <sys/spinlock.h>
65 #include <sys/spinlock2.h>
66 #include <sys/stat.h>
67 #include <sys/sysctl.h>
68 #include <sys/sysproto.h>
69 #include <sys/systm.h>
70 #include <sys/lock.h>
71 #include <sys/unistd.h>
72 #include <sys/vnode.h>
73 
74 /* System-wide limits. */
75 static u_int			mq_open_max = MQ_OPEN_MAX;
76 static u_int			mq_prio_max = MQ_PRIO_MAX;
77 static u_int			mq_max_msgsize = 16 * MQ_DEF_MSGSIZE;
78 static u_int			mq_def_maxmsg = 32;
79 static u_int			mq_max_maxmsg = 16 * 32;
80 
81 struct lock			mqlist_mtx;
82 static struct objcache *	mqmsg_cache;
83 static LIST_HEAD(, mqueue)	mqueue_head =
84 	LIST_HEAD_INITIALIZER(mqueue_head);
85 
86 typedef struct	file file_t;	/* XXX: Should we put this in sys/types.h ? */
87 
88 /* Function prototypes */
89 static int	mq_stat_fop(file_t *, struct stat *, struct ucred *cred);
90 static int	mq_close_fop(file_t *);
91 
92 /* Some time-related utility functions */
93 static int	itimespecfix(struct timespec *ts);
94 static int	tstohz(const struct timespec *ts);
95 
96 /* File operations vector */
97 static struct fileops mqops = {
98 	.fo_read = badfo_readwrite,
99 	.fo_write = badfo_readwrite,
100 	.fo_ioctl = badfo_ioctl,
101 	.fo_stat = mq_stat_fop,
102 	.fo_close = mq_close_fop,
103 	.fo_kqfilter = badfo_kqfilter,
104 	.fo_shutdown = badfo_shutdown
105 };
106 
107 /* Define a new malloc type for message queues */
108 MALLOC_DECLARE(M_MQBUF);
109 MALLOC_DEFINE(M_MQBUF, "mqueues", "Buffers to message queues");
110 
111 /* Malloc arguments for object cache */
112 struct objcache_malloc_args mqueue_malloc_args = {
113 	sizeof(struct mqueue), M_MQBUF };
114 
115 /*
116  * Initialize POSIX message queue subsystem.
117  */
118 void
119 mqueue_sysinit(void)
120 {
121 	mqmsg_cache = objcache_create("mqmsg_cache",
122 	    0,		/* infinite depot's capacity */
123 	    0,		/* default magazine's capacity */
124 	    NULL,	/* constructor */
125 	    NULL,	/* deconstructor */
126 	    NULL,
127 	    objcache_malloc_alloc,
128 	    objcache_malloc_free,
129 	    &mqueue_malloc_args);
130 
131 	lockinit(&mqlist_mtx, "mqlist_mtx", 0, LK_CANRECURSE);
132 }
133 
134 /*
135  * Free the message.
136  */
137 static void
138 mqueue_freemsg(struct mq_msg *msg, const size_t size)
139 {
140 
141 	if (size > MQ_DEF_MSGSIZE) {
142 		kfree(msg, M_MQBUF);
143 	} else {
144 		objcache_put(mqmsg_cache, msg);
145 	}
146 }
147 
148 /*
149  * Destroy the message queue.
150  */
151 static void
152 mqueue_destroy(struct mqueue *mq)
153 {
154 	struct mq_msg *msg;
155 	size_t msz;
156 	u_int i;
157 
158 	/* Note MQ_PQSIZE + 1. */
159 	for (i = 0; i < MQ_PQSIZE + 1; i++) {
160 		while ((msg = TAILQ_FIRST(&mq->mq_head[i])) != NULL) {
161 			TAILQ_REMOVE(&mq->mq_head[i], msg, msg_queue);
162 			msz = sizeof(struct mq_msg) + msg->msg_len;
163 			mqueue_freemsg(msg, msz);
164 		}
165 	}
166 	lockuninit(&mq->mq_mtx);
167 	kfree(mq, M_MQBUF);
168 }
169 
170 /*
171  * Lookup for file name in general list of message queues.
172  *  => locks the message queue
173  */
174 static void *
175 mqueue_lookup(char *name)
176 {
177 	struct mqueue *mq;
178 
179 	KKASSERT(lockstatus(&mqlist_mtx, curthread));
180 
181 	LIST_FOREACH(mq, &mqueue_head, mq_list) {
182 		if (strncmp(mq->mq_name, name, MQ_NAMELEN) == 0) {
183 			lockmgr(&mq->mq_mtx, LK_EXCLUSIVE);
184 			return mq;
185 		}
186 	}
187 
188 	return NULL;
189 }
190 
191 /*
192  * mqueue_get: get the mqueue from the descriptor.
193  *  => locks the message queue, if found.
194  *  => holds a reference on the file descriptor.
195  */
196 static int
197 mqueue_get(struct lwp *l, mqd_t mqd, file_t **fpr)
198 {
199 	struct mqueue *mq;
200 	file_t *fp;
201 
202 	fp = holdfp(curproc->p_fd, (int)mqd, -1);	/* XXX: Why -1 ? */
203 	if (__predict_false(fp == NULL))
204 		return EBADF;
205 
206 	if (__predict_false(fp->f_type != DTYPE_MQUEUE)) {
207 		fdrop(fp);
208 		return EBADF;
209 	}
210 	mq = fp->f_data;
211 	lockmgr(&mq->mq_mtx, LK_EXCLUSIVE);
212 
213 	*fpr = fp;
214 	return 0;
215 }
216 
217 /*
218  * mqueue_linear_insert: perform linear insert according to the message
219  * priority into the reserved queue (MQ_PQRESQ).  Reserved queue is a
220  * sorted list used only when mq_prio_max is increased via sysctl.
221  */
222 static inline void
223 mqueue_linear_insert(struct mqueue *mq, struct mq_msg *msg)
224 {
225 	struct mq_msg *mit;
226 
227 	TAILQ_FOREACH(mit, &mq->mq_head[MQ_PQRESQ], msg_queue) {
228 		if (msg->msg_prio > mit->msg_prio)
229 			break;
230 	}
231 	if (mit == NULL) {
232 		TAILQ_INSERT_TAIL(&mq->mq_head[MQ_PQRESQ], msg, msg_queue);
233 	} else {
234 		TAILQ_INSERT_BEFORE(mit, msg, msg_queue);
235 	}
236 }
237 
238 /*
239  * Validate input.
240  */
241 int
242 itimespecfix(struct timespec *ts)
243 {
244 	if (ts->tv_sec < 0 || ts->tv_nsec < 0 || ts->tv_nsec >= 1000000000)
245 		return (EINVAL);
246 	if (ts->tv_sec == 0 && ts->tv_nsec != 0 && ts->tv_nsec < nstick)
247 		ts->tv_nsec = nstick;
248 	return (0);
249 }
250 
251 /*
252  * Compute number of ticks in the specified amount of time.
253  */
254 int
255 tstohz(const struct timespec *ts)
256 {
257 	struct timeval tv;
258 
259 	/*
260 	 * usec has great enough resolution for hz, so convert to a
261 	 * timeval and use tvtohz() above.
262 	 */
263 	TIMESPEC_TO_TIMEVAL(&tv, ts);
264 	return tvtohz_high(&tv);	/* XXX Why _high() and not _low() ? */
265 }
266 
267 /*
268  * Converter from struct timespec to the ticks.
269  * Used by mq_timedreceive(), mq_timedsend().
270  */
271 int
272 abstimeout2timo(struct timespec *ts, int *timo)
273 {
274 	struct timespec tsd;
275 	int error;
276 
277 	error = itimespecfix(ts);
278 	if (error) {
279 		return error;
280 	}
281 	getnanotime(&tsd);
282 	timespecsub(ts, &tsd);
283 	if (ts->tv_sec < 0 || (ts->tv_sec == 0 && ts->tv_nsec <= 0)) {
284 		return ETIMEDOUT;
285 	}
286 	*timo = tstohz(ts);
287 	KKASSERT(*timo != 0);
288 
289 	return 0;
290 }
291 
292 static int
293 mq_stat_fop(file_t *fp, struct stat *st, struct ucred *cred)
294 {
295 	struct mqueue *mq = fp->f_data;
296 
297 	(void)memset(st, 0, sizeof(*st));
298 
299 	lockmgr(&mq->mq_mtx, LK_EXCLUSIVE);
300 	st->st_mode = mq->mq_mode;
301 	st->st_uid = mq->mq_euid;
302 	st->st_gid = mq->mq_egid;
303 	st->st_atimespec = mq->mq_atime;
304 	st->st_mtimespec = mq->mq_mtime;
305 	/*st->st_ctimespec = st->st_birthtimespec = mq->mq_btime;*/
306 	st->st_uid = fp->f_cred->cr_uid;
307 	st->st_gid = fp->f_cred->cr_svgid;
308 	lockmgr(&mq->mq_mtx, LK_RELEASE);
309 
310 	return 0;
311 }
312 
313 static int
314 mq_close_fop(file_t *fp)
315 {
316 	struct proc *p = curproc;
317 	struct mqueue *mq = fp->f_data;
318 	bool destroy;
319 
320 	lockmgr(&mqlist_mtx, LK_EXCLUSIVE);
321 	lockmgr(&mq->mq_mtx, LK_EXCLUSIVE);
322 
323 	/* Decrease the counters */
324 	p->p_mqueue_cnt--;
325 	mq->mq_refcnt--;
326 
327 	/* Remove notification if registered for this process */
328 	if (mq->mq_notify_proc == p)
329 		mq->mq_notify_proc = NULL;
330 
331 	/*
332 	 * If this is the last reference and mqueue is marked for unlink,
333 	 * remove and later destroy the message queue.
334 	 */
335 	if (mq->mq_refcnt == 0 && (mq->mq_attrib.mq_flags & MQ_UNLINK)) {
336 		LIST_REMOVE(mq, mq_list);
337 		destroy = true;
338 	} else
339 		destroy = false;
340 
341 	lockmgr(&mq->mq_mtx, LK_RELEASE);
342 	lockmgr(&mqlist_mtx, LK_RELEASE);
343 
344 	if (destroy)
345 		mqueue_destroy(mq);
346 
347 	return 0;
348 }
349 
350 /*
351  * General mqueue system calls.
352  */
353 
354 int
355 sys_mq_open(struct mq_open_args *uap)
356 {
357 	/* {
358 		syscallarg(const char *) name;
359 		syscallarg(int) oflag;
360 		syscallarg(mode_t) mode;
361 		syscallarg(struct mq_attr) attr;
362 	} */
363 	struct thread *td = curthread;
364 	struct proc *p = td->td_proc;
365 	struct filedesc *fdp = p->p_fd;
366 	struct mqueue *mq, *mq_new = NULL;
367 	file_t *fp;
368 	char *name;
369 	int mqd, error, oflag;
370 
371 	/* Check access mode flags */
372 	oflag = SCARG(uap, oflag);
373 	if ((oflag & O_ACCMODE) == (O_WRONLY | O_RDWR)) {
374 		return EINVAL;
375 	}
376 
377 	/* Get the name from the user-space */
378 	name = kmalloc(MQ_NAMELEN, M_MQBUF, M_WAITOK | M_ZERO);
379 	error = copyinstr(SCARG(uap, name), name, MQ_NAMELEN - 1, NULL);
380 	if (error) {
381 		kfree(name, M_MQBUF);
382 		return error;
383 	}
384 
385 	if (oflag & O_CREAT) {
386 		struct mq_attr attr;
387 		u_int i;
388 
389 		/* Check the limit */
390 		if (p->p_mqueue_cnt == mq_open_max) {
391 			kfree(name, M_MQBUF);
392 			return EMFILE;
393 		}
394 
395 		/* Empty name is invalid */
396 		if (name[0] == '\0') {
397 			kfree(name, M_MQBUF);
398 			return EINVAL;
399 		}
400 
401 		/* Check for mqueue attributes */
402 		if (SCARG(uap, attr)) {
403 			error = copyin(SCARG(uap, attr), &attr,
404 				sizeof(struct mq_attr));
405 			if (error) {
406 				kfree(name, M_MQBUF);
407 				return error;
408 			}
409 			if (attr.mq_maxmsg <= 0 ||
410 			    attr.mq_maxmsg > mq_max_maxmsg ||
411 			    attr.mq_msgsize <= 0 ||
412 			    attr.mq_msgsize > mq_max_msgsize) {
413 				kfree(name, M_MQBUF);
414 				return EINVAL;
415 			}
416 			attr.mq_curmsgs = 0;
417 		} else {
418 			memset(&attr, 0, sizeof(struct mq_attr));
419 			attr.mq_maxmsg = mq_def_maxmsg;
420 			attr.mq_msgsize =
421 			    MQ_DEF_MSGSIZE - sizeof(struct mq_msg);
422 		}
423 
424 		/*
425 		 * Allocate new mqueue, initialize data structures,
426 		 * copy the name, attributes and set the flag.
427 		 */
428 		mq_new = kmalloc(sizeof(struct mqueue), M_MQBUF, M_WAITOK | M_ZERO);
429 
430 		lockinit(&mq_new->mq_mtx, "mq_new->mq_mtx", 0, LK_CANRECURSE);
431 		for (i = 0; i < (MQ_PQSIZE + 1); i++) {
432 			TAILQ_INIT(&mq_new->mq_head[i]);
433 		}
434 
435 		strlcpy(mq_new->mq_name, name, MQ_NAMELEN);
436 		memcpy(&mq_new->mq_attrib, &attr, sizeof(struct mq_attr));
437 
438 		/*CTASSERT((O_MASK & (MQ_UNLINK | MQ_RECEIVE)) == 0);*/
439 		/* mq_new->mq_attrib.mq_flags = (O_MASK & oflag); */
440 		mq_new->mq_attrib.mq_flags = oflag;
441 
442 		/* Store mode and effective UID with GID */
443 		mq_new->mq_mode = ((SCARG(uap, mode) &
444 		    ~p->p_fd->fd_cmask) & ALLPERMS) & ~S_ISTXT;
445 		mq_new->mq_euid = td->td_ucred->cr_uid;
446 		mq_new->mq_egid = td->td_ucred->cr_svgid;
447 	}
448 
449 	/* Allocate file structure and descriptor */
450 	error = falloc(td->td_lwp, &fp, &mqd);
451 	if (error) {
452 		if (mq_new)
453 			mqueue_destroy(mq_new);
454 		kfree(name, M_MQBUF);
455 		return error;
456 	}
457 	fp->f_type = DTYPE_MQUEUE;
458 	fp->f_flag = FFLAGS(oflag) & (FREAD | FWRITE);
459 	fp->f_ops = &mqops;
460 
461 	/* Look up for mqueue with such name */
462 	lockmgr(&mqlist_mtx, LK_EXCLUSIVE);
463 	mq = mqueue_lookup(name);
464 	if (mq) {
465 		int acc_mode;
466 
467 		KKASSERT(lockstatus(&mq->mq_mtx, curthread));
468 
469 		/* Check if mqueue is not marked as unlinking */
470 		if (mq->mq_attrib.mq_flags & MQ_UNLINK) {
471 			error = EACCES;
472 			goto exit;
473 		}
474 		/* Fail if O_EXCL is set, and mqueue already exists */
475 		if ((oflag & O_CREAT) && (oflag & O_EXCL)) {
476 			error = EEXIST;
477 			goto exit;
478 		}
479 
480 		/*
481 		 * Check the permissions. Note the difference between
482 		 * VREAD/VWRITE and FREAD/FWRITE.
483 		 */
484 		acc_mode = 0;
485 		if (fp->f_flag & FREAD) {
486 			acc_mode |= VREAD;
487 		}
488 		if (fp->f_flag & FWRITE) {
489 			acc_mode |= VWRITE;
490 		}
491 		if (vaccess(VNON, mq->mq_mode, mq->mq_euid, mq->mq_egid,
492 			acc_mode, td->td_ucred)) {
493 
494 			error = EACCES;
495 			goto exit;
496 		}
497 	} else {
498 		/* Fail if mqueue neither exists, nor we create it */
499 		if ((oflag & O_CREAT) == 0) {
500 			lockmgr(&mqlist_mtx, LK_RELEASE);
501 			KKASSERT(mq_new == NULL);
502 			fsetfd(fdp, NULL, mqd);
503 			fp->f_ops = &badfileops;
504 			fdrop(fp);
505 			kfree(name, M_MQBUF);
506 			return ENOENT;
507 		}
508 
509 		/* Check the limit */
510 		if (p->p_mqueue_cnt == mq_open_max) {
511 			error = EMFILE;
512 			goto exit;
513 		}
514 
515 		/* Insert the queue to the list */
516 		mq = mq_new;
517 		lockmgr(&mq->mq_mtx, LK_EXCLUSIVE);
518 		LIST_INSERT_HEAD(&mqueue_head, mq, mq_list);
519 		mq_new = NULL;
520 		getnanotime(&mq->mq_btime);
521 		mq->mq_atime = mq->mq_mtime = mq->mq_btime;
522 	}
523 
524 	/* Increase the counters, and make descriptor ready */
525 	p->p_mqueue_cnt++;
526 	mq->mq_refcnt++;
527 	fp->f_data = mq;
528 exit:
529 	lockmgr(&mq->mq_mtx, LK_RELEASE);
530 	lockmgr(&mqlist_mtx, LK_RELEASE);
531 
532 	if (mq_new)
533 		mqueue_destroy(mq_new);
534 	if (error) {
535 		fsetfd(fdp, NULL, mqd);
536 		fp->f_ops = &badfileops;
537 	} else {
538 		fsetfd(fdp, fp, mqd);
539 		uap->sysmsg_result = mqd;
540 	}
541 	fdrop(fp);
542 	kfree(name, M_MQBUF);
543 
544 	return error;
545 }
546 
547 int
548 sys_mq_close(struct mq_close_args *uap)
549 {
550 	return sys_close((void *)uap);
551 }
552 
553 /*
554  * Primary mq_receive1() function.
555  */
556 int
557 mq_receive1(struct lwp *l, mqd_t mqdes, void *msg_ptr, size_t msg_len,
558     unsigned *msg_prio, struct timespec *ts, ssize_t *mlen)
559 {
560 	file_t *fp = NULL;
561 	struct mqueue *mq;
562 	struct mq_msg *msg = NULL;
563 	struct mq_attr *mqattr;
564 	u_int idx;
565 	int error;
566 
567 	/* Get the message queue */
568 	error = mqueue_get(l, mqdes, &fp);
569 	if (error) {
570 		return error;
571 	}
572 	mq = fp->f_data;
573 	if ((fp->f_flag & FREAD) == 0) {
574 		error = EBADF;
575 		goto error;
576 	}
577 	getnanotime(&mq->mq_atime);
578 	mqattr = &mq->mq_attrib;
579 
580 	/* Check the message size limits */
581 	if (msg_len < mqattr->mq_msgsize) {
582 		error = EMSGSIZE;
583 		goto error;
584 	}
585 
586 	/* Check if queue is empty */
587 	while (mqattr->mq_curmsgs == 0) {
588 		int t;
589 
590 		if (mqattr->mq_flags & O_NONBLOCK) {
591 			error = EAGAIN;
592 			goto error;
593 		}
594 		if (ts) {
595 			error = abstimeout2timo(ts, &t);
596 			if (error)
597 				goto error;
598 		} else
599 			t = 0;
600 		/*
601 		 * Block until someone sends the message.
602 		 * While doing this, notification should not be sent.
603 		 */
604 		mqattr->mq_flags |= MQ_RECEIVE;
605 		error = lksleep(&mq->mq_send_cv, &mq->mq_mtx, PCATCH, "mqsend", t);
606 		mqattr->mq_flags &= ~MQ_RECEIVE;
607 		if (error || (mqattr->mq_flags & MQ_UNLINK)) {
608 			error = (error == EWOULDBLOCK) ? ETIMEDOUT : EINTR;
609 			goto error;
610 		}
611 	}
612 
613 
614 	/*
615 	 * Find the highest priority message, and remove it from the queue.
616 	 * At first, reserved queue is checked, bitmap is next.
617 	 */
618 	msg = TAILQ_FIRST(&mq->mq_head[MQ_PQRESQ]);
619 	if (__predict_true(msg == NULL)) {
620 		idx = ffs(mq->mq_bitmap);
621 		msg = TAILQ_FIRST(&mq->mq_head[idx]);
622 		KKASSERT(msg != NULL);
623 	} else {
624 		idx = MQ_PQRESQ;
625 	}
626 	TAILQ_REMOVE(&mq->mq_head[idx], msg, msg_queue);
627 
628 	/* Unmark the bit, if last message. */
629 	if (__predict_true(idx) && TAILQ_EMPTY(&mq->mq_head[idx])) {
630 		KKASSERT((MQ_PQSIZE - idx) == msg->msg_prio);
631 		mq->mq_bitmap &= ~(1 << --idx);
632 	}
633 
634 	/* Decrement the counter and signal waiter, if any */
635 	mqattr->mq_curmsgs--;
636 	wakeup_one(&mq->mq_recv_cv);
637 
638 	/* Ready for sending now */
639 	get_mplock();
640 	selwakeup(&mq->mq_wsel);
641 	rel_mplock();
642 error:
643 	lockmgr(&mq->mq_mtx, LK_RELEASE);
644 	fdrop(fp);
645 	if (error)
646 		return error;
647 
648 	/*
649 	 * Copy the data to the user-space.
650 	 * Note: According to POSIX, no message should be removed from the
651 	 * queue in case of fail - this would be violated.
652 	 */
653 	*mlen = msg->msg_len;
654 	error = copyout(msg->msg_ptr, msg_ptr, msg->msg_len);
655 	if (error == 0 && msg_prio)
656 		error = copyout(&msg->msg_prio, msg_prio, sizeof(unsigned));
657 	mqueue_freemsg(msg, sizeof(struct mq_msg) + msg->msg_len);
658 
659 	return error;
660 }
661 
662 int
663 sys_mq_receive(struct mq_receive_args *uap)
664 {
665 	/* {
666 		syscallarg(mqd_t) mqdes;
667 		syscallarg(char *) msg_ptr;
668 		syscallarg(size_t) msg_len;
669 		syscallarg(unsigned *) msg_prio;
670 	} */
671 	ssize_t mlen;
672 	int error;
673 
674 	error = mq_receive1(curthread->td_lwp, SCARG(uap, mqdes), SCARG(uap, msg_ptr),
675 	    SCARG(uap, msg_len), SCARG(uap, msg_prio), 0, &mlen);
676 	if (error == 0)
677 		uap->sysmsg_result = mlen;
678 
679 	return error;
680 }
681 
682 int
683 sys_mq_timedreceive(struct mq_timedreceive_args *uap)
684 {
685 	/* {
686 		syscallarg(mqd_t) mqdes;
687 		syscallarg(char *) msg_ptr;
688 		syscallarg(size_t) msg_len;
689 		syscallarg(unsigned *) msg_prio;
690 		syscallarg(const struct timespec *) abs_timeout;
691 	} */
692 	int error;
693 	ssize_t mlen;
694 	struct timespec ts, *tsp;
695 
696 	/* Get and convert time value */
697 	if (SCARG(uap, abs_timeout)) {
698 		error = copyin(SCARG(uap, abs_timeout), &ts, sizeof(ts));
699 		if (error)
700 			return error;
701 		tsp = &ts;
702 	} else {
703 		tsp = NULL;
704 	}
705 
706 	error = mq_receive1(curthread->td_lwp, SCARG(uap, mqdes), SCARG(uap, msg_ptr),
707 	    SCARG(uap, msg_len), SCARG(uap, msg_prio), tsp, &mlen);
708 	if (error == 0)
709 		uap->sysmsg_result = mlen;
710 
711 	return error;
712 }
713 
714 /*
715  * Primary mq_send1() function.
716  */
717 int
718 mq_send1(struct lwp *l, mqd_t mqdes, const char *msg_ptr, size_t msg_len,
719     unsigned msg_prio, struct timespec *ts)
720 {
721 	file_t *fp = NULL;
722 	struct mqueue *mq;
723 	struct mq_msg *msg;
724 	struct mq_attr *mqattr;
725 	struct proc *notify = NULL;
726 	/*ksiginfo_t ksi;*/
727 	size_t size;
728 	int error;
729 
730 	/* Check the priority range */
731 	if (msg_prio >= mq_prio_max)
732 		return EINVAL;
733 
734 	/* Allocate a new message */
735 	size = sizeof(struct mq_msg) + msg_len;
736 	if (size > mq_max_msgsize)
737 		return EMSGSIZE;
738 
739 	if (size > MQ_DEF_MSGSIZE) {
740 		msg = kmalloc(size, M_MQBUF, M_WAITOK);
741 	} else {
742 		msg = objcache_get(mqmsg_cache, M_WAITOK);
743 	}
744 
745 	/* Get the data from user-space */
746 	error = copyin(msg_ptr, msg->msg_ptr, msg_len);
747 	if (error) {
748 		mqueue_freemsg(msg, size);
749 		return error;
750 	}
751 	msg->msg_len = msg_len;
752 	msg->msg_prio = msg_prio;
753 
754 	/* Get the mqueue */
755 	error = mqueue_get(l, mqdes, &fp);
756 	if (error) {
757 		mqueue_freemsg(msg, size);
758 		return error;
759 	}
760 	mq = fp->f_data;
761 	if ((fp->f_flag & FWRITE) == 0) {
762 		error = EBADF;
763 		goto error;
764 	}
765 	getnanotime(&mq->mq_mtime);
766 	mqattr = &mq->mq_attrib;
767 
768 	/* Check the message size limit */
769 	if (msg_len <= 0 || msg_len > mqattr->mq_msgsize) {
770 		error = EMSGSIZE;
771 		goto error;
772 	}
773 
774 	/* Check if queue is full */
775 	while (mqattr->mq_curmsgs >= mqattr->mq_maxmsg) {
776 		int t;
777 
778 		if (mqattr->mq_flags & O_NONBLOCK) {
779 			error = EAGAIN;
780 			goto error;
781 		}
782 		if (ts) {
783 			error = abstimeout2timo(ts, &t);
784 			if (error)
785 				goto error;
786 		} else
787 			t = 0;
788 		/* Block until queue becomes available */
789 		error = lksleep(&mq->mq_recv_cv, &mq->mq_mtx, PCATCH, "mqrecv", t);
790 		if (error || (mqattr->mq_flags & MQ_UNLINK)) {
791 			error = (error == EWOULDBLOCK) ? ETIMEDOUT : error;
792 			goto error;
793 		}
794 	}
795 	KKASSERT(mq->mq_attrib.mq_curmsgs < mq->mq_attrib.mq_maxmsg);
796 
797 	/*
798 	 * Insert message into the queue, according to the priority.
799 	 * Note the difference between index and priority.
800 	 */
801 	if (__predict_true(msg_prio < MQ_PQSIZE)) {
802 		u_int idx = MQ_PQSIZE - msg_prio;
803 
804 		KKASSERT(idx != MQ_PQRESQ);
805 		TAILQ_INSERT_TAIL(&mq->mq_head[idx], msg, msg_queue);
806 		mq->mq_bitmap |= (1 << --idx);
807 	} else {
808 		mqueue_linear_insert(mq, msg);
809 	}
810 
811 	/* Check for the notify */
812 	if (mqattr->mq_curmsgs == 0 && mq->mq_notify_proc &&
813 	    (mqattr->mq_flags & MQ_RECEIVE) == 0 &&
814 	    mq->mq_sig_notify.sigev_notify == SIGEV_SIGNAL) {
815 		/* Initialize the signal */
816 		/*KSI_INIT(&ksi);*/
817 		/*ksi.ksi_signo = mq->mq_sig_notify.sigev_signo;*/
818 		/*ksi.ksi_code = SI_MESGQ;*/
819 		/*ksi.ksi_value = mq->mq_sig_notify.sigev_value;*/
820 		/* Unregister the process */
821 		notify = mq->mq_notify_proc;
822 		mq->mq_notify_proc = NULL;
823 	}
824 
825 	/* Increment the counter and signal waiter, if any */
826 	mqattr->mq_curmsgs++;
827 	wakeup_one(&mq->mq_send_cv);
828 
829 	/* Ready for receiving now */
830 	get_mplock();
831 	selwakeup(&mq->mq_rsel);
832 	rel_mplock();
833 error:
834 	lockmgr(&mq->mq_mtx, LK_RELEASE);
835 	fdrop(fp);
836 
837 	if (error) {
838 		mqueue_freemsg(msg, size);
839 	} else if (notify) {
840 		/* Send the notify, if needed */
841 		lwkt_gettoken(&proc_token);
842 		/*kpsignal(notify, &ksi, NULL);*/
843 		ksignal(notify, mq->mq_sig_notify.sigev_signo);
844 		lwkt_reltoken(&proc_token);
845 	}
846 
847 	return error;
848 }
849 
850 int
851 sys_mq_send(struct mq_send_args *uap)
852 {
853 	/* {
854 		syscallarg(mqd_t) mqdes;
855 		syscallarg(const char *) msg_ptr;
856 		syscallarg(size_t) msg_len;
857 		syscallarg(unsigned) msg_prio;
858 	} */
859 
860 	return mq_send1(curthread->td_lwp, SCARG(uap, mqdes), SCARG(uap, msg_ptr),
861 	    SCARG(uap, msg_len), SCARG(uap, msg_prio), 0);
862 }
863 
864 int
865 sys_mq_timedsend(struct mq_timedsend_args *uap)
866 {
867 	/* {
868 		syscallarg(mqd_t) mqdes;
869 		syscallarg(const char *) msg_ptr;
870 		syscallarg(size_t) msg_len;
871 		syscallarg(unsigned) msg_prio;
872 		syscallarg(const struct timespec *) abs_timeout;
873 	} */
874 	struct timespec ts, *tsp;
875 	int error;
876 
877 	/* Get and convert time value */
878 	if (SCARG(uap, abs_timeout)) {
879 		error = copyin(SCARG(uap, abs_timeout), &ts, sizeof(ts));
880 		if (error)
881 			return error;
882 		tsp = &ts;
883 	} else {
884 		tsp = NULL;
885 	}
886 
887 	return mq_send1(curthread->td_lwp, SCARG(uap, mqdes), SCARG(uap, msg_ptr),
888 	    SCARG(uap, msg_len), SCARG(uap, msg_prio), tsp);
889 }
890 
891 int
892 sys_mq_notify(struct mq_notify_args *uap)
893 {
894 	/* {
895 		syscallarg(mqd_t) mqdes;
896 		syscallarg(const struct sigevent *) notification;
897 	} */
898 	file_t *fp = NULL;
899 	struct mqueue *mq;
900 	struct sigevent sig;
901 	int error;
902 
903 	if (SCARG(uap, notification)) {
904 		/* Get the signal from user-space */
905 		error = copyin(SCARG(uap, notification), &sig,
906 		    sizeof(struct sigevent));
907 		if (error)
908 			return error;
909 		if (sig.sigev_notify == SIGEV_SIGNAL &&
910 		    (sig.sigev_signo <= 0 || sig.sigev_signo >= NSIG))
911 			return EINVAL;
912 	}
913 
914 	error = mqueue_get(curthread->td_lwp, SCARG(uap, mqdes), &fp);
915 	if (error)
916 		return error;
917 	mq = fp->f_data;
918 
919 	if (SCARG(uap, notification)) {
920 		/* Register notification: set the signal and target process */
921 		if (mq->mq_notify_proc == NULL) {
922 			memcpy(&mq->mq_sig_notify, &sig,
923 			    sizeof(struct sigevent));
924 			mq->mq_notify_proc = curproc;
925 		} else {
926 			/* Fail if someone else already registered */
927 			error = EBUSY;
928 		}
929 	} else {
930 		/* Unregister the notification */
931 		mq->mq_notify_proc = NULL;
932 	}
933 	lockmgr(&mq->mq_mtx, LK_RELEASE);
934 	fdrop(fp);
935 
936 	return error;
937 }
938 
939 int
940 sys_mq_getattr(struct mq_getattr_args *uap)
941 {
942 	/* {
943 		syscallarg(mqd_t) mqdes;
944 		syscallarg(struct mq_attr *) mqstat;
945 	} */
946 	file_t *fp = NULL;
947 	struct mqueue *mq;
948 	struct mq_attr attr;
949 	int error;
950 
951 	/* Get the message queue */
952 	error = mqueue_get(curthread->td_lwp, SCARG(uap, mqdes), &fp);
953 	if (error)
954 		return error;
955 	mq = fp->f_data;
956 	memcpy(&attr, &mq->mq_attrib, sizeof(struct mq_attr));
957 	lockmgr(&mq->mq_mtx, LK_RELEASE);
958 	fdrop(fp);
959 
960 	return copyout(&attr, SCARG(uap, mqstat), sizeof(struct mq_attr));
961 }
962 
963 int
964 sys_mq_setattr(struct mq_setattr_args *uap)
965 {
966 	/* {
967 		syscallarg(mqd_t) mqdes;
968 		syscallarg(const struct mq_attr *) mqstat;
969 		syscallarg(struct mq_attr *) omqstat;
970 	} */
971 	file_t *fp = NULL;
972 	struct mqueue *mq;
973 	struct mq_attr attr;
974 	int error, nonblock;
975 
976 	error = copyin(SCARG(uap, mqstat), &attr, sizeof(struct mq_attr));
977 	if (error)
978 		return error;
979 	nonblock = (attr.mq_flags & O_NONBLOCK);
980 
981 	/* Get the message queue */
982 	error = mqueue_get(curthread->td_lwp, SCARG(uap, mqdes), &fp);
983 	if (error)
984 		return error;
985 	mq = fp->f_data;
986 
987 	/* Copy the old attributes, if needed */
988 	if (SCARG(uap, omqstat)) {
989 		memcpy(&attr, &mq->mq_attrib, sizeof(struct mq_attr));
990 	}
991 
992 	/* Ignore everything, except O_NONBLOCK */
993 	if (nonblock)
994 		mq->mq_attrib.mq_flags |= O_NONBLOCK;
995 	else
996 		mq->mq_attrib.mq_flags &= ~O_NONBLOCK;
997 
998 	lockmgr(&mq->mq_mtx, LK_RELEASE);
999 	fdrop(fp);
1000 
1001 	/*
1002 	 * Copy the data to the user-space.
1003 	 * Note: According to POSIX, the new attributes should not be set in
1004 	 * case of fail - this would be violated.
1005 	 */
1006 	if (SCARG(uap, omqstat))
1007 		error = copyout(&attr, SCARG(uap, omqstat),
1008 		    sizeof(struct mq_attr));
1009 
1010 	return error;
1011 }
1012 
1013 int
1014 sys_mq_unlink(struct mq_unlink_args *uap)
1015 {
1016 	/* {
1017 		syscallarg(const char *) name;
1018 	} */
1019 	struct thread *td = curthread;
1020 	struct mqueue *mq;
1021 	char *name;
1022 	int error, refcnt = 0;
1023 
1024 	/* Get the name from the user-space */
1025 	name = kmalloc(MQ_NAMELEN, M_MQBUF, M_WAITOK | M_ZERO);
1026 	error = copyinstr(SCARG(uap, name), name, MQ_NAMELEN - 1, NULL);
1027 	if (error) {
1028 		kfree(name, M_MQBUF);
1029 		return error;
1030 	}
1031 
1032 	/* Lookup for this file */
1033 	lockmgr(&mqlist_mtx, LK_EXCLUSIVE);
1034 	mq = mqueue_lookup(name);
1035 	if (mq == NULL) {
1036 		error = ENOENT;
1037 		goto error;
1038 	}
1039 
1040 	/* Check the permissions */
1041 	if (td->td_ucred->cr_uid != mq->mq_euid &&
1042 	    priv_check(td, PRIV_ROOT) != 0) {
1043 		lockmgr(&mq->mq_mtx, LK_RELEASE);
1044 		error = EACCES;
1045 		goto error;
1046 	}
1047 
1048 	/* Mark message queue as unlinking, before leaving the window */
1049 	mq->mq_attrib.mq_flags |= MQ_UNLINK;
1050 
1051 	/* Wake up all waiters, if there are such */
1052 	wakeup(&mq->mq_send_cv);
1053 	wakeup(&mq->mq_recv_cv);
1054 
1055 	get_mplock();
1056 	selwakeup(&mq->mq_rsel);
1057 	selwakeup(&mq->mq_wsel);
1058 	rel_mplock();
1059 
1060 	refcnt = mq->mq_refcnt;
1061 	if (refcnt == 0)
1062 		LIST_REMOVE(mq, mq_list);
1063 
1064 	lockmgr(&mq->mq_mtx, LK_RELEASE);
1065 error:
1066 	lockmgr(&mqlist_mtx, LK_RELEASE);
1067 
1068 	/*
1069 	 * If there are no references - destroy the message
1070 	 * queue, otherwise, the last mq_close() will do that.
1071 	 */
1072 	if (error == 0 && refcnt == 0)
1073 		mqueue_destroy(mq);
1074 
1075 	kfree(name, M_MQBUF);
1076 	return error;
1077 }
1078 
1079 /*
1080  * SysCtl.
1081  */
1082 SYSCTL_NODE(_kern, OID_AUTO, mqueue,
1083     CTLFLAG_RW, 0, "Message queue options");
1084 
1085 SYSCTL_INT(_kern_mqueue, OID_AUTO, mq_open_max,
1086     CTLFLAG_RW, &mq_open_max, 0,
1087     "Maximal number of message queue descriptors per process");
1088 
1089 SYSCTL_INT(_kern_mqueue, OID_AUTO, mq_prio_max,
1090     CTLFLAG_RW, &mq_prio_max, 0,
1091     "Maximal priority of the message");
1092 
1093 SYSCTL_INT(_kern_mqueue, OID_AUTO, mq_max_msgsize,
1094     CTLFLAG_RW, &mq_max_msgsize, 0,
1095     "Maximal allowed size of the message");
1096 
1097 SYSCTL_INT(_kern_mqueue, OID_AUTO, mq_def_maxmsg,
1098     CTLFLAG_RW, &mq_def_maxmsg, 0,
1099     "Default maximal message count");
1100 
1101 SYSCTL_INT(_kern_mqueue, OID_AUTO, mq_max_maxmsg,
1102     CTLFLAG_RW, &mq_max_maxmsg, 0,
1103     "Maximal allowed message count");
1104 
1105 SYSINIT(sys_mqueue_init, SI_SUB_PRE_DRIVERS, SI_ORDER_ANY, mqueue_sysinit, NULL);
1106