RK0 V0.8.0 – A Message-Passing kernel, if you want.

The changes in RK0 V0.8.0 were to be published a couple of months ago, but were not for reasons. What is really worth commenting on is the Message Passing.

Unified Message-Queue

On version 0.8.0 (except for the MRM Protocol), all message passing is constructed over a single abstraction: a Message Queue that passes data by copy.

PORTS: formal client–server endpoints

A Port is an endpoint for a server that executes a Remote Invocation on behalf of a client, loosely an RPC (Remote Procedure Call).

Symmetric client-priority execution

With Ports + RPC, the server executes a request at the client’s priority:

If client > server, the server is boosted (as before when Message Queue had ownership).
If client < server, the server is capped/demoted to the client’s priority for the duration of the call.
After the transaction (receive+reply) finishes, the server’s priority is restored.

Work is prioritised according to the requester’s needs, by design. That prevents a not-so-obvious priority inversion case. A server can enqueue several clients’ requests on its Port.

API and Usage Examples

kMesgQueue/Send/Recv/Peek/Jam/PostOvw/Query/Reset
kPortInit(queue, task) to create a server endpoint.
Client remote invocation: kPortSendRecv(...) (send request + wait for reply)
Server receive and return: kPortRecv(...), kPortReplyDone(...).
Back-compat convenience macros for mailbox-style usage still exist; they map cleanly onto the new queue.
Port Message Types:
- RK_PORT_MESG_2/4/8WORD (first two words are for meta-data, so 2WORD is when no payload is needed)
- If needing to send more than 8 words: RK_PORT_MESG_COOKIE (payload is an opaque (VOID*) pointer)

/* PORT RPC-FLOW SEQUENCE */
/*
The ServerTask calls PortRecv to get requests (kernel will enforce that only the correct server listens). When a mesg is recvd, the server executes _at the client's priority and processes the request_. It then calls PortReplyDone to send back a result code and to _signal_ the end of the RPC. The ClientTask uses PortSendRecv and block waiting for the reply.  Reply comes via mailbox, unblocking the client and, if there is, conveying a ret code.
*/

/* 
The example below computes a CRC on the server for the client’s payload and returns it as the reply code.
Logically, this is unbuffered: 
the client blocks until it gets the reply.
Note the server has its priority demoted while serving the the client. Upon finishing its priority is restored. 
(ret vals are not checked to reduce clutter)
*/

#include <application.h>
#include <logger.h>

/* aux functions, implementation omitted */
static inline UINT crc32(const VOID *data, ULONG size);
static inline BYTE xorshift8(void);


#define STACKSIZE     256
#define LOG_TASK_PRIO 3
/* covenience macro to declare task objects: handle name, entry function, stack storage and stack size  - no trailing ';'  */
RK_DECLARE_TASK(serverHandle, ServerTask, stack1, STACKSIZE)
RK_DECLARE_TASK(clientHandle, ClientTask, stack2, STACKSIZE)

/* declare port */
#define PORT_MESG_WORDS 4U   /* 2 words meta + 2 words payload */
#define PORT_CAPACITY 16
static RK_PORT serverPort;
/* conv macro to declare the port storage */
RK_DECLARE_PORT_BUF(portBuf, RK_PORT_MESG_4WORD, PORT_CAPACITY)


VOID kApplicationInit(void)
{
    kCreateTask(&serverHandle, ServerTask, RK_NO_ARGS,
               "Server", stack1, STACKSIZE, 1, RK_PREEMPT);
    kCreateTask(&clientHandle, ClientTask, RK_NO_ARGS,
               "Client", stack2, STACKSIZE, 2, RK_PREEMPT);
    /* init a PORT and attach a server task */
    kPortInit(&serverPort, portBuf, PORT_MESG_WORDS, PORT_CAPACITY, serverHandle);
  
    logInit(LOG_TASK_PRIO);
}

VOID ClientTask(VOID *args)
{
    RK_UNUSEARGS
    static RK_MAILBOX replyBox;  /* server replies here */
    kMailboxInit(&replyBox); /* init is just that */
    static BYTE vec[8];  
    for (UINT i = 0; i < 8; ++i) 
        vec[i] = xorshift8();
    static RK_PORT_MESG_4WORD mesg = {0};
    mesg.payload[0] = (ULONG) vec;  /* pointer as one word */
    mesg.payload[1] = 8;            /* number of bytes */
    UINT reply = 0;
    while(1)
    {
     /* Synchronous Call: send vector and expect CRC reply */
     UINT want = crc32(vec, 8);
     kPortSendRecv(&serverPort, (ULONG*)&mesg, &replyBox, &reply, RK_WAIT_FOREVER);
        logPost("[CLIENT] Need=0x%04X | Recvd=0x%04X", want, reply);
        /* if reply is correct, generate a new payload */
        if (want == reply)
        {
            for (UINT i = 0; i < 8; ++i) 
            {
               vec[i] = xorshift8();
            }
        }
        else
        {
           kassert(0);
        }
        kSleepPeriod(1000);
    }
}

VOID ServerTask(VOID *args)
{
    RK_UNUSEARGS
    static RK_PORT_MESG_4WORD mesg;
    while(1)
    {
        /* let this fella block here */
        kPortRecv(&serverPort, &mesg, RK_WAIT_FOREVER);
        BYTE  *vector = (BYTE*) mesg.payload[0];
        ULONG   size  =          mesg.payload[1];
        UINT    crc   = crc32(vector, size);
        
        /* it might have adopted client's priority */
        logPost("[SERVER] Will Reply CRC=0x%04X | Eff Prio=%d | Real Prio=%d", crc, RK_RUNNING_PRIO, RK_RUNNING_REAL_PRIO);

        /* to finish transaction: */
        kPortReplyDone(&serverPort, (ULONG const*)&mesg, crc);

        /* we should see priority restored */
        logPost("[SERVER] Finished. | Eff Prio: %d | Real Prio: %d", RK_RUNNING_PRIO, RK_RUNNING_REAL_PRIO);    
    }
}

As the title suggests, these features provide sufficient mechanisms for writing code around the kernel (possibly the entire operating system) using message passing, on a client-server paradigm.

For a straight application case, below is the Synchronisation Barrier that originally used CondVar Monitors (the application example that is shipped within every RK0 repo):

/* Synch barrier example using Message-Passing */

#include <application.h>
#include <logger.h>

#define LOG_PRIORITY 5
#define STACKSIZE 128

RK_DECLARE_TASK(task1Handle, Task1, stack1, STACKSIZE)
RK_DECLARE_TASK(task2Handle, Task2, stack2, STACKSIZE)
RK_DECLARE_TASK(task3Handle, Task3, stack3, STACKSIZE)
RK_DECLARE_TASK(barrierHandle, BarrierServer, stackB, STACKSIZE)

#define N_BARR_TASKS 3    /* logical barrier size */
#define PORT_CAPACITY 3U   
#define PORT_MESG_WORDS 2U /* meta-only (no payload) */

static RK_PORT barrierPort;
RK_DECLARE_PORT_BUF(barrierBuf, RK_PORT_MESG_2WORD, PORT_CAPACITY)

static inline VOID BarrierWaitPort(VOID)
{
    RK_MAILBOX replyBox ={0}; /* note the transient object per call */
    kMailboxInit(&replyBox); 

    RK_PORT_MESG_2WORD call = {0};
    UINT ack = 0;
    /* Send and pend for reply from server */
    kassert(!kPortSendRecv(&barrierPort, (ULONG *)&call, &replyBox, &ack,
                           RK_WAIT_FOREVER));
    K_UNUSE(ack); /* reply code unused (presence is the sync) */
}

VOID BarrierServer(VOID *args)
{
    RK_UNUSEARGS

    /* Store meta of blocked callers so we can reply later */
    RK_PORT_MESG_2WORD waiters[N_BARR_TASKS];
    UINT arrived = 0;

    while (1)
    {
        RK_PORT_MESG_2WORD mesg; /* meta-only message from a caller */
        kassert(!kPortRecv(&barrierPort, &mesg, RK_WAIT_FOREVER));
        
 
        if (arrived + 1U == N_BARR_TASKS)
        {
            /*  reply to all previous waiters ... */
            for (UINT i = 0; i < arrived; ++i)
                kassert(!kPortReply(&barrierPort, (ULONG const *)&waiters[i], 1U));

            /* ... and reply to the last one, ending the server transaction */
            kassert(!kPortReplyDone(&barrierPort, (ULONG const *)&mesg, 1U));

            arrived = 0;
         }
        else
        {
            waiters[arrived++] = mesg; /* keep caller meta for later reply */
        }
    }
}

VOID kApplicationInit(VOID)
{
    /*  server adopts callers during service... no mutex needed for prio inheritance */
    kassert(!kCreateTask(&barrierHandle, BarrierServer, RK_NO_ARGS,
                         "Barrier", stackB, STACKSIZE, 4, RK_PREEMPT));

    /* create a server : port + owner */
    kassert(!kPortInit(&barrierPort, barrierBuf, PORT_MESG_WORDS, PORT_CAPACITY,
                       barrierHandle));

    /* clients */
    kassert(!kCreateTask(&task1Handle, Task1, RK_NO_ARGS,
                         "Task1", stack1, STACKSIZE, 2, RK_PREEMPT));
    kassert(!kCreateTask(&task2Handle, Task2, RK_NO_ARGS,
                         "Task2", stack2, STACKSIZE, 3, RK_PREEMPT));
    kassert(!kCreateTask(&task3Handle, Task3, RK_NO_ARGS,
                         "Task3", stack3, STACKSIZE, 1, RK_PREEMPT));

    logInit(LOG_PRIORITY);
}

 VOID Task1(VOID *args)
{
    RK_UNUSEARGS
    while (1)
    {
        logPost("Task 1 is waiting at the barrier...");
        BarrierWaitPort();
        logPost("Task 1 passed the barrier!");
        kSleep(800);
    }
}

VOID Task2(VOID *args)
{
    RK_UNUSEARGS;
    while (1)
    {
        logPost("Task 2 is waiting at the barrier...");
        BarrierWaitPort();
        logPost("Task 2 passed the barrier!");
        kSleep(500);
    }
}

VOID Task3(VOID *args)
{
    RK_UNUSEARGS
    while (1)
    {
        logPost("Task 3 is waiting at the barrier...");
        BarrierWaitPort();
        logPost("Task 3 passed the barrier!");
        kSleep(300);
    }
}

The Message-Passing solution, for sure, has more overhead in size and might be a bit slower, given all the queue work. Yet, synchronising and exchanging data between tasks is handled only by the message-passing interface.

Use a Mailbox when you want last-message semantics (overwrite) or a first-message rendezvous (note, it is valid to have no interest in the message, only in the rendezvous).

Use a Port (RPC) when a single owner should own the state, and clients should call into it. You’ll get automatic client-priority execution and clean reply semantics.

Use a Message Queue when you want buffering on both the sender/receiver sides, such as for batching and decoupling the producer/consumer. Ownership and priority boosting can still apply on the receiving side.

RK0