Inter-Task and Inter-Process Communication in VxWorks

Click “Read the original text” to access more VxWorks resources

VxWorks provides several different mechanisms for inter-task communication. The different methods available include:

Semaphores

Semaphores are the primary means of synchronization between tasks in VxWorks. There are three different types of semaphores: binary semaphores, counting semaphores, and mutex semaphores. Binary semaphores are used for simple synchronization between tasks for critical regions. Counting semaphores allow multiple simultaneous access to critical regions representing resources (which can be acquired in multiple ways). Mutex semaphores are unique in that they provide solutions to some inherent issues of basic semaphore types (including increased priority inversion protection, deletion safety, and return access to resources). All semaphores in VxWorks also allow for timeouts related to resource acquisition to prevent indefinite waiting.

When establishing a semaphore, the program can also specify a wake-up mechanism for tasks waiting for the semaphore (i.e., about to become available). There are two different wake-up mechanisms available: a priority-based wake-up mechanism that wakes the highest priority task, and a first-in-first-out (FIFO) wake-up mechanism that wakes the task that has been waiting the longest (regardless of the priority of other blocked tasks). VxWorks also provides access to POSIX semaphores.

Message Queues and Pipes

VxWorks provides message queues as the primary means of inter-task communication. Message queues can hold a variable number of messages, each of different lengths. Application messages are queued in the message queue using the msgQSend function. The receiver will remain blocked until there is available information. When information is available, the blocked receiver will be awakened using one of the following two wake-up mechanisms: based on priority or FIFO.

VxWorks’s native message queues only provide two similar priorities: normal priority and urgent priority (messages are stored at the front of the queue when sent). VxWorks also provides POSIX message queues (mq_*) with 32 different priorities.

Pipes provide another means of inter-process communication. Pipes offer a “select” function that allows tasks to wait for data from any set of file descriptors (input/output devices). QNX Neutrino provides a pipe interface implementation that meets the requirements of the POSIX 1003.1 interface specification.

Example of VxWorks Message Queue

This example illustrates how to implement synchronization and communication between tasks in VxWorks using message queues. First, we need to establish a message queue that will be shared among different tasks. This results in four tasks: two sender tasks and two receiver tasks.

#include "msgQLib.h"
#include "stdio.h"
#include "sysLib.h"
#include "string.h"
#include "taskLib.h"
#define MAX_NUM_MSG 5
#define MAX_MSG_LEN 100
#define NUM_MSGS 2*MAX_NUM_MSG
MSG_Q_ID msgQId;
void sendTask1(void)
{
    static char buf[MAX_MSG_LEN];
    int i;

    for (i = 0; i < NUM_MSGS; i++) {
        /* Yield to allow other tasks to run. Otherwise FIFO
           means that this task will run until blocked. */
        taskDelay(0);
        /* Write and send a message. */
        sprintf(buf, "Send1 %d", i);
        if (msgQSend(msgQId, buf, strlen(buf) + 1, WAIT_FOREVER, MSG_PRI_NORMAL) == ERROR) {
            printf("Send 1 msgQSend failed!\n");
            return;
        }
        if (i == 3) {
            /* Send an urgent message. */
            sprintf(buf, "Send1 URG %d", i);
            if (msgQSend(msgQId, buf, strlen(buf) + 1, WAIT_FOREVER, MSG_PRI_URGENT) == ERROR) {
                printf("Send 1 msgQSend failed!\n");
                return;
            }
        }
    }
}

void sendTask2(void)
{
    static char buf[MAX_MSG_LEN];
    int i;

    taskDelay(sysClkRateGet());
    for (i = 0; i < NUM_MSGS; i++) {
        taskDelay(0);
        sprintf(buf, "Send2 %d", i);
        if (msgQSend(msgQId, buf, strlen(buf) + 1, WAIT_FOREVER, MSG_PRI_NORMAL) == ERROR) {
            printf("Send 2 msgQSend failed!\n");
            return;
        }
        if (i == 5) {
            sprintf(buf, "Send2 URG %d", i);
            if (msgQSend(msgQId, buf, strlen(buf) + 1, WAIT_FOREVER, MSG_PRI_URGENT) == ERROR) {
                printf("Send 2 msgQSend failed!\n");
                return;
            }
        }
    }
}

void receiveTask1(void)
{
    char buf[MAX_MSG_LEN];
    int i;

    for (i = 0; i < NUM_MSGS; i++) {
        if (msgQReceive(msgQId, buf, MAX_MSG_LEN, WAIT_FOREVER) == ERROR) {
            printf("Rx task 1 msgQReceive failed.\n");
        } else {
            printf("Task 1 received %s\n", buf);
        }
    }
}

void receiveTask2(void)
{
    char buf[MAX_MSG_LEN];
    int i;

    for (i = 0; i < 10; i++) {
        if (msgQReceive(msgQId, buf, MAX_MSG_LEN, WAIT_FOREVER) == ERROR) {
            printf("Rx task 2 msgQReceive failed.\n");
        } else {
            printf("Task 2 received %s\n", buf);
        }
    }
}

void msgQ_init(void)
{
    int pri;
    taskPriorityGet(0, &pri);
    pri++;                      /* Priority is less than init’s priority, so init
                                   can run to completion. */
    if ((msgQId = msgQCreate(MAX_NUM_MSG, MAX_MSG_LEN, MSG_Q_FIFO)) == NULL) {
        printf("msgQCreate failed!\n");
        return;
    }
    if (taskSpawn("task1Tx", pri, 0, 1000, (FUNCPTR) sendTask1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) == ERROR) {
        printf("taskSpawn Tx1 failed!\n");
    }
    if (taskSpawn("task2Tx", pri, 0, 1000, (FUNCPTR) sendTask2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) == ERROR) {
        printf("taskSpawn Tx 2 failed!\n");
    }
    if (taskSpawn("task1Rx", pri, 0, 1000, (FUNCPTR) receiveTask1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) == ERROR) {
        printf("taskSpawn Rx 1 failed!\n");
    }
    if (taskSpawn("task2Rx", pri, 0, 1000, (FUNCPTR) receiveTask2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) == ERROR) {
        printf("taskSpawn Rx 2 failed!\n");
    }
}