/* Copyright 2002, Red Hat Inc. */
|
/* Copyright 2002, Red Hat Inc. */
|
|
|
#include <mqueue.h>
|
#include <mqueue.h>
|
#include <fcntl.h>
|
#include <fcntl.h>
|
#include <unistd.h>
|
#include <unistd.h>
|
#include <errno.h>
|
#include <errno.h>
|
#include <sys/stat.h>
|
#include <sys/stat.h>
|
#include <sys/mman.h>
|
#include <sys/mman.h>
|
#include <sys/ipc.h>
|
#include <sys/ipc.h>
|
#include <sys/sem.h>
|
#include <sys/sem.h>
|
#include <string.h>
|
#include <string.h>
|
#include <stdlib.h>
|
#include <stdlib.h>
|
#include <time.h>
|
#include <time.h>
|
#include <stdarg.h>
|
#include <stdarg.h>
|
#include <machine/weakalias.h>
|
#include <machine/weakalias.h>
|
#define _LIBC 1
|
#define _LIBC 1
|
#include <sys/lock.h>
|
#include <sys/lock.h>
|
#undef _LIBC
|
#undef _LIBC
|
|
|
#include "mqlocal.h"
|
#include "mqlocal.h"
|
|
|
#define NHASH 32 /* Num of hash lists, must be a power of 2 */
|
#define NHASH 32 /* Num of hash lists, must be a power of 2 */
|
#define LOCHASH(i) ((i)&(NHASH-1))
|
#define LOCHASH(i) ((i)&(NHASH-1))
|
|
|
static long mq_index; /* Index of next entry */
|
static long mq_index; /* Index of next entry */
|
static struct libc_mq *mq_hash[NHASH]; /* Hash list heads for mqopen_infos */
|
static struct libc_mq *mq_hash[NHASH]; /* Hash list heads for mqopen_infos */
|
|
|
__LOCK_INIT(static, mq_hash_lock);
|
__LOCK_INIT(static, mq_hash_lock);
|
|
|
mqd_t
|
mqd_t
|
mq_open (const char *name, int oflag, ...)
|
mq_open (const char *name, int oflag, ...)
|
{
|
{
|
MSG *wrbuf = NULL;
|
MSG *wrbuf = NULL;
|
MSG *rdbuf = NULL;
|
MSG *rdbuf = NULL;
|
int msgqid = -1;
|
int msgqid = -1;
|
int rc = -1;
|
int rc = -1;
|
int fd = -1;
|
int fd = -1;
|
int semid = -1;
|
int semid = -1;
|
int created = 0;
|
int created = 0;
|
key_t key = (key_t)-1;
|
key_t key = (key_t)-1;
|
struct mq_attr *attr = (struct mq_attr *)MAP_FAILED;
|
struct mq_attr *attr = (struct mq_attr *)MAP_FAILED;
|
struct sembuf sb = {0, 0, 0};
|
struct sembuf sb = {0, 0, 0};
|
mode_t mode = 0;
|
mode_t mode = 0;
|
int size;
|
int size;
|
int i, index, saved_errno;
|
int i, index, saved_errno;
|
char *real_name;
|
char *real_name;
|
char *ptr;
|
char *ptr;
|
struct mq_attr *user_attr = NULL;
|
struct mq_attr *user_attr = NULL;
|
struct libc_mq *info;
|
struct libc_mq *info;
|
union semun arg;
|
union semun arg;
|
|
|
/* ignore opening slash if present */
|
/* ignore opening slash if present */
|
if (*name == '/')
|
if (*name == '/')
|
++name;
|
++name;
|
size = strlen(name);
|
size = strlen(name);
|
|
|
if ((real_name = (char *)malloc (size + sizeof(MSGQ_PREFIX))) == NULL ||
|
if ((real_name = (char *)malloc (size + sizeof(MSGQ_PREFIX))) == NULL ||
|
(info = (struct libc_mq *)malloc (sizeof(struct libc_mq))) == NULL)
|
(info = (struct libc_mq *)malloc (sizeof(struct libc_mq))) == NULL)
|
{
|
{
|
errno = ENOSPC;
|
errno = ENOSPC;
|
if (real_name)
|
if (real_name)
|
free (real_name);
|
free (real_name);
|
return (mqd_t)-1;
|
return (mqd_t)-1;
|
}
|
}
|
|
|
/* use given name to create shared memory file name - we convert any
|
/* use given name to create shared memory file name - we convert any
|
slashes to underscores so we don't have to create directories */
|
slashes to underscores so we don't have to create directories */
|
memcpy (real_name, MSGQ_PREFIX, sizeof(MSGQ_PREFIX) - 1);
|
memcpy (real_name, MSGQ_PREFIX, sizeof(MSGQ_PREFIX) - 1);
|
memcpy (real_name + sizeof(MSGQ_PREFIX) - 1, name, size + 1);
|
memcpy (real_name + sizeof(MSGQ_PREFIX) - 1, name, size + 1);
|
ptr = real_name + sizeof(MSGQ_PREFIX) - 1;
|
ptr = real_name + sizeof(MSGQ_PREFIX) - 1;
|
for (i = 0; i < size; ++i)
|
for (i = 0; i < size; ++i)
|
{
|
{
|
if (*ptr == '/')
|
if (*ptr == '/')
|
*ptr = '_';
|
*ptr = '_';
|
++ptr;
|
++ptr;
|
}
|
}
|
|
|
/* open shared memory file based on msg queue open flags and then use memory
|
/* open shared memory file based on msg queue open flags and then use memory
|
file to create a unique key to use for semaphores, etc.. */
|
file to create a unique key to use for semaphores, etc.. */
|
if (oflag & O_CREAT)
|
if (oflag & O_CREAT)
|
{
|
{
|
va_list list;
|
va_list list;
|
va_start (list, oflag);
|
va_start (list, oflag);
|
|
|
saved_errno = errno;
|
saved_errno = errno;
|
mode = (mode_t)va_arg (list, int);
|
mode = (mode_t)va_arg (list, int);
|
user_attr = va_arg(list,struct mq_attr *);
|
user_attr = va_arg(list,struct mq_attr *);
|
va_end (list);
|
va_end (list);
|
|
|
/* attempt to open the shared memory file for exclusive create so we know
|
/* attempt to open the shared memory file for exclusive create so we know
|
whether we are the owners or not */
|
whether we are the owners or not */
|
fd = open (real_name, O_RDWR | O_CREAT | O_EXCL, mode);
|
fd = open (real_name, O_RDWR | O_CREAT | O_EXCL, mode);
|
if (fd < 0 && (oflag & O_EXCL))
|
if (fd < 0 && (oflag & O_EXCL))
|
{
|
{
|
/* we failed and the user wanted exclusive create */
|
/* we failed and the user wanted exclusive create */
|
free (real_name);
|
free (real_name);
|
free (info);
|
free (info);
|
return (mqd_t)-1;
|
return (mqd_t)-1;
|
}
|
}
|
errno = saved_errno;
|
errno = saved_errno;
|
/* check if we created the file or not */
|
/* check if we created the file or not */
|
if (fd >= 0)
|
if (fd >= 0)
|
created = 1;
|
created = 1;
|
}
|
}
|
|
|
if (fd < 0)
|
if (fd < 0)
|
fd = open (real_name, O_RDWR, 0);
|
fd = open (real_name, O_RDWR, 0);
|
|
|
if (fd >= 0)
|
if (fd >= 0)
|
key = ftok(real_name, 255);
|
key = ftok(real_name, 255);
|
|
|
if (key != (key_t)-1)
|
if (key != (key_t)-1)
|
/* memory map the shared memory file so we have a global shared data area to use */
|
/* memory map the shared memory file so we have a global shared data area to use */
|
attr = (struct mq_attr *)mmap (0, sizeof(struct mq_attr), PROT_READ | PROT_WRITE,
|
attr = (struct mq_attr *)mmap (0, sizeof(struct mq_attr), PROT_READ | PROT_WRITE,
|
MAP_SHARED, fd, 0);
|
MAP_SHARED, fd, 0);
|
|
|
if (attr != (struct mq_attr *)MAP_FAILED)
|
if (attr != (struct mq_attr *)MAP_FAILED)
|
{
|
{
|
/* we need semaphores to prevent multi-process race conditions on the
|
/* we need semaphores to prevent multi-process race conditions on the
|
shared storage which contains a shared structure. The following
|
shared storage which contains a shared structure. The following
|
are the ones we need.
|
are the ones we need.
|
|
|
0 = open semaphore
|
0 = open semaphore
|
1 = number of opens
|
1 = number of opens
|
2 = number of writes left until queue is full
|
2 = number of writes left until queue is full
|
3 = number of reads available in queue
|
3 = number of reads available in queue
|
4 = notify semaphore
|
4 = notify semaphore
|
5 = number of readers */
|
5 = number of readers */
|
arg.val = 0;
|
arg.val = 0;
|
/* make sure the creator of the shared memory file also is the creator of the
|
/* make sure the creator of the shared memory file also is the creator of the
|
semaphores...this will ensure that it also creates the message queue */
|
semaphores...this will ensure that it also creates the message queue */
|
if (created)
|
if (created)
|
{
|
{
|
saved_errno = errno;
|
saved_errno = errno;
|
semid = semget (key, 6, IPC_CREAT | IPC_EXCL | mode);
|
semid = semget (key, 6, IPC_CREAT | IPC_EXCL | mode);
|
errno = saved_errno;
|
errno = saved_errno;
|
/* now that we have created the semaphore, we should initialize it */
|
/* now that we have created the semaphore, we should initialize it */
|
if (semid != -1)
|
if (semid != -1)
|
semctl (semid, 0, SETVAL, arg);
|
semctl (semid, 0, SETVAL, arg);
|
}
|
}
|
else
|
else
|
{
|
{
|
/* if we didn't create the shared memory file but have gotten to here, we want
|
/* if we didn't create the shared memory file but have gotten to here, we want
|
to ensure we haven't gotten ahead of the creator temporarily so we will
|
to ensure we haven't gotten ahead of the creator temporarily so we will
|
loop until the semaphore exists. This ensures that the creator will be the
|
loop until the semaphore exists. This ensures that the creator will be the
|
one to create the message queue with the correct mode and we will be blocked
|
one to create the message queue with the correct mode and we will be blocked
|
by the open semaphore 0. We impose a time limit to ensure something terrible
|
by the open semaphore 0. We impose a time limit to ensure something terrible
|
hasn't gone wrong. */
|
hasn't gone wrong. */
|
struct timespec tms;
|
struct timespec tms;
|
int i;
|
int i;
|
|
|
tms.tv_sec = 0;
|
tms.tv_sec = 0;
|
tms.tv_nsec = 10000; /* 10 microseconds */
|
tms.tv_nsec = 10000; /* 10 microseconds */
|
for (i = 0; i < 100; ++i)
|
for (i = 0; i < 100; ++i)
|
{
|
{
|
if ((semid = semget (key, 6, 0)) != -1)
|
if ((semid = semget (key, 6, 0)) != -1)
|
break;
|
break;
|
/* sleep in case we our a higher priority process */
|
/* sleep in case we our a higher priority process */
|
nanosleep (&tms, NULL);
|
nanosleep (&tms, NULL);
|
}
|
}
|
}
|
}
|
}
|
}
|
|
|
if (semid != -1)
|
if (semid != -1)
|
{
|
{
|
/* acquire main open semaphore if we didn't create it */
|
/* acquire main open semaphore if we didn't create it */
|
if (!created)
|
if (!created)
|
{
|
{
|
sb.sem_op = -1;
|
sb.sem_op = -1;
|
rc = semop (semid, &sb, 1);
|
rc = semop (semid, &sb, 1);
|
}
|
}
|
else
|
else
|
rc = 0; /* need this to continue below */
|
rc = 0; /* need this to continue below */
|
}
|
}
|
|
|
if (rc == 0)
|
if (rc == 0)
|
{
|
{
|
if (created)
|
if (created)
|
{
|
{
|
/* the creator must get here first so the message queue will be created */
|
/* the creator must get here first so the message queue will be created */
|
msgqid = msgget (key, IPC_CREAT | mode);
|
msgqid = msgget (key, IPC_CREAT | mode);
|
if (msgqid >= 0)
|
if (msgqid >= 0)
|
{
|
{
|
/* we have created the message queue so check and set the attributes */
|
/* we have created the message queue so check and set the attributes */
|
if ((wrbuf = (MSG *)malloc (user_attr->mq_msgsize + sizeof(int))) == NULL ||
|
if ((wrbuf = (MSG *)malloc (user_attr->mq_msgsize + sizeof(int))) == NULL ||
|
(rdbuf = (MSG *)malloc (user_attr->mq_msgsize + sizeof(int))) == NULL ||
|
(rdbuf = (MSG *)malloc (user_attr->mq_msgsize + sizeof(int))) == NULL ||
|
user_attr == NULL || user_attr->mq_msgsize <= 0 || user_attr->mq_maxmsg <= 0)
|
user_attr == NULL || user_attr->mq_msgsize <= 0 || user_attr->mq_maxmsg <= 0)
|
{
|
{
|
/* we're out of space and we created the message queue so we should
|
/* we're out of space and we created the message queue so we should
|
try to remove it */
|
try to remove it */
|
msgctl (msgqid, IPC_RMID, NULL);
|
msgctl (msgqid, IPC_RMID, NULL);
|
msgqid = -1; /* allow clean up to occur below */
|
msgqid = -1; /* allow clean up to occur below */
|
if (wrbuf && rdbuf)
|
if (wrbuf && rdbuf)
|
errno = EINVAL;
|
errno = EINVAL;
|
else
|
else
|
errno = ENOSPC;
|
errno = ENOSPC;
|
}
|
}
|
else /* valid attributes */
|
else /* valid attributes */
|
{
|
{
|
write (fd, user_attr, sizeof(struct mq_attr));
|
write (fd, user_attr, sizeof(struct mq_attr));
|
attr->mq_curmsgs = 0;
|
attr->mq_curmsgs = 0;
|
attr->mq_flags = oflag & O_NONBLOCK;
|
attr->mq_flags = oflag & O_NONBLOCK;
|
arg.val = 0;
|
arg.val = 0;
|
semctl (semid, 1, SETVAL, arg); /* number of opens starts at 0 */
|
semctl (semid, 1, SETVAL, arg); /* number of opens starts at 0 */
|
semctl (semid, 3, SETVAL, arg); /* number of reads available starts at 0 */
|
semctl (semid, 3, SETVAL, arg); /* number of reads available starts at 0 */
|
semctl (semid, 5, SETVAL, arg); /* number of readers starts at 0 */
|
semctl (semid, 5, SETVAL, arg); /* number of readers starts at 0 */
|
arg.val = 1;
|
arg.val = 1;
|
semctl (semid, 4, SETVAL, arg); /* notify semaphore */
|
semctl (semid, 4, SETVAL, arg); /* notify semaphore */
|
arg.val = user_attr->mq_maxmsg;
|
arg.val = user_attr->mq_maxmsg;
|
semctl (semid, 2, SETVAL, arg); /* number of writes left starts at mq_maxmsg */
|
semctl (semid, 2, SETVAL, arg); /* number of writes left starts at mq_maxmsg */
|
}
|
}
|
}
|
}
|
}
|
}
|
else /* just open it */
|
else /* just open it */
|
{
|
{
|
msgqid = msgget (key, 0);
|
msgqid = msgget (key, 0);
|
wrbuf = (MSG *)malloc (attr->mq_msgsize + sizeof(int));
|
wrbuf = (MSG *)malloc (attr->mq_msgsize + sizeof(int));
|
rdbuf = (MSG *)malloc (attr->mq_msgsize + sizeof(int));
|
rdbuf = (MSG *)malloc (attr->mq_msgsize + sizeof(int));
|
}
|
}
|
|
|
/* release semaphore acquired earlier */
|
/* release semaphore acquired earlier */
|
sb.sem_op = 1;
|
sb.sem_op = 1;
|
semop (semid, &sb, 1);
|
semop (semid, &sb, 1);
|
}
|
}
|
|
|
/* if we get here and we haven't got a message queue id, then we need to clean up
|
/* if we get here and we haven't got a message queue id, then we need to clean up
|
our mess and return failure */
|
our mess and return failure */
|
if (msgqid < 0)
|
if (msgqid < 0)
|
{
|
{
|
if (fd >= 0)
|
if (fd >= 0)
|
close (fd);
|
close (fd);
|
if (attr != (struct mq_attr *)MAP_FAILED)
|
if (attr != (struct mq_attr *)MAP_FAILED)
|
munmap (attr, sizeof(struct mq_attr));
|
munmap (attr, sizeof(struct mq_attr));
|
if (created)
|
if (created)
|
{
|
{
|
unlink (real_name);
|
unlink (real_name);
|
if (semid != -1)
|
if (semid != -1)
|
semctl (semid, 0, IPC_RMID);
|
semctl (semid, 0, IPC_RMID);
|
}
|
}
|
free (real_name);
|
free (real_name);
|
free (info);
|
free (info);
|
if (wrbuf)
|
if (wrbuf)
|
free (wrbuf);
|
free (wrbuf);
|
if (rdbuf)
|
if (rdbuf)
|
free (rdbuf);
|
free (rdbuf);
|
return (mqd_t)-1;
|
return (mqd_t)-1;
|
}
|
}
|
|
|
/* we are successful so register the message queue */
|
/* we are successful so register the message queue */
|
|
|
/* up the count of msg queue opens */
|
/* up the count of msg queue opens */
|
sb.sem_op = 1;
|
sb.sem_op = 1;
|
sb.sem_num = 1;
|
sb.sem_num = 1;
|
semop (semid, &sb, 1);
|
semop (semid, &sb, 1);
|
|
|
/* success, translate into index into mq_info array */
|
/* success, translate into index into mq_info array */
|
__lock_acquire(mq_hash_lock);
|
__lock_acquire(mq_hash_lock);
|
index = mq_index++;
|
index = mq_index++;
|
info->index = index;
|
info->index = index;
|
info->msgqid = msgqid;
|
info->msgqid = msgqid;
|
info->name = real_name;
|
info->name = real_name;
|
info->semid = semid;
|
info->semid = semid;
|
info->fd = fd;
|
info->fd = fd;
|
info->oflag = oflag;
|
info->oflag = oflag;
|
info->wrbuf = wrbuf;
|
info->wrbuf = wrbuf;
|
info->rdbuf = rdbuf;
|
info->rdbuf = rdbuf;
|
info->cleanup_notify = NULL;
|
info->cleanup_notify = NULL;
|
info->next = mq_hash[LOCHASH(index)];
|
info->next = mq_hash[LOCHASH(index)];
|
info->attr = attr;
|
info->attr = attr;
|
mq_hash[LOCHASH(index)] = info;
|
mq_hash[LOCHASH(index)] = info;
|
__lock_release(mq_hash_lock);
|
__lock_release(mq_hash_lock);
|
|
|
return (mqd_t)index;
|
return (mqd_t)index;
|
}
|
}
|
|
|
struct libc_mq *
|
struct libc_mq *
|
__find_mq (mqd_t mq)
|
__find_mq (mqd_t mq)
|
{
|
{
|
struct libc_mq *ptr;
|
struct libc_mq *ptr;
|
|
|
__lock_acquire(mq_hash_lock);
|
__lock_acquire(mq_hash_lock);
|
|
|
ptr = mq_hash[LOCHASH((int)mq)];
|
ptr = mq_hash[LOCHASH((int)mq)];
|
|
|
while (ptr)
|
while (ptr)
|
{
|
{
|
if (ptr->index == (int)mq)
|
if (ptr->index == (int)mq)
|
break;
|
break;
|
ptr = ptr->next;
|
ptr = ptr->next;
|
}
|
}
|
|
|
__lock_release(mq_hash_lock);
|
__lock_release(mq_hash_lock);
|
|
|
return ptr;
|
return ptr;
|
}
|
}
|
|
|
void
|
void
|
__cleanup_mq (mqd_t mq)
|
__cleanup_mq (mqd_t mq)
|
{
|
{
|
struct libc_mq *ptr;
|
struct libc_mq *ptr;
|
struct libc_mq *prev;
|
struct libc_mq *prev;
|
int semid;
|
int semid;
|
struct sembuf sb = {0, 0, 0};
|
struct sembuf sb = {0, 0, 0};
|
|
|
__lock_acquire(mq_hash_lock);
|
__lock_acquire(mq_hash_lock);
|
|
|
ptr = mq_hash[LOCHASH((int)mq)];
|
ptr = mq_hash[LOCHASH((int)mq)];
|
prev = NULL;
|
prev = NULL;
|
|
|
while (ptr)
|
while (ptr)
|
{
|
{
|
if (ptr->index == (int)mq)
|
if (ptr->index == (int)mq)
|
break;
|
break;
|
prev = ptr;
|
prev = ptr;
|
ptr = ptr->next;
|
ptr = ptr->next;
|
}
|
}
|
|
|
if (ptr != NULL)
|
if (ptr != NULL)
|
{
|
{
|
if (ptr->cleanup_notify != NULL)
|
if (ptr->cleanup_notify != NULL)
|
ptr->cleanup_notify (ptr);
|
ptr->cleanup_notify (ptr);
|
if (prev != NULL)
|
if (prev != NULL)
|
prev->next = ptr->next;
|
prev->next = ptr->next;
|
else
|
else
|
mq_hash[LOCHASH((int)mq)] = NULL;
|
mq_hash[LOCHASH((int)mq)] = NULL;
|
munmap (ptr->attr, sizeof(struct mq_attr));
|
munmap (ptr->attr, sizeof(struct mq_attr));
|
close (ptr->fd);
|
close (ptr->fd);
|
free (ptr->name);
|
free (ptr->name);
|
free (ptr->wrbuf);
|
free (ptr->wrbuf);
|
free (ptr->rdbuf);
|
free (ptr->rdbuf);
|
semid = ptr->semid;
|
semid = ptr->semid;
|
free (ptr);
|
free (ptr);
|
/* lower the count of msg queue opens */
|
/* lower the count of msg queue opens */
|
sb.sem_op = -1;
|
sb.sem_op = -1;
|
sb.sem_num = 1;
|
sb.sem_num = 1;
|
sb.sem_flg = IPC_NOWAIT;
|
sb.sem_flg = IPC_NOWAIT;
|
semop (semid, &sb, 1);
|
semop (semid, &sb, 1);
|
}
|
}
|
|
|
__lock_release(mq_hash_lock);
|
__lock_release(mq_hash_lock);
|
}
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
