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