1 files changed, 128 insertions, 0 deletions
diff --git a/indra/llcommon/workqueue.cpp b/indra/llcommon/workqueue.cpp
new file mode 100644
index 0000000000..ffc9a97dc0
--- /dev/null
+++ b/indra/llcommon/workqueue.cpp
@@ -0,0 +1,128 @@
+/**
+ * @file   workqueue.cpp
+ * @author Nat Goodspeed
+ * @date   2021-10-06
+ * @brief  Implementation for WorkQueue.
+ * 
+ * $LicenseInfo:firstyear=2021&license=viewerlgpl$
+ * Copyright (c) 2021, Linden Research, Inc.
+ * $/LicenseInfo$
+ */
+
+// Precompiled header
+#include "linden_common.h"
+// associated header
+#include "workqueue.h"
+// STL headers
+// std headers
+// external library headers
+// other Linden headers
+#include "llcoros.h"
+#include LLCOROS_MUTEX_HEADER
+#include "llerror.h"
+#include "llexception.h"
+#include "stringize.h"
+
+using Mutex = LLCoros::Mutex;
+using Lock  = LLCoros::LockType;
+
+LL::WorkQueue::WorkQueue(const std::string& name):
+    super(makeName(name))
+{
+    // TODO: register for "LLApp" events so we can implicitly close() on
+    // viewer shutdown.
+}
+
+void LL::WorkQueue::close()
+{
+    mQueue.close();
+}
+
+void LL::WorkQueue::runUntilClose()
+{
+    try
+    {
+        for (;;)
+        {
+            callWork(mQueue.pop());
+        }
+    }
+    catch (const Queue::Closed&)
+    {
+    }
+}
+
+bool LL::WorkQueue::runPending()
+{
+    for (Work work; mQueue.tryPop(work); )
+    {
+        callWork(work);
+    }
+    return ! mQueue.done();
+}
+
+bool LL::WorkQueue::runOne()
+{
+    Work work;
+    if (mQueue.tryPop(work))
+    {
+        callWork(work);
+    }
+    return ! mQueue.done();
+}
+
+bool LL::WorkQueue::runUntil(const TimePoint& until)
+{
+    // Should we subtract some slop to allow for typical Work execution time?
+    // How much slop?
+    Work work;
+    while (TimePoint::clock::now() < until && mQueue.tryPopUntil(until, work))
+    {
+        callWork(work);
+    }
+    return ! mQueue.done();
+}
+
+std::string LL::WorkQueue::makeName(const std::string& name)
+{
+    if (! name.empty())
+        return name;
+
+    static U32 discriminator = 0;
+    static Mutex mutex;
+    U32 num;
+    {
+        // Protect discriminator from concurrent access by different threads.
+        // It can't be thread_local, else two racing threads will come up with
+        // the same name.
+        Lock lk(mutex);
+        num = discriminator++;
+    }
+    return STRINGIZE("WorkQueue" << num);
+}
+
+void LL::WorkQueue::callWork(const Queue::DataTuple& work)
+{
+    // ThreadSafeSchedule::pop() always delivers a tuple, even when
+    // there's only one data field per item, as for us.
+    callWork(std::get<0>(work));
+}
+
+void LL::WorkQueue::callWork(const Work& work)
+{
+    try
+    {
+        work();
+    }
+    catch (...)
+    {
+        // No matter what goes wrong with any individual work item, the worker
+        // thread must go on! Log our own instance name with the exception.
+        LOG_UNHANDLED_EXCEPTION(getKey());
+    }
+}
+
+void LL::WorkQueue::error(const std::string& msg)
+{
+    LL_ERRS("WorkQueue") << msg << LL_ENDL;
+}