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;
struct LogInfo_struct
{
LogIndex_t m_need; //declaration of space needed
LogIndex_t m_head; //current head of buffered log data
LogIndex_t m_alloc; //allocated out beyond working area
LogIndex_t m_filled; //committed complete data
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);
LogVar_t need, target, head, newHead, zero, alloc;
struct timespec timeStamp;
int i;
memset(openLoopCount,0,3*sizeof(int));
//first, push the needed position out by the size required
need = atomic_fetch_add(&g_log->m_need,size) + size;
//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
{
head = newHead = atomic_load(&g_log->m_head);
if (((newHead - zero) & LOG_SIZE_MASK) >= target) break;
do
{
newHead += g_log->m_logBuffer[newHead & LOG_SIZE_MASK] + 1 + sizeof(struct timespec);
}
while( ((newHead - zero) & LOG_SIZE_MASK) < target );
}
while(!atomic_compare_exchange_strong(&g_log->m_head, &head, newHead));
//either I moved the head down beyond all allocated areas, or someone else did
do
{
alloc = atomic_load(&g_log->m_alloc);
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
g_log->m_logBuffer[alloc & LOG_SIZE_MASK] = bodySize;
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;
newFilled = atomic_fetch_add(&g_log->m_filled,size) + size;
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
{
tail = atomic_load(&g_log->m_tail);
need = atomic_load(&g_log->m_need);
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;
LogVar_t head, tail;
struct timespec captureTime;
do
{
clock_gettime(CLOCK_MONOTONIC_RAW, &captureTime);
head = atomic_load(&m_log->m_head);
tail = atomic_load(&m_log->m_tail);
uint8_t *tmps;
//pull out a chunk of pending entries
if ( head == tail )
{
didWork = false;
break;
}
else if ( head > tail )
{
int len = (LOG_SIZE - head) & LOG_SIZE_MASK;
if (len > 1000) len = 1000;
memcpy(work,&m_log->m_logBuffer[head & LOG_SIZE_MASK],len);
tmps = &m_log->m_logBuffer[0];
addLen = tail & LOG_SIZE_MASK;
}
else
{
len = 0;
tmps = &m_log->m_logBuffer[head & LOG_SIZE_MASK];
addLen = (tail - head) & LOG_SIZE_MASK;
}
if (addLen > 1000 - len) addLen = 1000 - len;
if (addLen) memcpy(&work[len], tmps, addLen);
len += addLen;
if ( len >= 1000 )
{
//might have truncated, find proper new head
addLen = 0;
while( (int)(addLen) < len )
{
int fsize = work[addLen] + 1 + sizeof(struct timespec);
if (addLen + fsize <= len)
addLen += fsize;
else
break;
}
tail = head + addLen; //new
}
}
while(!atomic_compare_exchange_strong(&m_log->m_head,&head,tail));
