HH snippet

gistfile1.txt

#include <pthread.h>

struct osx_thread_control
{
    pthread_mutex_t *SuspendMutex;
    pthread_cond_t *ResumeCond;
};

struct work_queue_entry_storage
{
    void *UserPointer;
};

struct work_queue
{
    uint32 volatile EntryCompletionCount;
    uint32 volatile NextEntryToDo;
    uint32 volatile EntryCount;

    work_queue_entry_storage Entries[256];
};

struct work_queue_entry
{
    void *Data;
    bool32 IsValid;
};

inline uint32
InterlockedIncrement(uint32 volatile *p)
{
    return __atomic_add_fetch(p, 1, __ATOMIC_SEQ_CST);
}

internal void
AddWorkQueueEntry(work_queue *Queue, osx_thread_control *Control, void *Pointer)
{
    Assert(Queue->EntryCount < ArrayCount(Queue->Entries));
    Queue->Entries[Queue->EntryCount].UserPointer = Pointer;

    asm volatile("" ::: "memory");
    _mm_sfence();

    ++Queue->EntryCount;

    pthread_mutex_lock(Control->SuspendMutex);
    pthread_cond_broadcast(Control->ResumeCond);
    pthread_mutex_unlock(Control->SuspendMutex);
}

internal work_queue_entry
CompleteAndGetNextWorkQueueEntry(work_queue *Queue, work_queue_entry Completed)
{
    work_queue_entry Result;
    Result.IsValid = false;

    if (Completed.IsValid)
    {
        InterlockedIncrement(&Queue->EntryCompletionCount);
    }

    if (Queue->NextEntryToDo < Queue->EntryCount)
    {
        uint32 Index = InterlockedIncrement(&Queue->NextEntryToDo) - 1;
        Result.Data = Queue->Entries[Index].UserPointer;
        Result.IsValid = true;

        asm volatile("" ::: "memory");
    }

    return Result;
}

internal bool32
QueueWorkStillInProgress(work_queue *Queue)
{
    bool32 Result = (Queue->EntryCount != Queue->EntryCompletionCount);

    return Result;
}

inline void
DoWorkerWork(work_queue_entry Entry, int LogicalThreadIndex)
{
    Assert(Entry.IsValid);
    printf("Thread %u: %s\n", LogicalThreadIndex, (char *)Entry.Data);
}

struct osx_thread_info
{
    int LogicalThreadIndex;
    pthread_t ThreadID;
    work_queue *Queue;
    osx_thread_control *Control;
};

void *
ThreadProc(void *lpParameter)
{
    osx_thread_info *ThreadInfo = (osx_thread_info *)lpParameter;
    work_queue_entry Entry = {};

    for(;;)
    {
        Entry = CompleteAndGetNextWorkQueueEntry(ThreadInfo->Queue, Entry);

        if (Entry.IsValid)
        {
            DoWorkerWork(Entry, ThreadInfo->LogicalThreadIndex);
        } else {
            pthread_mutex_lock(ThreadInfo->Control->SuspendMutex);
            pthread_cond_wait(ThreadInfo->Control->ResumeCond, ThreadInfo->Control->SuspendMutex);
            pthread_mutex_unlock(ThreadInfo->Control->SuspendMutex);
        }
    }
}

internal void
PushString(work_queue *Queue, osx_thread_control *Control, char *String)
{
    AddWorkQueueEntry(Queue, Control, String);
}

int main() {
    /* init hi-res time */

    osxInitHrtime();

    /* multithreading */

    pthread_mutex_t SuspendMutex;
    pthread_cond_t ResumeCond;

    pthread_mutex_init(&SuspendMutex, NULL);
    pthread_cond_init(&ResumeCond, NULL);

    osx_thread_control ThreadControl = {};
    ThreadControl.SuspendMutex = &SuspendMutex;
    ThreadControl.ResumeCond = &ResumeCond;

    osx_thread_info ThreadInfo[7];
    work_queue Queue = {};

    uint32 InitialCount = 1;
    uint32 ThreadCount = ArrayCount(ThreadInfo);

    for(uint32 ThreadIndex = 0;
        ThreadIndex < ThreadCount;
        ++ThreadIndex)
    {
        osx_thread_info *Info = ThreadInfo + ThreadIndex;

        Info->Queue = &Queue;
        Info->Control = &ThreadControl;
        Info->LogicalThreadIndex = ThreadIndex;

        pthread_create(&Info->ThreadID, NULL, &ThreadProc, Info);
    }

    PushString(&Queue, &ThreadControl, (char *)"String A0");
    PushString(&Queue, &ThreadControl, (char *)"String A1");
    PushString(&Queue, &ThreadControl, (char *)"String A2");
    PushString(&Queue, &ThreadControl, (char *)"String A3");
    PushString(&Queue, &ThreadControl, (char *)"String A4");
    PushString(&Queue, &ThreadControl, (char *)"String A5");
    PushString(&Queue, &ThreadControl, (char *)"String A6");
    PushString(&Queue, &ThreadControl, (char *)"String A7");
    PushString(&Queue, &ThreadControl, (char *)"String A8");
    PushString(&Queue, &ThreadControl, (char *)"String A9");

    PushString(&Queue, &ThreadControl, (char *)"String B0");
    PushString(&Queue, &ThreadControl, (char *)"String B1");
    PushString(&Queue, &ThreadControl, (char *)"String B2");
    PushString(&Queue, &ThreadControl, (char *)"String B3");
    PushString(&Queue, &ThreadControl, (char *)"String B4");
    PushString(&Queue, &ThreadControl, (char *)"String B5");
    PushString(&Queue, &ThreadControl, (char *)"String B6");
    PushString(&Queue, &ThreadControl, (char *)"String B7");
    PushString(&Queue, &ThreadControl, (char *)"String B8");
    PushString(&Queue, &ThreadControl, (char *)"String B9");

    work_queue_entry Entry = {};

    while (QueueWorkStillInProgress(&Queue))
    {
        Entry = CompleteAndGetNextWorkQueueEntry(&Queue, Entry);

        if (Entry.IsValid)
        {
            DoWorkerWork(Entry, 7);
        }
    }

    // ...

    for(uint ThreadIndex = 0;
        ThreadIndex < ThreadCount;
        ++ThreadIndex)
    {
        osx_thread_info *Info = ThreadInfo + ThreadIndex;
        pthread_cancel(Info->ThreadID);
    }

    pthread_mutex_destroy(&SuspendMutex);
    pthread_cond_destroy(&ResumeCond);

    mainWindow.close();

    return 0;
}