In a previous post, the assignment of priorities was discussed. There are a couple of axioms involved. One must mention at least these two:
- No resource sharing, locks or busy-waits.
- Static priorities (i.e. the task with the highest static priority that is runnable immediately preempts all other tasks, and the task priority does not change dynamically)
Any kind of (spin)lock, mutex and or semaphore which is used to protect critical sections with regard to a shared resource could have severe impact on the real-time property of the system. At all cost we want to avoid priority inversion problems and hazardous deadlocks. A real-time system must be predictable and deterministic. Over the years, many different solutions have been proposed to deal with this problem.
Event passing
We advocate the use of lock free algorithms in combination with message or event passing. The same mechanism can be used to split a larger real-time task with a certain deadline, into several parts of smaller real-time or low priority tasks, i.e. real-time top-halves, real-time bottom-halves, background post-processing, etc…
The basic element that makes this possible is the lock-less single writer – single reader circular buffer or FIFO. This basic element used to be simple but nowadays – in the world of out-of-order code generation and out-of-order execution – it can be tricky to get it right. Memory barriers to the rescue… What follows here is the description of the algorithm in its simplest in-order-executed form.
The Pointer Queue or FIFO contains four members; the read index, the write index, the pointer to the array of pointers and the size of the array of pointers in number of elements.
typedef struct _PointerQueue
{
//! start offset of queue
volatile BYTE start;
//! end offset of queue
volatile BYTE end;
//! pointer to queue data
void** dataP;
//! event Queue size
BYTE size;
}
PointerQueue;
The FIFO is initialized by the PTRQ_Init call which sets the read and the write index to zero (and fills in the pointer reference to the array storage and its size).
void PTRQ_Init(PointerQueuePtr pQ,
void* pStorageArray,
BYTE nStorageSize)
{
pQ->start = 0;
pQ->end = 0;
pQ->dataP = pStorageArray;
pQ->size = nStorageSize;
}
There exist different options but the FIFO is defined to be empty when start (=read) index equals end (=write) index.
BOOLEAN PTRQ_IsEmpty(IN PointerQueuePtr pQ)
{
if (pQ->end == pQ->start)
{
return TRUE;
}
return FALSE;
}
This definition implies that at most (size – 1) elements can be stored in the FIFO.
BYTE PTRQ_GetUsedSpace(PointerQueuePtr pQ)
{
SCHAR usedBytes;
usedBytes = (SCHAR) (pQ->end - pQ->start);
if (usedBytes < 0)
{
usedBytes = (SCHAR) (usedBytes + pQ->size);
}
return (BYTE) usedBytes;
}
BYTE PTRQ_GetFreeSpace(PointerQueuePtr pQ)
{
BYTE freeBytes;
freeBytes = (BYTE) (pQ->size - PTRQ_GetUsedSpace(pQ) - 1);
return freeBytes;
}
When adding an element (or pushing one) to the FIFO, the order of code execution matters. First, it is determined that there is enough space in the FIFO for the additional element, then the element is added to the FIFO, and only then the end (write) index is changed. If the end of the array is reached, the index is reset to 0 (the beginning of the storage array): Standard circular buffer implementation.
BOOLEAN PTRQ_Push(PointerQueuePtr pQ, void* pEvent)
{
if (PTRQ_GetFreeSpace(pQ))
{
// get the end
BYTE end = pQ->end;
// Copy event into buffer
pQ->dataP[end] = pEvent;
++end;
if (end == pQ->size)
{
end = 0;
}
// Update queue end after the write!
pQ->end = end;
// return success
return TRUE;
}
return FALSE;
}
Removing an element from the FIFO is very similar. First the element is read out, and only then the start (read) index is altered.
BOOLEAN PTRQ_Pop(PointerQueuePtr pQ, void** pEvent)
{
if (NULL == pEvent)
{
return FALSE; // invalid event pointer or queue empty!
}
if(PTRQ_IsEmpty(pQ))
{
return FALSE; // invalid event pointer or queue empty!
}
else
{
// get the start
BYTE start = pQ->start;
// Get the character out of the buffer
*pEvent = pQ->dataP[start];
++start;
// Calculate next 'start' offset
if (start == pQ->size)
{
start = 0;
}
// Update queue start after the write!
pQ->start = start;
// return success
return TRUE;
}
}
So this simple setup allows us to move pre-allocated chunks of memory (or events or message structures) from writer to reader. The writer can be an interrupt function or a thread task. That does not matter, as long as its single writer — single reader, no locks are required. In an out-of-order world, the order must be enforced. The optimizing compiler can reorder code (which is partly remedied by using volatile variables), but also the instruction scheduler in the processor(s) can change the order in which instructions are executed. It is an excellent exercise (left to our blog readers) to figure out where to place memory barrier instructions like gnu gcc’s __sync_synchronize().
To move the pre-allocated chunks of memory or events structures or message structures between interrupts and tasks, two FIFOs are typically required. One containing the list of ‘free’ chunks, and one containing the list of ‘used’ chunks. The ‘writer’ pops a ‘free’ one from the ‘free chunk FIFO’, writes its info in the structure, and then pushes it to the ‘used’ list. The ‘reader’ pops the pointer to the structure from the ‘used chunk FIFO’, handles the info and when done, re-inserts the now free chunk into the ‘free chunk FIFO’.
About rtos.be
Speak Your Mind