[posix] posix 的 mqueue 对接到文件系统的 fd (#7768)

components/dfs/Kconfig

@@ -173,6 +173,11 @@ endif
         bool "Enable TMP file system"
         default n
 
+    config RT_USING_DFS_MQUEUE
+        bool "Enable MQUEUE file system"
+        select RT_USING_DEV_BUS
+        default n
+
 if RT_USING_DFS_V1
     config RT_USING_DFS_NFS
         bool "Using NFS v3 client file system"

components/dfs/dfs_v1/filesystems/mqueue/SConscript

@@ -0,0 +1,11 @@
+# RT-Thread building script for component
+
+from building import *
+
+cwd = GetCurrentDir()
+src = Glob('*.c')
+CPPPATH = [cwd]
+
+group = DefineGroup('Filesystem', src, depend = ['RT_USING_DFS', 'RT_USING_DFS_MQUEUE'], CPPPATH = CPPPATH)
+
+Return('group')

components/dfs/dfs_v1/filesystems/mqueue/dfs_mqueue.c

@@ -0,0 +1,165 @@
+/*
+ * Copyright (c) 2006-2023, RT-Thread Development Team
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Change Logs:
+ * Date           Author       Notes
+ * 2023-07-04     zhkag        first Version
+ */
+
+#include <rtthread.h>
+#include <rthw.h>
+#include <dfs_file.h>
+#include "dfs_mqueue.h"
+
+static rt_list_t _mqueue_file_list = RT_LIST_OBJECT_INIT(_mqueue_file_list);
+struct rt_spinlock mqueue_lock;
+
+void dfs_mqueue_insert_after(rt_list_t *n) {
+    rt_spin_lock(&mqueue_lock);
+    rt_list_insert_after(&(_mqueue_file_list), n);
+    rt_spin_unlock(&mqueue_lock);
+}
+
+struct mqueue_file *dfs_mqueue_lookup(const char *path, rt_size_t *size) {
+    struct mqueue_file *file;
+    rt_list_t *node;
+    rt_spin_lock(&mqueue_lock);
+    rt_list_for_each(node, &_mqueue_file_list) {
+        file = rt_list_entry(node, struct mqueue_file, list);
+
+        if (rt_strncmp(file->name, path, RT_NAME_MAX) == 0) {
+            *size = file->size;
+            rt_spin_unlock(&mqueue_lock);
+            return file;
+        }
+    }
+    rt_spin_unlock(&mqueue_lock);
+    return RT_NULL;
+}
+
+int dfs_mqueue_mount(struct dfs_filesystem *fs, unsigned long rwflag, const void *data) {
+    return RT_EOK;
+}
+
+int dfs_mqueue_unmount(struct dfs_filesystem *fs) { return RT_EOK; }
+
+int dfs_mqueue_statfs(struct dfs_filesystem *fs, struct statfs *buf) { return RT_EOK; }
+
+int dfs_mqueue_close(struct dfs_file *file) { return RT_EOK; }
+
+int dfs_mqueue_open(struct dfs_file *file) {
+    rt_size_t size;
+    if ((file->vnode->path[0] == '/') && (file->vnode->path[1] == '\0'))
+        return 0;
+
+    if (file->flags & O_DIRECTORY)
+        return -ENOENT;
+
+    struct mqueue_file *mq_file;
+
+    mq_file = dfs_mqueue_lookup(file->vnode->path + 1, &size);
+    if (mq_file == RT_NULL && !(file->flags & O_CREAT))
+        return -ENOENT;
+    if (mq_file == RT_NULL) {
+        mq_file = (struct mqueue_file *)rt_malloc(sizeof(struct mqueue_file));
+        if (mq_file == RT_NULL) {
+            return -ENFILE;
+        }
+        mq_file->msg_size = 8192;
+        mq_file->max_msgs = 10;
+        strncpy(mq_file->name, file->vnode->path + 1, RT_NAME_MAX);
+        dfs_mqueue_insert_after(&(mq_file->list));
+    }
+
+    if (file->flags & O_CREAT) {
+        rt_mq_t mq = rt_mq_create(file->vnode->path + 1, mq_file->msg_size, mq_file->max_msgs,
+                                  RT_IPC_FLAG_FIFO);
+        mq_file->data = (void *)mq;
+        file->vnode->data = mq_file;
+        file->vnode->size = 0;
+    }
+
+    return 0;
+}
+
+int dfs_mqueue_stat(struct dfs_filesystem *fs, const char *path, struct stat *st) {
+    st->st_dev = 0;
+    st->st_mode = S_IFREG | S_IRUSR | S_IRGRP | S_IROTH | S_IWUSR | S_IWGRP | S_IWOTH;
+    st->st_size = 0;
+    st->st_mtime = 0;
+    return RT_EOK;
+}
+
+int dfs_mqueue_getdents(struct dfs_file *file, struct dirent *dirp, uint32_t count) {
+    rt_size_t index, end;
+    struct dirent *d;
+    count = (count / sizeof(struct dirent));
+    end = file->pos + count;
+    index = 0;
+    count = 0;
+    struct mqueue_file *mq_file;
+    rt_list_t *node;
+    rt_spin_lock(&mqueue_lock);
+    rt_list_for_each(node, &_mqueue_file_list) {
+        if (index >= (rt_size_t)file->pos) {
+            mq_file = rt_list_entry(node, struct mqueue_file, list);
+            d = dirp + count;
+            d->d_namlen = RT_NAME_MAX;
+            d->d_reclen = (rt_uint16_t)sizeof(struct dirent);
+            rt_strncpy(d->d_name, mq_file->name, RT_NAME_MAX);
+            count += 1;
+            file->pos += 1;
+        }
+        index += 1;
+        if (index >= end) {
+            break;
+        }
+    }
+    rt_spin_unlock(&mqueue_lock);
+    return count * sizeof(struct dirent);
+}
+
+int dfs_mqueue_unlink(struct dfs_filesystem *fs, const char *path) {
+    rt_size_t size;
+    struct mqueue_file *mq_file;
+    mq_file = dfs_mqueue_lookup(path + 1, &size);
+    if (mq_file == RT_NULL)
+        return -ENOENT;
+    rt_list_remove(&(mq_file->list));
+    if (mq_file->data != RT_NULL)
+        rt_mq_delete((rt_mq_t)mq_file->data);
+    rt_free(mq_file);
+    return RT_EOK;
+}
+
+static const struct dfs_file_ops _mqueue_fops = {
+    .open = dfs_mqueue_open,
+    .close = dfs_mqueue_close,
+    .getdents = dfs_mqueue_getdents,
+};
+
+static const struct dfs_filesystem_ops _mqueue = {
+    .name = "mqueue",
+    .flags = DFS_FS_FLAG_DEFAULT,
+    .fops = &_mqueue_fops,
+
+    .mount = dfs_mqueue_mount,
+    .unmount = dfs_mqueue_unmount,
+    .statfs = dfs_mqueue_statfs,
+
+    .unlink = dfs_mqueue_unlink,
+    .stat = dfs_mqueue_stat,
+};
+
+int dfs_mqueue_init(void) {
+    /* register mqueue file system */
+    dfs_register(&_mqueue);
+    mkdir("/dev/mqueue", 0x777);
+    if (dfs_mount(RT_NULL, "/dev/mqueue", "mqueue", 0, 0) != 0) {
+        rt_kprintf("Dir /dev/mqueue mount failed!\n");
+    }
+    return 0;
+}
+INIT_COMPONENT_EXPORT(dfs_mqueue_init);

components/dfs/dfs_v1/filesystems/mqueue/dfs_mqueue.h

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2006-2023, RT-Thread Development Team
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Change Logs:
+ * Date           Author       Notes
+ * 2023-07-04     zhkag        first Version
+ */
+
+#ifndef __DFS_MQUEUE_H__
+#define __DFS_MQUEUE_H__
+
+#include <rtthread.h>
+
+struct mqueue_file {
+    char name[RT_NAME_MAX]; /* file name */
+    rt_uint16_t msg_size;   /**< message size of each message */
+    rt_uint16_t max_msgs;   /**< max number of messages */
+    rt_list_t list;
+    rt_uint8_t *data; /* file date ptr */
+    rt_size_t size;   /* file size */
+};
+
+struct mqueue_file *dfs_mqueue_lookup(const char *path, rt_size_t *size);
+void dfs_mqueue_insert_after(rt_list_t *n);
+
+#endif

components/dfs/dfs_v2/filesystems/mqueue/SConscript

@@ -0,0 +1,11 @@
+# RT-Thread building script for component
+
+from building import *
+
+cwd = GetCurrentDir()
+src = Glob('*.c')
+CPPPATH = [cwd]
+
+group = DefineGroup('Filesystem', src, depend = ['RT_USING_DFS', 'RT_USING_DFS_MQUEUE'], CPPPATH = CPPPATH)
+
+Return('group')

components/dfs/dfs_v2/filesystems/mqueue/dfs_mqueue.c

@@ -0,0 +1,240 @@
+/*
+ * Copyright (c) 2006-2023, RT-Thread Development Team
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Change Logs:
+ * Date           Author       Notes
+ * 2023-07-04     zhkag        first Version
+ */
+
+#include <rtthread.h>
+#include <rthw.h>
+#include <dfs.h>
+#include <dfs_fs.h>
+#include <dfs_file.h>
+#include <dfs_dentry.h>
+#include <dfs_mnt.h>
+#include "dfs_mqueue.h"
+
+static rt_list_t _mqueue_file_list = RT_LIST_OBJECT_INIT(_mqueue_file_list);
+struct rt_spinlock mqueue_lock;
+
+void dfs_mqueue_insert_after(rt_list_t *n) {
+    rt_spin_lock(&mqueue_lock);
+    rt_list_insert_after(&(_mqueue_file_list), n);
+    rt_spin_unlock(&mqueue_lock);
+}
+
+struct mqueue_file *dfs_mqueue_lookup(const char *path, rt_size_t *size) {
+    struct mqueue_file *file;
+    rt_list_t *node;
+    rt_spin_lock(&mqueue_lock);
+    rt_list_for_each(node, &_mqueue_file_list) {
+        file = rt_list_entry(node, struct mqueue_file, list);
+
+        if (rt_strncmp(file->name, path, RT_NAME_MAX) == 0) {
+            *size = file->size;
+            rt_spin_unlock(&mqueue_lock);
+            return file;
+        }
+    }
+    rt_spin_unlock(&mqueue_lock);
+    return RT_NULL;
+}
+
+int dfs_mqueue_mount(struct dfs_filesystem *fs, unsigned long rwflag, const void *data) {
+    return RT_EOK;
+}
+
+int dfs_mqueue_umount(struct dfs_filesystem *fs) { return RT_EOK; }
+
+int dfs_mqueue_statfs(struct dfs_filesystem *fs, struct statfs *buf) { return RT_EOK; }
+
+int dfs_mqueue_close(struct dfs_file *file) { return RT_EOK; }
+
+int dfs_mqueue_open(struct dfs_file *file) {
+    rt_size_t size;
+    if ((file->dentry->pathname[0] == '/') && (file->dentry->pathname[1] == '\0'))
+        return 0;
+
+}
+
+int dfs_mqueue_stat(struct dfs_dentry *dentry, struct stat *st) {
+    const char *path = RT_NULL;
+    struct dfs_vnode *vnode = RT_NULL;
+    if (dentry && dentry->vnode) {
+        path = dentry->pathname;
+        vnode = dentry->vnode;
+        st->st_dev = 0;
+        st->st_gid = vnode->gid;
+        st->st_uid = vnode->uid;
+        st->st_ino = 0;
+        st->st_mode = vnode->mode;
+        st->st_nlink = vnode->nlink;
+        st->st_size = vnode->size;
+        st->st_mtim.tv_nsec = vnode->mtime.tv_nsec;
+        st->st_mtim.tv_sec = vnode->mtime.tv_sec;
+        st->st_ctim.tv_nsec = vnode->ctime.tv_nsec;
+        st->st_ctim.tv_sec = vnode->ctime.tv_sec;
+        st->st_atim.tv_nsec = vnode->atime.tv_nsec;
+        st->st_atim.tv_sec = vnode->atime.tv_sec;
+    }
+    return RT_EOK;
+}
+
+int dfs_mqueue_getdents(struct dfs_file *file, struct dirent *dirp, uint32_t count) {
+    rt_size_t index, end;
+    struct dirent *d;
+    count = (count / sizeof(struct dirent));
+    end = file->fpos + count;
+    index = 0;
+    count = 0;
+    struct mqueue_file *mq_file;
+    rt_list_t *node;
+    rt_spin_lock(&mqueue_lock);
+    rt_list_for_each(node, &_mqueue_file_list) {
+        if (index >= (rt_size_t)file->fpos) {
+            mq_file = rt_list_entry(node, struct mqueue_file, list);
+            d = dirp + count;
+            d->d_namlen = RT_NAME_MAX;
+            d->d_reclen = (rt_uint16_t)sizeof(struct dirent);
+            rt_strncpy(d->d_name, mq_file->name, RT_NAME_MAX);
+            count += 1;
+            file->fpos += 1;
+        }
+        index += 1;
+        if (index >= end) {
+            break;
+        }
+    }
+    rt_spin_unlock(&mqueue_lock);
+    return count * sizeof(struct dirent);
+}
+
+int dfs_mqueue_unlink(struct dfs_dentry *dentry) {
+    rt_size_t size;
+    struct mqueue_file *mq_file;
+    mq_file = dfs_mqueue_lookup(dentry->pathname + 1, &size);
+    if (mq_file == RT_NULL)
+        return -ENOENT;
+    rt_list_remove(&(mq_file->list));
+    if (mq_file->data != RT_NULL)
+        rt_mq_delete((rt_mq_t)mq_file->data);
+    rt_free(mq_file);
+    return RT_EOK;
+}
+
+static struct dfs_vnode *dfs_mqueue_create_vnode(struct dfs_dentry *dentry, int type, mode_t mode) {
+    struct dfs_vnode *vnode = RT_NULL;
+    rt_size_t size;
+    struct mqueue_file *mq_file;
+
+    if (dentry == NULL || dentry->mnt == NULL) {
+        return NULL;
+    }
+
+    vnode = dfs_vnode_create();
+    if (vnode) {
+        mq_file = dfs_mqueue_lookup(dentry->pathname + 1, &size);
+        if (mq_file == RT_NULL) {
+            mq_file = (struct mqueue_file *)rt_malloc(sizeof(struct mqueue_file));
+            if (mq_file == RT_NULL) {
+                return -ENFILE;
+            }
+            mq_file->msg_size = 8192;
+            mq_file->max_msgs = 10;
+            strncpy(mq_file->name, dentry->pathname + 1, RT_NAME_MAX);
+            dfs_mqueue_insert_after(&(mq_file->list));
+        }
+
+        vnode->mode = S_IFREG | mode;
+        vnode->type = FT_REGULAR;
+        rt_mq_t mq = rt_mq_create(dentry->pathname + 1, mq_file->msg_size, mq_file->max_msgs,
+                                  RT_IPC_FLAG_FIFO);
+        mq_file->data = (void *)mq;
+        vnode->data = mq_file;
+        vnode->size = 0;
+    }
+
+    return vnode;
+}
+
+static int dfs_mqueue_free_vnode(struct dfs_vnode *vnode) {
+    /* nothing to be freed */
+    if (vnode && vnode->ref_count <= 1) {
+        vnode->data = NULL;
+    }
+    return 0;
+}
+
+static const struct dfs_file_ops _mqueue_fops = {
+    .open = dfs_mqueue_open,
+    .close = dfs_mqueue_close,
+    .getdents = dfs_mqueue_getdents,
+};
+
+struct dfs_vnode *_dfs_mqueue_lookup(struct dfs_dentry *dentry) {
+    struct dfs_vnode *vnode = RT_NULL;
+    rt_size_t size;
+    // struct tmpfs_sb *superblock;
+    struct mqueue_file *mq_file;
+
+    if (dentry == NULL || dentry->mnt == NULL) {
+        return NULL;
+    }
+
+    if (dentry->pathname[0] == '/' && dentry->pathname[1] == '\0') {
+    }
+
+    mq_file = dfs_mqueue_lookup(dentry->pathname + 1, &size);
+
+    vnode = dfs_vnode_create();
+    if (mq_file && mq_file->data) {
+        vnode->mode = S_IFREG | S_IRUSR | S_IWUSR | S_IXUSR;
+        vnode->type = FT_REGULAR;
+        vnode->mnt = dentry->mnt;
+        vnode->data = mq_file;
+        vnode->size = mq_file->size;
+    } else {
+        vnode->size = 0;
+        vnode->nlink = 1;
+        vnode->fops = &_mqueue_fops;
+        vnode->mnt = dentry->mnt;
+        vnode->type = FT_DIRECTORY;
+        vnode->mode = S_IFDIR | S_IRUSR | S_IWUSR | S_IXUSR;
+    }
+    return vnode;
+}
+
+static const struct dfs_filesystem_ops _mqueue_ops = {
+    .name = "mqueue",
+    .flags = DFS_FS_FLAG_DEFAULT,
+    .default_fops = &_mqueue_fops,
+
+    .mount = dfs_mqueue_mount,
+    .umount = dfs_mqueue_umount,
+    .statfs = dfs_mqueue_statfs,
+
+    .unlink = dfs_mqueue_unlink,
+    .stat = dfs_mqueue_stat,
+
+    .lookup = _dfs_mqueue_lookup,
+    .create_vnode = dfs_mqueue_create_vnode,
+    .free_vnode = dfs_mqueue_free_vnode
+};
+
+static struct dfs_filesystem_type _mqueue = {
+    .fs_ops = &_mqueue_ops,
+};
+
+int dfs_mqueue_init(void) {
+    /* register mqueue file system */
+    dfs_register(&_mqueue);
+    mkdir("/dev/mqueue", 0x777);
+    if (dfs_mount(RT_NULL, "/dev/mqueue", "mqueue", 0, 0) != 0) {
+        rt_kprintf("Dir /dev/mqueue mount failed!\n");
+    }
+    return 0;
+}
+INIT_COMPONENT_EXPORT(dfs_mqueue_init);

components/dfs/dfs_v2/filesystems/mqueue/dfs_mqueue.h

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2006-2023, RT-Thread Development Team
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Change Logs:
+ * Date           Author       Notes
+ * 2023-07-04     zhkag        first Version
+ */
+
+#ifndef __DFS_MQUEUE_H__
+#define __DFS_MQUEUE_H__
+
+#include <rtthread.h>
+
+struct mqueue_file {
+    char name[RT_NAME_MAX]; /* file name */
+    rt_uint16_t msg_size;   /**< message size of each message */
+    rt_uint16_t max_msgs;   /**< max number of messages */
+    rt_list_t list;
+    rt_uint8_t *data; /* file date ptr */
+    rt_size_t size;   /* file size */
+};
+
+struct mqueue_file *dfs_mqueue_lookup(const char *path, rt_size_t *size);
+void dfs_mqueue_insert_after(rt_list_t *n);
+
+#endif

components/libc/posix/ipc/Kconfig

@@ -26,6 +26,7 @@ config RT_USING_POSIX_MESSAGE_QUEUE
     bool "Enable posix message queue <mqueue.h>"
     select RT_USING_POSIX_CLOCK
     select RT_USING_MESSAGEQUEUE_PRIORITY
+    select RT_USING_DFS_MQUEUE
     default n
 
 config RT_USING_POSIX_MESSAGE_SEMAPHORE

components/libc/posix/ipc/SConscript

@@ -9,7 +9,7 @@ inc        = [cwd]
 #     src += Glob('system-v/*.c')
 #     inc += [cwd + '/system-v']
 
-if GetDepend('RT_USING_POSIX_MESSAGE_QUEUE'):
+if GetDepend(['RT_USING_POSIX_MESSAGE_QUEUE', 'RT_USING_DFS_MQUEUE']):
     src += ['mqueue.c']
 
 if GetDepend('RT_USING_POSIX_MESSAGE_SEMAPHORE'):

components/libc/posix/ipc/mqueue.c

@@ -7,93 +7,10 @@
  * Date           Author       Notes
  */
 
-#include <string.h>
-#include <fcntl.h>
-#include <sys/signal.h>
-#include <sys/time.h>
-#include <sys/errno.h>
-#include <rtthread.h>
-#include <limits.h>
+#include <dfs_file.h>
+#include <unistd.h>
 #include "mqueue.h"
 
-static mqdes_t posix_mq_list = RT_NULL;
-static struct rt_semaphore posix_mq_lock;
-
-/* initialize posix mqueue */
-static int posix_mq_system_init(void)
-{
-    rt_sem_init(&posix_mq_lock, "pmq", 1, RT_IPC_FLAG_FIFO);
-    return 0;
-}
-INIT_COMPONENT_EXPORT(posix_mq_system_init);
-
-rt_inline void posix_mq_insert(mqdes_t pmq)
-{
-    if (posix_mq_list == RT_NULL)
-        pmq->mq_id = 1;
-    else
-        pmq->mq_id = posix_mq_list->mq_id + 1;
-    pmq->next = posix_mq_list;
-    posix_mq_list = pmq;
-}
-
-static void posix_mq_delete(mqdes_t pmq)
-{
-    mqdes_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 mqdes_t posix_mq_find(const char *name)
-{
-    mqdes_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;
-        }
-    }
-
-    return RT_NULL;
-}
-
-static mqdes_t posix_mq_id_find(mqd_t id)
-{
-    for (mqdes_t iter = posix_mq_list; iter != RT_NULL; iter = iter->next)
-        if (iter->mq_id == id)
-            return iter;
-    return RT_NULL;
-}
-
 int mq_setattr(mqd_t                 id,
                const struct mq_attr *mqstat,
                struct mq_attr       *omqstat)
@@ -109,18 +26,18 @@ RTM_EXPORT(mq_setattr);
 
 int mq_getattr(mqd_t id, struct mq_attr *mqstat)
 {
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
-    mqdes_t mqdes = posix_mq_id_find(id);
-    rt_sem_release(&posix_mq_lock);
-    if ((mqdes == RT_NULL) || mqstat == RT_NULL)
+    rt_mq_t mq;
+    struct mqueue_file *mq_file;
+    mq_file = fd_get(id)->vnode->data;
+    mq = (rt_mq_t)mq_file->data;
+    if ((mq == RT_NULL) || mqstat == RT_NULL)
     {
         rt_set_errno(EBADF);
-
         return -1;
     }
 
-    mqstat->mq_maxmsg = mqdes->mq->max_msgs;
-    mqstat->mq_msgsize = mqdes->mq->msg_size;
+    mqstat->mq_maxmsg = mq->max_msgs;
+    mqstat->mq_msgsize = mq->msg_size;
     mqstat->mq_curmsgs = 0;
     mqstat->mq_flags = 0;
 
@@ -130,109 +47,88 @@ RTM_EXPORT(mq_getattr);
 
 mqd_t mq_open(const char *name, int oflag, ...)
 {
+    int mq_fd;
     va_list arg;
     mode_t mode;
-    mqdes_t mqdes = RT_NULL;
     struct mq_attr *attr = RT_NULL;
+    va_start(arg, oflag);
+    mode = (mode_t)va_arg(arg, unsigned int);
+    mode = (mode_t)mode; /* self-assignment avoids compiler optimization */
+    attr = (struct mq_attr *)va_arg(arg, struct mq_attr *);
+    attr = (struct mq_attr *)attr; /* self-assignment avoids compiler optimization */
+    va_end(arg);
+    if(*name == '/')
+    {
+        name++;
+    }
 
-    /* lock posix mqueue list */
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
     int len = rt_strlen(name);
     if (len > RT_NAME_MAX)
     {
         rt_set_errno(ENAMETOOLONG);
-        goto __return;
+        return (mqd_t)(-1);
     }
-
-    mqdes = posix_mq_find(name);
-    if (mqdes != RT_NULL)
+    rt_size_t size;
+    struct mqueue_file *mq_file;
+    mq_file = dfs_mqueue_lookup(name, &size);
+    if(mq_file != RT_NULL)
     {
         if (oflag & O_CREAT && oflag & O_EXCL)
         {
             rt_set_errno(EEXIST);
-            rt_sem_release(&posix_mq_lock);
             return (mqd_t)(-1);
         }
-        mqdes->refcount++; /* increase reference count */
     }
     else if (oflag & O_CREAT)
     {
-        va_start(arg, oflag);
-        mode = (mode_t)va_arg(arg, unsigned int);
-        mode = (mode_t)mode; /* self-assignment avoids compiler optimization */
-        attr = (struct mq_attr *)va_arg(arg, struct mq_attr *);
-        attr = (struct mq_attr *)attr; /* self-assignment avoids compiler optimization */
-        va_end(arg);
-
         if (attr->mq_maxmsg <= 0)
         {
             rt_set_errno(EINVAL);
-            goto __return;
-        }
-
-        mqdes = (mqdes_t) rt_malloc (sizeof(struct mqdes));
-        if (mqdes == RT_NULL)
-        {
-            rt_set_errno(ENFILE);
-            goto __return;
+            return (mqd_t)(-1);
         }
+        struct mqueue_file *mq_file;
+        mq_file = (struct mqueue_file *) rt_malloc (sizeof(struct mqueue_file));
 
-        /* create RT-Thread message queue */
-        mqdes->mq = rt_mq_create(name, attr->mq_msgsize, attr->mq_maxmsg, RT_IPC_FLAG_FIFO);
-        if (mqdes->mq == RT_NULL) /* create failed */
+        if (mq_file == RT_NULL)
         {
             rt_set_errno(ENFILE);
-            goto __return;
+            return (mqd_t)(-1);
         }
-        /* initialize reference count */
-        mqdes->refcount = 1;
-        mqdes->unlinked = 0;
-
-        /* insert mq to posix mq list */
-        posix_mq_insert(mqdes);
+        mq_file->msg_size = attr->mq_msgsize;
+        mq_file->max_msgs = attr->mq_maxmsg;
+        mq_file->data = RT_NULL;
+        strncpy(mq_file->name, name, RT_NAME_MAX);
+        dfs_mqueue_insert_after(&(mq_file->list));
     }
     else
     {
         rt_set_errno(ENOENT);
-        goto __return;
+        return (mqd_t)(-1);
     }
-    rt_sem_release(&posix_mq_lock);
 
-    return (mqd_t)(mqdes->mq_id);
+    const char* mq_path = "/dev/mqueue/";
+    char mq_name[RT_NAME_MAX + 12] = {0};
+    rt_sprintf(mq_name, "%s%s", mq_path, name);
+    mq_fd = open(mq_name, oflag);
 
-__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 */
-            rt_mq_delete(mqdes->mq);
-        }
-        rt_free(mqdes);
-    }
-    return (mqd_t)(-1);
+    return (mqd_t)(mq_fd);
 }
 RTM_EXPORT(mq_open);
 
 ssize_t mq_receive(mqd_t id, char *msg_ptr, size_t msg_len, unsigned *msg_prio)
 {
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
-    mqdes_t mqdes = posix_mq_id_find(id);
-    rt_sem_release(&posix_mq_lock);
+    rt_mq_t mq;
     rt_err_t result;
-
-    if ((mqdes == RT_NULL) || (msg_ptr == RT_NULL))
+    struct mqueue_file *mq_file;
+    mq_file = fd_get(id)->vnode->data;
+    mq = (rt_mq_t)mq_file->data;
+    if ((mq == RT_NULL) || (msg_ptr == RT_NULL))
     {
         rt_set_errno(EINVAL);
-
         return -1;
     }
 
-    result = rt_mq_recv_prio(mqdes->mq, msg_ptr, msg_len, (rt_int32_t *)msg_prio, RT_WAITING_FOREVER, RT_UNINTERRUPTIBLE);
+    result = rt_mq_recv_prio(mq, msg_ptr, msg_len, (rt_int32_t *)msg_prio, RT_WAITING_FOREVER, RT_UNINTERRUPTIBLE);
     if (result >= 0)
         return rt_strlen(msg_ptr);
 
@@ -243,19 +139,18 @@ RTM_EXPORT(mq_receive);
 
 int mq_send(mqd_t id, const char *msg_ptr, size_t msg_len, unsigned msg_prio)
 {
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
-    mqdes_t mqdes = posix_mq_id_find(id);
-    rt_sem_release(&posix_mq_lock);
+    rt_mq_t mq;
     rt_err_t result;
+    struct mqueue_file *mq_file;
+    mq_file = fd_get(id)->vnode->data;
+    mq = (rt_mq_t)mq_file->data;
 
-    if ((mqdes == RT_NULL) || (msg_ptr == RT_NULL))
+    if ((mq == RT_NULL) || (msg_ptr == RT_NULL))
     {
         rt_set_errno(EINVAL);
-
         return -1;
     }
-
-    result = rt_mq_send_wait_prio(mqdes->mq, (void *)msg_ptr, msg_len, msg_prio, 0, RT_UNINTERRUPTIBLE);
+    result = rt_mq_send_wait_prio(mq, (void *)msg_ptr, msg_len, msg_prio, 0, RT_UNINTERRUPTIBLE);
     if (result == RT_EOK)
         return 0;
 
@@ -271,23 +166,22 @@ ssize_t mq_timedreceive(mqd_t                  id,
                         unsigned              *msg_prio,
                         const struct timespec *abs_timeout)
 {
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
-    mqdes_t mqdes = posix_mq_id_find(id);
-    rt_sem_release(&posix_mq_lock);
-    int tick = 0;
+    rt_mq_t mq;
     rt_err_t result;
-
+    int tick = 0;
+    struct mqueue_file *mq_file;
+    mq_file = fd_get(id)->vnode->data;
+    mq = (rt_mq_t)mq_file->data;
     /* parameters check */
-    if ((mqdes == RT_NULL) || (msg_ptr == RT_NULL))
+    if ((mq == RT_NULL) || (msg_ptr == RT_NULL))
     {
         rt_set_errno(EINVAL);
-
         return -1;
     }
     if (abs_timeout != RT_NULL)
         tick = rt_timespec_to_tick(abs_timeout);
 
-    result = rt_mq_recv_prio(mqdes->mq, msg_ptr, msg_len, (rt_int32_t *)msg_prio, tick, RT_UNINTERRUPTIBLE);
+    result = rt_mq_recv_prio(mq, msg_ptr, msg_len, (rt_int32_t *)msg_prio, tick, RT_UNINTERRUPTIBLE);
 
     if (result >= 0)
         return rt_strlen(msg_ptr);
@@ -316,10 +210,11 @@ RTM_EXPORT(mq_timedsend);
 
 int mq_notify(mqd_t id, const struct sigevent *notification)
 {
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
-    mqdes_t mqdes = posix_mq_id_find(id);
-    rt_sem_release(&posix_mq_lock);
-    if (mqdes == RT_NULL || mqdes->refcount == 0)
+    rt_mq_t mq;
+    struct mqueue_file *mq_file;
+    mq_file = fd_get(id)->vnode->data;
+    mq = (rt_mq_t)mq_file->data;
+    if (mq == RT_NULL)
     {
         rt_set_errno(EBADF);
         return -1;
@@ -332,27 +227,7 @@ RTM_EXPORT(mq_notify);
 
 int mq_close(mqd_t id)
 {
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
-    mqdes_t mqdes = posix_mq_id_find(id);
-    rt_sem_release(&posix_mq_lock);
-    if (mqdes == RT_NULL)
-    {
-        rt_set_errno(EBADF);
-        return -1;
-    }
-
-    /* lock posix mqueue list */
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
-    mqdes->refcount --;
-    if (mqdes->refcount == 0)
-    {
-        /* delete from posix mqueue list */
-        if (mqdes->unlinked)
-            posix_mq_delete(mqdes);
-    }
-    rt_sem_release(&posix_mq_lock);
-
-    return 0;
+    return close(id);
 }
 RTM_EXPORT(mq_close);
 
@@ -392,28 +267,13 @@ RTM_EXPORT(mq_close);
  */
 int mq_unlink(const char *name)
 {
-    mqdes_t pmq;
-
-    /* lock posix mqueue list */
-    rt_sem_take(&posix_mq_lock, RT_WAITING_FOREVER);
-    pmq = posix_mq_find(name);
-    if (pmq != RT_NULL)
+    if(*name == '/')
     {
-        pmq->unlinked = 1;
-        if (pmq->refcount == 0)
-        {
-            /* remove this mqueue */
-            posix_mq_delete(pmq);
-        }
-        rt_sem_release(&posix_mq_lock);
-
-        return 0;
+        name++;
     }
-    rt_sem_release(&posix_mq_lock);
-
-    /* no this entry */
-    rt_set_errno(ENOENT);
-
-    return -1;
+    const char *mq_path = "/dev/mqueue/";
+    char mq_name[RT_NAME_MAX + 12] = {0};
+    rt_sprintf(mq_name, "%s%s", mq_path, name);
+    return unlink(mq_name);
 }
 RTM_EXPORT(mq_unlink);

components/libc/posix/ipc/mqueue.h

@@ -10,24 +10,12 @@
 #ifndef __MQUEUE_H__
 #define __MQUEUE_H__
 
+#include <rtthread.h>
 #include <sys/signal.h>
-#include <sys/time.h>
-#include <rtdef.h>
 
-struct mqdes
-{
-    /* reference count and unlinked */
-    rt_uint16_t refcount;
-    rt_uint16_t unlinked;
-
-    /* RT-Thread message queue */
-    rt_mq_t mq;
-
-    int mq_id;
-    /* next posix mqueue */
-    struct mqdes* next;
-};
-typedef struct mqdes* mqdes_t;
+#ifdef RT_USING_DFS_MQUEUE
+#include "dfs_mqueue.h"
+#endif
 
 typedef int mqd_t;