src/ledger/bin/storage/impl/btree/diff.cc - fuchsia - Git at Google

 // Copyright 2017 The Fuchsia Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 #include "src/ledger/bin/storage/impl/btree/diff.h"

 #include <lib/fit/function.h>

 #include <utility>

 #include "src/ledger/bin/storage/impl/btree/internal_helper.h"
 #include "src/ledger/bin/storage/impl/btree/iterator.h"
 #include "src/ledger/bin/storage/impl/btree/synchronous_storage.h"
 #include "src/ledger/bin/storage/impl/object_digest.h"

 namespace storage {
 namespace btree {
 namespace {
 // Aggregates 2 |BTreeIterator|s and allows to walk through these concurrently
 // to compute the diff. |on_next| will be called for each diff entry.
 class IteratorPair {
  public:
   IteratorPair(
       SynchronousStorage* storage,
       fit::function<bool(std::unique_ptr<Entry>, std::unique_ptr<Entry>)>
           on_next)
       : on_next_(std::move(on_next)), left_(storage), right_(storage) {}

   // Initialize the pair with the ids of both roots.
   Status Init(ObjectIdentifier left_node_identifier,
               ObjectIdentifier right_node_identifier, fxl::StringView min_key) {
     RETURN_ON_ERROR(left_.Init(left_node_identifier));
     RETURN_ON_ERROR(right_.Init(right_node_identifier));
     if (!min_key.empty()) {
       RETURN_ON_ERROR(SkipIteratorsTo(min_key));
     }
     Normalize();
     if (!Finished() && !HasDiff()) {
       RETURN_ON_ERROR(Advance());
     }

     return Status::OK;
   }

   bool Finished() const {
     FXL_DCHECK(IsNormalized());
     return right_.Finished();
   }

   // Send the actual diff to the client. Returns |false| if the iteration must
   // be stopped.
   bool SendDiff() {
     FXL_DCHECK(HasDiff());

     // If the 2 iterators are on 2 equals values, nothing to do.
     if (left_.HasValue() && right_.HasValue() &&
         left_.CurrentEntry() == right_.CurrentEntry()) {
       return true;
     }

     if (HasSameNextChild()) {
       // If the 2 iterators are on the same child, send a diff for each
       // iterator that is currently on a value.
       if (right_.HasValue()) {
         if (!SendRight()) {
           return false;
         }
       }
       if (left_.HasValue() &&
           (!right_.HasValue() ||
            left_.CurrentEntry().key != right_.CurrentEntry().key)) {
         if (!SendLeft()) {
           return false;
         }
       }
       return true;
     }

     // Otherwise, just send the diff of the right node.
     return SendRight();
   }

   // Advance the iterator until there is potentially a diff to send.
   Status Advance() {
     FXL_DCHECK(!Finished());
     do {
       FXL_DCHECK(IsNormalized());

       // If the 2 next children are identical, skip these.
       if (HasSameNextChild()) {
         right_.SkipNextSubTree();
         left_.SkipNextSubTree();
         Normalize();
         continue;
       }

       // If both iterators are sitting on a value for the same key, both need
       // to be advanced.
       if (right_.HasValue() && left_.HasValue() &&
           right_.CurrentEntry().key == left_.CurrentEntry().key) {
         RETURN_ON_ERROR(right_.Advance());
         Swap();
       }

       RETURN_ON_ERROR(right_.Advance());
       Normalize();
     } while (!Finished() && !HasDiff());
     return Status::OK;
   }

  private:
   // Advances the two iterators so that they are both in the first entry that 1)
   // is greater than or equal to min_key and 2) might be different between the
   // two iterators. We consider that the two entries might be different, if they
   // are in btree nodes with different ids.
   Status SkipIteratorsTo(fxl::StringView min_key) {
     for (;;) {
       if (left_.SkipToIndex(min_key)) {
         return right_.SkipTo(min_key);
       }
       if (right_.SkipToIndex(min_key)) {
         left_.SkipToIndex(min_key);
         return Status::OK;
       }

       auto left_child = left_.GetNextChild();
       auto right_child = right_.GetNextChild();
       if (!left_child) {
         return right_.SkipTo(min_key);
       }
       if (!right_child) {
         return left_.SkipTo(min_key);
       }
       if (left_child == right_child) {
         return Status::OK;
       }
       // Same nodes might be in different depths of the btrees. Only descend in
       // each if their current level is the same as or greater than the other
       // one's.
       uint8_t level_left = left_.GetLevel();
       uint8_t level_right = right_.GetLevel();
       if (level_left >= level_right) {
         RETURN_ON_ERROR(left_.Advance());
       }
       if (level_right >= level_left) {
         RETURN_ON_ERROR(right_.Advance());
       }
     }
   }

   // Ensure that the representation of the pair of iterator is normalized
   // according to the following rules:
   // If only one iterator is finished, it is always the left one.
   // If only one iterator is on a value, it is always the left one.
   // If both iterator are on a value, the left one has a key greater or equals
   //     to the right one, and if the keys are equals, the iterator are in
   //     their original order.
   // If none of the iterators is on a value, the right one has a level
   //     greather or equals to the left one.
   //
   // When the iterator is normalized, the different algorithms can cut the
   // number of case it needs to consider.
   void Normalize() {
     if (left_.Finished()) {
       return;
     }
     if (right_.Finished()) {
       Swap();
       return;
     }

     if (right_.HasValue() && left_.HasValue()) {
       if (left_.CurrentEntry().key < right_.CurrentEntry().key) {
         Swap();
         return;
       }
       if (left_.CurrentEntry().key == right_.CurrentEntry().key) {
         ResetSwap();
       }
       return;
     }

     if (left_.HasValue()) {
       return;
     }
     if (right_.HasValue()) {
       Swap();
       return;
     }

     if (left_.GetLevel() > right_.GetLevel()) {
       Swap();
     }
   }

   // Returns if the iterator is normalized. See |Normalize| for the
   // definition. This is only used in DCHECKs.
   bool IsNormalized() const {
     if (left_.Finished() || right_.Finished()) {
       return left_.Finished();
     }

     if (left_.HasValue()) {
       if (!right_.HasValue()) {
         return true;
       }

       if (right_.CurrentEntry().key > left_.CurrentEntry().key) {
         return false;
       }

       if (right_.CurrentEntry().key == left_.CurrentEntry().key) {
         if (!diff_from_left_to_right_)
           return diff_from_left_to_right_;
       }

       return true;
     }

     if (right_.HasValue()) {
       return false;
     }

     return right_.GetLevel() >= left_.GetLevel();
   }

   // Returns whether there is a potential diff to send at the current state.
   bool HasDiff() const {
     FXL_DCHECK(IsNormalized());
     return (right_.HasValue() && (left_.Finished() || left_.HasValue())) ||
            (left_.HasValue() && HasSameNextChild());
   }

   // Returns whether the 2 iterators have the same next child in the
   // iteration. This allows to skip part of the 2 btrees when they are
   // identicals.
   bool HasSameNextChild() const {
     if (left_.Finished()) {
       return false;
     }
     auto right_child = right_.GetNextChild();
     if (!right_child) {
       return false;
     }
     auto left_child = left_.GetNextChild();
     if (!left_child) {
       return false;
     }
     return *right_child == *left_child;
   }

   // Swaps the 2 iterators. This is useful to reduce the number of case to
   // consider during the iteration.
   void Swap() {
     std::swap(left_, right_);
     diff_from_left_to_right_ = !diff_from_left_to_right_;
   }

   // Reset the iterators so that they are back in the original order.
   void ResetSwap() {
     if (!diff_from_left_to_right_) {
       Swap();
     }
   }

   // Send a diff using the right iterator.
   bool SendRight() { return Send(&right_, &left_, !diff_from_left_to_right_); }

   // Send a diff using the left iterator.
   bool SendLeft() { return Send(&left_, &right_, diff_from_left_to_right_); }

   bool Send(BTreeIterator* it1, BTreeIterator* it2, bool it1_to_it2) {
     std::unique_ptr<Entry> it1_entry, it2_entry;
     it1_entry = std::make_unique<Entry>(it1->CurrentEntry());
     if (!it2->Finished() && it2->HasValue() &&
         it1->CurrentEntry().key == it2->CurrentEntry().key) {
       it2_entry = std::make_unique<Entry>(it2->CurrentEntry());
     }

     if (it1_to_it2) {
       return on_next_(std::move(it1_entry), std::move(it2_entry));
     }
     return on_next_(std::move(it2_entry), std::move(it1_entry));
   }

   fit::function<bool(std::unique_ptr<Entry>, std::unique_ptr<Entry>)> on_next_;
   BTreeIterator left_;
   BTreeIterator right_;
   // Keep track whether the change is from left to right, or right to left.
   // This allows to switch left and right during the algorithm to handle less
   // cases.
   bool diff_from_left_to_right_ = true;
 };

 // Iterator that does a three-way diff by using two IteratorPair objects in
 // parallel.
 // Here is an attempt to convey what this iterator should be doing:
 // - It creates an IteratorPair (IP thereafter) for each side of the diff
 //   (base-to-left and base-to-right).
 // - At the initialization time, it advances each internal IP to their first
 //   diff. Each IP (as viewed from here) is on one key: the key of the latest
 //   diff it returned.
 // - We always advance the IP with the lowest key, or the one not finished yet.
 //   If both are on the same key, we advance both.
 // - The current key considered by the ThreeWayIterator is the lowest key of the
 //   latest left and right diffs. If one IP is finished, then the current key is
 //   the key of the other IP's diff.
 // - When sending the ThreeWayIterator diff, we consider the current key (per
 //   above). If both IPs are on the same key, the diff is easy (we just send
 //   everything). However, if the IPs are on different keys, or one of them is
 //   finished, we have to consider multiple cases:
 //   - If the base entry is present, it means the key/value was present in the
 //     base revision. Given that the other IP moved past this key, there is no
 //     diff on that side and we copy the base entry to that side entry within
 //     the tree-way diff change.
 //   - If the base entry is not present, it means the key/value was not present
 //     in the base revision and it is an addition.
 class ThreeWayIterator {
  public:
   explicit ThreeWayIterator(SynchronousStorage* storage) {
     base_left_iterators_ = std::make_unique<IteratorPair>(
         storage,
         [this](std::unique_ptr<Entry> base, std::unique_ptr<Entry> left) {
           left_advanced_ = true;
           base_left_.swap(base);
           left_.swap(left);
           return true;
         });
     base_right_iterators_ = std::make_unique<IteratorPair>(
         storage,
         [this](std::unique_ptr<Entry> base, std::unique_ptr<Entry> right) {
           right_advanced_ = true;
           base_right_.swap(base);
           right_.swap(right);
           return true;
         });
   }

   Status Init(ObjectIdentifier base_node_identifier,
               ObjectIdentifier left_node_identifier,
               ObjectIdentifier right_node_identifier, fxl::StringView min_key) {
     RETURN_ON_ERROR(base_left_iterators_->Init(base_node_identifier,
                                                left_node_identifier, min_key));
     RETURN_ON_ERROR(base_right_iterators_->Init(
         base_node_identifier, right_node_identifier, min_key));
     if (!Finished()) {
       RETURN_ON_ERROR(AdvanceLeft());
       RETURN_ON_ERROR(AdvanceRight());
     }
     return Status::OK;
   }

   bool Finished() const {
     return base_left_iterators_->Finished() &&
            base_right_iterators_->Finished() && !base_left_ && !left_ &&
            !base_right_ && !right_;
   }

   Status Advance() {
     FXL_DCHECK(!Finished());
     if (base_left_iterators_->Finished() && !base_left_ && !left_) {
       RETURN_ON_ERROR(AdvanceRight());
     } else if (base_right_iterators_->Finished() && !base_right_ && !right_) {
       RETURN_ON_ERROR(AdvanceLeft());
     } else if (GetLeftKey() < GetRightKey()) {
       RETURN_ON_ERROR(AdvanceLeft());
     } else if (GetLeftKey() > GetRightKey()) {
       RETURN_ON_ERROR(AdvanceRight());
     } else {
       RETURN_ON_ERROR(AdvanceLeft());
       RETURN_ON_ERROR(AdvanceRight());
     }
     return Status::OK;
   }

   ThreeWayChange GetCurrentDiff() {
     FXL_DCHECK(!Finished());
     ThreeWayChange change;
     change.base = GetBase();
     change.left = GetEntry(change.base, base_left_iterators_, left_, right_);
     change.right = GetEntry(change.base, base_right_iterators_, right_, left_);
     return change;
   }

  private:
   // GetLeftKey (resp. GetRightKey) should not be called if the left (resp.
   // right) IteratorPair is finished.
   const std::string& GetLeftKey() {
     FXL_DCHECK(base_left_ || left_);
     if (base_left_) {
       return base_left_->key;
     }
     return left_->key;
   }

   const std::string& GetRightKey() {
     FXL_DCHECK(base_right_ || right_);
     if (base_right_) {
       return base_right_->key;
     }
     return right_->key;
   }

   Status AdvanceLeft() {
     left_advanced_ = false;
     while (!base_left_iterators_->Finished() && !left_advanced_) {
       if (!base_left_iterators_->SendDiff()) {
         return Status::OK;
       }
       RETURN_ON_ERROR(base_left_iterators_->Advance());
     }
     if (!left_advanced_ && base_left_iterators_->Finished()) {
       base_left_.reset();
       left_.reset();
     }
     return Status::OK;
   }

   Status AdvanceRight() {
     right_advanced_ = false;
     while (!base_right_iterators_->Finished() && !right_advanced_) {
       if (!base_right_iterators_->SendDiff()) {
         return Status::OK;
       }
       RETURN_ON_ERROR(base_right_iterators_->Advance());
     }
     if (!right_advanced_ && base_right_iterators_->Finished()) {
       base_right_.reset();
       right_.reset();
     }
     return Status::OK;
   }

   std::unique_ptr<Entry> GetBase() {
     if (base_left_ && base_right_) {
       if (base_left_->key < base_right_->key) {
         return std::make_unique<Entry>(*base_left_);
       }
       return std::make_unique<Entry>(*base_right_);
     }
     if (!base_right_ && base_left_ &&
         (!right_ || base_left_->key < right_->key)) {
       return std::make_unique<Entry>(*base_left_);
     }
     if (!base_left_ && base_right_ &&
         (!left_ || base_right_->key < left_->key)) {
       return std::make_unique<Entry>(*base_right_);
     }
     return std::unique_ptr<Entry>();
   }

   std::unique_ptr<Entry> GetEntry(
       const std::unique_ptr<Entry>& base,
       const std::unique_ptr<IteratorPair>& this_iterator,
       const std::unique_ptr<Entry>& this_entry,
       const std::unique_ptr<Entry>& other_entry) {
     if (base) {
       if (this_entry && base->key == this_entry->key) {
         return std::make_unique<Entry>(*this_entry);
       }
       if (this_entry && base->key < this_entry->key) {
         return std::make_unique<Entry>(*base);
       }
       if (this_iterator->Finished()) {
         return std::make_unique<Entry>(*base);
       }
       return std::unique_ptr<Entry>();
     }
     if (!this_entry || (other_entry && other_entry->key < this_entry->key)) {
       return std::unique_ptr<Entry>();
     }
     return std::make_unique<Entry>(*this_entry);
   }

   bool left_advanced_ = false;
   bool right_advanced_ = false;

   std::unique_ptr<Entry> base_left_;
   std::unique_ptr<Entry> base_right_;
   std::unique_ptr<Entry> left_;
   std::unique_ptr<Entry> right_;

   std::unique_ptr<IteratorPair> base_left_iterators_;
   std::unique_ptr<IteratorPair> base_right_iterators_;
 };

 Status ForEachDiffInternal(SynchronousStorage* storage,
                            ObjectIdentifier left_node_identifier,
                            ObjectIdentifier right_node_identifier,
                            std::string min_key,
                            fit::function<bool(EntryChange)> on_next) {
   FXL_DCHECK(storage::IsDigestValid(left_node_identifier.object_digest()));
   FXL_DCHECK(storage::IsDigestValid(right_node_identifier.object_digest()));

   if (left_node_identifier == right_node_identifier) {
     return Status::OK;
   }

   auto wrapped_next = [on_next = std::move(on_next)](
                           std::unique_ptr<Entry> base,
                           std::unique_ptr<Entry> other) {
     if (other) {
       return on_next({std::move(*other), false});
     }
     return on_next({std::move(*base), true});
   };

   IteratorPair iterators(storage, std::move(wrapped_next));
   RETURN_ON_ERROR(
       iterators.Init(left_node_identifier, right_node_identifier, min_key));

   while (!iterators.Finished()) {
     if (!iterators.SendDiff()) {
       return Status::OK;
     }
     RETURN_ON_ERROR(iterators.Advance());
   }

   return Status::OK;
 }  // namespace

 Status ForEachThreeWayDiffInternal(
     SynchronousStorage* storage, ObjectIdentifier base_node_identifier,
     ObjectIdentifier left_node_identifier,
     ObjectIdentifier right_node_identifier, std::string min_key,
     fit::function<bool(ThreeWayChange)> on_next) {
   FXL_DCHECK(IsDigestValid(base_node_identifier.object_digest()));
   FXL_DCHECK(IsDigestValid(left_node_identifier.object_digest()));
   FXL_DCHECK(IsDigestValid(right_node_identifier.object_digest()));

   if (left_node_identifier == right_node_identifier) {
     return Status::OK;
   }

   ThreeWayIterator iterator(storage);
   RETURN_ON_ERROR(iterator.Init(base_node_identifier, left_node_identifier,
                                 right_node_identifier, min_key));

   while (!iterator.Finished()) {
     if (!on_next(iterator.GetCurrentDiff())) {
       return Status::OK;
     }
     RETURN_ON_ERROR(iterator.Advance());
   }

   return Status::OK;
 }

 }  // namespace

 void ForEachDiff(coroutine::CoroutineService* coroutine_service,
                  PageStorage* page_storage,
                  ObjectIdentifier base_root_identifier,
                  ObjectIdentifier other_root_identifier, std::string min_key,
                  fit::function<bool(EntryChange)> on_next,
                  fit::function<void(Status)> on_done) {
   FXL_DCHECK(storage::IsDigestValid(base_root_identifier.object_digest()));
   FXL_DCHECK(storage::IsDigestValid(other_root_identifier.object_digest()));
   coroutine_service->StartCoroutine(
       [page_storage, base_root_identifier = std::move(base_root_identifier),
        other_root_identifier = std::move(other_root_identifier),
        on_next = std::move(on_next), min_key = std::move(min_key),
        on_done =
            std::move(on_done)](coroutine::CoroutineHandler* handler) mutable {
         SynchronousStorage storage(page_storage, handler);

         on_done(ForEachDiffInternal(&storage, base_root_identifier,
                                     other_root_identifier, std::move(min_key),
                                     std::move(on_next)));
       });
 }

 void ForEachThreeWayDiff(coroutine::CoroutineService* coroutine_service,
                          PageStorage* page_storage,
                          ObjectIdentifier base_root_identifier,
                          ObjectIdentifier left_root_identifier,
                          ObjectIdentifier right_root_identifier,
                          std::string min_key,
                          fit::function<bool(ThreeWayChange)> on_next,
                          fit::function<void(Status)> on_done) {
   FXL_DCHECK(storage::IsDigestValid(base_root_identifier.object_digest()));
   FXL_DCHECK(storage::IsDigestValid(left_root_identifier.object_digest()));
   FXL_DCHECK(storage::IsDigestValid(right_root_identifier.object_digest()));
   coroutine_service->StartCoroutine(
       [page_storage, base_root_identifier = std::move(base_root_identifier),
        left_root_identifier = std::move(left_root_identifier),
        right_root_identifier = std::move(right_root_identifier),
        on_next = std::move(on_next), min_key = std::move(min_key),
        on_done =
            std::move(on_done)](coroutine::CoroutineHandler* handler) mutable {
         SynchronousStorage storage(page_storage, handler);

         on_done(ForEachThreeWayDiffInternal(
             &storage, base_root_identifier, left_root_identifier,
             right_root_identifier, std::move(min_key), std::move(on_next)));
       });
 }

 }  // namespace btree
 }  // namespace storage
	// Copyright 2017 The Fuchsia Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	#include "src/ledger/bin/storage/impl/btree/diff.h"

	#include <lib/fit/function.h>

	#include <utility>

	#include "src/ledger/bin/storage/impl/btree/internal_helper.h"
	#include "src/ledger/bin/storage/impl/btree/iterator.h"
	#include "src/ledger/bin/storage/impl/btree/synchronous_storage.h"
	#include "src/ledger/bin/storage/impl/object_digest.h"

	namespace storage {
	namespace btree {
	namespace {
	// Aggregates 2 \|BTreeIterator\|s and allows to walk through these concurrently
	// to compute the diff. \|on_next\| will be called for each diff entry.
	class IteratorPair {
	public:
	IteratorPair(
	SynchronousStorage* storage,
	fit::function<bool(std::unique_ptr<Entry>, std::unique_ptr<Entry>)>
	on_next)
	: on_next_(std::move(on_next)), left_(storage), right_(storage) {}

	// Initialize the pair with the ids of both roots.
	Status Init(ObjectIdentifier left_node_identifier,
	ObjectIdentifier right_node_identifier, fxl::StringView min_key) {
	RETURN_ON_ERROR(left_.Init(left_node_identifier));
	RETURN_ON_ERROR(right_.Init(right_node_identifier));
	if (!min_key.empty()) {
	RETURN_ON_ERROR(SkipIteratorsTo(min_key));
	}
	Normalize();
	if (!Finished() && !HasDiff()) {
	RETURN_ON_ERROR(Advance());
	}

	return Status::OK;
	}

	bool Finished() const {
	FXL_DCHECK(IsNormalized());
	return right_.Finished();
	}

	// Send the actual diff to the client. Returns \|false\| if the iteration must
	// be stopped.
	bool SendDiff() {
	FXL_DCHECK(HasDiff());

	// If the 2 iterators are on 2 equals values, nothing to do.
	if (left_.HasValue() && right_.HasValue() &&
	left_.CurrentEntry() == right_.CurrentEntry()) {
	return true;
	}

	if (HasSameNextChild()) {
	// If the 2 iterators are on the same child, send a diff for each
	// iterator that is currently on a value.
	if (right_.HasValue()) {
	if (!SendRight()) {
	return false;
	}
	}
	if (left_.HasValue() &&
	(!right_.HasValue() \|\|
	left_.CurrentEntry().key != right_.CurrentEntry().key)) {
	if (!SendLeft()) {
	return false;
	}
	}
	return true;
	}

	// Otherwise, just send the diff of the right node.
	return SendRight();
	}

	// Advance the iterator until there is potentially a diff to send.
	Status Advance() {
	FXL_DCHECK(!Finished());
	do {
	FXL_DCHECK(IsNormalized());

	// If the 2 next children are identical, skip these.
	if (HasSameNextChild()) {
	right_.SkipNextSubTree();
	left_.SkipNextSubTree();
	Normalize();
	continue;
	}

	// If both iterators are sitting on a value for the same key, both need
	// to be advanced.
	if (right_.HasValue() && left_.HasValue() &&
	right_.CurrentEntry().key == left_.CurrentEntry().key) {
	RETURN_ON_ERROR(right_.Advance());
	Swap();
	}

	RETURN_ON_ERROR(right_.Advance());
	Normalize();
	} while (!Finished() && !HasDiff());
	return Status::OK;
	}

	private:
	// Advances the two iterators so that they are both in the first entry that 1)
	// is greater than or equal to min_key and 2) might be different between the
	// two iterators. We consider that the two entries might be different, if they
	// are in btree nodes with different ids.
	Status SkipIteratorsTo(fxl::StringView min_key) {
	for (;;) {
	if (left_.SkipToIndex(min_key)) {
	return right_.SkipTo(min_key);
	}
	if (right_.SkipToIndex(min_key)) {
	left_.SkipToIndex(min_key);
	return Status::OK;
	}

	auto left_child = left_.GetNextChild();
	auto right_child = right_.GetNextChild();
	if (!left_child) {
	return right_.SkipTo(min_key);
	}
	if (!right_child) {
	return left_.SkipTo(min_key);
	}
	if (left_child == right_child) {
	return Status::OK;
	}
	// Same nodes might be in different depths of the btrees. Only descend in
	// each if their current level is the same as or greater than the other
	// one's.
	uint8_t level_left = left_.GetLevel();
	uint8_t level_right = right_.GetLevel();
	if (level_left >= level_right) {
	RETURN_ON_ERROR(left_.Advance());
	}
	if (level_right >= level_left) {
	RETURN_ON_ERROR(right_.Advance());
	}
	}
	}

	// Ensure that the representation of the pair of iterator is normalized
	// according to the following rules:
	// If only one iterator is finished, it is always the left one.
	// If only one iterator is on a value, it is always the left one.
	// If both iterator are on a value, the left one has a key greater or equals
	// to the right one, and if the keys are equals, the iterator are in
	// their original order.
	// If none of the iterators is on a value, the right one has a level
	// greather or equals to the left one.
	//
	// When the iterator is normalized, the different algorithms can cut the
	// number of case it needs to consider.
	void Normalize() {
	if (left_.Finished()) {
	return;
	}
	if (right_.Finished()) {
	Swap();
	return;
	}

	if (right_.HasValue() && left_.HasValue()) {
	if (left_.CurrentEntry().key < right_.CurrentEntry().key) {
	Swap();
	return;
	}
	if (left_.CurrentEntry().key == right_.CurrentEntry().key) {
	ResetSwap();
	}
	return;
	}

	if (left_.HasValue()) {
	return;
	}
	if (right_.HasValue()) {
	Swap();
	return;
	}

	if (left_.GetLevel() > right_.GetLevel()) {
	Swap();
	}
	}

	// Returns if the iterator is normalized. See \|Normalize\| for the
	// definition. This is only used in DCHECKs.
	bool IsNormalized() const {
	if (left_.Finished() \|\| right_.Finished()) {
	return left_.Finished();
	}

	if (left_.HasValue()) {
	if (!right_.HasValue()) {
	return true;
	}

	if (right_.CurrentEntry().key > left_.CurrentEntry().key) {
	return false;
	}

	if (right_.CurrentEntry().key == left_.CurrentEntry().key) {
	if (!diff_from_left_to_right_)
	return diff_from_left_to_right_;
	}

	return true;
	}

	if (right_.HasValue()) {
	return false;
	}

	return right_.GetLevel() >= left_.GetLevel();
	}

	// Returns whether there is a potential diff to send at the current state.
	bool HasDiff() const {
	FXL_DCHECK(IsNormalized());
	return (right_.HasValue() && (left_.Finished() \|\| left_.HasValue())) \|\|
	(left_.HasValue() && HasSameNextChild());
	}

	// Returns whether the 2 iterators have the same next child in the
	// iteration. This allows to skip part of the 2 btrees when they are
	// identicals.
	bool HasSameNextChild() const {
	if (left_.Finished()) {
	return false;
	}
	auto right_child = right_.GetNextChild();
	if (!right_child) {
	return false;
	}
	auto left_child = left_.GetNextChild();
	if (!left_child) {
	return false;
	}
	return right_child == left_child;
	}

	// Swaps the 2 iterators. This is useful to reduce the number of case to
	// consider during the iteration.
	void Swap() {
	std::swap(left_, right_);
	diff_from_left_to_right_ = !diff_from_left_to_right_;
	}

	// Reset the iterators so that they are back in the original order.
	void ResetSwap() {
	if (!diff_from_left_to_right_) {
	Swap();
	}
	}

	// Send a diff using the right iterator.
	bool SendRight() { return Send(&right_, &left_, !diff_from_left_to_right_); }

	// Send a diff using the left iterator.
	bool SendLeft() { return Send(&left_, &right_, diff_from_left_to_right_); }

	bool Send(BTreeIterator* it1, BTreeIterator* it2, bool it1_to_it2) {
	std::unique_ptr<Entry> it1_entry, it2_entry;
	it1_entry = std::make_unique<Entry>(it1->CurrentEntry());
	if (!it2->Finished() && it2->HasValue() &&
	it1->CurrentEntry().key == it2->CurrentEntry().key) {
	it2_entry = std::make_unique<Entry>(it2->CurrentEntry());
	}

	if (it1_to_it2) {
	return on_next_(std::move(it1_entry), std::move(it2_entry));
	}
	return on_next_(std::move(it2_entry), std::move(it1_entry));
	}

	fit::function<bool(std::unique_ptr<Entry>, std::unique_ptr<Entry>)> on_next_;
	BTreeIterator left_;
	BTreeIterator right_;
	// Keep track whether the change is from left to right, or right to left.
	// This allows to switch left and right during the algorithm to handle less
	// cases.
	bool diff_from_left_to_right_ = true;
	};

	// Iterator that does a three-way diff by using two IteratorPair objects in
	// parallel.
	// Here is an attempt to convey what this iterator should be doing:
	// - It creates an IteratorPair (IP thereafter) for each side of the diff
	// (base-to-left and base-to-right).
	// - At the initialization time, it advances each internal IP to their first
	// diff. Each IP (as viewed from here) is on one key: the key of the latest
	// diff it returned.
	// - We always advance the IP with the lowest key, or the one not finished yet.
	// If both are on the same key, we advance both.
	// - The current key considered by the ThreeWayIterator is the lowest key of the
	// latest left and right diffs. If one IP is finished, then the current key is
	// the key of the other IP's diff.
	// - When sending the ThreeWayIterator diff, we consider the current key (per
	// above). If both IPs are on the same key, the diff is easy (we just send
	// everything). However, if the IPs are on different keys, or one of them is
	// finished, we have to consider multiple cases:
	// - If the base entry is present, it means the key/value was present in the
	// base revision. Given that the other IP moved past this key, there is no
	// diff on that side and we copy the base entry to that side entry within
	// the tree-way diff change.
	// - If the base entry is not present, it means the key/value was not present
	// in the base revision and it is an addition.
	class ThreeWayIterator {
	public:
	explicit ThreeWayIterator(SynchronousStorage* storage) {
	base_left_iterators_ = std::make_unique<IteratorPair>(
	storage,
	[this](std::unique_ptr<Entry> base, std::unique_ptr<Entry> left) {
	left_advanced_ = true;
	base_left_.swap(base);
	left_.swap(left);
	return true;
	});
	base_right_iterators_ = std::make_unique<IteratorPair>(
	storage,
	[this](std::unique_ptr<Entry> base, std::unique_ptr<Entry> right) {
	right_advanced_ = true;
	base_right_.swap(base);
	right_.swap(right);
	return true;
	});
	}

	Status Init(ObjectIdentifier base_node_identifier,
	ObjectIdentifier left_node_identifier,
	ObjectIdentifier right_node_identifier, fxl::StringView min_key) {
	RETURN_ON_ERROR(base_left_iterators_->Init(base_node_identifier,
	left_node_identifier, min_key));
	RETURN_ON_ERROR(base_right_iterators_->Init(
	base_node_identifier, right_node_identifier, min_key));
	if (!Finished()) {
	RETURN_ON_ERROR(AdvanceLeft());
	RETURN_ON_ERROR(AdvanceRight());
	}
	return Status::OK;
	}

	bool Finished() const {
	return base_left_iterators_->Finished() &&
	base_right_iterators_->Finished() && !base_left_ && !left_ &&
	!base_right_ && !right_;
	}

	Status Advance() {
	FXL_DCHECK(!Finished());
	if (base_left_iterators_->Finished() && !base_left_ && !left_) {
	RETURN_ON_ERROR(AdvanceRight());
	} else if (base_right_iterators_->Finished() && !base_right_ && !right_) {
	RETURN_ON_ERROR(AdvanceLeft());
	} else if (GetLeftKey() < GetRightKey()) {
	RETURN_ON_ERROR(AdvanceLeft());
	} else if (GetLeftKey() > GetRightKey()) {
	RETURN_ON_ERROR(AdvanceRight());
	} else {
	RETURN_ON_ERROR(AdvanceLeft());
	RETURN_ON_ERROR(AdvanceRight());
	}
	return Status::OK;
	}

	ThreeWayChange GetCurrentDiff() {
	FXL_DCHECK(!Finished());
	ThreeWayChange change;
	change.base = GetBase();
	change.left = GetEntry(change.base, base_left_iterators_, left_, right_);
	change.right = GetEntry(change.base, base_right_iterators_, right_, left_);
	return change;
	}

	private:
	// GetLeftKey (resp. GetRightKey) should not be called if the left (resp.
	// right) IteratorPair is finished.
	const std::string& GetLeftKey() {
	FXL_DCHECK(base_left_ \|\| left_);
	if (base_left_) {
	return base_left_->key;
	}
	return left_->key;
	}

	const std::string& GetRightKey() {
	FXL_DCHECK(base_right_ \|\| right_);
	if (base_right_) {
	return base_right_->key;
	}
	return right_->key;
	}

	Status AdvanceLeft() {
	left_advanced_ = false;
	while (!base_left_iterators_->Finished() && !left_advanced_) {
	if (!base_left_iterators_->SendDiff()) {
	return Status::OK;
	}
	RETURN_ON_ERROR(base_left_iterators_->Advance());
	}
	if (!left_advanced_ && base_left_iterators_->Finished()) {
	base_left_.reset();
	left_.reset();
	}
	return Status::OK;
	}

	Status AdvanceRight() {
	right_advanced_ = false;
	while (!base_right_iterators_->Finished() && !right_advanced_) {
	if (!base_right_iterators_->SendDiff()) {
	return Status::OK;
	}
	RETURN_ON_ERROR(base_right_iterators_->Advance());
	}
	if (!right_advanced_ && base_right_iterators_->Finished()) {
	base_right_.reset();
	right_.reset();
	}
	return Status::OK;
	}

	std::unique_ptr<Entry> GetBase() {
	if (base_left_ && base_right_) {
	if (base_left_->key < base_right_->key) {
	return std::make_unique<Entry>(*base_left_);
	}
	return std::make_unique<Entry>(*base_right_);
	}
	if (!base_right_ && base_left_ &&
	(!right_ \|\| base_left_->key < right_->key)) {
	return std::make_unique<Entry>(*base_left_);
	}
	if (!base_left_ && base_right_ &&
	(!left_ \|\| base_right_->key < left_->key)) {
	return std::make_unique<Entry>(*base_right_);
	}
	return std::unique_ptr<Entry>();
	}

	std::unique_ptr<Entry> GetEntry(
	const std::unique_ptr<Entry>& base,
	const std::unique_ptr<IteratorPair>& this_iterator,
	const std::unique_ptr<Entry>& this_entry,
	const std::unique_ptr<Entry>& other_entry) {
	if (base) {
	if (this_entry && base->key == this_entry->key) {
	return std::make_unique<Entry>(*this_entry);
	}
	if (this_entry && base->key < this_entry->key) {
	return std::make_unique<Entry>(*base);
	}
	if (this_iterator->Finished()) {
	return std::make_unique<Entry>(*base);
	}
	return std::unique_ptr<Entry>();
	}
	if (!this_entry \|\| (other_entry && other_entry->key < this_entry->key)) {
	return std::unique_ptr<Entry>();
	}
	return std::make_unique<Entry>(*this_entry);
	}

	bool left_advanced_ = false;
	bool right_advanced_ = false;

	std::unique_ptr<Entry> base_left_;
	std::unique_ptr<Entry> base_right_;
	std::unique_ptr<Entry> left_;
	std::unique_ptr<Entry> right_;

	std::unique_ptr<IteratorPair> base_left_iterators_;
	std::unique_ptr<IteratorPair> base_right_iterators_;
	};

	Status ForEachDiffInternal(SynchronousStorage* storage,
	ObjectIdentifier left_node_identifier,
	ObjectIdentifier right_node_identifier,
	std::string min_key,
	fit::function<bool(EntryChange)> on_next) {
	FXL_DCHECK(storage::IsDigestValid(left_node_identifier.object_digest()));
	FXL_DCHECK(storage::IsDigestValid(right_node_identifier.object_digest()));

	if (left_node_identifier == right_node_identifier) {
	return Status::OK;
	}

	auto wrapped_next = [on_next = std::move(on_next)](
	std::unique_ptr<Entry> base,
	std::unique_ptr<Entry> other) {
	if (other) {
	return on_next({std::move(*other), false});
	}
	return on_next({std::move(*base), true});
	};

	IteratorPair iterators(storage, std::move(wrapped_next));
	RETURN_ON_ERROR(
	iterators.Init(left_node_identifier, right_node_identifier, min_key));

	while (!iterators.Finished()) {
	if (!iterators.SendDiff()) {
	return Status::OK;
	}
	RETURN_ON_ERROR(iterators.Advance());
	}

	return Status::OK;
	} // namespace

	Status ForEachThreeWayDiffInternal(
	SynchronousStorage* storage, ObjectIdentifier base_node_identifier,
	ObjectIdentifier left_node_identifier,
	ObjectIdentifier right_node_identifier, std::string min_key,
	fit::function<bool(ThreeWayChange)> on_next) {
	FXL_DCHECK(IsDigestValid(base_node_identifier.object_digest()));
	FXL_DCHECK(IsDigestValid(left_node_identifier.object_digest()));
	FXL_DCHECK(IsDigestValid(right_node_identifier.object_digest()));

	if (left_node_identifier == right_node_identifier) {
	return Status::OK;
	}

	ThreeWayIterator iterator(storage);
	RETURN_ON_ERROR(iterator.Init(base_node_identifier, left_node_identifier,
	right_node_identifier, min_key));

	while (!iterator.Finished()) {
	if (!on_next(iterator.GetCurrentDiff())) {
	return Status::OK;
	}
	RETURN_ON_ERROR(iterator.Advance());
	}

	return Status::OK;
	}

	} // namespace

	void ForEachDiff(coroutine::CoroutineService* coroutine_service,
	PageStorage* page_storage,
	ObjectIdentifier base_root_identifier,
	ObjectIdentifier other_root_identifier, std::string min_key,
	fit::function<bool(EntryChange)> on_next,
	fit::function<void(Status)> on_done) {
	FXL_DCHECK(storage::IsDigestValid(base_root_identifier.object_digest()));
	FXL_DCHECK(storage::IsDigestValid(other_root_identifier.object_digest()));
	coroutine_service->StartCoroutine(
	[page_storage, base_root_identifier = std::move(base_root_identifier),
	other_root_identifier = std::move(other_root_identifier),
	on_next = std::move(on_next), min_key = std::move(min_key),
	on_done =
	std::move(on_done)](coroutine::CoroutineHandler* handler) mutable {
	SynchronousStorage storage(page_storage, handler);

	on_done(ForEachDiffInternal(&storage, base_root_identifier,
	other_root_identifier, std::move(min_key),
	std::move(on_next)));
	});
	}

	void ForEachThreeWayDiff(coroutine::CoroutineService* coroutine_service,
	PageStorage* page_storage,
	ObjectIdentifier base_root_identifier,
	ObjectIdentifier left_root_identifier,
	ObjectIdentifier right_root_identifier,
	std::string min_key,
	fit::function<bool(ThreeWayChange)> on_next,
	fit::function<void(Status)> on_done) {
	FXL_DCHECK(storage::IsDigestValid(base_root_identifier.object_digest()));
	FXL_DCHECK(storage::IsDigestValid(left_root_identifier.object_digest()));
	FXL_DCHECK(storage::IsDigestValid(right_root_identifier.object_digest()));
	coroutine_service->StartCoroutine(
	[page_storage, base_root_identifier = std::move(base_root_identifier),
	left_root_identifier = std::move(left_root_identifier),
	right_root_identifier = std::move(right_root_identifier),
	on_next = std::move(on_next), min_key = std::move(min_key),
	on_done =
	std::move(on_done)](coroutine::CoroutineHandler* handler) mutable {
	SynchronousStorage storage(page_storage, handler);

	on_done(ForEachThreeWayDiffInternal(
	&storage, base_root_identifier, left_root_identifier,
	right_root_identifier, std::move(min_key), std::move(on_next)));
	});
	}

	} // namespace btree
	} // namespace storage