19 Control algorithm using merge and factorisation

In the following we describe a control algorithm that avoids the assumption that composite operations are implemented as contextually serialised lists of atomic operations - and in particular a linear list of atomic operations that comprise a local history buffer, or a list of atomic operations for a transient HB-suffix.

Instead the proposed control algorithm makes use of abstractly defined composite operations that represent arbitrarily large changes to the database, and provide the IT, merge and factorization operators. The potential benefit is the avoidance of the quadratic complexity implicit with using linear lists of atomic operations to represent composite operations.

Persistent working set state:

struct WorkingSet
{
    SiteId s;
    int t;
    Operation hb;
    Mutex m;
};

void DoLocalOperation(WorkingSet& w,Operation o)
{
    o.Init(w.s, w.t++);       // Set (s,t) in all intervals
    {
        Lock lock(w.m);
        w.hb = o;
        o.Execute();
    }
}

void RunSession(WorkingSet& w, OutStream& os, InStream& is)
{
    {
        Lock lock(w.m);
        os << vout(w.hb);  // Send vector time describing content of local HB
    }

    VectorTime vr;
    is >> vr;          // Receive vector time describing content of remote HB

    async repeat       // Thread to send operations
    {
        Lock lock(w.m);
        os << Rf(w.hb, vout(w.hb)  vr);
        vr = vout(w.hb);
    }
    async repeat       // Thread to receive operations
    {
        Operation o;
        is >> o;
        {
            Lock lock(w.m);
            VectorTime v = vout(w.hb)  vout(o);
            o = IT(Rf(o,v),Rf(w.hb,v);
            w.hb = o;
            o.Execute();
        }
    }
}

Notes:

• A session is always between a pair of sites. A site can hold any number of sessions with its sites at the same time.
• A given communication channel is associated with a sender site that uses an output stream ‘os’ to write vector times or operations, and another receiver site that uses an input stream ‘is’ to read vector times or operations. It is assumed that the channel behaves in the manner of a reliable transient FIFO queue. For simplicity we have ignored issues of send and receive buffers, flushing etc.
• A session involves full duplex communication using two independent channels as described above. The protocol involves the sending of a single vector time following by a stream of composite operations.
• When a session begins, each site sends its local v_out(hb) which summarises the set of operations that have been applied at that site. By the time this is received by the other site it may only represent an underestimate of what operations are present.
• Each session hosts two worker threads – one to repeatedly send operations and another to repeatedly receive operations.
• The working set contains a mutex to serialise access to all of its state. Both worker threads need to lock the mutex before they are granted access.
• Since we assume that a given site only has an underestimate of what’s already present on the remote site, it is necessary to allow for the possibility that an operation o is received that (perhaps partially) contains atomic operations that are already present. This is achieved by taking the RFactor of o with respect to v_out(hb) in order to remove from o, all those atomic operations that are already contained within the HB. This could potentially render o empty! Note that the assertion in the call to Rf2 won’t fail – ie v_in(o) ≤ v_out(hb) because the sender can never send an operation performed in the context of atomic operations that are not already present on the receiver.
• An atomic operation is never sent more than once. This follows from the fact that we send an RFactor of the HB with respect to a vector time vr that is subsequently unioned with v_out(hb). Therefore any existing atomic operations in the HB will not be sent again.
• The execution context of sent operations can only increase over the life of the session. ie if O₁ is sent before O₂ then v_in(O₁) ≤ v_in(O₂). This is obvious from the fact that we send an operation O calculated as the RFactor of the HB with respect to vr so therefore v_in(O) = vr. The result follows from the fact that vr is only changed by assignment from a union with itself so the new vr contains the previous vr.