43 AsyncObjectLoader

There is one AsyncObjectLoader per PSpace. Its purpose is to asynchronously load IPersistable objects from the LSS.

Public interface

Instances of AsyncObjectLoader are dynamically allocated and reference counted using std::shared_ptr. 

class AsyncObjectLoader
{
public:
    AsyncObjectLoader(PSpace* pspace, ROT* rot, boost::asio::io_context& context);

    // Must be called by a thread that has locked the PSpace
    void Push(OID oid);

    // Must be called by a thread without a CSpace lock or else there will be a dead-lock
    void Stop();

private:
    ...
};

Push is called to a add a request to asynchronously load the object with the given OID. Push must be called by a thread that has locked the CSpace mutex.

AsyncObjectLoader uses the boost::asio::io_context provided in the constructor to load the objects. It is assumed the io_context is only run by a single thread. This is vitally important, otherwise there will be undefined behaviour due to race conditions.

It is assumed class ROT implements the method Update: 

struct LoadObjectResult
{
    OID oid;
    EAsyncBindResult result;
    unique_interface_ptr<IPersistable> po;
};

class ROT
{
    void Update(std::vector<LoadObjectResult>& loaded);
    ...
};

ROT::Update is called by the AsyncObjectLoader context thread after it has loaded a "batch" of IPersistable objects and it has acquired a lock on the CSpace. ROT::Update is passed a vector of IPersistable objects that have been dynamically allocated and deserialised from serial elements in the LSS. Each object is owned by a unique_interface_ptr<IPersistable> and has not been registered in a CSpace.

Implementation

Conceptually we need a Single Producer Single Consumer (SPSC) queue of OIDs to pass the requests to asynchronously load objects. It is single producer because only threads that have locked the CSpace mutex can call Push. It is single consumer because only the single thread executing the boost asio io_context is popping the oids. Therefore we implement a "double buffered" approach, using one std::vector<OID> for the producer and one std::vector<OID> for the consumer. Having separate vectors allows the producer to push oids concurrently with the consumer popping oids. When the consumer has run out of oids to process (i.e. its vector is empty), it very briefly locks the CSpace mutex giving it exclusive access to the producer's vector. That allows it to swap the two vectors then release the CSpace mutex.

AsyncProcess_ is a private asynchronous function which posts a task to process the producer vector as follows:

  1. Lock the CSpace, swap the producer vector with its empty consumer vector and unlock the CSpace
  2. Iterate through all the oids in the consumer vector and load them into memory.
    Currently the io_context blocks on I/O when it calls the synchronous method LoadObject on the PersistStore which in turn calls the synchronous method ReadContiguousSerialElement on the LSS. Perhaps one day we will support asynchronous I/O in the LSS. This would involve asynchronous loading of segments into the LSS Segment Cache. It's not clear how to achieve that in a platform independent manner. boost asio doesn't seem to support asynchronous disk I/O.
  3. Clear the consumer vector
  4. Lock the CSpace, update the ROT with all the loaded objects and unlock the CSpace. Note that unique_interface_ptr<IPersistable> supports move assignment, so it is possible to move the objects into the CSpace as it were, where they are owned by the CSpace garbage collector.
  5. Clear the loaded objects vector

The consumer vector could have been a std::vector<OID> declared on the frame for the posted task, but that would lead to more heap allocations because the capacity of the vector would not be reused, so instead it is a member variable of AsyncObjectLoader.

This is also done for the std::vector<LoadObjectResult> used to record the set of objects that have been loaded in the batch.

AsyncProcess_ is called from Push whenever the producer vector transitions from empty to non-empty.

The Stop method is called without a lock on the CSpace mutex when the PSpace is being closed. It needs to synchronously put the AsyncObjectLoader into a state where no posted task can be running the 5 steps described above.

AsyncObjectLoader.h

Source: Ceda/cxPersistStore/src/AsyncObjectLoader.h

AsyncObjectLoader.cpp

Source: Ceda/cxPersistStore/src/AsyncObjectLoader.cpp