# Real-Time MP/SC Lock Free Ring

This is (the significant) part of an implementation of a logging mechanism for recording events in a real-time control system. The basic requirements are:

• No dynamic memory on either side (producers or consumer),
• Lock-free
• variable length event entries
• strictly sequential timestamps
• multi-processor safe
• tolerant of overruns/data drops
• usable from both kernel and user space code

As I've never written a lock free algorithm before, I'm hoping for some comment on any concurrency problems.I have this running, but I am very occasionally seeing a corrupted log message, which seems to have been published (tail moved forwards) before the body is completed. I'm hoping I'm missing something simple.

The system holds log data in a shared memory area (or several, depending on the use case). Typically each process in the system will create one log buffer of the form:

typedef atomic_uint_least32_t LogIndex_t;
typedef uint32_t LogVar_t;

{
LogIndex_t m_need;                     //declaration of space needed
LogIndex_t m_alloc;                    //allocated out beyond working area
LogIndex_t m_tail;                     //safe location at tail of buffered log data
uint8_t    m_logBuffer[LOG_SIZE];      //head of log buffer
char       m_strings[1<<12][64];       //copied and validated table of strings
} *g_log = 0;


Where LOG_SIZE is a power of 2, and the string table is used to encode longer, reused items. I won't be discussing the shared memory setup or the actual content of the message body, since this is unimportant here. The structure of each log entry is:

• 1 byte length of body
• 2-16 byte timestamp (depending on requirements)
• X bytes of arbitrary data (0-255 bytes)

Logging involves computing the body length and content, calling openEntry, writing the body content into the provided frame, and calling closeEntry.

static inline void bh_log_getTimeStamp( struct timespec *time )
{
#if defined(__KERNEL__)
getrawmonotonic(time);
#else
clock_gettime(CLOCK_MONOTONIC_RAW, time);
#endif
}

static LogVar_t bh_log_openEntry( uint8_t bodySize )
{
uint16_t size = bodySize + 1 + sizeof(struct timespec);
struct timespec timeStamp;
int i;
memset(openLoopCount,0,3*sizeof(int));

//first, push the needed position out by the size required

//then verify that the head is beyond the need position (using tail as reference)
zero = atomic_load(&g_log->m_tail) + 1;    //tail can't move until we commit
target = (need - zero) & LOG_SIZE_MASK;

//ensure the head is moved "far enough" to free up space for allocation
do
{

do
{
}
}
//either I moved the head down beyond all allocated areas, or someone else did

do
{
target = alloc + size;
bh_log_getTimeStamp(&timeStamp);
}
while( !atomic_compare_exchange_strong(&g_log->m_alloc,  &alloc,  target) );
//alloc == index of allocated location, timestamp = captured timestamp to use

for(i=0; i < sizeof(struct timespec); ++i)
{
g_log->m_logBuffer[(alloc + 1 + i) & LOG_SIZE_MASK] = ((uint8_t*)(&timeStamp))[i];
}
return alloc + 1 + sizeof(struct timespec);
}

static void bh_log_closeEntry( uint8_t bodySize )
{
uint16_t size = bodySize + 1 + sizeof(struct timespec);
LogVar_t need, newFilled, tail;

do      //I'm not sure this loop is needed.  If tail moved, then someone else passed
//this logic in the interim.  We can possibly just try once and give up
{

if (newFilled != need) break;
}
while( !atomic_compare_exchange_strong(&g_log->m_tail,&tail,newFilled) );
}


On the backend I have a (non-realtime) task which pulls out chunks of entries (anything [m_head..m_tail)) and moves m_head forward (ideally, faster than the m_tail is moving on average). This backend service collates multiple log buffers, converts the log entries into human readable format, and does UI and file system interactions.

The relevant consumer portion of that code looks like this:

    uint8_t work[1000];
bool didWork = true;
int len = 0, addLen = 0;
struct timespec captureTime;

do
{
clock_gettime(CLOCK_MONOTONIC_RAW, &captureTime);

uint8_t *tmps;

//pull out a chunk of pending entries
if ( head == tail )
{
didWork = false;
break;
}
else if ( head > tail )
{
if (len > 1000) len = 1000;
tmps = &m_log->m_logBuffer[0];
}
else
{
len = 0;
}

if ( len >= 1000 )
{
//might have truncated, find proper new head
{
int fsize = work[addLen] + 1 + sizeof(struct timespec);
if (addLen + fsize <= len)
else
break;
}
}
}

• Do you assume that this program is running on an x86 platform or not? – JS1 Sep 10 '15 at 7:12
• This is currently running on an x86_64 platform, but I'd like to have it be portable if that's practical. – Speed8ump Sep 10 '15 at 12:55
• Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers. – Simon Forsberg Sep 10 '15 at 13:10

# Consumer bug

Unless I am misunderstanding something, there seems to be a big problem in the consumer. Your variables m_head and m_tail appear to start at 0 and move forward indefinitely, without wrapping around at LOG_SIZE (which is fine). But the way the consumer interprets head and tail looks like it expected them to wrap around. For example, I don't see how this if could ever be true:

    else if ( head > tail )


And when the code inevitably falls into the next case:

    else
{
len = 0;
}


You will be copying past the end of m_logBuffer whenever head gets close to the end of the buffer. For example, suppose LOG_SIZE is 32768, head is 32700, and tail is 32800. The above code will copy 33 bytes past the end of the buffer.
• Good catch. The else if ( head > tail ) looks to be the reason for the occasional bug mentioned. Everywhere else that uses any of the index values very carefully applies the LOG_SIZE_MASK. The test should be ( (head & LOG_SIZE_MASK) > (tail & LOG_SIZE_MASK) ). – Speed8ump Sep 10 '15 at 12:58