2 Design notes and project history

Can we go full asynchronous?

As of August 2021 CEDA has been made largely asynchronous by using boost asio. However, at present there needs to be at least one worker thread, because the main thread tends to synchronously close the LSS, CSpaces, sockets etc, and this is often achieved by waiting for a ManualResetEvent to be signalled to indicate that posted tasks will no longer execute.

Fully asynchronous implies we can't use these ManualResetEvents and we can't implement synchronous Close methods.

Consider that the LSS itself is reference counted using shared_ptr. We can ensure that all asynchronous tasks that access the LSS uses a shared_ptr to the LSS.

When the LSS is closed, we can synchronously close the file and set an atomic boolean flag so that asynchronous tasks abort as soon as they can. We don't delete this, instead we can use the trick of having the LSS store a shared_ptr which points back at itself, and Close clears the shared_ptr.

Similarly a CSpace can be reference counted using shared_ptr and Close is asynchronous.

Consider a test which creates a client and server in a loop. How do we make sure listening on the port doesn't fail because of the previous iteration? It seems that the execution of each test needs to be invoked after the previous execution completed. This is not too difficult - just use a lambda for an execution and post it to start it and have it posted again when the appropriate conditions are met.

TODO

  1. Shouldn't need to capture a shared_ptr in all the Stop functions because we always wait on the task that was posted?
  2. Deprecate ThreadBlockerWhileUsed and ThreadBlockerWhileUsedByStrand
  3. Deprecate the memory reclaimer?
  4. Deprecate AsyncTimeOut and AsyncRepeatedTask (AsyncPeriodicTimer is sufficient)
  5. Use the terms "periodic" and "non-periodic" to refer to timers that repeatedly execute the task versus ones that don't.
  6. Use IoContextPool instead of threads throughout the ceda core libraries
  7. Create folder under ceda-implementation cxThread for discarded code
  8. Move ThreadBlockerWhileUsedByStrand, ThreadBlockerWhileUsedFast etc into discarded code

DONE

  • Implement AsyncTaskExecuter, which is needed for a CSpace
  • Allow SetThreadDescription() to be used to set the thread name under Windows
  • Support default singleton IoContextPool
  • Move IoContextPool from cxMessage2 to cxThread
  • Implement AsyncSignaledTask
  • Implement AsyncPeriodicTimer
  • Implement AsyncRepeatedTask
  • Implement AsyncTimeOut
  • Move thread related functionality from cxUtils to cxThread

Some general comments on writing correct programs

Writing correct multi-threaded programs is difficult, and it's all too easy to have race conditions or dead-locks. The ceda-core must limit itself to simple, and obviously correct designs.

A good principle to go by is to tend to write classes which don't have threads, thread pools, task executors, mutexes, waitable events, condition variables etc. Rather it is assumed that all the method calls on a given object instance are serialised. There can be no race conditions and no dead-locks. The class can be deterministically unit tested. This makes the class easy to understand and easy to document.

Similarly we can have the same idea for large sets of objects - such as all the objects in a CSpace! If every object in the CSpace avoids all mention of threads, mutexes, condition variables etc, and all access to the objects in the CSpace (including the execution of object constructors and destructors) is protected by the CSpace mutex, then we know that those objects don't cause race conditions or dead-locks.

But in various ways we haven't been doing this, here are some examples:

  • CSpace
    • has three mutexes
    • has two task executors
    • has a std::thread
    • has an auto reset event
    • StopGcAndEvictAllDgsNodes() closes its task executors and calls join() on its GC thread
  • LSS
    • has a mutex for LssTxn
    • has a mutex for doing check points
    • has a mutex for the RPM
    • has a mutex for the SUT
    • has a mutex for the FileRAS
    • has a mutex for the SegmentCache
    • has a mutex for the SegmentWriter
    • has a mutex in the LazyWriter
    • has a thread and a mutex in the LazyWriterQueue
    • has a thread, atomic int, volatile bool, auto reset event in a TimeOutTask in the LazyFlusher
    • has a thread, atomic int, volatile bool, auto reset event in a TimeOutTask in the LazyCleaner
    • has a thread, volatile bool, auto reset event in a SignaledTask in the LazyCheckPointer
  • PersistStore
    • has a mutex for the CidMap
    • has a mutex, thread, atomic int, volatile bool, auto reset event in the DosWriter
    • has a thread, manual reset event, mutex in the AsyncJobQueue for loading objects
    • has a thread, manual reset event, mutex in the AsyncJobQueue for updating the ROT
    • has an atomic bool in the PSpaceDos
    • has a mutex in the PSpaceMap
    • has a mutex in the PSpaceTxnGroupMgr
    • has an atomic int, volatile bool in DGIndepNodeForAsyncPref
    • has a ThreadBlockerWhileUsed, AutoResetEvent, array of 16 std::mutex in the ROT
    • has a mutex in the TypeOpsMap
  • cxRmi
    • has a mutex, volatile bool, ManualResetEvent in a ThreadBlocker
    • has a mutex in a ResponseQueue
    • has an atomic int, ThreadBlocker in RmiCaller
  • cxOperation
    • RodSession
      • is created by a thread without a lock on the CSpace
      • is closed by a thread without a lock on the CSpace
      • has a ITaskExecuter member
      • ~RodSession() waits for its task executor to close
    • has a mutex for WorkingSetMachine
    • has a mutex for DeltaWriter

Avoid using cxUtils for anything thread related

cxUtils should be avoided when more specific libraries can be used. Thread support should be in the cxThread library, not cxUtils.

We should consider moving the following files from cxUtils to cxThread:

  • ThreadBlockerWhileUsed.h
  • ThreadName.cpp/h
  • SignaledTask.cpp/h
  • Event.h
  • AsyncJobQueue.h
  • LruCache.h (uses std::mutex, std::condition_variable)
  • CacheMap.h (uses std::mutex and ManualResetEvent)
  • MemoryReclaimer.cpp/h

LruCache and CacheMap are standalone and easy to move.

Review of thread usage

As of June 2021 when running CEDA applications and tests a lot of threads are created:

  • 4 threads per LSS (for lazy cleaner, checkpointer, flusher and writer)
  • cxMessage2 creates a thread for each context in the IoContextPool
  • CSpace has a std::thread member for the GC thread
  • PSpace DosWriter has a std::thread member
  • AsyncJobQueue in cxUtils has a std::thread member. This is used by AsyncLoadJob and AsyncUpdateRotJob in cxPersistStore.
  • SignaledTask in cxUtils has a std::thread member. SignaledTask is used by LazyCheckPointer, LazyWriter.
  • TimeOutTask in cxUtils has a std::thread member. TimeOutTask is used by LazyCleaner, LazyFlusher.
  • cxThread implements IThreadPool which has a set of threads
  • MemoryReclaimer in cxUtils has a std::thread member

ThreadPoolMixin defined in cxThread is not currently used.

A cxThread thread pool is created by calling CreateThreadPool(). This is currently only used by cxObject to create two singleton thread pools - one for I/O and one for CPU. These are obtained with the functions GetTheThreadPool() and GetTheThreadPoolForIO(). These are in turn used to implement CSpace::PostTaskForIO() and CSpace::PostTask(), and also the RodSession calls GetTheThreadPool().

CSpace::PostTaskForIO() and CSpace::PostTask() are used by the DGS for async dependents.

Using boost::asio::io_context for all our needs?

boost::asio::io_context can be used for the following:

  • Implementing TCP servers, clients and sessions as in cxMessage2
  • posting tasks to be done asynchronously as soon as possible (using boost::asio::post)
  • posting tasks to be done asynchronously after a delay (using boost::asio::deadline_timer)

That seems to cover everything we need for a thread pool.

Here is some example code using a boost::asio::deadline_timer 


void handler(const boost::system::error_code& error)
{
  if (!error)
  {
    // Timer expired.
  }
}

// Construct a timer with an absolute expiry time.
boost::asio::deadline_timer timer(my_context, boost::posix_time::time_from_string("2005-12-07 23:59:59.000"));

// Start an asynchronous wait.
timer.async_wait(handler);

Aborting a boost::asio::deadline_timer

In order for an application to shut down quickly it needs to be able to abort async waits on boost::asio::deadline_timer objects.

boost::asio::deadline_timer::cancel() can be called for this purpose.

boost::asio::deadline_timer is not thread-safe, so therefore it seems necessary to post a task to call cancel() on the basic_deadline_timer, to the same io_context on which async_wait was called.

For example LazyFlusher::Close() might resemble TcpMsgClient::Close() in the way it declares a ManualResetEvent on the frame then uses boost::asio::post to post a task to the io_context. The task would call cancel() on the deadline_timer.

Waiting until async tasks have finished

What about the concept of a TaskExecuter, and closing it to wait on all the tasks that were posted to it?

The cxThread TaskExecuter has a ManualResetEvent which is used to block threads while there are tasks assigned to the TaskExecuter still running.

But maybe this is a bad idea in the first place? It seems a lot safer to declare a ManualResetEvent on the frame when a thread needs to wait until a counter falls to zero. This ensures the thread doing the waiting is the thread that deletes the condition variable and mutex. Compare that to when the thread that finds the counter falls to zero tries to delete the condition variable and mutex.

Issue with async load from LSS

Consider that the LSS provides an AsyncLoad() method, which takes an oid and a callback function. Now suppose the client code wants to close the LSS. What happens? Or consider that the LazyFlusher has posted an async dead-line timer to the IoContextPool and then the LSS is closed. Closing the LSS must synchronously close the file. Do we have to wait for the async call to complete (aborting if possible), or do we put the LSS into a state where the callbacks are safe and innocuous?

Proposed Changes

Note that we want to avoid including boost headers in Ceda public headers if we can, because forcing users of ceda to also need to install boost makes ceda that much harder to package and use.

cxThread implements an IoContextPool. It provides the following functions to create/close an IoContextPool: 


IoContextPool* CreateIoContextPool(int numThreads);
void Close(IoContextPool*);

cxThread Source code

The historical source tree is represented by the source excerpts embedded in this document.