How can a task wait on multiple vxworks Queues?

Question

We have a vxWorks design which requires one task to process both high and low priority messages sent over two message queues.
The messages for a given priority have to be processed in FIFO order.

For example, process all the high priority messages in the order they were received, then process the low priority messages. If there is no high priority message, then process the low priority message immediately.

Is there a way to do this?

Answer 1

If you use named pipes (pipeDevCreate(), write(), read()) instead of message queues, you can use select() to block until there are messages in either pipe.

Whenever select() triggers, you process all messages in the high priority pipe. Then you process a single message from the low priority pipe. Then call select again (loop).

Example Code snippets:

 // Initialization: Create high and low priority named pipes
 pipeDrv(); //initialize pipe driver
 int fdHi = pipeDevCreate("/pipe/high",numMsgs,msgSize);
 int fdLo = pipeDevCreate("/pipe/low",numMsgs,msgSize);

 ...

 // Message sending thread: Add messages to pipe
 write(fdHi, buf, sizeof(buf));

 ...

 // Message processing Thread: select loop
 fd_set rdFdSet;

 while(1)
 {
     FD_ZERO(&rdFdSet);
     FD_SET(fdHi, &rdFdSet);
     FD_SET(fdLo, &rdFdSet;

     if (select(FD_SETSIZE, &rdFdSet, NULL, NULL, NULL) != ERROR)
     {
         if (FD_ISSET(fdHi, &rdFdSet))
         {
             // process all high-priority messages
             while(read(fdHi,buf,size) > 0)
             {
                 //process high-priority
             }
         }

         if (FD_ISSET(fdLo, &rdFdSet))
         {
             // process a single low priority message
             if (read(fdLo,buf,size) > 0)
             {
                 // process low priority
             }
         }
     }
 }

Answer 2

In vxWorks, you can't wait directly on multiple queues. You can however use the OS events (from eventLib) to achieve this result. Here is a simple code snippet:

MSG_Q_ID lowQ, hiQ;

void Init() {
// Task Initialization Code.  This should be called from the task that will
// be receiving the messages
...
hiQ = msgQCreate(...);
lowQ = msgQCreate(...);
msgQEvStart(hiQ, VX_EV01);  // Event 1 sent when hiQ receives message
msgQEvStart(loQ, VX_EV02);  // Event 2 sent when loQ receives message
...
}
void RxMessages() {
...
  UINT32 ev;   // Event received

   // Blocks until we receive Event 1 or 2
   eventReceive(VX_EV01 | VX_EV02, EVENT_WAIT_ANY, WAIT_FOREVER, &ev);
   if(ev & VX_EV01) {
      msgQReceive(hiQ, ...);
   }
   if(ev & VX_EV02) {
      msgQReceive(loQ, ...);
   }
}

Note that you need to modify that code to make sure you drain all your queues in case there is more than one message that was received.

The same mechanism can also be applied to Binary semaphores using the semEvStart() function.