lkml.org 
[lkml]   [2003]   [Oct]   [3]   [last100]   RSS Feed
Views: [wrap][no wrap]   [headers]  [forward] 
 
Messages in this thread
Patch in this message
/
Subject[PATCH] [2/2] posix message queues
From
Date
[2/2] implements the interface in ipc/posixmsg.c through a set of
new syscalls and a new pseudo filesystem msgfs
the implementation passes the posixtestsuite 1.2 and has SIGEV_THREAD
support through userspace (taken from Benedyczak,Wronski)



diff -X dontdiff -Nur vanilla-2.6.0-test6/arch/i386/kernel/entry.S linux-2.6.0-test6/arch/i386/kernel/entry.S
--- vanilla-2.6.0-test6/arch/i386/kernel/entry.S 2003-08-23 01:53:39.000000000 +0200
+++ linux-2.6.0-test6/arch/i386/kernel/entry.S 2003-09-07 21:48:07.000000000 +0200
@@ -879,5 +879,12 @@
.long sys_tgkill /* 270 */
.long sys_utimes
.long sys_fadvise64_64
+ .long sys_mq_open
+ .long sys_mq_unlink
+ .long sys_mq_timedsend /* 275 */
+ .long sys_mq_timedreceive
+ .long sys_mq_notify
+ .long sys_mq_getattr
+ .long sys_mq_setattr

nr_syscalls=(.-sys_call_table)/4
diff -X dontdiff -Nur vanilla-2.6.0-test6/CREDITS linux-2.6.0-test6/CREDITS
--- vanilla-2.6.0-test6/CREDITS 2003-10-01 20:10:35.000000000 +0200
+++ linux-2.6.0-test6/CREDITS 2003-10-03 14:10:31.000000000 +0200
@@ -3286,6 +3295,14 @@
S: 5554 GG Valkenswaard
S: The Netherlands

+N: Peter Wächtler
+E: pwaechtler@mac.com
+W: http://homepage.mac.com/pwaechtler/
+D: Posix message queues
+S: Fliederstr. 3
+S: 30167 Hannover
+S: Germany
+
N: Peter Shaobo Wang
E: pwang@mmdcorp.com
W: http://www.mmdcorp.com/pw/linux
diff -X dontdiff -Nur vanilla-2.6.0-test6/include/asm-i386/unistd.h linux-2.6.0-test6/include/asm-i386/unistd.h
--- vanilla-2.6.0-test6/include/asm-i386/unistd.h 2003-08-23 01:57:19.000000000 +0200
+++ linux-2.6.0-test6/include/asm-i386/unistd.h 2003-09-07 21:48:07.000000000 +0200
@@ -278,8 +278,15 @@
#define __NR_tgkill 270
#define __NR_utimes 271
#define __NR_fadvise64_64 272
+#define __NR_sys_mq_open 273
+#define __NR_sys_mq_unlink (__NR_sys_mq_open+1)
+#define __NR_mq_timedsend (__NR_sys_mq_open+2)
+#define __NR_mq_timedreceive (__NR_sys_mq_open+3)
+#define __NR_mq_notify (__NR_sys_mq_open+4)
+#define __NR_mq_getattr (__NR_sys_mq_open+5)
+#define __NR_mq_setattr (__NR_sys_mq_open+6)

-#define NR_syscalls 273
+#define NR_syscalls 279

/* user-visible error numbers are in the range -1 - -124: see <asm-i386/errno.h> */

diff -X dontdiff -Nur vanilla-2.6.0-test6/include/linux/pxqueue.h linux-2.6.0-test6/include/linux/pxqueue.h
--- vanilla-2.6.0-test6/include/linux/pxqueue.h 1970-01-01 01:00:00.000000000 +0100
+++ linux-2.6.0-test6/include/linux/pxqueue.h 2003-09-28 22:30:48.000000000 +0200
@@ -0,0 +1,48 @@
+#ifndef _LINUX_MQUEUE_H
+#define _LINUX_MQUEUE_H
+
+#define MQ_MAXMSG 40 /* max number of messages in each queue */
+#define MQ_MAXSYSSIZE 1048576 /* max size that all m.q. can have together */
+#define MQ_PRIO_MAX 256 /* max priority */
+
+#define MSGFS_MAGIC 0x4D455347
+#define _POSIX_MESSAGE_PASSING 1
+
+typedef int mqd_t; /* message queue descriptor */
+
+struct mq_attr {
+ long mq_flags; /* message queue flags */
+ long mq_maxmsg; /* maximum number of messages */
+ long mq_msgsize; /* maximum message size */
+ long mq_curmsgs; /* number of messages currently queued */
+};
+
+asmlinkage mqd_t sys_mq_open(const char *u_path, int oflag, mode_t mode,
+ struct mq_attr *u_attr);
+asmlinkage int sys_mq_close(mqd_t mqdes);
+asmlinkage int sys_mq_unlink(const char *u_name);
+asmlinkage int sys_mq_timedsend(mqd_t mqdes, const char *msg_ptr,
+ size_t msg_len, unsigned int msg_prio, struct timespec *utime);
+asmlinkage ssize_t sys_mq_timedreceive(mqd_t mqdes, char *msg_ptr,
+ size_t msg_len, unsigned int *msg_prio, struct timespec *utime);
+asmlinkage int sys_mq_notify(mqd_t mqdes,
+ const struct sigevent *u_notification);
+asmlinkage int sys_mq_getattr(mqd_t mqdes, struct mq_attr *u_mqstat);
+asmlinkage int sys_mq_setattr(mqd_t mqdes, const struct mq_attr *u_mqstat,
+ struct mq_attr *u_omqstat);
+
+#ifdef __KERNEL__
+
+struct mqueue_ds { /* queue */
+ struct mq_attr attr;
+ struct msg_queue queue; /* ipc/msg */
+ spinlock_t lock;
+ wait_queue_head_t wait_recv;
+ wait_queue_head_t wait_send;
+
+ pid_t notify_pid; /* who we have to notify (or 0) */
+ struct sigevent notify; /* notification */
+};
+#endif /* __KERNEL__ */
+
+#endif
diff -X dontdiff -Nur vanilla-2.6.0-test6/init/Kconfig linux-2.6.0-test6/init/Kconfig
--- vanilla-2.6.0-test6/init/Kconfig 2003-10-01 20:10:58.000000000 +0200
+++ linux-2.6.0-test6/init/Kconfig 2003-10-01 20:05:34.000000000 +0200
@@ -91,6 +91,15 @@
section 6.4 of the Linux Programmer's Guide, available from
<http://www.tldp.org/docs.html#guide>.

+config POSIXMSG
+ bool "POSIX message queues"
+ ---help---
+ This gives you POSIX compliant interfaces for message queues.
+ For userspace stub look at
+ <http://homepage.mac.com/pwaechtler/mqueue.html>.
+
+
config BSD_PROCESS_ACCT
bool "BSD Process Accounting"
help
diff -X dontdiff -Nur vanilla-2.6.0-test6/ipc/Makefile linux-2.6.0-test6/ipc/Makefile
--- vanilla-2.6.0-test6/ipc/Makefile 2003-08-23 01:52:27.000000000 +0200
+++ linux-2.6.0-test6/ipc/Makefile 2003-09-07 23:07:47.000000000 +0200
@@ -5,3 +5,5 @@
obj-y := util.o

obj-$(CONFIG_SYSVIPC) += msg.o sem.o shm.o
+obj-$(CONFIG_POSIXMSG) += posixmsg.o
diff -X dontdiff -Nur vanilla-2.6.0-test6/ipc/posixmsg.c linux-2.6.0-test6/ipc/posixmsg.c
--- vanilla-2.6.0-test6/ipc/posixmsg.c 1970-01-01 01:00:00.000000000 +0100
+++ linux-2.6.0-test6/ipc/posixmsg.c 2003-10-03 13:24:54.000000000 +0200
@@ -0,0 +1,945 @@
+/*
+ * linux/ipc/posixmsg.c
+ *
+ * Copyright 2002,2003 Peter Wächtler <pwaechtler@mac.com>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ *
+ * root can mount the filesystem to see the names of the currently
+ * used mqueues. The i_size shows the number of msgs in the queue.
+ * The only shell operation on the "fifos" is an unlink() via rm(1)
+ * The queues are accessed through syscalls, poll is supported
+ *
+ * CHANGES:
+ * put MSGFS_MAGIC into mqueue.h
+ * update inode->m_time on send (yes, no posix semantics on this)
+ * update inode->a_time on recv
+ * translate absolute timeouts in kernel to prevent too long sleeps
+ * corrected spinlocks + [gs]etattr/mq_flags
+ * replaced spinlocks with inode->i_sem
+ * (don't know yet what's broken on SMP with spinlocks)
+ * use spinlocks, where was the problem?
+ * open coded the wait_event_interruptible_timeouts to avoid races
+ * fixed accmode tests
+ * fixed freespace check (meant a per queue limit?)
+ * fixed msg_len check in sys_mq_timedreceive
+ * defer check for negative timeout on mq_timedreceive
+ * don't put_user in timedreceive if msg_prio is NULL
+ * add macros to choose fget/fput or not (read comment at get_filp)
+ *
+ * * passes the posixtestsuite 1.2
+ *
+ * TODO:
+ * check for more sysv msg limits (or add some new, e.g. MQ_PRIO_MAX)?
+ * think about prio based waitqueues: simply enqueue them in order?
+ * bits/posix_opt.h claims to support _POSIX_PRIORITY_SCHEDULING
+ * make wake_up_interruptible_all sysctl'able
+ */
+#include <linux/kernel.h>
+#include <linux/init.h>
+#include <linux/slab.h>
+#include <linux/fs.h>
+#include <linux/mount.h>
+#include <linux/file.h>
+#include <linux/namei.h>
+#include <linux/pagemap.h> /* PAGE_CACHE_SIZE */
+#include <linux/poll.h>
+#include <linux/security.h>
+#include <linux/pxqueue.h>
+#include <linux/msg.h>
+#include <asm/uaccess.h>
+#include "util.h"
+
+
+extern int msg_ctlmnb; /* default max size of a message queue (all msgs sum up) */
+extern int msg_ctlmni; /* max # of msg queue identifiers */
+extern int msg_ctlmax; /* max size of one message (bytes) */
+
+static atomic_t msg_bytes = ATOMIC_INIT(0);
+static struct vfsmount *msg_mnt;
+
+#define get_mqueue(filp)\
+ ((filp) ? filp->f_dentry->d_inode->u.generic_ip: filp)
+
+static struct msg_queue *
+init_queue(struct msg_queue *queue)
+{
+ int retval;
+
+ queue->q_perm.mode = 0;
+ queue->q_perm.key = 0;
+ queue->q_perm.security = NULL;
+ retval = security_msg_queue_alloc(queue);
+ if (retval)
+ return ERR_PTR(retval);;
+
+ queue->q_stime = queue->q_rtime = 0;
+ queue->q_qbytes = queue->q_cbytes = queue->q_qnum = 0;
+ queue->q_lspid = queue->q_lrpid = 0;
+ INIT_LIST_HEAD(&queue->q_messages);
+ INIT_LIST_HEAD(&queue->q_receivers); /* unused */
+ INIT_LIST_HEAD(&queue->q_senders); /* unused */
+
+ return queue;
+}
+
+static int
+mqueue_unlink (struct inode *dir, struct dentry *dentry)
+{
+ dentry->d_inode->i_nlink--;
+ dput(dentry);
+ return 0;
+}
+
+/* cleans up after close() or exit() if the refount drops to zero */
+static int
+mqueue_close(struct inode *inode, struct file *filp)
+{
+ struct mqueue_ds *q = inode->u.generic_ip;
+
+ if (!q || !filp)
+ return -EBADFD;
+
+ spin_lock(&q->lock);
+ /* remove possible notification;
+ * sys_getpid() returns the tgid if multithreaded */
+ if (q->notify_pid == current->pid) {
+ q->notify_pid = 0;
+ q->notify.sigev_signo = 0;
+ q->notify.sigev_notify = SIGEV_NONE;
+ }
+ spin_unlock(&q->lock);
+ return 0;
+}
+
+static unsigned int
+mqueue_poll(struct file *filp, struct poll_table_struct *wait)
+{
+ struct mqueue_ds *q = get_mqueue(filp);
+ int ret = 0;
+
+ poll_wait(filp, &q->wait_recv, wait);
+ poll_wait(filp, &q->wait_send, wait);
+
+ if (q->queue.q_qnum)
+ ret = POLLIN | POLLRDNORM;
+
+ if (q->queue.q_qnum < (unsigned long)q->attr.mq_maxmsg)
+ ret |= POLLOUT | POLLWRNORM;
+
+ return ret;
+}
+
+static struct file_operations msg_fops = {
+ .llseek = no_llseek,
+ .release = mqueue_close,
+ .poll = mqueue_poll,
+};
+static struct inode_operations msg_dir_inode_operations = {
+ .lookup = simple_lookup,
+ .unlink = mqueue_unlink,
+};
+
+static struct inode *
+get_msg_inode(struct super_block *sb, mode_t mode)
+{
+ struct mqueue_ds *q;
+ struct inode *inode = new_inode(sb);
+
+ if (inode) {
+ inode->i_mode = mode;
+ inode->i_uid = current->fsuid;
+ inode->i_gid = current->fsgid;
+ inode->i_atime = inode->i_mtime = inode->i_ctime = CURRENT_TIME;
+ inode->i_blksize = 1024;
+ switch (mode & S_IFMT) {
+ default:
+ case S_IFIFO:
+ case S_IFREG:
+ inode->i_fop = &msg_fops;
+ if ((q = kmalloc(sizeof (struct mqueue_ds), GFP_KERNEL))) {
+ if (!init_queue(&q->queue)) {
+ iput(inode);
+ inode = ERR_PTR(-EACCES);
+ }
+ inode->u.generic_ip = q;
+ spin_lock_init(&q->lock);
+ } else {
+ iput(inode);
+ inode = ERR_PTR(-ENOSPC);
+ }
+ break;
+ case S_IFDIR:
+ inode->i_op = &msg_dir_inode_operations;
+ inode->i_fop = &simple_dir_operations;
+ /* directory inodes start off with i_nlink == 2 (for "." entry) */
+ inode->i_nlink++;
+ break;
+ case S_IFLNK:
+ break;
+ }
+ } else
+ inode = ERR_PTR(-ENOSPC);
+
+ return inode;
+}
+
+#if 0
+/* don't use fget() to avoid the fput() for speed reason
+ * on create/open the refcount is 1 and decremented on close
+ * if you have a multithreaded app where one thread closes
+ * the mqueue while another thread operates on it -> possible crash
+ * the spec says the behavior is undefined
+ * separate processes are not affected
+ */
+static struct file *
+get_filp(mqd_t fd)
+{
+ struct files_struct *files = current->files;
+ struct file *rc = NULL;
+
+ spin_lock(&files->file_lock);
+ if (fd < files->max_fds && fd >= 0)
+ rc = files->fd[fd];
+ spin_unlock(&files->file_lock);
+ return rc;
+
+}
+#define ERROUT(x) return x;
+#define FPUTOUT
+
+#else
+
+#define get_filp(x) fget(x)
+#define ERROUT(x) { err = x; goto out_fput;}
+#define FPUTOUT out_fput: fput(filp)
+#endif
+
+static inline int
+freespace(struct mqueue_ds *q, size_t msg_len)
+{
+ return (int) ( q->queue.q_qnum < (unsigned long)q->attr.mq_maxmsg);
+}
+
+/* removes a queue - called from vfs when an inode gets deleted */
+static void
+mqueue_release(struct inode *inode)
+{
+ struct mqueue_ds *q = inode->u.generic_ip;
+ struct msg_queue *queue;
+ struct list_head *tmp;
+
+ if (q){
+ queue = &q->queue;
+ tmp = queue->q_messages.next;
+ while (tmp != &queue->q_messages) {
+ struct msg_msg *msg = list_entry(tmp, struct msg_msg, m_list);
+ tmp = tmp->next;
+ free_msg(msg);
+ }
+ atomic_sub(queue->q_cbytes, &msg_bytes);
+ security_msg_queue_free(queue);
+ inode->u.generic_ip = NULL;
+ kfree(q);
+ }
+ clear_inode(inode);
+}
+
+static int
+create_queue(struct dentry *dir, struct qstr *qname,
+ int oflag, mode_t mode, struct mq_attr *u_attr)
+{
+ int ret, fd;
+ struct file *filp;
+ struct mqueue_ds *q;
+ struct inode *inode;
+ struct dentry *dentry;
+
+ inode = get_msg_inode(msg_mnt->mnt_sb, S_IFIFO | (mode & S_IRWXUGO));
+ if (IS_ERR(inode)) {
+ ret = PTR_ERR(inode);
+ goto out_ret;
+ }
+ q = inode->u.generic_ip;
+
+ if (u_attr != NULL) {
+ if (copy_from_user(&q->attr, u_attr, sizeof (struct mq_attr))) {
+ ret = -EFAULT;
+ goto out_inode;
+ }
+ if (q->attr.mq_maxmsg <= 0
+ || q->attr.mq_msgsize <= 0
+ || q->attr.mq_maxmsg > MQ_MAXMSG
+ || q->attr.mq_msgsize > msg_ctlmax) {
+ ret = -EINVAL;
+ goto out_inode;
+ }
+ } else { /* implementation defined */
+ q->attr.mq_maxmsg = MQ_MAXMSG;
+ q->attr.mq_msgsize = 1024 /*msg_ctlmax */ ;
+ }
+ q->attr.mq_flags = oflag & O_ACCMODE;
+ q->notify_pid = 0;
+ q->notify.sigev_signo = 0;
+ q->notify.sigev_notify = SIGEV_NONE;
+ init_waitqueue_head(&q->wait_send);
+ init_waitqueue_head(&q->wait_recv);
+
+ ret = -ENFILE;
+ if ((fd = get_unused_fd()) < 0)
+ goto out_inode;
+ if (!(filp = get_empty_filp()))
+ goto out_fd;
+
+ qname->hash = full_name_hash(qname->name, qname->len);
+ dentry = d_alloc(dir, qname);
+ if (!dentry) {
+ ret = -ENOMEM;
+ goto out_filp;
+ }
+ d_add(dentry, inode);
+ ret = get_write_access(inode);
+ if (ret)
+ goto out_filp;
+
+ filp->f_vfsmnt = mntget(msg_mnt);
+ filp->f_dentry = dget(dentry); /* leave it active */
+ filp->f_op = &msg_fops;
+ filp->f_mode = (q->attr.mq_flags +1 ) & O_ACCMODE;
+ filp->f_flags = oflag;
+
+ /* Now we map fd to filp, so userspace can access it */
+ fd_install(fd, filp);
+ ret = fd;
+ goto out_ret;
+
+out_filp:
+ put_filp(filp);
+out_fd:
+ put_unused_fd(fd);
+out_inode:
+ iput(inode);
+out_ret:
+ return ret;
+}
+
+/**
+ * sys_mq_open - opens a message queue associated with @u_name
+ * @mqdes: descriptor of mqueue
+ * @oflag: flags like O_CREAT, O_EXCL, O_RDWR
+ * @mode: when O_CREAT is specified, the permission bits
+ * @u_attr: pointer to the attributes, like max msgsize, when creating
+ *
+ * returns a descriptor to the opened queue or negative value on error
+ */
+asmlinkage mqd_t
+sys_mq_open(const char *u_name, int oflag, mode_t mode, struct mq_attr * u_attr)
+{
+ struct file *filp;
+ struct dentry *dentry;
+ struct qstr this;
+ static int oflag2acc[O_ACCMODE] =
+ { MAY_READ, MAY_WRITE, MAY_READ | MAY_WRITE };
+ int fd, ret;
+
+ if (IS_ERR(this.name = getname(u_name)))
+ return PTR_ERR(this.name);
+ this.len = strlen(this.name);
+ down(&msg_mnt->mnt_root->d_inode->i_sem);
+ dentry = lookup_one_len(this.name, msg_mnt->mnt_root, this.len);
+
+ if (IS_ERR(dentry)) {
+ ret = PTR_ERR(dentry);
+ goto out_ret;
+ }
+ if (oflag & O_CREAT) {
+ if (dentry->d_inode) {
+ /* entry exists already */
+ if (oflag & O_EXCL) {
+ ret = -EEXIST;
+ } else {
+ goto open_existing;
+ }
+ goto out_dput;
+ } else {
+ ret = create_queue(msg_mnt->mnt_root, &this,
+ oflag, mode, u_attr);
+ }
+
+ } else { /* O_CREAT */
+ if (!dentry->d_inode) {
+ ret = -ENOENT;
+ } else {
+open_existing:
+ if (permission
+ (dentry->d_inode, oflag2acc[oflag & O_ACCMODE], NULL)) {
+ ret = -EACCES;
+ } else {
+ fd = get_unused_fd();
+ if (fd >= 0) {
+ mntget(msg_mnt);
+ filp =
+ dentry_open(dentry, msg_mnt, oflag);
+ filp->f_op = &msg_fops;
+ if (IS_ERR(filp)) {
+ ret = PTR_ERR(filp);
+ put_unused_fd(fd);
+ goto out_dput;
+ }
+ dget(dentry);
+ fd_install(fd, filp);
+ }
+ ret = fd;
+ }
+ }
+ }
+out_dput:
+ dput(dentry);
+out_ret:
+ up(&msg_mnt->mnt_root->d_inode->i_sem);
+ putname(this.name);
+
+ return ret;
+}
+
+/**
+ * sys_mq_unlink - removes a message queue from the namespace
+ *
+ * @u_name: pointer to the name
+ */
+asmlinkage int
+sys_mq_unlink(const char *u_name)
+{
+ int err=0;
+ struct dentry *dentry;
+ char *name = getname(u_name);
+
+ if (IS_ERR(name))
+ return PTR_ERR(name);
+
+ down(&msg_mnt->mnt_root->d_inode->i_sem);
+ dentry = lookup_one_len(name, msg_mnt->mnt_root, strlen(name));
+ if (IS_ERR(dentry)){
+ err= PTR_ERR(dentry);
+ goto out_unlock;
+ }
+ if (!dentry->d_inode){
+ err= -ENOENT;
+ goto out_unlock;
+ }
+ err = vfs_unlink(dentry->d_parent->d_inode, dentry);
+
+out_unlock:
+ up(&msg_mnt->mnt_root->d_inode->i_sem);
+ putname(name);
+ return err;
+}
+
+static inline long get_timeout( struct timespec *abs, int *err)
+{
+ struct timespec t;
+ *err = 0;
+ if (abs->tv_nsec >= 1000000000L || abs->tv_nsec < 0 || abs->tv_sec < 0)
+ return *err = -EINVAL;
+
+ t=current_kernel_time();
+ if (t.tv_sec > abs->tv_sec ||
+ (t.tv_sec == abs->tv_sec && t.tv_nsec > abs->tv_nsec))
+ return -ETIMEDOUT;
+
+ t.tv_sec = abs->tv_sec - t.tv_sec;
+ t.tv_nsec = abs->tv_nsec - t.tv_nsec;
+ if (t.tv_nsec < 0){
+ t.tv_sec--;
+ t.tv_nsec+= 1000000000;
+ }
+ return timespec_to_jiffies(&t) + 1;
+}
+
+static void local_add_wait_queue(wait_queue_head_t *q, wait_queue_t * wait)
+{
+ wait->flags &= ~WQ_FLAG_EXCLUSIVE;
+ spin_lock(&q->lock);
+ __add_wait_queue(q, wait);
+ spin_unlock(&q->lock);
+}
+
+static void local_remove_wait_queue(wait_queue_head_t *q, wait_queue_t * wait)
+{
+ spin_lock(&q->lock);
+ __remove_wait_queue(q, wait);
+ spin_unlock(&q->lock);
+}
+
+/**
+ * sys_mq_timedsend - send a message to the queue associated
+ * with the descriptor mqdes
+ * @mqdes: descriptor of mqueue
+ * @msg_ptr: pointer to buffer holding the message
+ * @msg_len: length of the message
+ * @msg_prio: the priority of the message
+ * @utime: if !NULL the function will only block for specified time
+ */
+asmlinkage int
+sys_mq_timedsend(mqd_t mqdes,
+ const char *msg_ptr, size_t msg_len,
+ unsigned int msg_prio, struct timespec *utime)
+{
+ struct msg_msg *msg;
+ struct list_head *p;
+ struct msg_queue *queue;
+ int err;
+ long timeout;
+ struct timespec ts;
+ struct mqueue_ds *q;
+ struct file *filp = get_filp(mqdes);
+ struct inode *inode;
+
+
+ if (!(q = get_mqueue(filp)))
+ return -EBADF;
+ if ( !(filp->f_mode & FMODE_WRITE))
+ ERROUT( -EBADF);
+ if ((unsigned int) msg_prio >= (unsigned int) MQ_PRIO_MAX)
+ ERROUT( -EINVAL);
+ if ((long)msg_len > q->attr.mq_msgsize)
+ ERROUT( -EMSGSIZE);
+
+ queue = &q->queue;
+ inode=filp->f_dentry->d_inode;
+ spin_lock(&q->lock);
+ if ((filp->f_flags & O_NONBLOCK) && !freespace(q, msg_len)){
+ spin_unlock(&q->lock);
+ ERROUT( -EAGAIN);
+ }
+ spin_unlock(&q->lock);
+ /* check if this message will exceed overall limit for messages */
+ if (atomic_read(&msg_bytes) + msg_len > MQ_MAXSYSSIZE)
+ ERROUT( -ENOMEM);
+
+ if (utime) {
+ if (copy_from_user(&ts, utime, sizeof (ts)))
+ ERROUT( -EFAULT);
+ timeout = get_timeout(&ts, &err);
+ if ( err == -EINVAL )
+ ERROUT( err);
+ /* don't timeout yet, check if queue is empty */
+ } else
+ timeout = 0L;
+
+ msg = get_msg((char *) msg_ptr, msg_len);
+ if (IS_ERR(msg))
+ ERROUT( PTR_ERR(msg));
+
+ msg->m_type = msg_prio;
+ msg->m_ts = msg_len;
+
+ if (!timeout)
+ timeout = MAX_SCHEDULE_TIMEOUT;
+
+ /* open coded wait_event_interruptible_timeout()
+ * to avoid race on freespace()
+ */
+ spin_lock(&q->lock);
+ if (!freespace(q, msg_len) ){
+ if ( unlikely(timeout < 0) ){
+ err = -ETIMEDOUT;
+ goto out_unlock;
+ }
+ wait_queue_t __wait;
+ init_waitqueue_entry(&__wait, current);
+
+ local_add_wait_queue(&q->wait_send, &__wait);
+ for (;;) {
+ set_current_state(TASK_INTERRUPTIBLE);
+ if ( freespace(q, msg_len))
+ break;
+ if (!signal_pending(current)) {
+ spin_unlock(&q->lock);
+ timeout = schedule_timeout(timeout);
+ spin_lock(&q->lock);
+ if (!timeout)
+ break;
+ continue;
+ }
+ timeout = -ERESTARTSYS;
+ break;
+ }
+ current->state = TASK_RUNNING;
+ local_remove_wait_queue(&q->wait_send, &__wait);
+
+ if (unlikely(timeout == -ERESTARTSYS)){
+ err = -EINTR;
+ goto out_unlock;
+ }
+ if (unlikely(timeout == 0)){
+ err = -ETIMEDOUT;
+ goto out_unlock;
+ }
+ }
+ err = 0;
+ /* enqueue message in prio order */
+ p = queue->q_messages.next; /* used as flag if msg was queued */
+ if (msg_prio > 0 && !list_empty(&queue->q_messages)) {
+
+ list_for_each(p, &queue->q_messages) {
+ struct msg_msg *tmp =
+ list_entry(p, struct msg_msg, m_list);
+ if (tmp->m_type < (long)msg_prio) {
+ list_add_tail(&msg->m_list, p);
+ p = NULL;
+ break;
+ }
+ }
+ }
+ if (p) /* ok, put it at the end */
+ list_add_tail(&msg->m_list, &queue->q_messages);
+
+ queue->q_lspid = current->pid;
+ queue->q_cbytes += msg_len;
+ atomic_add(msg_len, &msg_bytes);
+ queue->q_qnum++;
+ inode->i_size = queue->q_qnum;
+ inode->i_mtime = CURRENT_TIME;
+
+ if (waitqueue_active(&q->wait_recv)) {
+ /* wake up all waiters to serve the highest prio waiter */
+ wake_up_interruptible_all(&q->wait_recv);
+ } else {
+ /* since there was no synchronously waiting process for message
+ * we notify it when the state of queue changed from
+ * empty to not empty */
+ if (q->notify_pid != 0 && queue->q_qnum == 1) {
+ /* TODO: Add support for sigev_notify==SIGEV_THREAD
+ * we should create a thread in userspace
+ */
+ if (q->notify.sigev_notify == SIGEV_THREAD) {
+
+ err = -ENOSYS;
+ pr_info("mq_*send: SIGEV_THREAD not supported\n");
+ }
+ /* sends signal */
+ if (q->notify.sigev_notify == SIGEV_SIGNAL) {
+ struct siginfo sig_i;
+ sig_i.si_signo = q->notify.sigev_signo;
+ sig_i.si_errno = 0;
+ sig_i.si_code = SI_MESGQ;
+ sig_i.si_pid = current->pid;
+ sig_i.si_uid = current->uid;
+ kill_proc_info(q->notify.sigev_signo,
+ &sig_i, q->notify_pid);
+ }
+ /* after notification unregister process */
+ q->notify_pid = 0;
+ }
+ }
+out_unlock:
+ spin_unlock(&q->lock);
+FPUTOUT;
+ return err;
+}
+
+/**
+ * sys_mq_timedreceive - receive a message from the queue associated
+ * with the descriptor mqdes
+ * @mqdes: descriptor of mqueue
+ * @msg_ptr: pointer to buffer to hold the message
+ * @msg_len: length of the userspace buffer
+ * @msg_prio: will hold the priority if a message was received
+ * @utime: if !NULL the function will only block for specified time
+ */
+asmlinkage ssize_t
+sys_mq_timedreceive(mqd_t mqdes,
+ char *msg_ptr, size_t msg_len,
+ unsigned int *msg_prio, struct timespec * utime)
+{
+ struct msg_queue *queue;
+ struct msg_msg *msg;
+ int err;
+ long timeout;
+ struct timespec ts;
+ struct mqueue_ds *q;
+ struct file *filp = get_filp(mqdes);
+ struct inode *inode;
+
+ if (!(q = get_mqueue(filp)))
+ return -EBADF;
+ if ((long)msg_len < q->attr.mq_msgsize)
+ ERROUT( -EMSGSIZE);
+
+ if ( !(filp->f_mode & FMODE_READ))
+ ERROUT( -EBADF);
+
+ queue = &q->queue;
+ if ((filp->f_flags & O_NONBLOCK) && queue->q_qnum == 0)
+ ERROUT( -EAGAIN);
+
+ if (utime) {
+ if (copy_from_user(&ts, utime, sizeof (ts)))
+ ERROUT( -EFAULT);
+ timeout = get_timeout(&ts, &err);
+ if ( err == -EINVAL )
+ ERROUT( err);
+ /* we do not check yet for the timeout */
+ } else
+ timeout = 0L;
+ if (!timeout)
+ timeout = MAX_SCHEDULE_TIMEOUT;
+
+ spin_lock(&q->lock);
+wait_on_msg:
+ if (queue->q_qnum == 0 ){
+ if (unlikely(timeout < 0)){
+ err = -ETIMEDOUT;
+ goto out_unlock;
+ }
+ wait_queue_t __wait;
+ init_waitqueue_entry(&__wait, current);
+
+ local_add_wait_queue(&q->wait_recv, &__wait);
+ for (;;) {
+ set_current_state(TASK_INTERRUPTIBLE);
+ if ( queue->q_qnum > 0)
+ break;
+ if (!signal_pending(current)) {
+ spin_unlock(&q->lock);
+ timeout = schedule_timeout(timeout);
+ spin_lock(&q->lock);
+ if (!timeout)
+ break;
+ continue;
+ }
+ timeout = -ERESTARTSYS;
+ break;
+ }
+ current->state = TASK_RUNNING;
+ local_remove_wait_queue(&q->wait_recv, &__wait);
+
+ if (unlikely(timeout == -ERESTARTSYS)){
+ err = -EINTR;
+ goto out_unlock;
+ }
+ if (unlikely(timeout == 0)){
+ err = -ETIMEDOUT;
+ goto out_unlock;
+ }
+ }
+ if (unlikely(list_empty(&queue->q_messages)))
+ goto wait_on_msg; /* another task was faster and removed the msg */
+
+ msg=list_entry(queue->q_messages.next, struct msg_msg, m_list);
+ list_del(&msg->m_list);
+ queue->q_lrpid = current->pid;
+ queue->q_cbytes -= msg->m_ts;
+ atomic_sub(msg->m_ts, &msg_bytes);
+ queue->q_qnum--;
+ inode=filp->f_dentry->d_inode;
+ inode->i_size = queue->q_qnum;
+ inode->i_atime = CURRENT_TIME;
+ /* msg is removed from list, we're already safe */
+ spin_unlock(&q->lock);
+
+ wake_up_interruptible(&q->wait_send);
+
+ msg_len = (msg_len > (size_t)msg->m_ts) ? (size_t)msg->m_ts : msg_len;
+
+ if ( put_msg(msg_ptr, msg, msg_len) ||
+ ( msg_prio && put_user(msg->m_type, msg_prio))) {
+ msg_len = -EFAULT; /* hmh, now the msg is lost */
+ }
+ free_msg(msg);
+
+ fput(filp);
+ return msg_len;
+
+out_unlock:
+ spin_unlock(&q->lock);
+FPUTOUT;
+ return err;
+}
+
+/**
+ * sys_mq_notify - set or remove a notification on the queue associated
+ * with the descriptor mqdes
+ * @mqdes: descriptor of mqueue
+ * @u_notification: pointer to struct sigevent
+ *
+ */
+asmlinkage int
+sys_mq_notify(mqd_t mqdes, const struct sigevent *u_notification)
+{
+ struct sigevent notify;
+ struct file *filp = get_filp(mqdes);
+ struct mqueue_ds *q;
+ int err = 0;
+
+ if (!(q = get_mqueue(filp)))
+ return -EBADF;
+ if (u_notification != NULL)
+ if (copy_from_user
+ (&notify, u_notification, sizeof (struct sigevent)))
+ ERROUT( -EFAULT);
+
+ down(&filp->f_dentry->d_inode->i_sem);
+ if (q->notify_pid == current->pid
+ && (u_notification == NULL || notify.sigev_notify == SIGEV_NONE)) {
+ q->notify_pid = 0; /* remove notification */
+ q->notify.sigev_signo = 0;
+ q->notify.sigev_notify = SIGEV_NONE;
+ } else if (q->notify_pid > 0) {
+ err = -EBUSY;
+ } else if (u_notification != NULL) {
+ if (notify.sigev_notify == SIGEV_SIGNAL) {
+ /* add notification */
+ q->notify_pid = current->pid;
+ q->notify.sigev_signo = notify.sigev_signo;
+ q->notify.sigev_notify = notify.sigev_notify;
+ } else if (notify.sigev_notify == SIGEV_THREAD) {
+ err = -ENOSYS;
+ pr_info("mq_*send: SIGEV_THREAD not supported yet\n");
+ }
+ }
+ up(&filp->f_dentry->d_inode->i_sem);
+FPUTOUT;
+ return err;
+}
+
+static inline void
+fill_flags(struct mqueue_ds *q, struct file *filp)
+{
+ spin_lock(&q->lock);
+ q->attr.mq_flags = (filp->f_mode -1) & O_ACCMODE;
+ q->attr.mq_flags |= (filp->f_flags) & O_NONBLOCK;
+ q->attr.mq_curmsgs = q->queue.q_qnum;
+ spin_unlock(&q->lock);
+}
+
+/**
+ * sys_mq_getattr - get the attributes of the queue associated
+ * with the descriptor mqdes
+ * @mqdes: descriptor of mqueue
+ * @u_mqstat: pointer to struct holding the new values
+ *
+ */
+asmlinkage int
+sys_mq_getattr(mqd_t mqdes, struct mq_attr *u_mqstat)
+{
+ int err = 0;
+ struct mqueue_ds *q;
+ struct file *filp = get_filp(mqdes);
+
+ if (!(q = get_mqueue(filp)))
+ return -EBADF;
+
+ fill_flags(q,filp);
+ if (copy_to_user(u_mqstat, &q->attr, sizeof (struct mq_attr)))
+ err = -EFAULT;
+FPUTOUT;
+ return err;
+}
+
+/**
+ * sys_mq_setattr - set the attributes of the queue associated
+ * with the descriptor mqdes
+ * @mqdes: descriptor of mqueue
+ * @u_mqstat: pointer to struct holding the new values
+ * @u_omqstat: pointer to store the original attributes (if !NULL)
+ *
+ */
+asmlinkage int
+sys_mq_setattr(mqd_t mqdes, const struct mq_attr *u_mqstat,
+ struct mq_attr *u_omqstat)
+{
+ struct mq_attr mqstat;
+ struct mqueue_ds *q;
+ struct file *filp = get_filp(mqdes);
+ int err = 0;
+
+ if (!(q = get_mqueue(filp)))
+ return -EBADF;
+
+ if (u_omqstat != NULL) {
+ fill_flags(q,filp);
+ if (copy_to_user(u_omqstat, &q->attr, sizeof (struct mq_attr)))
+ ERROUT( -EFAULT);
+ }
+ if (copy_from_user(&mqstat, u_mqstat, sizeof (struct mq_attr)))
+ ERROUT( -EFAULT);
+ if (mqstat.mq_flags & O_NONBLOCK)
+ filp->f_flags |= O_NONBLOCK;
+ else
+ filp->f_flags &= ~O_NONBLOCK;
+
+FPUTOUT;
+ return err;
+}
+
+static struct super_operations msg_s_ops = {
+ .statfs = simple_statfs,
+ .drop_inode = generic_delete_inode,
+ .delete_inode = mqueue_release,
+};
+
+static int
+msg_fill_super(struct super_block *sb, void *data, int silent)
+{
+ struct inode *root;
+ struct dentry *root_dentry;
+ static const struct qstr this={ .name="msg:", .len=4, .hash=0 };
+
+ sb->s_blocksize = PAGE_CACHE_SIZE;
+ sb->s_blocksize_bits = PAGE_CACHE_SHIFT;
+ sb->s_magic = MSGFS_MAGIC;
+ sb->s_op = &msg_s_ops;
+
+ root = get_msg_inode(sb, S_IFDIR | S_IRWXUGO | S_ISVTX);
+ if (!root)
+ goto out;
+ root_dentry = d_alloc(NULL, &this);
+ if (!root_dentry)
+ goto out_iput;
+ sb->s_root = root_dentry;
+ sb->s_root->d_sb = sb;
+ sb->s_root->d_parent = sb->s_root;
+ d_instantiate(sb->s_root, root);
+ return 0;
+
+out_iput:
+ iput(root);
+out:
+ return -ENOMEM;
+}
+
+static struct super_block *
+msgfs_get_sb(struct file_system_type *fs_type,
+ int flags, const char *dev_name, void *data)
+{
+ return get_sb_single(fs_type, flags, data, msg_fill_super);
+}
+
+static struct file_system_type msg_fs_type = {
+ .name = "msgfs",
+ .get_sb = msgfs_get_sb,
+ .kill_sb = kill_anon_super,
+};
+
+static int __init
+mqueue_init(void)
+{
+ int err = register_filesystem(&msg_fs_type);
+
+ if (!err){
+ if (IS_ERR(msg_mnt = kern_mount(&msg_fs_type))){
+ unregister_filesystem(&msg_fs_type);
+ err= PTR_ERR(msg_mnt);
+ } else
+ err=0;
+ }
+
+ return err;
+}
+
+__initcall(mqueue_init);
diff -X dontdiff -Nur vanilla-2.6.0-test6/MAINTAINERS linux-2.6.0-test6/MAINTAINERS
--- vanilla-2.6.0-test6/MAINTAINERS 2003-10-01 20:10:36.000000000 +0200
+++ linux-2.6.0-test6/MAINTAINERS 2003-10-03 13:40:59.000000000 +0200
@@ -1558,6 +1558,11 @@
M: ambx1@neo.rr.com
S: Maintained

+POSIXMSG message queues
+P: Peter Wächtler
+M: pwaechtler@mac.com
+S: Maintained
+
PPP PROTOCOL DRIVERS AND COMPRESSORS
P: Paul Mackerras
M: paulus@samba.org
diff -X dontdiff -Nur vanilla-2.6.0-test6/ipc/util.c linux-2.6.0-test6/ipc/util.c
--- vanilla-2.6.0-test6/ipc/util.c 2003-10-03 17:17:58.000000000 +0200
+++ linux-2.6.0-test6/ipc/util.c 2003-10-03 14:58:35.000000000 +0200
@@ -24,6 +24,7 @@
#include <linux/security.h>
#include <linux/rcupdate.h>
#include <linux/workqueue.h>
+#include <linux/pxqueue.h>

#if defined(CONFIG_SYSVIPC)

@@ -731,3 +732,50 @@
}

#endif /* CONFIG_SYSVIPC */
+
+#if !defined(CONFIG_POSIXMSG)
+/*
+ * Dummy functions when POSIXMSG isn't configured
+ */
+
+asmlinkage mqd_t sys_mq_open(const char *u_path, int oflag, mode_t mode,
+ struct mq_attr *u_attr)
+{
+ return (mqd_t) -ENOSYS;
+}
+
+asmlinkage int sys_mq_unlink(const char *u_name)
+{
+ return -ENOSYS;
+}
+
+asmlinkage int sys_mq_timedsend(mqd_t mqdes, const char *msg_ptr,
+ size_t msg_len, unsigned int msg_prio, struct timespec *utime)
+{
+ return -ENOSYS;
+}
+
+asmlinkage ssize_t sys_mq_timedreceive(mqd_t mqdes, char *msg_ptr,
+ size_t msg_len, unsigned int *msg_prio, struct timespec *utime)
+{
+ return -ENOSYS;
+}
+
+asmlinkage int sys_mq_notify(mqd_t mqdes,
+ const struct sigevent *u_notification)
+{
+ return -ENOSYS;
+}
+
+asmlinkage int sys_mq_getattr(mqd_t mqdes, struct mq_attr *u_mqstat)
+{
+ return -ENOSYS;
+}
+
+asmlinkage int sys_mq_setattr(mqd_t mqdes, const struct mq_attr *u_mqstat,
+ struct mq_attr *u_omqstat)
+{
+ return -ENOSYS;
+}
+
+#endif /* CONFIG_POSIXMSG */
\
 
 \ /
  Last update: 2005-03-22 13:49    [W:0.240 / U:0.220 seconds]
©2003-2014 Jasper Spaans. Advertise on this site