3
\$\begingroup\$

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));
\$\endgroup\$
  • \$\begingroup\$ Do you assume that this program is running on an x86 platform or not? \$\endgroup\$ – JS1 Sep 10 '15 at 7:12
  • \$\begingroup\$ This is currently running on an x86_64 platform, but I'd like to have it be portable if that's practical. \$\endgroup\$ – Speed8ump Sep 10 '15 at 12:55
  • \$\begingroup\$ 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. \$\endgroup\$ – Simon Forsberg Sep 10 '15 at 13:10
2
\$\begingroup\$

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;
        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 barriers

Normally, 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.

\$\endgroup\$
  • \$\begingroup\$ 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) ). \$\endgroup\$ – Speed8ump Sep 10 '15 at 12:58
  • \$\begingroup\$ x86-TSO doesn't necessarily preclude the need for memory barriers. Stores to the same location have a total order across all processors, but that is not true of stores to different locations, even though there is a total order for all stores from any particular core. It seems plausible that given the MP/SC nature, many processors (across many sockets) are doing these multi-location stores, and memory barriers may still be needed to guarantee some of the state of the log structure used for producing / consuming. \$\endgroup\$ – dho Sep 12 '15 at 17:02
  • \$\begingroup\$ @dho I don't see how a memory fence on x86 would fix any of the potential problems you are talking about. \$\endgroup\$ – JS1 Sep 12 '15 at 18:59

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.