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-freevariable length event entriesstrictly sequential timestampsmulti-processor safetolerant of overruns/data dropsusable from both kernel and user space codeAs 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 body2-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));
Real-Time MP/SC Lock Free Ring
c;lock free
Consumer bugUnless 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;        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;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.I'm surprised that you haven't noticed this problem.  I wonder if your producer also copies past the end of the buffer?  You never showed us the part where the producer copies the body.  If that code had a similar problem, it could explain why the code might appear to be working.Memory barriersNormally, you would need to use memory barriers on both the producer and consumer sides to ensure that the content and tail updates are seen in the correct order.  However, I'm assuming this program is running on an x86 target, where these barriers are not needed due to the strong memory ordering guarantees on that architecture.