diff options
| author | Brad Davis <[email protected]> | 2013-06-27 11:25:32 -0800 |
|---|---|---|
| committer | Brad Davis <[email protected]> | 2013-06-28 10:47:29 -0700 |
| commit | ebefcc885f74461cd0e3f19b5ae3622dc6cf6dbc (patch) | |
| tree | 2b16db7350fce54c2e6c1b1c4020d67419cb1164 /LibOVR/Src/OVR_ThreadCommandQueue.cpp | |
| parent | 0ade748e1845694c5cbe562fb823e56f09773e27 (diff) | |
SDK 0.2.2
Diffstat (limited to 'LibOVR/Src/OVR_ThreadCommandQueue.cpp')
| -rw-r--r-- | LibOVR/Src/OVR_ThreadCommandQueue.cpp | 370 |
1 files changed, 370 insertions, 0 deletions
diff --git a/LibOVR/Src/OVR_ThreadCommandQueue.cpp b/LibOVR/Src/OVR_ThreadCommandQueue.cpp new file mode 100644 index 0000000..d9dbab3 --- /dev/null +++ b/LibOVR/Src/OVR_ThreadCommandQueue.cpp @@ -0,0 +1,370 @@ +/************************************************************************************
+
+PublicHeader: None
+Filename : OVR_ThreadCommandQueue.cpp
+Content : Command queue for operations executed on a thread
+Created : October 29, 2012
+
+Copyright : Copyright 2012 Oculus VR, Inc. All Rights reserved.
+
+Use of this software is subject to the terms of the Oculus license
+agreement provided at the time of installation or download, or which
+otherwise accompanies this software in either electronic or hard copy form.
+
+************************************************************************************/
+
+#include "OVR_ThreadCommandQueue.h"
+
+namespace OVR {
+
+
+//------------------------------------------------------------------------
+// ***** CircularBuffer
+
+// CircularBuffer is a FIFO buffer implemented in a single block of memory,
+// which allows writing and reading variable-size data chucks. Write fails
+// if buffer is full.
+
+class CircularBuffer
+{
+ enum {
+ AlignSize = 16,
+ AlignMask = AlignSize - 1
+ };
+
+ UByte* pBuffer;
+ UPInt Size;
+ UPInt Tail; // Byte offset of next item to be popped.
+ UPInt Head; // Byte offset of where next push will take place.
+ UPInt End; // When Head < Tail, this is used instead of Size.
+
+ inline UPInt roundUpSize(UPInt size)
+ { return (size + AlignMask) & ~(UPInt)AlignMask; }
+
+public:
+
+ CircularBuffer(UPInt size)
+ : Size(size), Tail(0), Head(0), End(0)
+ {
+ pBuffer = (UByte*)OVR_ALLOC_ALIGNED(roundUpSize(size), AlignSize);
+ }
+ ~CircularBuffer()
+ {
+ // For ThreadCommands, we must consume everything before shutdown.
+ OVR_ASSERT(IsEmpty());
+ OVR_FREE_ALIGNED(pBuffer);
+ }
+
+ bool IsEmpty() const { return (Head == Tail); }
+
+ // Allocates a state block of specified size and advances pointers,
+ // returning 0 if buffer is full.
+ UByte* Write(UPInt size);
+
+ // Returns a pointer to next available data block; 0 if none available.
+ UByte* ReadBegin()
+ { return (Head != Tail) ? (pBuffer + Tail) : 0; }
+ // Consumes data of specified size; this must match size passed to Write.
+ void ReadEnd(UPInt size);
+};
+
+
+// Allocates a state block of specified size and advances pointers,
+// returning 0 if buffer is full.
+UByte* CircularBuffer::Write(UPInt size)
+{
+ UByte* p = 0;
+
+ size = roundUpSize(size);
+ // Since this is circular buffer, always allow at least one item.
+ OVR_ASSERT(size < Size/2);
+
+ if (Head >= Tail)
+ {
+ OVR_ASSERT(End == 0);
+
+ if (size <= (Size - Head))
+ {
+ p = pBuffer + Head;
+ Head += size;
+ }
+ else if (size < Tail)
+ {
+ p = pBuffer;
+ End = Head;
+ Head = size;
+ OVR_ASSERT(Head != Tail);
+ }
+ }
+ else
+ {
+ OVR_ASSERT(End != 0);
+
+ if ((Tail - Head) > size)
+ {
+ p = pBuffer + Head;
+ Head += size;
+ OVR_ASSERT(Head != Tail);
+ }
+ }
+
+ return p;
+}
+
+void CircularBuffer::ReadEnd(UPInt size)
+{
+ OVR_ASSERT(Head != Tail);
+ size = roundUpSize(size);
+
+ Tail += size;
+ if (Tail == End)
+ {
+ Tail = End = 0;
+ }
+ else if (Tail == Head)
+ {
+ OVR_ASSERT(End == 0);
+ Tail = Head = 0;
+ }
+}
+
+
+//-------------------------------------------------------------------------------------
+// ***** ThreadCommand
+
+ThreadCommand::PopBuffer::~PopBuffer()
+{
+ if (Size)
+ Destruct<ThreadCommand>(toCommand());
+}
+
+void ThreadCommand::PopBuffer::InitFromBuffer(void* data)
+{
+ ThreadCommand* cmd = (ThreadCommand*)data;
+ OVR_ASSERT(cmd->Size <= MaxSize);
+
+ if (Size)
+ Destruct<ThreadCommand>(toCommand());
+ Size = cmd->Size;
+ memcpy(Buffer, (void*)cmd, Size);
+}
+
+void ThreadCommand::PopBuffer::Execute()
+{
+ ThreadCommand* command = toCommand();
+ OVR_ASSERT(command);
+ command->Execute();
+ if (NeedsWait())
+ GetEvent()->PulseEvent();
+}
+
+//-------------------------------------------------------------------------------------
+
+class ThreadCommandQueueImpl : public NewOverrideBase
+{
+ typedef ThreadCommand::NotifyEvent NotifyEvent;
+ friend class ThreadCommandQueue;
+
+public:
+
+ ThreadCommandQueueImpl(ThreadCommandQueue* queue)
+ : pQueue(queue), CommandBuffer(2048),
+ ExitEnqueued(false), ExitProcessed(false)
+ {
+ }
+ ~ThreadCommandQueueImpl();
+
+
+ bool PushCommand(const ThreadCommand& command);
+ bool PopCommand(ThreadCommand::PopBuffer* popBuffer);
+
+
+ // ExitCommand is used by notify us that Thread is shutting down.
+ struct ExitCommand : public ThreadCommand
+ {
+ ThreadCommandQueueImpl* pImpl;
+
+ ExitCommand(ThreadCommandQueueImpl* impl, bool wait)
+ : ThreadCommand(sizeof(ExitCommand), wait, true), pImpl(impl) { }
+
+ virtual void Execute() const
+ {
+ Lock::Locker lock(&pImpl->QueueLock);
+ pImpl->ExitProcessed = true;
+ }
+ virtual ThreadCommand* CopyConstruct(void* p) const
+ { return Construct<ExitCommand>(p, *this); }
+ };
+
+
+ NotifyEvent* AllocNotifyEvent_NTS()
+ {
+ NotifyEvent* p = AvailableEvents.GetFirst();
+
+ if (!AvailableEvents.IsNull(p))
+ p->RemoveNode();
+ else
+ p = new NotifyEvent;
+ return p;
+ }
+
+ void FreeNotifyEvent_NTS(NotifyEvent* p)
+ {
+ AvailableEvents.PushBack(p);
+ }
+
+ void FreeNotifyEvents_NTS()
+ {
+ while(!AvailableEvents.IsEmpty())
+ {
+ NotifyEvent* p = AvailableEvents.GetFirst();
+ p->RemoveNode();
+ delete p;
+ }
+ }
+
+ ThreadCommandQueue* pQueue;
+ Lock QueueLock;
+ volatile bool ExitEnqueued;
+ volatile bool ExitProcessed;
+ List<NotifyEvent> AvailableEvents;
+ List<NotifyEvent> BlockedProducers;
+ CircularBuffer CommandBuffer;
+};
+
+
+
+ThreadCommandQueueImpl::~ThreadCommandQueueImpl()
+{
+ Lock::Locker lock(&QueueLock);
+ OVR_ASSERT(BlockedProducers.IsEmpty());
+ FreeNotifyEvents_NTS();
+}
+
+bool ThreadCommandQueueImpl::PushCommand(const ThreadCommand& command)
+{
+ ThreadCommand::NotifyEvent* completeEvent = 0;
+ ThreadCommand::NotifyEvent* queueAvailableEvent = 0;
+
+ // Repeat writing command into buffer until it is available.
+ do {
+
+ { // Lock Scope
+ Lock::Locker lock(&QueueLock);
+
+ if (queueAvailableEvent)
+ {
+ FreeNotifyEvent_NTS(queueAvailableEvent);
+ queueAvailableEvent = 0;
+ }
+
+ // Don't allow any commands after PushExitCommand() is called.
+ if (ExitEnqueued && !command.ExitFlag)
+ return false;
+
+
+ bool bufferWasEmpty = CommandBuffer.IsEmpty();
+ UByte* buffer = CommandBuffer.Write(command.GetSize());
+ if (buffer)
+ {
+ ThreadCommand* c = command.CopyConstruct(buffer);
+ if (c->NeedsWait())
+ completeEvent = c->pEvent = AllocNotifyEvent_NTS();
+ // Signal-waker consumer when we add data to buffer.
+ if (bufferWasEmpty)
+ pQueue->OnPushNonEmpty_Locked();
+ break;
+ }
+
+ queueAvailableEvent = AllocNotifyEvent_NTS();
+ BlockedProducers.PushBack(queueAvailableEvent);
+ } // Lock Scope
+
+ queueAvailableEvent->Wait();
+
+ } while(1);
+
+ // Command was enqueued, wait if necessary.
+ if (completeEvent)
+ {
+ completeEvent->Wait();
+ Lock::Locker lock(&QueueLock);
+ FreeNotifyEvent_NTS(completeEvent);
+ }
+
+ return true;
+}
+
+
+// Pops the next command from the thread queue, if any is available.
+bool ThreadCommandQueueImpl::PopCommand(ThreadCommand::PopBuffer* popBuffer)
+{
+ Lock::Locker lock(&QueueLock);
+
+ UByte* buffer = CommandBuffer.ReadBegin();
+ if (!buffer)
+ {
+ // Notify thread while in lock scope, enabling initialization of wait.
+ pQueue->OnPopEmpty_Locked();
+ return false;
+ }
+
+ popBuffer->InitFromBuffer(buffer);
+ CommandBuffer.ReadEnd(popBuffer->GetSize());
+
+ if (!BlockedProducers.IsEmpty())
+ {
+ ThreadCommand::NotifyEvent* queueAvailableEvent = BlockedProducers.GetFirst();
+ queueAvailableEvent->RemoveNode();
+ queueAvailableEvent->PulseEvent();
+ // Event is freed later by waiter.
+ }
+ return true;
+}
+
+
+//-------------------------------------------------------------------------------------
+
+ThreadCommandQueue::ThreadCommandQueue()
+{
+ pImpl = new ThreadCommandQueueImpl(this);
+}
+ThreadCommandQueue::~ThreadCommandQueue()
+{
+ delete pImpl;
+}
+
+bool ThreadCommandQueue::PushCommand(const ThreadCommand& command)
+{
+ return pImpl->PushCommand(command);
+}
+
+bool ThreadCommandQueue::PopCommand(ThreadCommand::PopBuffer* popBuffer)
+{
+ return pImpl->PopCommand(popBuffer);
+}
+
+void ThreadCommandQueue::PushExitCommand(bool wait)
+{
+ // Exit is processed in two stages:
+ // - First, ExitEnqueued flag is set to block further commands from queuing up.
+ // - Second, the actual exit call is processed on the consumer thread, flushing
+ // any prior commands.
+ // IsExiting() only returns true after exit has flushed.
+ {
+ Lock::Locker lock(&pImpl->QueueLock);
+ if (pImpl->ExitEnqueued)
+ return;
+ pImpl->ExitEnqueued = true;
+ }
+
+ PushCommand(ThreadCommandQueueImpl::ExitCommand(pImpl, wait));
+}
+
+bool ThreadCommandQueue::IsExiting() const
+{
+ return pImpl->ExitProcessed;
+}
+
+
+} // namespace OVR
|