19.8 Dual inclusion transformation

Contract and interaction matrix

For context-equivalent concurrent operations O1 and O2, DualIT(O1,O2) changes them in place to O1'=IT(O1,O2) and O2'=IT(O2,O1). The identifiers must have different sites.

The rows and columns use the coordinate order C,D,I,X defined in Operations. A cell ab is the interaction of family a from O1 with family b from O2. The steps below do not evaluate sixteen unrelated cases: each step establishes compatible coordinates for the cells it processes and thereby makes the next part of the matrix available.

In a matrix that shows the named interactions, each two-letter cell label is a mnemonic. Read the first letter from the O1 column and the second from the O2 row: c means creation, d deletion, i move insertion and x move extraction. Thus ix is the interaction of I1 with X2, while xd is the interaction of X1 with D2. As a visual reminder, read the first letter down to the family label below its column, then read the second letter across to the family label beside its row.

The progress matrices replace interaction names with the following status and action mnemonics. The arrows show relationships along the matrix axes, not movement of text in a document. The algorithm and prose beneath each matrix define the exact operation.

Supporting algorithm: competing move extractions

Scan X1 and X2 by xq. Disjoint intervals advance normally. Split the four partial-overlap cases until the current spans are equal:

[1111111111111)     [2222222222222)     [111111111111)     [111111)
      [2222222)           [1111111)     [222222)           [222222222222)

Let the equal-span alias groups be g1,g2. If the head of g1 has e=0, queue all of g2 to track to that move's destination. Apply the symmetric rule for g2. The inner loop then treats the increasing e values as concurrent single-character insertions. It compares (s1+x2->e) - (s2+x1->e), uses site identifier on equality, and applies the opposite group's accumulated shift. The complete C++ below keeps this inner scan beside its outer interval scan.

void DualIT_xx(
    MoveInterval* x1,
    MoveInterval* x2,
    std::vector<SetNewExtractionPosCommand<MoveInterval> >& N1,
    std::vector<SetNewExtractionPosCommand<MoveInterval> >& N2)
{
    while (x1 && x2)
    {
        if (x1->xq + x1->size() <= x2->xq)
        {
            x1 = x1->nextX;
        }
        else if (x2->xq + x2->size() <= x1->xq)
        {
            x2 = x2->nextX;
        }
        else
        {
            if (x1->xq < x2->xq)
                x1 = SplitInterval(x1, x2->xq - x1->xq, false);
            else if (x2->xq < x1->xq)
                x2 = SplitInterval(x2, x1->xq - x2->xq, false);

            assert(x2->xq == x1->xq);
            if (x2->size() < x1->size())
                SplitInterval(x1, x2->size(), false);
            else if (x1->size() < x2->size())
                SplitInterval(x2, x1->size(), false);

            assert(x1->size() == x2->size());
            if (x1->e == 0)
                N2.push_back(SetNewExtractionPosCommand<MoveInterval>(
                    x2, x1->idoc, x1->iq));
            if (x2->e == 0)
                N1.push_back(SetNewExtractionPosCommand<MoveInterval>(
                    x1, x2->idoc, x2->iq));

            int xq = x1->xq;
            int s1 = 0;
            int s2 = 0;
            while (1)
            {
                assert(x1 && x2);
                assert(x1->xq == xq && x2->xq == xq);
                assert(x1->str == x2->str);

                int d = (s1 + x2->e) - (s2 + x1->e);
                if (d < 0 ||
                    (d == 0 && x2->opid.id < x1->opid.id))
                {
                    x2->e += s1;
                    ++s2;
                    x2 = x2->nextX;
                    if (!x2 || x2->xq != xq)
                    {
                        do
                        {
                            x1->e += s2;
                            x1 = x1->nextX;
                        }
                        while (x1 && x1->xq == xq);
                        break;
                    }
                }
                else
                {
                    x1->e += s2;
                    ++s1;
                    x1 = x1->nextX;
                    if (!x1 || x1->xq != xq)
                    {
                        do
                        {
                            x2->e += s1;
                            x2 = x2->nextX;
                        }
                        while (x2 && x2->xq == xq);
                        break;
                    }
                }
            }
        }
    }
}

The DualIT function

The following four subsections form the body of DualIT. The function first validates both operations. Each step then establishes the coordinates required by the next step; the final listing coalesces the results, validates them and closes the function.

void DualIT(Operation& O1, Operation& O2)
{
    O1.AssertValid();
    O2.AssertValid();

Step 1: creations and their position effects

For every document present in both operations, perform this exact sequence:

    for (auto& [doc, di2] : O2.documents)
    {
        auto m1 = O1.documents.find(doc);
        if (m1 != O1.documents.end())
        {
            DocIntervals& di1 = m1->second;
            DualIT_ii(di1.c, di2.c);
            IT_xi(di2.d, di1.c);
            IT_xi(di1.d, di2.c);
            DualIT_ii(di1.c, di2.i);
            DualIT_ii(di2.c, di1.i);
            IT_xi(di2.d, di1.i);
            IT_xi(di1.d, di2.i);
            IT_xi(di2.x, di1.c);
            IT_xi(di1.x, di2.c);
        }
    }

Step 2: deletions track enabled concurrent moves

    {
        std::vector<SetNewExtractionPosCommand<DeleteInterval> > ND1;
        std::vector<SetNewExtractionPosCommand<DeleteInterval> > ND2;

        for (auto& [doc, di2] : O2.documents)
        {
            auto m1 = O1.documents.find(doc);
            if (m1 != O1.documents.end())
            {
                DocIntervals& di1 = m1->second;
                IT_dx(di1.d, di2.x, ND1);
                IT_dx(di2.d, di1.x, ND2);
            }
        }

        ApplyNewExtractionPosCommands(O1, ND1);
        ApplyNewExtractionPosCommands(O2, ND2);
    }

At this point the relevant lists share post-C1,C2,I2 or symmetric coordinates. If a deletion overlaps the enabled move from the other operation, the deletion follows that character to the move destination.

Step 3: move insertions

    for (auto& [doc, di2] : O2.documents)
    {
        auto m1 = O1.documents.find(doc);
        if (m1 != O1.documents.end())
        {
            DocIntervals& di1 = m1->second;
            DualIT_ii(di1.i, di2.i);
            IT_xi(di2.x, di1.i);
            IT_xi(di1.x, di2.i);
        }
    }

Step 4: competing move extractions

The function calls that helper for every document common to the two operations. Relocations remain queued until the scan is complete:

    {
        std::vector<SetNewExtractionPosCommand<MoveInterval> > NM1;
        std::vector<SetNewExtractionPosCommand<MoveInterval> > NM2;

        for (auto& [doc, di2] : O2.documents)
        {
            auto m1 = O1.documents.find(doc);
            if (m1 != O1.documents.end())
            {
                DocIntervals& di1 = m1->second;
                DualIT_xx(di1.x, di2.x, NM1, NM2);
            }
        }

        ApplyNewExtractionPosCommands(O1, NM1);
        ApplyNewExtractionPosCommands(O2, NM2);
    }

The common epilogue restores compact form and checks the output invariants:

    O1.CoalesceAdjacentIntervals();
    O2.CoalesceAdjacentIntervals();
    O1.AssertValid();
    O2.AssertValid();
}