Posix Message Queues
Section 5.7 Posix Signals 105
5.8 Implementation Using Memory-Mapped
We now provide an implementation of Posix message queues using memory-mapped along with Posix mutexes and condition variables.
We cover mutexes and condition variables in Chapter 7 and memory-mapped in Chapters 12 and 13. You may wish to skip this section until you have read those chapters.
Figure shows a layout of the data structures that we use to implement Posix message queues. In this figure, we assume that the message queue was created to hold up to four messages of 7 bytes each.
Figure shows our mqueue
.
h header, which defines the fundamental structures for this implementation.Our message queue descriptor is just a pointer to an o structure. Each call to allocates one of these structures, and the pointer to this structure is what gets returned to the caller. This reiterates that a message queue descriptor need not be a small integer, like a file descriptor-the only Posix requirement is that this
cannot be an array type.
5.8 Implementation Using Memory-Mapped 107
t
t
msg-len m s g q r i o
msg-len
msg-len
m s g q r i o 1 byte pad
start of region
one for each
J
of message queue
1
one message one messageone message
one message
end of memory-mapped region
J
one memory-mapped file per message queue
Figure 5.19 Layout of data structures to implement message queues using a memory-mapped file.
108 Posix Message Queues Chapter 5 max size of a message (in bytes)
number of messages currently on queue * /
This structure appears at the beginning of the mapped file and contains all the per- queue information. The lags member of the structure is not used, because the flags (the nonblocking flag is the only one defined) must be maintained on a per-open basis, not on a per-queue basis. The flags are maintained in the
structure. We describe the remaining members of this structure as we use them in the various functions.
Note now that everything that we refer to as an index (the mqh-head and ree members of this structure, and the msg-next member of the next structure) contains byte indexes from the beginning of the mapped file. For example, the size of
5.8 Implementation Using Memory-Mapped 1 / 0 109
the structure under 2.6 is 96 bytes, so the index of the first message fol- lowing this header is 96. Each message in Figure 5.19 occupies 20 bytes (12 bytes for the structure and 8 bytes for the message data), so the indexes of the remaining three messages are 116,136, and 156, and the size of this mapped file is 176 bytes. These indexes are used to maintain two linked lists in the mapped file: one list
contains all the messages currently on the queue, and the other ree) contains all the free messages on the queue. We cannot use actual memory pointers (addresses) for these list pointers, because the mapped file can start at different memory addresses in each process that maps the file (as we show in Figure 13.6).
Structure
This structure appears at the beginning of each message in the mapped file. All messages are either on the message list or on the free list, and the member contains the index of the next message on the list (or if this message is the end of the list). is the actual length of the message data, which for our example in Fig- ure 5.19 can be between and 7 bytes, inclusive. is the priority assigned to the message by the caller of
structure
26-32 One of these structures is dynamically allocated by when a queue is opened, and freed by points to the mapped file (the starting address returned by A pointer to this structure is the fundamental
of our implementation, and this pointer is the return value from
The member contains once this structure has been initial- ized and is checked by each function that is passed an pointer, to make certain that the pointer really points to an o structure. l a g s contains the blocking flag for this open instance of the queue.
macro
33-34 For alignment purposes, we want each message in the mapped file to start on a long integer boundary. Therefore, if the maximum size of each message is not so aligned, we add between 1 and 3 bytes of padding to the data portion of each message, as shown in Figure 5.19. This assumes that the size of a long integer is 4 bytes (which is true for but if the size of a long integer is 8 bytes (as on Digital Unix then the amount of padding will be between 1 and 7 bytes.
Function
Figure 5.21 shows the first part of our function, which creates a new message queue or opens an existing message queue.
3 #include
4 #define 10 / * for waiting for initialization * / 5 struct defattr =
6 128, 1024,
110 Posix Message Queues Chapter 5
7
8 char int
9 I
int i, fd, nonblock, created, save-errno;
long msgsize, filesize, index;
ap;
i.8 Implementation Using Memory-Mapped 1 / 0 111
argument is of type mode-t, but this is a primitive system that can be any type of integer. The problem we encounter is on which defines this
as an u n s i g n e d s h o r t integer (occupying 16 bits). Since an integer on this implemen- tation occupies 32 bits, the C compiler expands an argument of this type from 16 to 32 bits, since all short integers are expanded to integers in the argument list. But if we specify mode-t in the call to it will step past 16 bits of argument on the stack, when the argument has been expanded to occupy 32 bits. Therefore, we must define our own datatype, va-mode-t, that is an integer under or of type mode-t under other systems. The following lines in our
.
h header (Figure C.l) handle this portability problem:#define va-mode-t int
#else
#define va-mode-t mode-t
We turn off the user-execute bit in the mode variable for reasons that we describe shortly.
Create a new message queue
A regular file is created with the name specified by the caller, and the user-execute bit is turned on.
Handle potential race condition
If we were to just open the file, memory map its contents, and initialize the mapped file (as described shortly) when the flag is specified by the caller, we would have a race condition. A message queue is initialized by only if is specified by the caller and the message queue does not already exist. That means we need some method of detecting whether the message queue already exists. To do so, we always specify when we open the file that will be memory-mapped. But an error return of EEXIST from open becomes an error from only if the caller specified Otherwise, if open returns an error of EEXIST, the file already exists and we just skip ahead to Figure 5.23 as if the flag was not specified.
The possible race condition is because our use of a memory-mapped file to repre- sent a message queue requires two steps to initialize a new message queue: first, the file must be created by open, and second, the contents of the file (described shortly) must be initialized. The problem occurs if two threads (in the same or different processes) call at about the same time. One thread can create the file, and then the sys- tem switches to the second thread before the first thread completes the initialization.
This second thread detects that the file already exists (using the flag to open) and immediately tries to use the message queue. But the message queue cannot be used until the first thread initializes the message queue. We use the user-execute bit of the file to indicate that the message queue initialized. This bit is enabled only by the thread that actually creates the file (using the flag to detect which thread creates the file), and that thread initializes the message queue and then turns off the user-execute bit. We encounter similar race conditions in Figures 10.43 and
112 Posix Message Queues Chapter 5
Check attributes
If the caller specifies a null pointer for the final argument, we use the default attributes shown at the beginning of this figure: 128 messages and 1024 bytes per mes- sage. If the caller specifies the attributes, we verify that and
are positive.
The second part of our function is shown in Figure 5.22; it completes the initialization of a new queue.
/ * calculate and set the file size * /
msghdr = (struct msg-hdr
= 0; / * end of free list * / / * initialize condition variable * /
if ( = !=
pthreaderr;
Section 5.8 Implementation Using Memory-Mapped 1 / 0 113
9 8 PTHREAD-PROCESS-SHARED);
99 =
Figure 5.22 Second part of function: complete initialization of new queue.
Set the file size
We calculate the size of each message, rounding up to the next multiple of the size of a long integer. To calculate the file size, we also allocate room for the struc- ture at the beginning of the file and the structure at the beginning of each message (Figure We set the size of the newly created file using lseek and then writing one byte of Just calling truncate (Section 13.3) would be easier, but we are not guaranteed that this works to increase the size of a file.
Memory map the file
59-63 The file is memory mapped by Allocate structure
Initialize mutex and condition variable
Since Posix message queues can be shared by any process that knows the message queue's name and has adequate permission, we must initialize the mutex and condition variable with the PTHREAD-PROCESS-SHARED attribute. To do so for the message queue, we first initialize the attributes by calling t, then call to set the process-shared attribute in this struc- ture, and then initialize the mutex by calling t. Nearly identical steps are done for the condition variable. We are careful to destroy the mutex or
114 Posix Message Queues Chapter 5
condition variable attributes that are initialized, even if an error occurs, because the calls
to or might allocate memory
(Exercise 7.3).
Turn off user-execute bit
Once the message queue is initialized, we turn off the user-execute bit. This indi- cates that the message queue has been initialized. We also close the file, since it has been memory mapped and there is no need to keep it open u p a descriptor).
Figure 5.23 shows the final part of our function, which opens an existing queue.
exists:
/ * open the file then memory map * / if ( (fd
if == ENOENT again;
err;
/ * make certain initialization is complete * / for = i
if == -1)
if == ENOENT again;
err;
if == 0)
break;
if == MAX-TRIES)
= ETIMEDOUT;
err;
=
mptr MAP-SHARED, fd,
if (mptr == MAP-FAILED) err;
close ;
/ * allocate one for each open * /
if = == NULL)
err;
= (struct mptr;
=
= nonblock;
return ( mqinfo) ;
116 Message Queues Chapter 5
Memory map file; ailocate and structure
The file is memory mapped, and the descriptor can then be closed. We allocate an o structure and initialize it. The return value is a pointer to the o struc- ture that was allocated.
Handie errors
When an error is detected earlier in the function, the label e r r is branched to, with set to the value to be returned by We are careful that the functions called to clean u p after the error is detected do not affect the returned by this function.
Function
Figure 5.24 shows our o s e function.
2 #include
.
h"3 int 4 5
6 long msgsize, filesize;
7 struct 8 struct 9 struct
10 mqinfo = mqd;
11 if ! =
12 = EBADF;
13 return (-1);
14
15 mqhdr = 16 attr =
17 if NULL) ! = 0) / * unregister calling process * / 18 return (-1) ;
19
2 0 = + *
21 + msgsize)
22 if filesize) == -1)
2 3 return (-1);
24 0; / * just in case * /
2 5 free o ;
2 6 return ;
27
Figure 5.24 function.
5.8 Implementation Using Memory-Mapped 1/0 117
Get pointers to structures
The argument is validated, and pointers are then obtained to the memory-mapped region and the attributes (in the structure).
Unregister calling process
We call i to unregister the calling process for this queue. If the process is registered, it will be unregistered, but if it is not registered, no error is returned.
Unmap region and free memory
We calculate the size of the file for munmap and then free the memory used by the o structure. Just in case the caller continues to use the message queue descrip- tor before that region of memory is reused by we set the magic number to so that our message queue functions will detect the error.
Note that if the process terminates without calling the same operations take place on process termination: the memory-mapped file is unmapped and the mem- ory is freed.
link Function
Our function shown in Figure 5.25 removes the name associated with our message queue. It just calls the u n l i n k function.
3 int
4 char
5
6 if == -1)
7 return
8 return ;
9
Figure 5.25 function.
tattr Function
Figure 5.26 shows our function, which returns the current attributes of the specified queue.
Acquire queue's mutex lock
We must acquire the message queue's mutex lock before fetching the attributes, in case some other thread is in the middle of changing them.
118 Message Queues Chapter 5
#include 2 #include 3 int
4 mqd, struct
5
int struct struct struct
mqinfo = mqd;
if !=
= EBADF;
return
I
mqhdr = attr =
if (n = ! = 0)
= n;
return
I
= / * per-open * /
= / * remaining three per-queue * /
=
=
return ;
I
Figure 5.26 function.
Function
Figure 5.27 shows our function, which sets the current attributes of the specified queue.
Return current attributes
22-27 If the third argument is a nonnull pointer, we return the previous attributes and cur- rent status before changing anything.
Change
28-31 The only attribute that can be changed with this function is lags, which we store in the o structure.
8 Implementation Using Memory-Mapped 1/0 119
3 int
4 mqd, struct
struct 7 int
8 struct ;
struct 10 struct
mqinfo = mqd;
if !=
= EBADF;
return
I
mqhdr = attr =
if ( (n = ! = 0)
= n;
return (-1) ;
I
if (omqstat ! = NULL)
= previous attributes * /
=
=
= / * and current status * /
I
if else
&=
32
3 3 return ;
Figure 5.27 function.
ify Function
The function shown in Figure 5.28 registers or unregisters the calling pro- cess for the queue. We keep track of the process currently registered for a queue by storing its process ID in the member of the structure. Only one pro- cess at a time can be registered for a given queue. When a process registers itself, we also save its specified sigevent structure in the structure.
Message Queues Chapter 5
3 int
4 mqd, struct sigevent *notification) 5
int
pid;
struct struct
mqinfo = mqd;
if
= EBADF;
return -1 ;
mqhdr =
if = 0)
return (-1 ;
I
pid = getpid ;
if (notification == NULL)
if ==
= 0; / * unregister calling process * /
I / * no error if caller not registered * /
else
if !=
if != -1 ! = ESRCH)
= EBUSY;
err;
I
= pid;
= *notification;
I
return ;
err : return
I
Figure 5.28 t i f function.
Unregister caliing process
If the second argument is a null pointer, the calling process is unregistered for this queue. Strangely, no error is specified if the calling process is not registered for this queue.
15.8 Implementation Using Memory-Mapped 1/0 121
Register caliing process
4 If some process is already registered, we check whether it still exists by sending it signal (called the null signal). This performs the normal error checking, but does not send a signal and returns an error of if the process does not exist. An error of is returned if the previously registered process still exists. Otherwise, the pro- cess ID is saved, along with the caller's s i g e v e n t structure.
Our test for whether the previously registered process exists is not perfect. This process can terminate and then have its process ID reused at some later time.
Function
Figure shows the first half of our function.
Pointers are obtained to the structures that we will use, and the mutex lock for the queue is obtained. A check is made that the size of the message does not exceed the maximum message size for this queue.
Check for empty queue and send notification if applicabie
If we are placing a message onto an empty queue, we check whether any process is registered for this queue and whether any thread is blocked in a call to
For the latter check, we will see that our function keeps a count of the number of threads blocked on the empty queue. If this counter is nonzero, we do not send any notification to the registered process. We handle a notifi- cation of and call to send the signal. The registered process is then unregistered.
Calling sigqueue to send the signal results in an of being passed to the signal handler in the structure (Section which is incorrect. Generating the correct of from a user process is implementation dependent. Page 433 of
19961 mentions that a hidden interface into the signal generation mechanism is required to generate this signal from a user library.
Check for fuil queue
If the queue is full but the flag has been set, we return an error of EAGAIN. Otherwise, we wait on the condition variable which we will see is signaled by our function when a message is read from a full queue.
implementation is simplistic with regard to returning an error of EINTR if this to is interrupted by a signal that is caught by the calling process. The problem is that does not return an error when the signal handler returns: it can either return a value of (which appears as a spurious or it need not return at all. Ways around this exist, all nontrivial.
Figure 5.30 shows the second half of our function. At this point, we know the queue has room for the new message.
122 Posix Message Queues Chapter 5
Implementation Using Memory-Mapped 1 / 0 123
Figure 5.30 function: second half.
Get index of free biock to use
Since the number of free messages created when the queue was initialized equals we should never have a situation where is less than with an empty free list.
Copy message
contains the address in the mapped memory of where the message is stored. The priority and length are stored in its m s g - h d r structure, and then the con- tents of the message are copied from the caller.
Message Queues Chapter 5
Place new message onto list in correct location
57- 74 The order of messages on our linked list is from highest priority at the front to lowest priority at the end. When a new message is added to the queue and one or more messages of the same priority are already on the queue, the new mes- sage is added after the last message with its priority. Using this ordering,
always returns the first message on the linked list (which is the oldest message of the highest priority on the queue). As we step through the linked list, pmsghdr contains the address of the previous message in the list, because its value will contain the index of the new message.
Our design can be slow when lots of messages are on the queue, forcing a traversal of a large number of entries each time a message is written to the queue. A separate index could be maintained that remembers the location of the last message for each possible priority.
Wake up anyone blocked in
75- 77 If the queue was empty before we placed the message onto the queue, we call to wake any thread that might be blocked in
78 The number of messages currently on the queue, is incremented.
Function
Figure 5.31 shows the first half of our function, which sets up the pointers that it needs, obtains the mutex lock, and verifies that the caller's buffer is large enough for the largest possible message.
Check for empty queue
If the queue is empty and the flag is set, an error of EAGAIN is returned. Otherwise, we increment the queue's counter, which was exam-
If the queue is empty and the flag is set, an error of EAGAIN is returned. Otherwise, we increment the queue's counter, which was exam-