rewrite posix semaphore and mqueue implementation.

git-svn-id: https://rt-thread.googlecode.com/svn/trunk@1103 bbd45198-f89e-11dd-88c7-29a3b14d5316

1 files changed, 199 insertions, 25 deletions

diff --git a/components/pthreads/mqueue.c b/components/pthreads/mqueue.c
index a3cc2c7c73..bd31b4450a 100644
--- a/components/pthreads/mqueue.c
+++ b/components/pthreads/mqueue.c
@@ -5,6 +5,65 @@
 #include <errno.h>

 #include <sys/fcntl.h>

 

+static posix_mq_t* posix_mq_list = RT_NULL;

+static struct rt_semaphore posix_mq_lock;

+void posix_mq_system_init()

+{

+	rt_sem_init(&posix_mq_lock, "pmq", 1, RT_IPC_FLAG_FIFO);

+}

+

+rt_inline void posix_mq_insert(posix_mq_t *pmq)

+{

+	pmq->next = posix_mq_list;

+	posix_mq_list = pmq;

+}

+

+static void posix_mq_delete(posix_mq_t *pmq)

+{

+	posix_mq_t *iter;

+	if (posix_mq_list == pmq)

+	{

+		posix_mq_list = pmq->next;

+

+		rt_mq_delete(pmq->mq);

+		rt_free(pmq);

+

+		return;

+	}

+	for (iter = posix_mq_list; iter->next != RT_NULL; iter = iter->next)

+	{

+		if (iter->next == pmq)

+		{

+			/* delete this mq */

+			if (pmq->next != RT_NULL)

+				iter->next = pmq->next;

+			else

+				iter->next = RT_NULL;

+

+			/* delete RT-Thread mqueue */

+			rt_mq_delete(pmq->mq);

+			rt_free(pmq);

+			return ;

+		}

+	}

+}

+

+static posix_mq_t *posix_mq_find(const char* name)

+{

+	posix_mq_t *iter;

+	rt_object_t object;

+

+	for (iter = posix_mq_list; iter != RT_NULL; iter = iter->next)

+	{

+		object = (rt_object_t)&(iter->mq);

+

+		if (strncmp(object->name, name, RT_NAME_MAX) == 0)

+		{

+			return iter;

+		}

+	}

+}

+

 int mq_setattr(mqd_t mqdes, const struct mq_attr *mqstat,

 		struct mq_attr *omqstat)

 {

@@ -20,8 +79,8 @@ int mq_getattr(mqd_t mqdes, struct mq_attr *mqstat)
 		return -1;

 	}

 

-	mqstat->mq_maxmsg = mqdes->max_msgs;

-	mqstat->mq_msgsize = mqdes->msg_size;

+	mqstat->mq_maxmsg = mqdes->mq->mq->max_msgs;

+	mqstat->mq_msgsize = mqdes->mq->mq->msg_size;

 	mqstat->mq_curmsgs = 0;

 	mqstat->mq_flags = 0;

 

@@ -30,11 +89,15 @@ int mq_getattr(mqd_t mqdes, struct mq_attr *mqstat)
 

 mqd_t mq_open(const char *name, int oflag, ...)

 {

-	rt_mq_t mq;

+	mqd_t mqdes;

 	va_list arg;

 	mode_t mode;

 	struct mq_attr *attr = RT_NULL;

 

+    /* lock posix mqueue list */

+    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);

+

+    mqdes = RT_NULL;

 	if (oflag & O_CREAT)

 	{

 	    va_start(arg, oflag);

@@ -42,21 +105,82 @@ mqd_t mq_open(const char *name, int oflag, ...)
 	    attr = (struct mq_attr *) va_arg(arg, struct mq_attr *);

 	    va_end(arg);

 

-		mq = rt_mq_create(name, attr->mq_msgsize, attr->mq_maxmsg, RT_IPC_FLAG_FIFO);

-		if (mq == RT_NULL) /* create failed */

+	    if (oflag & O_EXCL)

+	    {

+	    	if (posix_mq_find(name) != RT_NULL)

+	    	{

+	    		rt_set_errno(EEXIST);

+	    		goto __return;

+	    	}

+	    }

+	    mqdes = (mqd_t) rt_malloc (sizeof(struct mqdes));

+	    if (mqdes == RT_NULL)

+	    {

+	    	rt_set_errno(ENFILE);

+	    	goto __return;

+	    }

+

+	    mqdes->flags = oflag;

+	    mqdes->mq = (posix_mq_t*) rt_malloc (sizeof(posix_mq_t));

+	    if (mqdes->mq == RT_NULL)

+	    {

+	    	rt_set_errno(ENFILE);

+	    	goto __return;

+	    }

+

+	    /* create RT-Thread message queue */

+		mqdes->mq->mq = rt_mq_create(name, attr->mq_msgsize, attr->mq_maxmsg, RT_IPC_FLAG_FIFO);

+		if (mqdes->mq->mq == RT_NULL) /* create failed */

 		{

 			rt_set_errno(ENFILE);

-			return RT_NULL;

+			goto __return;

 		}

-	}

+		/* initialize reference count */

+		mqdes->mq->refcount = 1;

+		mqdes->mq->unlinked = 0;

 

-	if (oflag & O_EXCL)

+		/* insert mq to posix mq list */

+		posix_mq_insert(mqdes->mq);

+	}

+	else

 	{

-		mq = (rt_mq_t)rt_object_find(name, RT_Object_Class_MessageQueue);

-		if (mq == RT_NULL) rt_set_errno(ENOSPC);

+		posix_mq_t *mq;

+

+		/* find mqueue */

+		mq = posix_mq_find(name);

+		if (mq != RT_NULL)

+		{

+			mqdes = (mqd_t) rt_malloc (sizeof(struct mqdes));

+			mqdes->mq = mq;

+			mqdes->flags = oflag;

+			mq->refcount ++; /* increase reference count */

+		}

+		else

+		{

+			rt_set_errno(ENOENT);

+			goto __return;

+		}

 	}

+	rt_sem_release(&posix_mq_lock);

+	return mqdes;

 

-	return mq;

+__return:

+	/* release lock */

+	rt_sem_release(&posix_mq_lock);

+

+	/* release allocated memory */

+	if (mqdes != RT_NULL)

+	{

+		if (mqdes->mq != RT_NULL)

+		{

+			/* delete RT-Thread message queue */

+			if (mqdes->mq->mq != RT_NULL)

+				rt_mq_delete(mqdes->mq->mq);

+			rt_free(mqdes->mq);

+		}

+		rt_free(mqdes);

+	}

+	return RT_NULL;

 }

 

 ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned *msg_prio)

@@ -69,7 +193,14 @@ ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned *msg_pri
 		return -1;

 	}

 

-	result = rt_mq_recv(mqdes, msg_ptr, msg_len, RT_WAITING_FOREVER);

+	/* permission check */

+	if (!(mqdes->flags & O_RDONLY))

+	{

+		rt_set_errno(EBADF);

+		return -1;

+	}

+

+	result = rt_mq_recv(mqdes->mq->mq, msg_ptr, msg_len, RT_WAITING_FOREVER);

 	if (result == RT_EOK)

 		return msg_len;

 

@@ -87,7 +218,14 @@ int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned msg_prio)
 		return -1;

 	}

 

-	result = rt_mq_send(mqdes, msg_ptr, msg_len);

+	/* permission check */

+	if (!(mqdes->flags & O_WRONLY))

+	{

+		rt_set_errno(EBADF);

+		return -1;

+	}

+

+	result = rt_mq_send(mqdes->mq->mq, (void*)msg_ptr, msg_len);

 	if (result == RT_EOK)

 		return 0;

 

@@ -101,14 +239,22 @@ ssize_t mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
 	int tick;

 	rt_err_t result;

 

+	/* parameters check */

 	if ((mqdes == RT_NULL) || (msg_ptr == RT_NULL))

 	{

 		rt_set_errno(EINVAL);

 		return -1;

 	}

+	/* permission check */

+	if (!(mqdes->flags & O_RDONLY))

+	{

+		rt_set_errno(EBADF);

+		return -1;

+	}

+

 	tick = libc_time_to_tick(abs_timeout);

 

-	result = rt_mq_recv(mqdes, msg_ptr, msg_len, tick);

+	result = rt_mq_recv(mqdes->mq->mq, msg_ptr, msg_len, tick);

 	if (result == RT_EOK) return msg_len;

 

 	if (result == -RT_ETIMEOUT)

@@ -134,21 +280,49 @@ int mq_notify(mqd_t mqdes, const struct sigevent *notification)
 

 int mq_close(mqd_t mqdes)

 {

-	return 0;

+	if (mqdes == RT_NULL)

+	{

+		rt_set_errno(EINVAL);

+		return -1;

+	}

+

+    /* lock posix mqueue list */

+    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);

+    mqdes->mq->refcount --;

+    if (mqdes->mq->refcount == 0)

+    {

+    	/* delete from posix mqueue list */

+    	if (mqdes->mq->unlinked)

+    		posix_mq_delete(mqdes->mq);

+    	mqdes->mq = RT_NULL;

+    }

+    rt_sem_release(&posix_mq_lock);

+

+    rt_free(mqdes);

+    return 0;

 }

 

 int mq_unlink(const char *name)

 {

-	rt_mq_t mq;

+	posix_mq_t *pmq;

 

-	mq = (rt_mq_t)rt_object_find(name, RT_Object_Class_MessageQueue);

-	if (mq == RT_NULL)

-	{

-		rt_set_errno(ENOENT);

-		return -1;

-	}

+    /* lock posix mqueue list */

+    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);

+    pmq = posix_mq_find(name);

+    if (pmq != RT_NULL)

+    {

+    	pmq->unlinked = 1;

+    	if (pmq->refcount == 0)

+    	{

+    		/* remove this mqueue */

+    		posix_mq_delete(pmq);

+    	}

+        rt_sem_release(&posix_mq_lock);

+        return 0;

+    }

+    rt_sem_release(&posix_mq_lock);

 

-	/* delete this message queue */

-	rt_mq_delete(mq);

-	return 0;

+    /* no this entry */

+    rt_set_errno(ENOENT);

+    return -1;

 }