[PATCH] IPC namespace - msg
[linux-2.6.git] / ipc / msg.c
1 /*
2  * linux/ipc/msg.c
3  * Copyright (C) 1992 Krishna Balasubramanian
4  *
5  * Removed all the remaining kerneld mess
6  * Catch the -EFAULT stuff properly
7  * Use GFP_KERNEL for messages as in 1.2
8  * Fixed up the unchecked user space derefs
9  * Copyright (C) 1998 Alan Cox & Andi Kleen
10  *
11  * /proc/sysvipc/msg support (c) 1999 Dragos Acostachioaie <dragos@iname.com>
12  *
13  * mostly rewritten, threaded and wake-one semantics added
14  * MSGMAX limit removed, sysctl's added
15  * (c) 1999 Manfred Spraul <manfred@colorfullife.com>
16  *
17  * support for audit of ipc object properties and permission changes
18  * Dustin Kirkland <dustin.kirkland@us.ibm.com>
19  *
20  * namespaces support
21  * OpenVZ, SWsoft Inc.
22  * Pavel Emelianov <xemul@openvz.org>
23  */
24
25 #include <linux/capability.h>
26 #include <linux/slab.h>
27 #include <linux/msg.h>
28 #include <linux/spinlock.h>
29 #include <linux/init.h>
30 #include <linux/proc_fs.h>
31 #include <linux/list.h>
32 #include <linux/security.h>
33 #include <linux/sched.h>
34 #include <linux/syscalls.h>
35 #include <linux/audit.h>
36 #include <linux/seq_file.h>
37 #include <linux/mutex.h>
38 #include <linux/nsproxy.h>
39
40 #include <asm/current.h>
41 #include <asm/uaccess.h>
42 #include "util.h"
43
44 /*
45  * one msg_receiver structure for each sleeping receiver:
46  */
47 struct msg_receiver {
48         struct list_head        r_list;
49         struct task_struct      *r_tsk;
50
51         int                     r_mode;
52         long                    r_msgtype;
53         long                    r_maxsize;
54
55         volatile struct msg_msg *r_msg;
56 };
57
58 /* one msg_sender for each sleeping sender */
59 struct msg_sender {
60         struct list_head        list;
61         struct task_struct      *tsk;
62 };
63
64 #define SEARCH_ANY              1
65 #define SEARCH_EQUAL            2
66 #define SEARCH_NOTEQUAL         3
67 #define SEARCH_LESSEQUAL        4
68
69 static atomic_t msg_bytes =     ATOMIC_INIT(0);
70 static atomic_t msg_hdrs =      ATOMIC_INIT(0);
71
72 static struct ipc_ids init_msg_ids;
73
74 #define msg_ids(ns)     (*((ns)->ids[IPC_MSG_IDS]))
75
76 #define msg_lock(ns, id)        ((struct msg_queue*)ipc_lock(&msg_ids(ns), id))
77 #define msg_unlock(msq)         ipc_unlock(&(msq)->q_perm)
78 #define msg_rmid(ns, id)        ((struct msg_queue*)ipc_rmid(&msg_ids(ns), id))
79 #define msg_checkid(ns, msq, msgid)     \
80         ipc_checkid(&msg_ids(ns), &msq->q_perm, msgid)
81 #define msg_buildid(ns, id, seq) \
82         ipc_buildid(&msg_ids(ns), id, seq)
83
84 static void freeque (struct ipc_namespace *ns, struct msg_queue *msq, int id);
85 static int newque (struct ipc_namespace *ns, key_t key, int msgflg);
86 #ifdef CONFIG_PROC_FS
87 static int sysvipc_msg_proc_show(struct seq_file *s, void *it);
88 #endif
89
90 static void __ipc_init __msg_init_ns(struct ipc_namespace *ns, struct ipc_ids *ids)
91 {
92         ns->ids[IPC_MSG_IDS] = ids;
93         ns->msg_ctlmax = MSGMAX;
94         ns->msg_ctlmnb = MSGMNB;
95         ns->msg_ctlmni = MSGMNI;
96         ipc_init_ids(ids, ns->msg_ctlmni);
97 }
98
99 #ifdef CONFIG_IPC_NS
100 int msg_init_ns(struct ipc_namespace *ns)
101 {
102         struct ipc_ids *ids;
103
104         ids = kmalloc(sizeof(struct ipc_ids), GFP_KERNEL);
105         if (ids == NULL)
106                 return -ENOMEM;
107
108         __msg_init_ns(ns, ids);
109         return 0;
110 }
111
112 void msg_exit_ns(struct ipc_namespace *ns)
113 {
114         int i;
115         struct msg_queue *msq;
116
117         mutex_lock(&msg_ids(ns).mutex);
118         for (i = 0; i <= msg_ids(ns).max_id; i++) {
119                 msq = msg_lock(ns, i);
120                 if (msq == NULL)
121                         continue;
122
123                 freeque(ns, msq, i);
124         }
125         mutex_unlock(&msg_ids(ns).mutex);
126
127         kfree(ns->ids[IPC_MSG_IDS]);
128         ns->ids[IPC_MSG_IDS] = NULL;
129 }
130 #endif
131
132 void __init msg_init(void)
133 {
134         __msg_init_ns(&init_ipc_ns, &init_msg_ids);
135         ipc_init_proc_interface("sysvipc/msg",
136                                 "       key      msqid perms      cbytes       qnum lspid lrpid   uid   gid  cuid  cgid      stime      rtime      ctime\n",
137                                 IPC_MSG_IDS, sysvipc_msg_proc_show);
138 }
139
140 static int newque (struct ipc_namespace *ns, key_t key, int msgflg)
141 {
142         struct msg_queue *msq;
143         int id, retval;
144
145         msq = ipc_rcu_alloc(sizeof(*msq));
146         if (!msq)
147                 return -ENOMEM;
148
149         msq->q_perm.mode = msgflg & S_IRWXUGO;
150         msq->q_perm.key = key;
151
152         msq->q_perm.security = NULL;
153         retval = security_msg_queue_alloc(msq);
154         if (retval) {
155                 ipc_rcu_putref(msq);
156                 return retval;
157         }
158
159         id = ipc_addid(&msg_ids(ns), &msq->q_perm, ns->msg_ctlmni);
160         if (id == -1) {
161                 security_msg_queue_free(msq);
162                 ipc_rcu_putref(msq);
163                 return -ENOSPC;
164         }
165
166         msq->q_id = msg_buildid(ns, id, msq->q_perm.seq);
167         msq->q_stime = msq->q_rtime = 0;
168         msq->q_ctime = get_seconds();
169         msq->q_cbytes = msq->q_qnum = 0;
170         msq->q_qbytes = ns->msg_ctlmnb;
171         msq->q_lspid = msq->q_lrpid = 0;
172         INIT_LIST_HEAD(&msq->q_messages);
173         INIT_LIST_HEAD(&msq->q_receivers);
174         INIT_LIST_HEAD(&msq->q_senders);
175         msg_unlock(msq);
176
177         return msq->q_id;
178 }
179
180 static inline void ss_add(struct msg_queue *msq, struct msg_sender *mss)
181 {
182         mss->tsk = current;
183         current->state = TASK_INTERRUPTIBLE;
184         list_add_tail(&mss->list, &msq->q_senders);
185 }
186
187 static inline void ss_del(struct msg_sender *mss)
188 {
189         if (mss->list.next != NULL)
190                 list_del(&mss->list);
191 }
192
193 static void ss_wakeup(struct list_head *h, int kill)
194 {
195         struct list_head *tmp;
196
197         tmp = h->next;
198         while (tmp != h) {
199                 struct msg_sender *mss;
200
201                 mss = list_entry(tmp, struct msg_sender, list);
202                 tmp = tmp->next;
203                 if (kill)
204                         mss->list.next = NULL;
205                 wake_up_process(mss->tsk);
206         }
207 }
208
209 static void expunge_all(struct msg_queue *msq, int res)
210 {
211         struct list_head *tmp;
212
213         tmp = msq->q_receivers.next;
214         while (tmp != &msq->q_receivers) {
215                 struct msg_receiver *msr;
216
217                 msr = list_entry(tmp, struct msg_receiver, r_list);
218                 tmp = tmp->next;
219                 msr->r_msg = NULL;
220                 wake_up_process(msr->r_tsk);
221                 smp_mb();
222                 msr->r_msg = ERR_PTR(res);
223         }
224 }
225
226 /*
227  * freeque() wakes up waiters on the sender and receiver waiting queue,
228  * removes the message queue from message queue ID
229  * array, and cleans up all the messages associated with this queue.
230  *
231  * msg_ids.mutex and the spinlock for this message queue is hold
232  * before freeque() is called. msg_ids.mutex remains locked on exit.
233  */
234 static void freeque(struct ipc_namespace *ns, struct msg_queue *msq, int id)
235 {
236         struct list_head *tmp;
237
238         expunge_all(msq, -EIDRM);
239         ss_wakeup(&msq->q_senders, 1);
240         msq = msg_rmid(ns, id);
241         msg_unlock(msq);
242
243         tmp = msq->q_messages.next;
244         while (tmp != &msq->q_messages) {
245                 struct msg_msg *msg = list_entry(tmp, struct msg_msg, m_list);
246
247                 tmp = tmp->next;
248                 atomic_dec(&msg_hdrs);
249                 free_msg(msg);
250         }
251         atomic_sub(msq->q_cbytes, &msg_bytes);
252         security_msg_queue_free(msq);
253         ipc_rcu_putref(msq);
254 }
255
256 asmlinkage long sys_msgget(key_t key, int msgflg)
257 {
258         struct msg_queue *msq;
259         int id, ret = -EPERM;
260         struct ipc_namespace *ns;
261
262         ns = current->nsproxy->ipc_ns;
263         
264         mutex_lock(&msg_ids(ns).mutex);
265         if (key == IPC_PRIVATE) 
266                 ret = newque(ns, key, msgflg);
267         else if ((id = ipc_findkey(&msg_ids(ns), key)) == -1) { /* key not used */
268                 if (!(msgflg & IPC_CREAT))
269                         ret = -ENOENT;
270                 else
271                         ret = newque(ns, key, msgflg);
272         } else if (msgflg & IPC_CREAT && msgflg & IPC_EXCL) {
273                 ret = -EEXIST;
274         } else {
275                 msq = msg_lock(ns, id);
276                 BUG_ON(msq == NULL);
277                 if (ipcperms(&msq->q_perm, msgflg))
278                         ret = -EACCES;
279                 else {
280                         int qid = msg_buildid(ns, id, msq->q_perm.seq);
281
282                         ret = security_msg_queue_associate(msq, msgflg);
283                         if (!ret)
284                                 ret = qid;
285                 }
286                 msg_unlock(msq);
287         }
288         mutex_unlock(&msg_ids(ns).mutex);
289
290         return ret;
291 }
292
293 static inline unsigned long
294 copy_msqid_to_user(void __user *buf, struct msqid64_ds *in, int version)
295 {
296         switch(version) {
297         case IPC_64:
298                 return copy_to_user(buf, in, sizeof(*in));
299         case IPC_OLD:
300         {
301                 struct msqid_ds out;
302
303                 memset(&out, 0, sizeof(out));
304
305                 ipc64_perm_to_ipc_perm(&in->msg_perm, &out.msg_perm);
306
307                 out.msg_stime           = in->msg_stime;
308                 out.msg_rtime           = in->msg_rtime;
309                 out.msg_ctime           = in->msg_ctime;
310
311                 if (in->msg_cbytes > USHRT_MAX)
312                         out.msg_cbytes  = USHRT_MAX;
313                 else
314                         out.msg_cbytes  = in->msg_cbytes;
315                 out.msg_lcbytes         = in->msg_cbytes;
316
317                 if (in->msg_qnum > USHRT_MAX)
318                         out.msg_qnum    = USHRT_MAX;
319                 else
320                         out.msg_qnum    = in->msg_qnum;
321
322                 if (in->msg_qbytes > USHRT_MAX)
323                         out.msg_qbytes  = USHRT_MAX;
324                 else
325                         out.msg_qbytes  = in->msg_qbytes;
326                 out.msg_lqbytes         = in->msg_qbytes;
327
328                 out.msg_lspid           = in->msg_lspid;
329                 out.msg_lrpid           = in->msg_lrpid;
330
331                 return copy_to_user(buf, &out, sizeof(out));
332         }
333         default:
334                 return -EINVAL;
335         }
336 }
337
338 struct msq_setbuf {
339         unsigned long   qbytes;
340         uid_t           uid;
341         gid_t           gid;
342         mode_t          mode;
343 };
344
345 static inline unsigned long
346 copy_msqid_from_user(struct msq_setbuf *out, void __user *buf, int version)
347 {
348         switch(version) {
349         case IPC_64:
350         {
351                 struct msqid64_ds tbuf;
352
353                 if (copy_from_user(&tbuf, buf, sizeof(tbuf)))
354                         return -EFAULT;
355
356                 out->qbytes             = tbuf.msg_qbytes;
357                 out->uid                = tbuf.msg_perm.uid;
358                 out->gid                = tbuf.msg_perm.gid;
359                 out->mode               = tbuf.msg_perm.mode;
360
361                 return 0;
362         }
363         case IPC_OLD:
364         {
365                 struct msqid_ds tbuf_old;
366
367                 if (copy_from_user(&tbuf_old, buf, sizeof(tbuf_old)))
368                         return -EFAULT;
369
370                 out->uid                = tbuf_old.msg_perm.uid;
371                 out->gid                = tbuf_old.msg_perm.gid;
372                 out->mode               = tbuf_old.msg_perm.mode;
373
374                 if (tbuf_old.msg_qbytes == 0)
375                         out->qbytes     = tbuf_old.msg_lqbytes;
376                 else
377                         out->qbytes     = tbuf_old.msg_qbytes;
378
379                 return 0;
380         }
381         default:
382                 return -EINVAL;
383         }
384 }
385
386 asmlinkage long sys_msgctl(int msqid, int cmd, struct msqid_ds __user *buf)
387 {
388         struct kern_ipc_perm *ipcp;
389         struct msq_setbuf setbuf;
390         struct msg_queue *msq;
391         int err, version;
392         struct ipc_namespace *ns;
393
394         if (msqid < 0 || cmd < 0)
395                 return -EINVAL;
396
397         version = ipc_parse_version(&cmd);
398         ns = current->nsproxy->ipc_ns;
399
400         switch (cmd) {
401         case IPC_INFO:
402         case MSG_INFO:
403         {
404                 struct msginfo msginfo;
405                 int max_id;
406
407                 if (!buf)
408                         return -EFAULT;
409                 /*
410                  * We must not return kernel stack data.
411                  * due to padding, it's not enough
412                  * to set all member fields.
413                  */
414                 err = security_msg_queue_msgctl(NULL, cmd);
415                 if (err)
416                         return err;
417
418                 memset(&msginfo, 0, sizeof(msginfo));
419                 msginfo.msgmni = ns->msg_ctlmni;
420                 msginfo.msgmax = ns->msg_ctlmax;
421                 msginfo.msgmnb = ns->msg_ctlmnb;
422                 msginfo.msgssz = MSGSSZ;
423                 msginfo.msgseg = MSGSEG;
424                 mutex_lock(&msg_ids(ns).mutex);
425                 if (cmd == MSG_INFO) {
426                         msginfo.msgpool = msg_ids(ns).in_use;
427                         msginfo.msgmap = atomic_read(&msg_hdrs);
428                         msginfo.msgtql = atomic_read(&msg_bytes);
429                 } else {
430                         msginfo.msgmap = MSGMAP;
431                         msginfo.msgpool = MSGPOOL;
432                         msginfo.msgtql = MSGTQL;
433                 }
434                 max_id = msg_ids(ns).max_id;
435                 mutex_unlock(&msg_ids(ns).mutex);
436                 if (copy_to_user(buf, &msginfo, sizeof(struct msginfo)))
437                         return -EFAULT;
438                 return (max_id < 0) ? 0 : max_id;
439         }
440         case MSG_STAT:
441         case IPC_STAT:
442         {
443                 struct msqid64_ds tbuf;
444                 int success_return;
445
446                 if (!buf)
447                         return -EFAULT;
448                 if (cmd == MSG_STAT && msqid >= msg_ids(ns).entries->size)
449                         return -EINVAL;
450
451                 memset(&tbuf, 0, sizeof(tbuf));
452
453                 msq = msg_lock(ns, msqid);
454                 if (msq == NULL)
455                         return -EINVAL;
456
457                 if (cmd == MSG_STAT) {
458                         success_return = msg_buildid(ns, msqid, msq->q_perm.seq);
459                 } else {
460                         err = -EIDRM;
461                         if (msg_checkid(ns, msq, msqid))
462                                 goto out_unlock;
463                         success_return = 0;
464                 }
465                 err = -EACCES;
466                 if (ipcperms(&msq->q_perm, S_IRUGO))
467                         goto out_unlock;
468
469                 err = security_msg_queue_msgctl(msq, cmd);
470                 if (err)
471                         goto out_unlock;
472
473                 kernel_to_ipc64_perm(&msq->q_perm, &tbuf.msg_perm);
474                 tbuf.msg_stime  = msq->q_stime;
475                 tbuf.msg_rtime  = msq->q_rtime;
476                 tbuf.msg_ctime  = msq->q_ctime;
477                 tbuf.msg_cbytes = msq->q_cbytes;
478                 tbuf.msg_qnum   = msq->q_qnum;
479                 tbuf.msg_qbytes = msq->q_qbytes;
480                 tbuf.msg_lspid  = msq->q_lspid;
481                 tbuf.msg_lrpid  = msq->q_lrpid;
482                 msg_unlock(msq);
483                 if (copy_msqid_to_user(buf, &tbuf, version))
484                         return -EFAULT;
485                 return success_return;
486         }
487         case IPC_SET:
488                 if (!buf)
489                         return -EFAULT;
490                 if (copy_msqid_from_user(&setbuf, buf, version))
491                         return -EFAULT;
492                 break;
493         case IPC_RMID:
494                 break;
495         default:
496                 return  -EINVAL;
497         }
498
499         mutex_lock(&msg_ids(ns).mutex);
500         msq = msg_lock(ns, msqid);
501         err = -EINVAL;
502         if (msq == NULL)
503                 goto out_up;
504
505         err = -EIDRM;
506         if (msg_checkid(ns, msq, msqid))
507                 goto out_unlock_up;
508         ipcp = &msq->q_perm;
509
510         err = audit_ipc_obj(ipcp);
511         if (err)
512                 goto out_unlock_up;
513         if (cmd==IPC_SET) {
514                 err = audit_ipc_set_perm(setbuf.qbytes, setbuf.uid, setbuf.gid,
515                                          setbuf.mode);
516                 if (err)
517                         goto out_unlock_up;
518         }
519
520         err = -EPERM;
521         if (current->euid != ipcp->cuid &&
522             current->euid != ipcp->uid && !capable(CAP_SYS_ADMIN))
523                 /* We _could_ check for CAP_CHOWN above, but we don't */
524                 goto out_unlock_up;
525
526         err = security_msg_queue_msgctl(msq, cmd);
527         if (err)
528                 goto out_unlock_up;
529
530         switch (cmd) {
531         case IPC_SET:
532         {
533                 err = -EPERM;
534                 if (setbuf.qbytes > ns->msg_ctlmnb && !capable(CAP_SYS_RESOURCE))
535                         goto out_unlock_up;
536
537                 msq->q_qbytes = setbuf.qbytes;
538
539                 ipcp->uid = setbuf.uid;
540                 ipcp->gid = setbuf.gid;
541                 ipcp->mode = (ipcp->mode & ~S_IRWXUGO) |
542                              (S_IRWXUGO & setbuf.mode);
543                 msq->q_ctime = get_seconds();
544                 /* sleeping receivers might be excluded by
545                  * stricter permissions.
546                  */
547                 expunge_all(msq, -EAGAIN);
548                 /* sleeping senders might be able to send
549                  * due to a larger queue size.
550                  */
551                 ss_wakeup(&msq->q_senders, 0);
552                 msg_unlock(msq);
553                 break;
554         }
555         case IPC_RMID:
556                 freeque(ns, msq, msqid);
557                 break;
558         }
559         err = 0;
560 out_up:
561         mutex_unlock(&msg_ids(ns).mutex);
562         return err;
563 out_unlock_up:
564         msg_unlock(msq);
565         goto out_up;
566 out_unlock:
567         msg_unlock(msq);
568         return err;
569 }
570
571 static int testmsg(struct msg_msg *msg, long type, int mode)
572 {
573         switch(mode)
574         {
575                 case SEARCH_ANY:
576                         return 1;
577                 case SEARCH_LESSEQUAL:
578                         if (msg->m_type <=type)
579                                 return 1;
580                         break;
581                 case SEARCH_EQUAL:
582                         if (msg->m_type == type)
583                                 return 1;
584                         break;
585                 case SEARCH_NOTEQUAL:
586                         if (msg->m_type != type)
587                                 return 1;
588                         break;
589         }
590         return 0;
591 }
592
593 static inline int pipelined_send(struct msg_queue *msq, struct msg_msg *msg)
594 {
595         struct list_head *tmp;
596
597         tmp = msq->q_receivers.next;
598         while (tmp != &msq->q_receivers) {
599                 struct msg_receiver *msr;
600
601                 msr = list_entry(tmp, struct msg_receiver, r_list);
602                 tmp = tmp->next;
603                 if (testmsg(msg, msr->r_msgtype, msr->r_mode) &&
604                     !security_msg_queue_msgrcv(msq, msg, msr->r_tsk,
605                                                msr->r_msgtype, msr->r_mode)) {
606
607                         list_del(&msr->r_list);
608                         if (msr->r_maxsize < msg->m_ts) {
609                                 msr->r_msg = NULL;
610                                 wake_up_process(msr->r_tsk);
611                                 smp_mb();
612                                 msr->r_msg = ERR_PTR(-E2BIG);
613                         } else {
614                                 msr->r_msg = NULL;
615                                 msq->q_lrpid = msr->r_tsk->pid;
616                                 msq->q_rtime = get_seconds();
617                                 wake_up_process(msr->r_tsk);
618                                 smp_mb();
619                                 msr->r_msg = msg;
620
621                                 return 1;
622                         }
623                 }
624         }
625         return 0;
626 }
627
628 asmlinkage long
629 sys_msgsnd(int msqid, struct msgbuf __user *msgp, size_t msgsz, int msgflg)
630 {
631         struct msg_queue *msq;
632         struct msg_msg *msg;
633         long mtype;
634         int err;
635         struct ipc_namespace *ns;
636
637         ns = current->nsproxy->ipc_ns;
638
639         if (msgsz > ns->msg_ctlmax || (long) msgsz < 0 || msqid < 0)
640                 return -EINVAL;
641         if (get_user(mtype, &msgp->mtype))
642                 return -EFAULT;
643         if (mtype < 1)
644                 return -EINVAL;
645
646         msg = load_msg(msgp->mtext, msgsz);
647         if (IS_ERR(msg))
648                 return PTR_ERR(msg);
649
650         msg->m_type = mtype;
651         msg->m_ts = msgsz;
652
653         msq = msg_lock(ns, msqid);
654         err = -EINVAL;
655         if (msq == NULL)
656                 goto out_free;
657
658         err= -EIDRM;
659         if (msg_checkid(ns, msq, msqid))
660                 goto out_unlock_free;
661
662         for (;;) {
663                 struct msg_sender s;
664
665                 err = -EACCES;
666                 if (ipcperms(&msq->q_perm, S_IWUGO))
667                         goto out_unlock_free;
668
669                 err = security_msg_queue_msgsnd(msq, msg, msgflg);
670                 if (err)
671                         goto out_unlock_free;
672
673                 if (msgsz + msq->q_cbytes <= msq->q_qbytes &&
674                                 1 + msq->q_qnum <= msq->q_qbytes) {
675                         break;
676                 }
677
678                 /* queue full, wait: */
679                 if (msgflg & IPC_NOWAIT) {
680                         err = -EAGAIN;
681                         goto out_unlock_free;
682                 }
683                 ss_add(msq, &s);
684                 ipc_rcu_getref(msq);
685                 msg_unlock(msq);
686                 schedule();
687
688                 ipc_lock_by_ptr(&msq->q_perm);
689                 ipc_rcu_putref(msq);
690                 if (msq->q_perm.deleted) {
691                         err = -EIDRM;
692                         goto out_unlock_free;
693                 }
694                 ss_del(&s);
695
696                 if (signal_pending(current)) {
697                         err = -ERESTARTNOHAND;
698                         goto out_unlock_free;
699                 }
700         }
701
702         msq->q_lspid = current->tgid;
703         msq->q_stime = get_seconds();
704
705         if (!pipelined_send(msq, msg)) {
706                 /* noone is waiting for this message, enqueue it */
707                 list_add_tail(&msg->m_list, &msq->q_messages);
708                 msq->q_cbytes += msgsz;
709                 msq->q_qnum++;
710                 atomic_add(msgsz, &msg_bytes);
711                 atomic_inc(&msg_hdrs);
712         }
713
714         err = 0;
715         msg = NULL;
716
717 out_unlock_free:
718         msg_unlock(msq);
719 out_free:
720         if (msg != NULL)
721                 free_msg(msg);
722         return err;
723 }
724
725 static inline int convert_mode(long *msgtyp, int msgflg)
726 {
727         /*
728          *  find message of correct type.
729          *  msgtyp = 0 => get first.
730          *  msgtyp > 0 => get first message of matching type.
731          *  msgtyp < 0 => get message with least type must be < abs(msgtype).
732          */
733         if (*msgtyp == 0)
734                 return SEARCH_ANY;
735         if (*msgtyp < 0) {
736                 *msgtyp = -*msgtyp;
737                 return SEARCH_LESSEQUAL;
738         }
739         if (msgflg & MSG_EXCEPT)
740                 return SEARCH_NOTEQUAL;
741         return SEARCH_EQUAL;
742 }
743
744 asmlinkage long sys_msgrcv(int msqid, struct msgbuf __user *msgp, size_t msgsz,
745                            long msgtyp, int msgflg)
746 {
747         struct msg_queue *msq;
748         struct msg_msg *msg;
749         int mode;
750         struct ipc_namespace *ns;
751
752         if (msqid < 0 || (long) msgsz < 0)
753                 return -EINVAL;
754         mode = convert_mode(&msgtyp, msgflg);
755         ns = current->nsproxy->ipc_ns;
756
757         msq = msg_lock(ns, msqid);
758         if (msq == NULL)
759                 return -EINVAL;
760
761         msg = ERR_PTR(-EIDRM);
762         if (msg_checkid(ns, msq, msqid))
763                 goto out_unlock;
764
765         for (;;) {
766                 struct msg_receiver msr_d;
767                 struct list_head *tmp;
768
769                 msg = ERR_PTR(-EACCES);
770                 if (ipcperms(&msq->q_perm, S_IRUGO))
771                         goto out_unlock;
772
773                 msg = ERR_PTR(-EAGAIN);
774                 tmp = msq->q_messages.next;
775                 while (tmp != &msq->q_messages) {
776                         struct msg_msg *walk_msg;
777
778                         walk_msg = list_entry(tmp, struct msg_msg, m_list);
779                         if (testmsg(walk_msg, msgtyp, mode) &&
780                             !security_msg_queue_msgrcv(msq, walk_msg, current,
781                                                        msgtyp, mode)) {
782
783                                 msg = walk_msg;
784                                 if (mode == SEARCH_LESSEQUAL &&
785                                                 walk_msg->m_type != 1) {
786                                         msg = walk_msg;
787                                         msgtyp = walk_msg->m_type - 1;
788                                 } else {
789                                         msg = walk_msg;
790                                         break;
791                                 }
792                         }
793                         tmp = tmp->next;
794                 }
795                 if (!IS_ERR(msg)) {
796                         /*
797                          * Found a suitable message.
798                          * Unlink it from the queue.
799                          */
800                         if ((msgsz < msg->m_ts) && !(msgflg & MSG_NOERROR)) {
801                                 msg = ERR_PTR(-E2BIG);
802                                 goto out_unlock;
803                         }
804                         list_del(&msg->m_list);
805                         msq->q_qnum--;
806                         msq->q_rtime = get_seconds();
807                         msq->q_lrpid = current->tgid;
808                         msq->q_cbytes -= msg->m_ts;
809                         atomic_sub(msg->m_ts, &msg_bytes);
810                         atomic_dec(&msg_hdrs);
811                         ss_wakeup(&msq->q_senders, 0);
812                         msg_unlock(msq);
813                         break;
814                 }
815                 /* No message waiting. Wait for a message */
816                 if (msgflg & IPC_NOWAIT) {
817                         msg = ERR_PTR(-ENOMSG);
818                         goto out_unlock;
819                 }
820                 list_add_tail(&msr_d.r_list, &msq->q_receivers);
821                 msr_d.r_tsk = current;
822                 msr_d.r_msgtype = msgtyp;
823                 msr_d.r_mode = mode;
824                 if (msgflg & MSG_NOERROR)
825                         msr_d.r_maxsize = INT_MAX;
826                 else
827                         msr_d.r_maxsize = msgsz;
828                 msr_d.r_msg = ERR_PTR(-EAGAIN);
829                 current->state = TASK_INTERRUPTIBLE;
830                 msg_unlock(msq);
831
832                 schedule();
833
834                 /* Lockless receive, part 1:
835                  * Disable preemption.  We don't hold a reference to the queue
836                  * and getting a reference would defeat the idea of a lockless
837                  * operation, thus the code relies on rcu to guarantee the
838                  * existance of msq:
839                  * Prior to destruction, expunge_all(-EIRDM) changes r_msg.
840                  * Thus if r_msg is -EAGAIN, then the queue not yet destroyed.
841                  * rcu_read_lock() prevents preemption between reading r_msg
842                  * and the spin_lock() inside ipc_lock_by_ptr().
843                  */
844                 rcu_read_lock();
845
846                 /* Lockless receive, part 2:
847                  * Wait until pipelined_send or expunge_all are outside of
848                  * wake_up_process(). There is a race with exit(), see
849                  * ipc/mqueue.c for the details.
850                  */
851                 msg = (struct msg_msg*)msr_d.r_msg;
852                 while (msg == NULL) {
853                         cpu_relax();
854                         msg = (struct msg_msg *)msr_d.r_msg;
855                 }
856
857                 /* Lockless receive, part 3:
858                  * If there is a message or an error then accept it without
859                  * locking.
860                  */
861                 if (msg != ERR_PTR(-EAGAIN)) {
862                         rcu_read_unlock();
863                         break;
864                 }
865
866                 /* Lockless receive, part 3:
867                  * Acquire the queue spinlock.
868                  */
869                 ipc_lock_by_ptr(&msq->q_perm);
870                 rcu_read_unlock();
871
872                 /* Lockless receive, part 4:
873                  * Repeat test after acquiring the spinlock.
874                  */
875                 msg = (struct msg_msg*)msr_d.r_msg;
876                 if (msg != ERR_PTR(-EAGAIN))
877                         goto out_unlock;
878
879                 list_del(&msr_d.r_list);
880                 if (signal_pending(current)) {
881                         msg = ERR_PTR(-ERESTARTNOHAND);
882 out_unlock:
883                         msg_unlock(msq);
884                         break;
885                 }
886         }
887         if (IS_ERR(msg))
888                 return PTR_ERR(msg);
889
890         msgsz = (msgsz > msg->m_ts) ? msg->m_ts : msgsz;
891         if (put_user (msg->m_type, &msgp->mtype) ||
892             store_msg(msgp->mtext, msg, msgsz)) {
893                 msgsz = -EFAULT;
894         }
895         free_msg(msg);
896
897         return msgsz;
898 }
899
900 #ifdef CONFIG_PROC_FS
901 static int sysvipc_msg_proc_show(struct seq_file *s, void *it)
902 {
903         struct msg_queue *msq = it;
904
905         return seq_printf(s,
906                         "%10d %10d  %4o  %10lu %10lu %5u %5u %5u %5u %5u %5u %10lu %10lu %10lu\n",
907                         msq->q_perm.key,
908                         msq->q_id,
909                         msq->q_perm.mode,
910                         msq->q_cbytes,
911                         msq->q_qnum,
912                         msq->q_lspid,
913                         msq->q_lrpid,
914                         msq->q_perm.uid,
915                         msq->q_perm.gid,
916                         msq->q_perm.cuid,
917                         msq->q_perm.cgid,
918                         msq->q_stime,
919                         msq->q_rtime,
920                         msq->q_ctime);
921 }
922 #endif