Practical 2 continued: Message Queues, Blocking and Non-blocking

sender  : sent [item 10]
sender  : sent [item 20]
sender  : sent [item 30]
receiver: got 64 bytes, type 1, [item 10]
receiver: got 64 bytes, type 1, [item 20]
receiver: got 64 bytes, type 1, [item 30]

Three sends, three receives. Compare that with the pipe, where three writes became one read. Each msgrcv returned 64 bytes, which is the size of mtext, because the sender asked for all 64 to be sent: a message queue delivers exactly what was posted, and the length is part of the message.

msgget(IPC_PRIVATE, IPC_CREAT | 0600) takes only a key and flags, and no size: a queue grows as messages are put into it, up to a system limit. The permission bits are the same idea as for shared memory.

The wait(NULL) in the middle is there only to make the page's output tidy, by letting all three messages be sent before any is received. It is not needed for correctness, and that is the point of the next section.

Blocking, which is what message passing gives you for nothing

A receive on an empty queue blocks: the process is put to sleep by the kernel and woken when a message arrives. A send to a full queue blocks in the same way.

That single behaviour is the reason the producer-consumer over a message queue needs no semaphore, no shared counter, and no wait.

#include <stdio.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/wait.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define ITEMS 5

struct message {
    long mtype;
    int  item;
};

int main(void)
{
    int q = msgget(IPC_PRIVATE, IPC_CREAT | 0600);
    if (q == -1) {
        perror("msgget");
        return 1;
    }

    if (fork() == 0) {                          /* the producer */
        struct message m = { .mtype = 1, .item = 0 };
        for (int i = 1; i <= ITEMS; i++) {
            m.item = i * i;
            if (msgsnd(q, &m, sizeof m.item, 0) == -1) {
                perror("msgsnd");
                _exit(1);
            }
            printf("producer: produced %d\n", m.item);
            fflush(stdout);
        }
        m.mtype = 2;                            /* type 2 means "I am done" */
        msgsnd(q, &m, sizeof m.item, 0);
        _exit(0);
    }

    if (fork() == 0) {                          /* the consumer */
        struct message m;
        int total = 0;
        for (;;) {
            /* -2 means: the lowest type that is 2 or less, so an item
               (type 1) is always preferred over the end marker (type 2) */
            if (msgrcv(q, &m, sizeof m.item, -2, 0) == -1) {
                perror("msgrcv");
                _exit(1);
            }
            if (m.mtype == 2)
                break;
            printf("consumer: consumed %d\n", m.item);
            fflush(stdout);
            total = total + m.item;
        }
        printf("consumer: the sender has finished, the total is %d\n", total);
        fflush(stdout);
        _exit(0);
    }

    while (wait(NULL) > 0)
        ;
    msgctl(q, IPC_RMID, NULL);
    return 0;
}

producer: produced 1
consumer: consumed 1
producer: produced 4
consumer: consumed 4
producer: produced 9
consumer: consumed 9
producer: produced 16
consumer: consumed 16
producer: produced 25
consumer: consumed 25
consumer: the sender has finished, the total is 55

That order is one run's, not the program's. The two processes are racing, and nothing here makes them take turns. This run went back and forth, a produce and then a consume. Another run sends all five before the consumer is given the processor, and prints the five produced lines in a block. Both are correct, so the check on this page compares these lines without their order, on all ten runs.

What holds on every run is this. Each square is produced once and consumed once. The consumer takes the items in the order they were sent, because a queue is first in, first out. And the total is 55, which is 1 plus 4 plus 9 plus 16 plus 25, printed so that the program proves no item was lost or counted twice.

A tidy page is exactly what the wait(NULL) in the previous program was buying. Here there is no wait, the producer and the consumer are alive at the same time, and so the page cannot be tidy. That is not a fault to be repaired. It is the blocking receive doing its work, and it is the reason this producer-consumer needs no semaphore at all.

The end marker is the idea worth taking from this program. A pipe tells the reader it is finished by reaching end of file when the writer closes it. A message queue has no such thing: it does not know who is using it. So the producer sends one last message of a different type, and the consumer stops when it sees that type. It is the standard answer, and an examiner asks for it as "how does the consumer know when to stop".

The -2 in the consumer's msgrcv is doing real work, and the next section explains it.

Selecting a message by its type

The fourth argument of msgrcv is the type to ask for, and it has three quite different meanings depending on its sign. This program sends three messages of types 10, 20 and 30 and then takes them out in a deliberately awkward order.

#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <sys/ipc.h>
#include <sys/msg.h>

struct message {
    long mtype;
    char mtext[32];
};

int main(void)
{
    int q = msgget(IPC_PRIVATE, IPC_CREAT | 0600);
    if (q == -1) {
        perror("msgget");
        return 1;
    }

    struct message m;
    const char *body[] = { "for the teacher", "for the student", "for anybody" };
    long types[] = { 10, 20, 30 };

    for (int i = 0; i < 3; i++) {
        m.mtype = types[i];
        snprintf(m.mtext, sizeof m.mtext, "%s", body[i]);
        if (msgsnd(q, &m, sizeof m.mtext, 0) == -1) {
            perror("msgsnd");
            return 1;
        }
    }

    /* a positive type: exactly that type */
    msgrcv(q, &m, sizeof m.mtext, 20, 0);
    printf("asked for type 20, got type %ld: [%s]\n", m.mtype, m.mtext);

    /* a negative type: the LOWEST type that is not more than its size */
    msgrcv(q, &m, sizeof m.mtext, -25, 0);
    printf("asked for the lowest type up to 25, got type %ld: [%s]\n",
           m.mtype, m.mtext);

    /* zero: whatever is at the front of the queue */
    msgrcv(q, &m, sizeof m.mtext, 0, 0);
    printf("asked for any type, got type %ld: [%s]\n", m.mtype, m.mtext);

    /* and now the queue is empty, and we ask without waiting */
    ssize_t n = msgrcv(q, &m, sizeof m.mtext, 0, IPC_NOWAIT);
    printf("the queue is empty now: msgrcv returned %zd, errno says %s\n",
           n, strerror(errno));

    msgctl(q, IPC_RMID, NULL);
    return 0;
}

asked for type 20, got type 20: [for the student]
asked for the lowest type up to 25, got type 10: [for the teacher]
asked for any type, got type 30: [for anybody]
the queue is empty now: msgrcv returned -1, errno says No message of desired type

That output is the whole rule, and it is worth putting in a table because an examiner asks it directly.

The second line of the output is the one to look at twice. The queue held types 10, 20 and 30; type 20 had already been taken; asking for -25 means "the lowest type up to 25", and the answer was type 10, not type 20 and not the front of the queue. That is what makes a message type usable as a priority: send urgent work as type 1 and ordinary work as type 5, receive with a negative number, and the urgent work comes out first however late it arrived.

It is also why the producer-consumer above used -2: items are type 1 and the end marker is type 2, so as long as any item remains the consumer takes an item, and it sees the marker only when the items have run out. With a plain 0 there the consumer would take the marker as soon as it reached the front, and stop early.

Blocking against non-blocking

The last line of that program is the other half of MU's bullet. IPC_NOWAIT in the flags means do not block: if there is nothing to receive, fail immediately instead of waiting.

msgrcv returned -1, errno says No message of desired type

errno is ENOMSG, and the message is the system's own wording. The same flag on msgsnd means do not wait for room in a full queue, and then the error is EAGAIN.

Blocking is the right default, and a student should say so and know why: a blocked process uses no processor time at all. The kernel takes it off the run queue and puts it back when the message arrives. A non-blocking receive in a loop, by contrast, asks the kernel over and over, gets -1 every time, and keeps a whole processor busy doing nothing. That pattern is called a busy-wait or a spin, and it is almost always the wrong answer.

Non-blocking is right in exactly one situation: when the process has something else to do and wants to check the queue in passing. A program drawing a screen sixty times a second cannot afford to block, so it looks, finds nothing, and carries on drawing.

Looking at a queue from outside

ipcs -q lists the message queues on the machine, as ipcs -m does for shared memory, and msgctl with IPC_STAT lets the program ask about its own queue.

#include <stdio.h>
#include <string.h>
#include <sys/ipc.h>
#include <sys/msg.h>

struct message {
    long mtype;
    char mtext[16];
};

int main(void)
{
    int q = msgget(IPC_PRIVATE, IPC_CREAT | 0600);
    if (q == -1) {
        perror("msgget");
        return 1;
    }

    struct message m = { .mtype = 1, .mtext = "" };
    for (int i = 0; i < 4; i++) {
        snprintf(m.mtext, sizeof m.mtext, "message %d", i);
        msgsnd(q, &m, sizeof m.mtext, 0);
    }

    struct msqid_ds info;
    if (msgctl(q, IPC_STAT, &info) == -1) {
        perror("msgctl IPC_STAT");
        return 1;
    }
    printf("messages waiting in the queue : %lu\n", (unsigned long) info.msg_qnum);
    printf("the most bytes it will hold   : %lu\n", (unsigned long) info.msg_qbytes);
    printf("the permission bits           : %o\n", info.msg_perm.mode & 0777);

    msgctl(q, IPC_RMID, NULL);
    return 0;
}

messages waiting in the queue : 4
the most bytes it will hold   : 16384
the permission bits           : 600

Four messages are waiting, nobody having received any, and the queue will hold 16384 bytes in total, which is the Linux default and is the figure a send would block on. A queue is therefore a bounded buffer as well, exactly like a pipe, and this is its bound.

A note for anyone who goes looking in the manual page: struct msqid_ds also holds the number of bytes currently in the queue, and that member is not portable. On glibc it is spelled __msg_cbytes, with two leading underscores, and asking for msg_cbytes is an error: struct msqid_ds has no member named msg_cbytes; did you mean msg_qbytes?. The three members above are the ones to rely on.

A queue left behind shows in ipcs -q and is removed with ipcrm -q <msqid>, and the same warning applies as for shared memory: a laboratory full of leaked queues eventually stops working for everybody.

Shared memory against message passing, which is MU's own bullet

This is the table to learn. It is the answer to the most likely viva question on Practical 1 and Practical 2 together.

Two sentences to remember it by:

Shared memory is a noticeboard: everybody can read and write it, nobody is told when it

changes, and two people writing at once make a mess.

A message queue is a pigeonhole: things go in whole, come out whole, wait until they are

collected, and can be sorted by who they are for.

Procedure

1. Write the first program. Compile with gcc -Wall -Wextra -o mq1 mq1.c and run it. Count the

sends and the receives.

1. Change sizeof m.mtext to sizeof m in the msgsnd call and run it again. Note what the

receiver prints, and put it back.

1. Write the producer-consumer program. Confirm the total.
2. Change the consumer's -2 to 0 and run it. Explain what happened to the missing items.
3. Write the type-selection program and read the three answers carefully.
4. Write the IPC_STAT program. Then run it with the msgctl(q, IPC_RMID, NULL) line commented

out, and find the queue with ipcs -q. Remove it with ipcrm -q.

Result

Three messages were sent and received whole, each one delivered as it was posted, which a pipe cannot do. The producer-consumer problem was solved over a message queue with no semaphore and no shared counter, because a receive on an empty queue blocks. A message was selected by exact type, by lowest type and by position, and IPC_NOWAIT on an empty queue was shown to return -1 with errno set to ENOMSG instead of waiting.

Where marks are lost

• sizeof m instead of sizeof m.mtext. The size is of everything after mtype.
• mtype not first in the structure, or set to 0 or a negative number. The kernel requires a

positive type in the first member.

• Forgetting msgctl(q, IPC_RMID, NULL). The queue stays on the machine; ipcs -q shows it.
• No end marker, so the consumer blocks for ever after the last item and the program has to be

killed.

• Using 0 as the type in a consumer that also has an end marker, so it stops early.
• Saying "non-blocking is better". It is not: a blocked process uses no processor, and a

non-blocking receive in a loop is a busy-wait.

• Not checking msgsnd and msgrcv. msgrcv returns the number of bytes, and -1 on failure,

which is not the same as 0.

For the journal

Write the aim, MU's own wording, the first program and its output, and the producer-consumer program with its total. Then the type-selection program, and copy the four lines of its output into the journal because they are the answer to the viva question. Add the comparison table of shared memory against message passing; it is two marks on its own. The conclusion: a message queue keeps each message whole and blocks the receiver until one arrives, so it needs no semaphore, and it costs two copies of the data that shared memory does not.

Quick revision

• A message queue is a kernel-held list of whole messages. msgget, msgsnd, msgrcv, msgctl.
• The message structure starts with long mtype, which must be positive. The size passed to

msgsnd and msgrcv is the size of the rest, not of the whole structure.

• A message goes in whole and comes out whole, unlike a pipe, which is a byte stream.
• The type argument of msgrcv: positive means exactly that type, zero means the front of the

queue, negative means the lowest type not greater than its size. Negative is how priorities are done.

• A blocking receive on an empty queue sleeps and uses no processor. IPC_NOWAIT returns -1 at

once with ENOMSG; a non-blocking send to a full queue gives EAGAIN.

• A consumer knows the producer has finished because the producer sends a last message of a

different type. There is no end of file on a queue.

• msgctl with IPC_STAT fills a struct msqid_ds: msg_qnum is how many messages are waiting,

msg_qbytes is the capacity, 16384 by default on Linux.

• ipcs -q lists queues, ipcrm -q <id> removes one.
• Shared memory: no copies, no synchronisation. Message passing: two copies, synchronisation free.

Questions you should be able to answer

1. What must the first member of a message structure be, and why? A long holding the message type. The kernel reads it to decide which receiver the message can satisfy, and it must be a positive number.

2. What size is passed to msgsnd? The size of the structure after mtype, so sizeof m.mtext and not sizeof m.

3. Three writes were made. How many receives does a message queue need, and how many did a pipe need? The queue needs three, because each message is delivered whole. The pipe took one read of 24 bytes for three writes of 8, because it is a byte stream.

4. What are the three meanings of the type argument to msgrcv? A positive number asks for exactly that type; zero asks for the message at the front of the queue; a negative number asks for the lowest type that is not greater than its absolute value.

5. How would you give some messages priority over others? Send the urgent ones with a low type number and receive with a negative type. The lowest type present is delivered first, whenever it arrived.

6. What does a non-blocking receive return on an empty queue? Minus one, with errno set to ENOMSG, whose message is "No message of desired type".

7. Is non-blocking communication better than blocking? No. A blocked process is asleep and uses no processor time; a non-blocking receive in a loop is a busy-wait that keeps a processor working for nothing. Non-blocking is right only when the process has other work to do between checks.

8. How does a consumer on a message queue know the producer has finished? The producer sends a final message of a distinct type, which the consumer treats as an end marker. A queue has no end of file, because the kernel does not know who is using it.

9. Give the one-line difference between shared memory and message passing. Shared memory copies nothing and synchronises nothing; message passing copies the data twice and does the synchronising for you.