// VectorOT.cpp // // Author David Barrett-Lennard // (C)opyright Cedanet Pty Ltd 2009 @import "Stdafx.h" #include "Ceda/Core/cxUtils/CedaAssert.h" #include "Ceda/Core/cxUtils/Tracer.h" @import "../SiteIdMap.h" @import "VectorOT.h" namespace ceda { const bool trace_VectorOT = false; /* Calculation of effective siteids -------------------------------- For the given list, we conceptually break it up into maximal q-contiguous pieces. E.g. in the following there are 3 maximal q-contiguous pieces. 1 2 3 [------------)[---)[-----) [--------)[---)[--------------------) [---) <--------- q contiguous ----------> For each maximal q-contiguous piece calculate the effective siteids using a left to right scan. [Technically that's not quite true - there are possibly some right to left scans involved as well - maybe enough to make it O(n^2) in pathological cases] For q-contiguous intervals it is required that the *effective* siteids increase left to right: [------------)[---)[-----) s1 s2 s3 es1 es2 es3 Must satisfy es1 < es2 < es3. If the s1,s2,s3 on the intervals satisfy s1 < s2 < s3 (i.e. the siteids are increasing left to right), then this is the easy case. Result will be es1=s1, es2=s2, es3=s3. Here is a more complex example. Let the follow list of (s,u) values be for a maximal q-contiguous section: s 2 4 6 8 2 5 5 3 u 6 0 7 4 6 5 3 2 Consider that we iterate through the intervals from O in order of q-position. While the siteid monotone increases we simply assign es and eu from s,u es 2 4 6 8 eu 6 0 7 4 Now we want to add the interval with s=2, u=6. We can no longer simply assign es,eu from s,u because it will break the requirement that es values monotone increase from left to right. There are two possibilities, depending on the comparison of u coord to the eu of the last interval in the list we are building. In this case the new interval has a larger (ie losing) u coord so it inherits the es=8,eu=4 from the tail of the list we have formed so far: es 2 4 6 8 8 eu 6 0 7 4 4 Now we want to add s=5, u=5. Again we have a lower siteid than at the tail of the list we are building, so we must compare u values. Again we inherit es=8, eu=4: es 2 4 6 8 8 8 eu 6 0 7 4 4 4 Now we want to add the interval with s=5,u=3. This time the entry to be added dominates the tail of the list because it has a lower u coord. This dominating s,u is propagated, starting from the tail of the list using a right to left scan until the requirement that siteids monotone increase is established once again: es 2 4 5 5 5 5 5 eu 6 0 3 3 3 3 3 Finally we want to add the interval with s=3, u=2. Again this dominates intervals in the tail so we propagate es=3,eu=2 from right to left. Eventually we each the interval with es=4,eu=0. This dominates all the intervals that come after it, so we have to scan left to right again applying es=4, eu=0. The end result is: es 2 4 4 4 4 4 4 4 eu 6 0 0 0 0 0 0 0 It is conceivable that any risk of O(n2) behavior can be eliminated by using a representation that partitions this list into sub-lists and we avoid the need to explicitly iterate through a sub-list when applying a different (es,eu) Claim : Let O1,O2 be q-contiguous, O1 || O2, and O1 <> O2. Then ordering based on the effective siteids is equivalent to the order that arises from transforming a list against a list where each list is in temporal order. Proof : To keep it simple assume each interval has size 1. First assume O2 = [b] (a single interval). Let each interval be described by s and q. Let size = 1. A list L in temporal order is characterised by L[u].s and L[u].q. We assume L[0].q = 0. In spatial order we have M[q].s and M[q].u How to map from L to M: Note that s(u) is maintained in both representations. Putting it another way: Let s[u] provide s as a function of u = 0,1,2,.... Let Q[u] provide q as function of u, and u[q] provide u as a function of q = 0,1,2,... E.g. u 0 1 2 3 4 5 6 s[u] 4 5 8 5 2 2 6 Q[u] 0 1 1 2 2 0 2 q 0 1 2 3 4 5 6 s[q] 2 4 6 8 2 5 5 u[q] 5 0 6 2 4 3 1 a) Calculate u[q] from Q[u] We insert u=0 then u=1 then u=2 and so on at the position given by Q[u] u[q] 0 0 1 0 2 1 0 2 3 1 0 2 4 3 1 5 0 2 4 3 1 5 0 6 2 4 3 1 b) Calculate Q[u] from u[q] We extract u=0 as an L factor resulting in q that equals the number of elements to its left that have already been extracted (which must be 0). Then we extract u=1, and so on. u[q] q 5 *0 6 2 4 3 1 0 5 *0 6 2 4 3 *1 1 5 *0 6 *2 4 3 *1 1 5 *0 6 *2 4 *3 *1 2 5 *0 6 *2 *4 *3 *1 2 *5 *0 6 *2 *4 *3 *1 0 *5 *0 *6 *2 *4 *3 *1 2 Claim: Let s[q], u[q] be given for q=0,1,2,... For any given adjacent pair q,q+1, if s[q] > s[q+1] then let either q inherit from q+1 or vice versa, the winner being chosen according to smaller u. Then: the final result is independent of the order in which effective s,u are applied. In order to prove this we need to formalise this definition more carefully. Def: Let [] denote the empty list. Def: Let L1,L2 be lists. Then L1+L2 denotes the concatenation of L1,L2. + is associative but doesn't commute. Def: let L be a list. Then n*L denotes n additions of L+L+...+L i.e. 0*L = [] 1*L = L 2*L = L+L 3*L = L+L+L etc --------------------------------------------------------------------------------------------------- Def: Let L1,L2 be lists of triples (s,u,n) where n denotes the "multiplicity" of (s,u). Let L1, L2 have monotone increasing siteids. We define a binary merge operator @ (that preserves the monotone increase property) as follows: 1) L1 @ [] = L1 2) [] @ L2 = L2 3) (L1 + [(s1,u1,n1)]) @ ([(s2,u2,n2)] + L2) = L1 + [(s1,u1,n1)] + [(s2,u2,n2)] + L2 if s1 <= s2 (L1 + [(s1,u1,(n1+n2))]) @ L2 if s1 > s2 and u1 < u2 L1 @ ([(s2,u2,(n1+n2))] + L2) if s1 > s2 and u2 < u1 Note: a) This recursive definition necessarily terminates. Reason: Termination occurs when either L1=[] or L2=[] or else the siteids are ordered. Otherwise each recursive step necesarily decrements the value of |L1|+|L2|. Since the lists are finite the recursion must terminate. b) The definition unambiguously covers all cases The cases 1) 2), 3) cover all cases and the only overlap is where L1=L2=[] in which case 1) and 2) agree that []@[] = [] The cases in 3) are obviously mutually exclusive. To show that they conver all cases we first need the following lemma: Lemma: Let i1 = (s1,u1,n1) and (s2,u2,n2) be two adjacent intervals. Then i1 || i2 => s1 < s2 Proof: [informal] There is no landmark character separating them. Therefore since they are concurrent their order should depend on siteid comparisons Using the lemma we now show that s1 > s2 => u1 != u2, as follows s1 > s2 => not (i1 || i2) (by contrapositive of lemma) => i1 --> i2 or i2 --> i1 (by defn of i1 || i2) => u1 < u2 or u2 < u1 (temporal index preserves causality) => u1 != u2 Therefore there is no case of s1 > s2 and u1 = u2, so the cases in 3) cover all possibilities c) In the recursion to (L1 + [(s1,u1,(n1+n2))]) @ L2 i) L2 has monotone increasing siteids because it is a sublist of the original list on the RHS; and ii) L1 + [(s1,u1,(n1+n2))] has monotone increasing siteids because it has the same sequence of siteids as the original list on LHS. So this recursion necessarily preserves the assumption that the two lists have monotone increasing siteids. Similarly for the recursion to L1 @ ([(s2,u2,(n1+n2))] + L2). d) Conceptually when two lists are merged, elements can be "eaten" from where they touch. This process terminates when the overall list is ordered (which may happen trivially because one of the lists has been eaten away to []). Def: Let Elist(L1,L2) be the sublist of L1+L2 that either ended up eating a neighbour or else was eaten by a neighbour (or both) in the calculation of L1@L2. An element is only eaten if it has a higher u value. It follows that the winner in Elist(L1,L2) is always the s,u with the smallest u. Claim: L1 @ L2 preserves monotone increase of siteids. Proof: (by induction on n = |L1|+|L2|) n = 0 L1 @ L2 = [] n = 1 [x] @ [] = [x] [] @ [x] = [x] n --> n+1 (L1 + [(s1,u1,n1)]) @ ([(s2,u2,n2)] + L2) = L1 + [(s1,u1,n1)] + [(s2,u2,n2)] + L2 if s1 <= s2 - this necessarily monotone increases in s (L1 + [(s1,u1,(n1+n2))]) @ L2 if s1 > s2 and u1 < u2 - this necessarily monotone increases in s, by induction L1 @ ([(s2,u2,(n1+n2))] + L2) if s1 > s2 and u2 < u1 - this necessarily monotone increases in s, by induction Claim: @ is associative Proof: We show (L1 @ L2) @ L3 = L1 @ (L2 @ L3) (informal) Eating can occur at the boundary between L1,L2 and at the boundary between L2,L3. There are four Elists of interest: Elist(L1,L2), Elist(L1@L2,L3), Elist(L2,L3), Elist(L1,L2@L3). If exists x in L2 such that x not in Elist(L1,L2) and x not in Elist(L2,L3) L1 L2 L3 ------------ --------x-------- --------------- <------------> <------------> Elist(L1,L2) Elist(L2,L3) (the outcome is independent of the order because no interaction is possible) else (we have interaction) L1 L2 L3 ------------ ----------------- --------------- <--------------------> <-------------------> Elist(L1,L2) Elist(L2,L3) <--------------- E ------------------> Note that Elist(L1,L2) and Elist(L2,L3) don't necessarily overlap, but in that case they do touch. That still allows for them to interact. It seems readily apparent that the Elist with the smaller u will end up eating the other Elist, and possibly more as well. Furthermore the same outcome arises irrespective of the order. Let E1 = Elist(L1,L2) union Elist(L1@L2,L3) E2 = Elist(L1,L2@L3) union Elist(L2,L3) It is sufficient to show that E1 = E2, a sublist of L1+L2+L3. For then the winning (s,u) will be the one with minimum u. Let E = Elist(L1,L2) union Elist(L2,L3). Claim: E is a sublist of E1 and E is a sublist of E2. Let u1 = min u in Elist(L1,L2), u2 = min u in Elist(L2,L3). If u1 < u2 then in (L1 @ L2) @ L3, Elist(L1,L2) reduces to u1, and u1 then eats all elements of Elist(L2,L3) in L1 @ (L2 @ L3), Elist(L1,L2) reduces to u2, and u1 necessarily imposes itself on the RHS. ???? hardly a formal proof! (by induction on n = |L1|+|L2|+|L3|) n = 0: L1 @ (L2 @ L3) = [] = (L1 @ (L2) @ L3 n-->n+1: if s1 <= s2 L1 @ ((L2 + [(s2,u2,n2)]) @ ([(s3,u3,n3)] + L3)) = L1 @ (L2 + [(s2,u2,n2)] + [(s3,u3,n3)] + L3) = ???? */ struct s_u_n { SiteId s; int u; int n; }; /* Add (s,u,n) to an existing ordered list. We use vector not linked list for good cache coherency and to avoid heap allocations. Note that Add() is very efficient because it a) avoids recursion using tail call optimisation b) only erases the last element of the vector, which is very efficient and never results in memory copies or heap allocations */ // Let V = V @ [(s,u,n)]. void Add(xvector& V, s_u_n& sun) { again: if (V.empty()) { V.push_back(sun); } else { s_u_n& last = V.back(); if (last.s <= sun.s) { V.push_back(sun); } else if (last.u < sun.u) { // last eats (s,u,n) last.n += sun.n; } else { // (s,u,n) eats last cxAssert(sun.u < last.u); sun.n += last.n; V.pop_back(); goto again; } } } bool IsMonotoneIncreasingSiteId(const xvector& V) { if (V.size() > 1) { xvector::const_iterator i = V.begin(); xvector::const_iterator j = i; ++j; do { if (i->s > j->s) return false; ++i; ++j; } while(j != V.end()); } return true; } VectorCreateNode* InitV( mDeclareSiteIdMapArg xvector& V, VectorCreateNode* r) { V.clear(); cxAssert(r); int q = r->m_q; do { s_u_n sun; @if (mUseSiteIdMap) { cxVerify(siteIdMap.GetSiteId(r->m_opid.s, sun.s)); } @else { sun.s = r->m_opid.s; } sun.u = r->m_u; sun.n = 1; Add(V,sun); q += r->m_n; r = r->m_next; } while(r && r->m_q == q); cxAssert(!V.empty()); cxAssert(IsMonotoneIncreasingSiteId(V)); return r; } void DualIT_ii( mDeclareSiteIdMapArg VectorCreateNode* i1, VectorCreateNode* i2) { int s1 = 0; // Accumulated characters inserted by i1 int s2 = 0; // Accumulated characters inserted by i2 while(i1 && i2) { int d = (s1 + i2->m_q) - (s2 + i1->m_q); if (d == 0) { /* For a given list, we conceptually break it up into maximal q-contiguous pieces. E.g. in the following there are 3 maximal q-contiguous pieces. 1 2 3 [------------)[---)[-----) [--------)[---)[--------------------) [---) <--------- q contiguous ----------> For each maximal q-contiguous piece we can calculate the effective siteids */ /* Vector V1 on the frame is used to record effective site ids for the next maximal q-contiguous section from I1. Scan through forward linked list of the insertion intervals in I1, while they are q-contiguous. For each interval we have (s,u). Call Add(V1,s,u,1). */ xvector V1; VectorCreateNode* endi1 = InitV(mPassSiteIdMapArg(siteIdMap) V1,i1); // Similarly for V2 xvector V2; VectorCreateNode* endi2 = InitV(mPassSiteIdMapArg(siteIdMap) V2,i2); // Use forward iterators v1 in V1 and v2 in V2, as we merge q-contiguous // prefixes of I1,I2 as follows: xvector::const_iterator v1 = V1.begin(); xvector::const_iterator v2 = V2.begin(); while(1) { if (v1->s < v2->s) { // Insert v1->n intervals from I1 for (int i=0 ; i < v1->n ; ++i) { cxAssert(i1); i1->m_q += s2; s1 += i1->size(); i1 = i1->m_next; } if (++v1 == V1.end()) { cxAssert(i1 == endi1); // Pick up remainder in V2 while(i2 != endi2) { cxAssert(i2); i2->m_q += s1; s2 += i2->size(); i2 = i2->m_next; } break; } } else { cxAssert(v2->s < v1->s); // Insert v2->n intervals from I2 for (int i=0 ; i < v2->n ; ++i) { cxAssert(i2); i2->m_q += s1; s2 += i2->size(); i2 = i2->m_next; } if (++v2 == V2.end()) { cxAssert(i2 == endi2); // Pick up remainder in V1 while(i1 != endi1) { cxAssert(i1); i1->m_q += s2; s1 += i1->size(); i1 = i1->m_next; } break; } } } cxAssert(i1 == endi1); cxAssert(i2 == endi2); } else { if (d < 0) { // Process i2 i2->m_q += s1; s2 += i2->size(); i2 = i2->m_next; } else { cxAssert(d > 0); // Process i1 i1->m_q += s2; s1 += i1->size(); i1 = i1->m_next; } } } if (s1) { while (i2) { i2->m_q += s1; i2 = i2->m_next; } } if (s2) { while (i1) { i1->m_q += s2; i1 = i1->m_next; } } } /////////////////////////////////////////////////////////////////////////////////////////////////// // VectorInsertionInterval VectorInsertionInterval::VectorInsertionInterval() : m_q(-1), m_u(-1) { } bool VectorInsertionInterval::operator==(const VectorInsertionInterval& rhs) const { return m_opid == rhs.m_opid && m_q == rhs.m_q; } /* xostream& operator<<(xostream& os, const VectorInsertionInterval& e) { os << e.m_opid; if (e.m_es != e.m_opid.s) { os << '>' << e.m_es; } os << ':' << e.m_q << '@' << e.m_u; return os; } */ /////////////////////////////////////////////////////////////////////////////////////////////////// /* An insertion interval may only split so that the left and right parts are non-empty. The original interval becomes the left interval. The right interval is allocated from the heap and inserted into the linked list. Note that by tranferring identity of the original interval to the left piece, splitting never causes the pointer to the first interval in the linked list to need to be updated. */ void SplitInterval(VectorCreateList& list, VectorCreateNode* r1, int n1) { cxAssert(r1); cxAssert(list.HasElement(r1)); cxAssert(n1 > 0); int n2 = r1->m_n - n1; cxAssert(n2 > 0); VectorCreateNode* r2 = new VectorCreateNode; r2->m_opid = r1->m_opid; r2->m_u = r1->m_u; r1->m_n = n1; r2->m_n = n2; cxAssert(r2->m_n == n2); cxAssert(r1->m_n == n1); r2->m_q = r1->m_q + n1; // Insert r2 after r1 list.InsertAfter(r1,r2); } /////////////////////////////////////////////////////////////////////////////////////////////////// // VectorInsertions VectorInsertions::VectorInsertions() : m_u2(0) { } bool VectorInsertions::operator==(const VectorInsertions& rhs) const { return m_c == rhs.m_c; } // Insert an interval [q,q+n) with the given opid. void VectorInsertions::LocalInsertInterval(const Opid& opid, int q, int n) { /* Record the insertion by inserting an interval for the given field. r [------) [------------------) [----) ^ /|\ | insert n elements at position q, for (s,t) = (opid.s, opid.t) Let r be one of the existing intervals. It can be assumed that r.n > 0. There are 4 cases: a) Insert before r if 1) q < r.q ; or 2) q = r.q and r.m_opid != opid b) Merge with r if 1) r.q <= q <= r.q + r.n; and 2) r.m_opid = opid c) Nontrivially split r if 1) r.q < q < r.q + r.n ; and 2) r.m_opid != opid d) Insert after r if 1) q > r.q + r.n; or 2) q = r.q + r.n and r.m_opid != opid Claim: These 4 cases cover all possibilities and are mutually exclusive ----- Proof ----- if (q < r.q) then a if (q == r.q) then if (r.m_opid = m_opid) then b else a if (r.q < q < r.q + r.n) then if (r.m_opid = m_opid) then b else c if (q == r.q + r.n) then if (r.m_opid = m_opid) then b else d if (q > r.q + r.n) then d */ // todo: It would seem better to employ a right to left scan - that way we can shift // intervals as we scan past them. However, currently we don't record a pointer to // the end of the list so there's no advantage. m_c.ValidateChain(); // Insert according to the q-position. This will cause // q-positions to the right of the insertion to be pushed further to the right. // [-----) [------------) [--------) [-----) [------) // r1 r2 ^ // q // // Scan forwards over intervals r2 for which it is sufficient to know based on // position and size that the insertion comes after that interval. // Note that q > r2.q + r2.n is sufficient condition to insert after r2 // May exit the loop with r1 or r2 (or both) set to NULL Node* r1 = NULL; Node* r2 = m_c.m_first; while(r2) { cxAssert(r2->m_prev == r1); if (r2->m_q + r2->m_n >= q) break; r1 = r2; r2 = r2->m_next; } // Note that both r1 == NULL and r2 == NULL is possible cxAssert(!r1 || r1->m_next == r2); cxAssert(!r2 || r2->m_prev == r1); cxAssert(!r2 || q <= r2->m_q + r2->m_n); // If r1 exists then insertion must come after r1 cxAssert(!r1 || q > r1->m_q + r1->m_n); // Note that it is possible to exit with q = r2->m_q + r2->m_n, and in fact there // could be an interval r3 = r2->m_next with r3->m_q = q. Therefore it is possible // that we may need to merge with r3. // // r1 r2 r3 // [--------) [-------)[----) // ^ // q // By default we assume we cannot merge the insertion into an existing interval. bool merge = false; // The following "if-else" block prepares r1,r2, and the 'merge' flag ready for two // possible actions to be performed immediately after the block depending on the // 'merge' flag: // // merge = true => merge the insertion into r2 // merge = false => create a new interval to be inserted between r1,r2. if (!r2 || q < r2->m_q) { // Insert between r1,r2 } else if (r2->m_opid == opid) { cxAssert(q >= r2->m_q); cxAssert(q <= r2->m_q + r2->m_n); // Merge into r2 merge = true; } else if (q == r2->m_q) { // Insert between r1,r2 } else if (q == r2->m_q + r2->m_n) { r1 = r2; r2 = r2->m_next; if (r2 && r2->m_q == q && r2->m_opid == opid) { // Merge into r2 merge = true; } // else insert between r1,r2 } else { // Split r2. After splitting, r2 points at the left piece. SplitInterval(m_c, r2, q - r2->m_q); r1 = r2; r2 = r2->m_next; // Insert between r1,r2 } if (merge) { // Merge into r2 r2->m_n += n; r2 = r2->m_next; } else { // Insert between r1,r2 Node* cn = new Node; cn->m_opid = opid; cn->m_q = q; cn->m_n = n; cn->m_u = m_u2++; if (r1) { // Insertion is not at the head of the list cxAssert(r1->m_next == r2); r1->m_next = cn; } else { // Insertion is at the head of list cxAssert(m_c.m_first == r2); m_c.m_first = cn; } if (r2) { cxAssert(r2->m_prev == r1); r2->m_prev = cn; } cn->m_prev = r1; cn->m_next = r2; } // Shift remaining intervals (starting from r2) to the right according to the number of elements // that have been inserted while(r2) { r2->m_q += n; r2 = r2->m_next; } //Tracer() << "oi = " << oi << '\n'; m_c.ValidateChain(); } /////////////////////////////////////////////////////////////////////////////////////////////////// // Convert v to equivalent representation as an interval set w void AsIntervalSet(IntervalSet& w, const VectorInsertions& v) { w.clear(); const VectorInsertions::Node* cn = v.m_c.m_first; while(cn) { // Note that we may need to coalesce adjacent intervals. This can happen quite often // because of the need to record (s,t) in each interval. w.AddBack( HalfOpenInterval(cn->m_q,cn->m_q + cn->m_n) ); cn = cn->m_next; } } void ExtractRFactor(VectorInsertions& O, const VectorTime& v, VectorInsertions& Rf) { typedef VectorInsertions::Node Node; cxAssert(O.m_u2 >= 0); // Accumulates the number of characters inserted by Rf int offset = 0; Rf.m_u2 = O.m_u2; // The following implementation makes use of the original intervals to avoid // heap deallocations and heap allocations. // We assign O.m_c.m_first = NULL to orphan the list from O. This allows us // to iterate through the intervals and add them either to O or Rf. Rf.m_c.Clear(); Node* cn = O.m_c.m_first; O.m_c.m_first = NULL; Node* last1 = NULL; Node* last2 = NULL; while(cn) { Node* next = cn->m_next; // Record next before we overwrite it with NULL cn->m_prev = NULL; cn->m_next = NULL; /* if (bTraceMergeOidOps) { Tracer() << " O.m_u2 = " << O.m_u2 << " cn->m_u = " << cn->m_u << " cn = " << *cn << '\n'; } */ cxAssert(cn->m_u < O.m_u2); if (v.ExtentContains(cn->m_opid)) { // Add cn to Lf cn->m_q -= offset; O.m_c.PushBack(last1,cn); } else { // Add cn to Rf offset += cn->m_n; Rf.m_c.PushBack(last2,cn); } cn = next; } } template void MergeContextSerialised_ii_T(L& list, T*& fi1, T*& fi2) { T* i1 = fi1; T* i2 = fi2; if (i1) { int s = 0; // Accumulated shift from I2 T* previ1 = NULL; while(i1 && i2) { int d = i2->m_q - (s + i1->m_q); if (d <= 0) { // [1111111) // ^ // [22222) T* nexti2 = i2->m_next; // Insert i2 immediately before i1 in I1 if (previ1) { cxAssert(previ1->m_next == i1); previ1->m_next = i2; } else { cxAssert(fi1 == i1); fi1 = i2; } i2->m_prev = previ1; i2->m_next = i1; i1->m_prev = i2; previ1 = i2; s += i2->m_n; i2 = nexti2; } else { if (d < i1->m_n) { // [111111) // ^ // [22222) SplitInterval(list,i1,d); } // [11111111) // ^ // [22222) i1->m_q += s; previ1 = i1; i1 = i1->m_next; } } if (i2) { // Append intervals from i2 onwards onto end of I1 cxAssert(!i1); cxAssert(previ1); previ1->m_next = i2; i2->m_prev = previ1; } else { // Apply shift to remaining intervals in I1 if (s) { while (i1) { i1->m_q += s; i1 = i1->m_next; } } } } else { // Simply transfer all entries from I2 fi1 = i2; } fi2 = NULL; } void MergeContextSerialised_ii( VectorCreateList& list, VectorCreateNode*& fi1, VectorCreateNode*& fi2) { MergeContextSerialised_ii_T(list,fi1,fi2); } /* Given insertions I1,I2 satisfying I1 >< I2 let I1 += I2, and clear I2 todo: arguably we would expect the intervals to be mutually exclusive. So why is there a need to split intervals? */ template void Mergeii_noshift_T(List& list,T*& fi1, T*& fi2) { T* i1 = fi1; T* i2 = fi2; if (i1) { T* previ1 = NULL; while(i1 && i2) { //Tracer() << " Mergeii : i1 = " << *i1 << " i2 = " << *i2 << '\n'; int d = i2->m_q - i1->m_q; //Tracer() << " Mergeii : d = " << d << '\n'; if (d <= 0) { // [1111111) // ^ // [22222) T* nexti2 = i2->m_next; // Insert i2 immediately before i1 in I1 if (previ1) { cxAssert(previ1->m_next == i1); previ1->m_next = i2; } else { cxAssert(fi1 == i1); fi1 = i2; } i2->m_prev = previ1; i2->m_next = i1; i1->m_prev = i2; previ1 = i2; i2 = nexti2; } else { if (d < i1->m_n) { // [111111) // ^ // [22222) SplitInterval(list,i1,d); } // [11111111) // ^ // [22222) previ1 = i1; i1 = i1->m_next; } } if (i2) { // Append intervals from i2 onwards onto end of I2 cxAssert(!i1); cxAssert(previ1); previ1->m_next = i2; i2->m_prev = previ1; } } else { // Simply transfer all entries from I2 fi1 = i2; } fi2 = NULL; } void Mergeii_noshift(VectorCreateList& list,VectorCreateNode*& fi1, VectorCreateNode*& fi2) { Mergeii_noshift_T(list,fi1,fi2); } /////////////////////////////////////////////////////////////////////////////////////////////////// // VectorDeletionInterval VectorDeletionInterval::VectorDeletionInterval() : m_q(-1), m_n(-1) { } bool VectorDeletionInterval::operator==(const VectorDeletionInterval& rhs) const { return m_opid == rhs.m_opid && m_q == rhs.m_q && m_n == rhs.m_n; } /////////////////////////////////////////////////////////////////////////////////////////////////// // VectorDeletions // Convert to equivalent representation as an interval set w void VectorDeletions::AsIntervalSet(IntervalSet& w) const { w.clear(); const Node* cn = m_first; while(cn) { w.AddBack( HalfOpenInterval(cn->m_q,cn->m_q + cn->m_n) ); cn = cn->m_next; } } /* An interval may only split so that the left and right parts are non-empty. The original interval becomes the left interval. The right interval is allocated from the heap and inserted into the linked list. Note that by tranferring identity of the original interval to the left piece, splitting never causes the pointer to the first interval in the linked list to need to be updated. */ void VectorDeletions::SplitInterval(Node* r1, int n1) { cxAssert(r1); cxAssert(HasElement(r1)); cxAssert(n1 > 0); int n2 = r1->m_n - n1; cxAssert(n2 > 0); Node* r2 = new Node; r2->m_opid = r1->m_opid; r1->m_n = n1; r2->m_n = n2; r2->m_q = r1->m_q + n1; // Insert r2 after r1 InsertAfter(r1,r2); } /* The insertion of n bytes at position q causes all deletion intervals to the right of this position to be shifted to the right. An interval may need to be split. */ void VectorDeletions::LocalInsertInterval(int q, int n) { cxAssert(0 <= q); cxAssert(0 < n); if (trace_VectorOT) { Tracer() << "VectorDeletions::LocalInsertInterval() q = " << q << " n = " << n << '\n'; } IndentTrace indent; if (trace_VectorOT) { Tracer() << "deletions = " << *this << '\n'; } // todo: It would seem better to employ a right to left scan - that way we can shift // intervals as we scan past them. However, currently we don't record a pointer to // the end of the list so there's no advantage. ValidateChain(); // Insert according to the q-position. This will cause // q-positions to the right of the insertion to be pushed further to the right. // [-----) [------------) [--------) [-----) [------) // r ^ // q // // Scan forwards over intervals r for which it is sufficient to know based on // position and size that the insertion comes after that interval. // Note that q >= r2.q + r2.n is sufficient condition to insert after r Node* r = m_first; while(r) { if (q < r->m_q + r->m_n) { if (r->m_q < q) { // Split r. After splitting, r points at the left piece. SplitInterval(r, q - r->m_q); r = r->m_next; } // Shift intervals (starting from r) to the right according to the number of elements // that have been inserted while(r) { r->m_q += n; r = r->m_next; } break; } r = r->m_next; } if (trace_VectorOT) { Tracer() << "---> deletions = " << *this << '\n'; } ValidateChain(); } // Transform these deletions past the given insertions void VectorDeletions::ITPastInsertions(const VectorInsertions& vi) { if (trace_VectorOT) { Tracer() << "VectorDeletions::ITPastInsertions()\n"; } IndentTrace indent; if (trace_VectorOT) { Tracer() << "I = " << vi << '\n'; Tracer() << "D = " << *this << '\n'; } if (Node* x = m_first) { if (VectorInsertions::Node* i = vi.m_c.m_first) { int shift = 0; // Accumulated shift from insertions while(1) { const int d = i->m_q - (shift + x->m_q); if (d <= 0) { // [2222222) // 1 // Process next insertion shift += i->m_n; i = i->m_next; if (!i) { if (shift) { do { x->m_q += shift; x = x->m_next; } while (x); } return; } } else { if (d < x->m_n) { // [222222) // 1 SplitInterval(x,d); } // [2222222) // 1 // Process next extraction x->m_q += shift; x = x->m_next; if (!x) return; } } } } } VectorDeletions::Node* VectorDeletions::InsertAfter_2(Node* r, const Opid& opid, int q, int n) { // Append new interval at the end of the list, which is just after r1 Node* rnew = new Node; rnew->m_opid = opid; rnew->m_q = q; rnew->m_n = n; if (r) { InsertAfter(r,rnew); } else { PushFront(rnew); } return rnew; } /* Merge in the given delete. */ void VectorDeletions::LocalDeleteInterval(const Opid& opid, int q, int n) { if (trace_VectorOT) { Tracer() << "VectorDeletions::LocalDeleteInterval() q = " << q << " n = " << n << '\n'; } IndentTrace indent; if (trace_VectorOT) { Tracer() << "D = " << *this << '\n'; } cxAssert(0 <= q); cxAssert(0 < n); /* It is allowable for a local delete to pretend to delete characters that have already been deleted! When adding an interval we choose to merge it with any existing adjacent intervals. The result is that intervals never touch! This is similar to the canonical condition imposed on an IntervalSet. r1 r2 r3 r4 [-----) [------) [----) [--------) [------) [------) [--------------s---------------) q q+n --> [-----) [---------------------s--------------) [------) Note that although local delete operations are contiguous in p-space, they are not necessarily contiguous in q-space. Therefore there is no point trying to write an optimised version of this algorithm that assumes no characters in [q,q+n) have been deleted already. */ /* gap > 0 r1 | | [-------) | q r1 = last interval before position q r2 = r1+1 */ Node* r1 = NULL; Node* r2 = m_first; while(r2 && r2->m_q + r2->m_n < q) { r1 = r2; r2 = r2->m_next; } if (r2) { cxAssert(r2->m_prev == r1); int q2 = q+n; if (r2->m_q > q2) { // Insert between r1,r2 InsertAfter_2(r1,opid,q,n); } else { // r2 overlaps or touches [q,q+n) if (r2->m_q <= q && q2 <= r2->m_q + r2->m_n) { // r2 contains [q,q+n) so the delete is redundant. There is nothing to do } else { /* gap > 0 | | r4 | [---------) q+n r4 = first interval past q+n r3 = r4-1 */ Node* r3 = r2; Node* r4 = r2->m_next; while(r4 && r4->m_q <= q2) { r3 = r4; r4 = r4->m_next; } // Modify r2 cxAssert(r3); const int r3q2 = r3->m_q + r3->m_n; if (r3q2 > q2) q2 = r3q2; if (q < r2->m_q) r2->m_q = q; r2->m_n = q2 - r2->m_q; cxAssert(r2->m_n > 0); r2->m_opid = opid; if (r2 != r3) { EraseInclusive(r2->m_next,r3); } } } } else { // Insert at the end InsertAfter_2(r1,opid,q,n); } if (trace_VectorOT) { Tracer() << "--> D = " << *this << '\n'; } } template void ScanTwoIntervalSets_2(Range1 r1, Range2 r2, Out& out) { // Set to false for slightly higher performance if it can be assumed that consecutive // intervals in r1 are separated by nonzero gaps, and similarly for r2. const bool allowGaps = true; typedef Out::Interval Interval; typedef Interval::value_type T; if (!r1) goto end1; if (!r2) goto end2; while(1) { if (r1->end() <= r2->begin()) { /* r1 comes before r2 (possibly adjacent) r1: [-----) r2: [-----) */ out.Add1(*r1, r1->begin(), r1->end()); ++r1; if (!r1) goto end1; } else if (r2->end() <= r1->begin()) { /* r2 comes before r1 (possibly adjacent) r1: [-----) r2: [-----) */ out.Add2(*r2, r2->begin(), r2->end()); ++r2; if (!r2) goto end2; } else { T t1; if (r1->begin() < r2->begin()) { out.Add1(*r1, r1->begin(),r2->begin()); t1 = r2->begin(); } else { if (r2->begin() < r1->begin()) { out.Add2(*r2, r2->begin(),r1->begin()); } t1 = r1->begin(); } if (r2->end() < r1->end()) goto stepw2; while(1) { /* Step through r1 coming before r2->end() W1: [----) [---) [---) [-------------) W2: [-------------------------------) r2->end() */ while(1) { cxAssert(r1->end() <= r2->end()); if (t1 < r1->end()) out.Add3(*r1, *r2, t1,r1->end()); t1 = r1->end(); if (!++r1) { if (t1 < r2->end()) out.Add2(*r2,t1,r2->end()); ++r2; goto end1; // Will add remainder of W2 if any } if (r1->end() > r2->end()) break; if (!allowGaps || t1 < r1->begin()) { cxAssert(t1 < r1->begin()); out.Add2(*r2,t1,r1->begin()); } t1 = r1->begin(); } if (r1->begin() > r2->end()) { /* r1 comes after r2 with a nonzero gap. Therefore r2->end() marks the end of h W1: [------) W2: [-------------------------------) r2->end() */ if (t1 < r2->end()) out.Add2(*r2,t1,r2->end()); if (!++r2) goto end2; break; // Break out of inner loop, continue with outer loop } else { if (!allowGaps || t1 < r1->begin()) { cxAssert(t1 < r1->begin()); out.Add2(*r2,t1,r1->begin()); } t1 = r1->begin(); } stepw2: /* Step through r2 coming before r1->end() W2: [----) [---) [---) [------) W1: [-------------------------------) r1->end() */ while(1) { cxAssert(r2->end() <= r1->end()); if (t1 < r2->end()) out.Add3(*r1,*r2,t1,r2->end()); t1 = r2->end(); if (!++r2) { if (t1 < r1->end()) out.Add1(*r1,t1,r1->end()); ++r1; goto end2; // Will add remainder of W1 if any } if (r2->end() > r1->end()) break; if (!allowGaps || t1 < r2->begin()) { cxAssert(t1 < r2->begin()); out.Add1(*r1,t1,r2->begin()); } t1 = r2->begin(); } if (r2->begin() > r1->end()) { /* r2 comes after r1 with a nonzero gap. W2: [------) W1: [-------------------------------) r1->end() */ if (t1 < r1->end()) out.Add1(*r1,t1,r1->end()); if (!++r1) goto end1; break; // Break out of inner loop, continue with outer loop } else { if (!allowGaps || t1 < r2->begin()) { cxAssert(t1 < r2->begin()); out.Add1(*r1,t1,r2->begin()); } t1 = r2->begin(); } } } } end1: cxAssert(!r1); while(r2) { out.Add2(*r2,r2->begin(), r2->end()); ++r2; } return; end2: cxAssert(!r2); while(r1) { out.Add1(*r1,r1->begin(), r1->end()); ++r1; } } /* When merging remote delete operations we have three regions to consider Dlocal \ Dremote Dremote \ Dlocal Dlocal & Dremote In the third case we have a choice for whether to use (s,t) of Dlocal or Dremote. It should make no difference to correctness. However it could make a different to performance. We choose the option that leads to the fewest number of intervals in the result. Consider that we only use Dlocal. Then a site that never generates local deletes but receives and applies many individual deletes will never merge (s,t) values. This is even though Dremote intervals arrive that are very large and could be used to achieve substantial compression. This suggests using Dremote is typically the better choice. How to implement: Process Dlocal, Dremote Add1 --> add with Dlocal Add2 --> add with Dremote Add3 --> add with Dremote Add function tries to add the the last interval if it has the same (s,t). */ struct MergeDeletionProcessor { typedef IntervalSet::Interval Interval; typedef RangeFromBiDirectionalIterators Range; typedef VectorDeletions::Node Node; MergeDeletionProcessor() : m_last(NULL) {} void Add(const Opid& opid, int q1, int q2) { if (m_last != NULL && m_last->m_opid == opid && m_last->m_q + m_last->m_n == q1) { m_last->m_n = q2 - m_last->m_q; } else { Node* rnew = new Node; rnew->m_opid = opid; rnew->m_q = q1; rnew->m_n = q2-q1; m_result.PushBack(m_last, rnew); } } // Process interval in Dlocal\Dremote void Add1(const VectorDeletionInterval& r1, int q1, int q2) { if (trace_VectorOT) { Tracer() << "Add1 q1 = " << q1 << " q2 = " << q2 << '\n'; } Add(r1.m_opid, q1, q2); } // Process interval in Dremote\Dlocal void Add2(const VectorDeletionInterval& r2, int q1, int q2) { if (trace_VectorOT) { Tracer() << "Add2 q1 = " << q1 << " q2 = " << q2 << '\n'; } Add(r2.m_opid, q1, q2); } // Process interval in Dlocal&\Dremote void Add3(const VectorDeletionInterval&, const VectorDeletionInterval& r2, int q1, int q2) { if (trace_VectorOT) { Tracer() << "Add3 q1 = " << q1 << " q2 = " << q2 << '\n'; } Add(r2.m_opid, q1, q2); } VectorDeletions m_result; Node* m_last; }; // *this += rhs void VectorDeletions::MergeRemoteDeletions(const VectorDeletions& rhs) { if (trace_VectorOT) { Tracer() << "VectorDeletions::MergeDeletions() rhs = " << rhs << '\n'; } IndentTrace indent; if (trace_VectorOT) { Tracer() << "D = " << *this << '\n'; } MergeDeletionProcessor out; ScanTwoIntervalSets_2( MakeRangeFromBiDirectionalIterators(rhs), MakeRangeFromBiDirectionalIterators(*this), out); swap(out.m_result); if (trace_VectorOT) { Tracer() << "--> D = " << *this << '\n'; } } /////////////////////////////////////////////////////////////////////////////////////////////////// // Map FieldId --> VectorInsertions @if (mWorkingSetUseBPlusTree) { static inline int CompareKeys(const FieldId& k1,const FieldId& k2) { // todo: inefficient if (k1 < k2) return -1; else if (k1 == k2) return 0; else return +1; } mWorkingSetBPlusTree(false,FieldIdToVectorInsertionsMap,FieldId,VectorInsertions) mWorkingSetBPlusTree(false,FieldIdToVectorInsertionsAndDeletionsMap,FieldId,VectorInsertionsAndDeletions) } } // namespace ceda /* The old way to calculate effective siteids. This transient state was recorded in the persistent intervals, and calculated before performing DualIT. */ @if (false) { // Calculate m_es,m_eu of each insertion interval in list. template void CalculateEffectiveSiteIds(List& list) { typedef List::Node Node; /* if (bTraceMergeOidOps) { Tracer() << "\nCalculateEffectiveSiteIds on list = " << list << '\n'; } */ Node* r1 = list.m_first; while(r1) { /* if (bTraceMergeOidOps) { Tracer() << " r1 = " << *r1 << '\n'; } */ r1->m_es = r1->m_opid.s; r1->m_eu = r1->m_u; // Process maximal q-contiguous piece that starts at r1 Node* rp = r1; cxAssert(rp); int q = r1->m_q + r1->m_n; Node* rn = rp->m_next; while(rn && rn->m_q == q) { /* if (bTraceMergeOidOps) { Tracer() << " rp = " << *rp << '\n'; Tracer() << " rn = " << *rn << '\n'; } */ // Process rn cxAssert(rp); cxAssert(rn); if (rp->m_es <= rn->m_opid.s) { // Siteids are in order rn->m_es = rn->m_opid.s; rn->m_eu = rn->m_u; } else { // Siteids are decreasing which is no good cxAssert(rp->m_eu != rn->m_u); if (rp->m_eu < rn->m_u) { // rp dominates rn, so rn inherits effective s,u from rp rn->m_es = rp->m_es; rn->m_eu = rp->m_eu; } else { // rn dominates rp, so we must scan right to left as far as // needed to propagate rn's s,u to the intervals that precede // it according to q-position SiteId s = rn->m_opid.s; int u = rn->m_u; rn->m_es = s; rn->m_eu = u; Node* r = rp; while(1) { cxAssert(r); if (r->m_es > s) { if (r->m_eu < u) { // r dominates rn, so we now have to scan left to right // from r propagating the s,u from r s = r->m_es; u = r->m_eu; while(1) { r = r->m_next; cxAssert(r); r->m_es = s; r->m_eu = u; if (r == rn) break; } break; } } else { break; } r->m_es = rn->m_opid.s; r->m_eu = rn->m_u; if (r == r1) break; r = r->m_prev; }; } } q += rn->m_n; rp = rn; rn = rn->m_next; } // Advance to the start of the next maximal q-contiguous piece cxAssert(rp); r1 = rp->m_next; } } } @if (false) { /////////////////////////////////////////////////////////////////////////////////////////////// // Attempt 1: Assumes it is possible to merge with all adjacent intervals void VectorDeletions::MergeRemoteDeletions(const VectorDeletions& rhs) { if (trace_VectorOT) { Tracer() << "VectorDeletions::MergeDeletions() rhs = " << rhs << '\n'; } IndentTrace indent; if (trace_VectorOT) { Tracer() << "D = " << *this << '\n'; } const Node* x = rhs.m_first; Node* r1 = NULL; Node* r2 = m_first; while(x) // Merge each interval from rhs in turn { /* gap > 0 r1 | | [-------) | q r1 = last interval before position q r2 = r1+1 */ while(r2 && r2->m_q + r2->m_n < x->m_q) { r1 = r2; r2 = r2->m_next; } if (r2) { cxAssert(r2->m_prev == r1); int q2 = x->m_q + x->m_n; if (r2->m_q > q2) { // Insert between r1,r2 r1 = InsertAfter_2(r1,x->m_opid,x->m_q,x->m_n); } else { // r2 overlaps or touches [x->m_q,x->m_q + x->m_n) if (r2->m_q <= x->m_q && q2 <= r2->m_q + r2->m_n) { // r2 contains [q,q+n) so the delete is redundant. There is nothing to do } else { /* gap > 0 | | r4 | [---------) q+n r4 = first interval past q+n r3 = r4-1 */ Node* r3 = r2; Node* r4 = r2->m_next; while(r4 && r4->m_q <= q2) { r3 = r4; r4 = r4->m_next; } // Modify r2 cxAssert(r3); const int r3q2 = r3->m_q + r3->m_n; if (r3q2 > q2) q2 = r3q2; if (x->m_q < r2->m_q) r2->m_q = x->m_q; r2->m_n = q2 - r2->m_q; cxAssert(r2->m_n > 0); r2->m_opid = x->m_opid; if (r2 != r3) { EraseInclusive(r2->m_next,r3); } } } } else { // Insert at the end r1 = InsertAfter_2(r1,x->m_opid,x->m_q,x->m_n); } x = x->m_next; } if (trace_VectorOT) { Tracer() << "--> D = " << *this << '\n'; } } /////////////////////////////////////////////////////////////////////////////////////////////// // Attempt 2: Never allows remote delete to override local deletes template void SubtractIntervalSets_2(Range1 r1, Range2 r2, Out& out) { typedef Out::Interval Interval; typedef Interval::value_type T; /* W = W1 \ W2 If intervals don't overlap then we can scan through the intervals of W1, W2, stepping iterator w1 or w2 depending on which interval comes next - in the manner of merge sort. Intervals from W1 are added to W whereas intervals from W2 are otherwise ignored. W1: [------) [---) [---) [---) W2: [---) W: [------) [---) [---) [---) If an interval of W1 intersects one or more intervals from W2 then we tend to keep the gaps between the intervals of W2. W1: [-----------) W2: [---) [--) [----) W: [-) [--) */ if (!r1 || !r2) goto end; while(1) { if (r1->end() <= r2->begin()) { /* r1 comes before r2 (possibly adjacent) W1: [-----) W2: [-----) */ out.Add(r2, *r1, r1->begin(), r1->end()); if (!++r1) return; } else if (r2->end() <= r1->begin()) { /* r2 comes before r1 (possibly adjacent) W1: [-----) W2: [-----) */ if (!++r2) break; } else { /* r1 and r2 overlap - i.e. they have a non-empty intersection. We need to work out what parts of r1 to keep (if any). W1: [-----------) W2: [---) [--) [----) W: [-) [--) */ if (r2->begin() > r1->begin()) { /* There is a prefix of W1 to be added to W W1: [-----------------------) W2: [---)... W: [---) ... */ out.Add(r2, *r1, r1->begin(),r2->begin()); } while(r2->end() < r1->end()) { // There are necessarily one or more gaps to be added to W T gapStart = r2->end(); if (!++r2) { /* W1: [-----------r1-----------) W2: [------r2-------) | | gap */ out.Add(r2, *r1, gapStart,r1->end()); ++r1; goto end; } if (r2->begin() < r1->end()) { /* W1: [-----------r1---------------) W2: [---------------) [--r2 | | gap */ out.Add(r2, *r1, gapStart,r2->begin()); } else { /* W1: [-----------r1---------------) W2: [---------------) [--r2 | | gap */ out.Add(r2, *r1,gapStart,r1->end()); break; } } // No more of r1 to add to W if (!++r1) return; } } end: cxAssert(!r1 || !r2); while(r1) { out.Add(r2, *r1, r1->begin(), r1->end()); ++r1; } } struct MergeDeletionProcessor { typedef IntervalSet::Interval Interval; typedef RangeFromBiDirectionalIterators Range; MergeDeletionProcessor(VectorDeletions& vd) : m_vd(vd) {} // Process interval in Dremote/Dlocal void Add(Range& r2, const VectorDeletionInterval& d, int q1, int q2) { if (trace_VectorOT) { Tracer() << "Add q1 = " << q1 << " q2 = " << q2 << '\n'; } int size = q2-q1; cxAssert(size > 0); // todo: need to create a new node with // m_opid from d, // m_q = q1 // m_n = size // Inserted just before r2.m_it1 VectorDeletions::Node* rnew = new VectorDeletions::Node; rnew->m_opid = d.m_opid; rnew->m_q = q1; rnew->m_n = size; if (r2.GetIt1() != m_vd.end()) { m_vd.InsertBefore(r2.GetIt1(), rnew); } else { m_vd.PushBack(rnew); } /* if (r) { InsertAfter(r,rnew); } else { PushFront(rnew); } */ } VectorDeletions& m_vd; }; void VectorDeletions::MergeRemoteDeletions(const VectorDeletions& rhs) { if (trace_VectorOT) { Tracer() << "VectorDeletions::MergeDeletions() rhs = " << rhs << '\n'; } IndentTrace indent; if (trace_VectorOT) { Tracer() << "D = " << *this << '\n'; } MergeDeletionProcessor out(*this); SubtractIntervalSets_2( MakeRangeFromBiDirectionalIterators(rhs), MakeRangeFromBiDirectionalIterators(*this), out); if (trace_VectorOT) { Tracer() << "--> D = " << *this << '\n'; } } }