Initial UniquePositions implementation

sync/internal_api/public/base/unique_position.cc

+// Copyright (c) 2012 The Chromium 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 "sync/internal_api/public/base/unique_position.h"
+
+#include "base/logging.h"
+#include "base/string_number_conversions.h"
+
+namespace syncer {
+
+const size_t UniquePosition::kSuffixLength = 28;
+
+// static.
+bool UniquePosition::IsValidSuffix(const std::string& suffix) {
+  // The suffix must be exactly the specified length, otherwise unique suffixes
+  // are not sufficient to guarantee unique positions (because prefix + suffix
+  // == p + refixsuffix).
+  return suffix.length() == kSuffixLength;
+}
+
+// static.
+bool UniquePosition::IsValidBytes(const std::string& bytes) {
+  // The first condition ensures that our suffix uniqueness is sufficient to
+  // guarantee position uniqueness.  Otherwise, it's possible the end of some
+  // prefix + some short suffix == some long suffix.
+  // The second condition ensures that FindSmallerWithSuffix can always return a
+  // result.
+  return bytes.length() >= kSuffixLength
+      && bytes[bytes.length()-1] != 0;
+}
+
+// static.
+UniquePosition UniquePosition::CreateInvalid() {
+  UniquePosition pos;
+  DCHECK(!pos.IsValid());
+  return pos;
+}
+
+// static.
+UniquePosition UniquePosition::FromBytes(
+    const std::string& bytes) {

[akalin, 2012/12/29 10:17:13]

fits on previous line?

[rlarocque, 2013/01/07 23:22:12]

Done.

+  UniquePosition result(bytes);
+  return result;
+}
+
+// static.
+UniquePosition UniquePosition::FromInt64(
+    int64 x, const std::string& suffix) {
+  uint64 y = static_cast<uint64>(x);
+  y ^= 0x8000000000000000ULL; // Make it non-negative.
+  std::string bytes(8, 0);
+  for (int i = 7; i >= 0; --i) {
+    bytes[i] = static_cast<uint8>(y);
+    y >>= 8;
+  }
+  return UniquePosition(bytes, suffix);
+}
+
+// static.
+UniquePosition UniquePosition::InitialPosition(
+    const std::string& suffix) {
+  DCHECK(IsValidSuffix(suffix));
+  return UniquePosition("", suffix);
+}
+
+// static.
+UniquePosition UniquePosition::Before(
+    const UniquePosition& x,
+    const std::string& suffix) {
+  DCHECK(IsValidSuffix(suffix));
+  DCHECK(x.IsValid());
+  const std::string& before = FindSmallerWithSuffix(x.bytes_, suffix);
+  return UniquePosition(before, suffix);
+}
+
+// static.
+UniquePosition UniquePosition::After(
+    const UniquePosition& x,
+    const std::string& suffix) {
+  DCHECK(IsValidSuffix(suffix));
+  DCHECK(x.IsValid());
+  const std::string& after = FindGreaterWithSuffix(x.bytes_, suffix);
+  return UniquePosition(after, suffix);
+}
+
+// static.
+UniquePosition UniquePosition::Between(
+    const UniquePosition& before,
+    const UniquePosition& after,
+    const std::string& suffix) {
+  DCHECK(before.IsValid());
+  DCHECK(after.IsValid());
+  DCHECK(before.LessThan(after));
+  DCHECK(IsValidSuffix(suffix));
+  const std::string& mid =
+      FindBetweenWithSuffix(before.bytes_, after.bytes_, suffix);
+  return UniquePosition(mid, suffix);
+}
+
+UniquePosition::UniquePosition() : is_valid_(false) { };

[akalin, 2012/12/29 10:17:13]

"{ };" -> "{}"

[rlarocque, 2013/01/07 23:22:12]

Done.

+
+bool UniquePosition::LessThan(const UniquePosition& other) const {
+  DCHECK(this->IsValid());
+  DCHECK(other.IsValid());
+  return bytes_ < other.bytes_;
+}
+
+bool UniquePosition::Equals(const UniquePosition& other) const {
+  if (!this->IsValid() && !other.IsValid())
+    return true;
+
+  return bytes_ == other.bytes_;
+}
+
+const std::string& UniquePosition::ToInternalValue() const {
+  return bytes_;
+}
+
+int64 UniquePosition::ToInt64() const {
+  uint64 y = 0;
+  const std::string& s = ToInternalValue();
+  size_t l = sizeof(int64);
+  if (s.length() < l) {
+    NOTREACHED();
+    l = s.length();
+  }
+  for (size_t i = 0; i < l; ++i) {
+    const uint8 byte = s[l - i - 1];
+    y |= static_cast<uint64>(byte) << (i * 8);
+  }
+  y ^= 0x8000000000000000ULL;
+  // This is technically implementation-defined if y > INT64_MAX, so
+  // we're assuming that we're on a twos-complement machine.
+  return static_cast<int64>(y);
+}
+
+bool UniquePosition::IsValid() const {
+  return is_valid_;
+}
+
+std::string UniquePosition::ToDebugString() const {
+  std::string debug_string = base::HexEncode(bytes_.data(), bytes_.length());
+  if (!IsValid()) {
+    debug_string = "INVALID[" + debug_string + "]";
+  }
+  return debug_string;;
+}
+
+std::string UniquePosition::FindSmallerWithSuffix(
+    const std::string& reference,
+    const std::string& suffix) {
+  size_t ref_zeroes = reference.find_first_not_of('\0');
+  size_t suffix_zeroes = suffix.find_first_not_of('\0');
+
+  // Neither of our inputs are allowed to have trailing zeroes, so the following
+  // must be true.
+  DCHECK(ref_zeroes != std::string::npos);

[akalin, 2012/12/29 10:17:13]

DCHECK_NE?

[rlarocque, 2013/01/07 23:22:12]

Done.

+  DCHECK(suffix_zeroes != std::string::npos);
+
+  if (suffix_zeroes > ref_zeroes) {
+    // Implies ref < suffix.

[akalin, 2012/12/29 10:17:13]

backwards -- suffix < ref

[rlarocque, 2013/01/07 23:22:12]

Done.

+    return "";
+  }
+
+  if (suffix.substr(suffix_zeroes) < reference.substr(ref_zeroes)) {
+    // Prepend zeroes so the result has as many zero digits as |reference|.
+    return std::string(ref_zeroes - suffix_zeroes, '\0');
+  } else if (suffix_zeroes > 1) {
+    // Prepend zeroes so the result has one more zero digit than |reference|.
+    // We could also take the "else" branch below, but taking this branch will
+    // give us a smaller result.
+    return std::string(ref_zeroes - suffix_zeroes + 1, '\0');
+  } else {
+    // Prepend zeroes to match those in the |reference|, then something smaller
+    // than the first non-zero digit in |reference|.
+    char lt_digit = (uint8)reference[ref_zeroes]/2;

[akalin, 2012/12/29 10:17:13]

static_cast<uint8>?

[rlarocque, 2013/01/07 23:22:12]

Done.

+    return std::string(ref_zeroes, '\0') + lt_digit;
+  }
+}
+
+// static
+std::string UniquePosition::FindGreaterWithSuffix(
+    const std::string& reference,
+    const std::string& suffix) {
+  size_t ref_FFs = reference.find_first_not_of(kuint8max);
+  size_t suffix_FFs = suffix.find_first_not_of(kuint8max);
+
+  if (ref_FFs == std::string::npos) {
+    ref_FFs = reference.length();
+  }
+  if (suffix_FFs == std::string::npos) {
+    suffix_FFs = suffix.length();
+  }
+
+  if (suffix_FFs > ref_FFs) {
+    // Implies suffix > reference

[akalin, 2012/12/29 10:17:13]

append .

[rlarocque, 2013/01/07 23:22:12]

Done.

+    return "";
+  }
+
+  if (suffix.substr(suffix_FFs) > reference.substr(ref_FFs)) {
+    // Prepend FF digits to match those in |reference|.
+    return std::string(ref_FFs - suffix_FFs, kuint8max);
+  } else if (suffix_FFs > 1) {
+    // Prepend enough leading FF digits so result has one more of them than
+    // |reference| does.  We could also take the "else" branch below, but this
+    // gives us a smaller result.
+    return std::string(ref_FFs - suffix_FFs + 1, kuint8max);
+  } else {
+    // Prepend FF digits to match those in |reference|, then something larger
+    // than the first non-FF digit in |reference|.
+    char gt_digit = (uint8)reference[ref_FFs] +

[akalin, 2012/12/29 10:17:13]

static_cast<uint8>

[rlarocque, 2013/01/07 23:22:12]

Done.

+        (kuint8max - (uint8)reference[ref_FFs] + 1) / 2;
+    return std::string(ref_FFs, kuint8max) + gt_digit;
+  }
+}
+
+// static
+std::string UniquePosition::FindBetweenWithSuffix(
+    const std::string& before,
+    const std::string& after,
+    const std::string& suffix) {
+  DCHECK(IsValidSuffix(suffix));
+  DCHECK_NE(before, after);
+  DCHECK_LT(before, after);
+
+  std::string mid;
+
+  // Sometimes our suffix puts us where we want to be.
+  if (before < suffix && suffix < after) {
+    return "";
+  }
+
+  size_t i = 0;
+  for ( ; i < std::min(before.length(), after.length()); ++i) {
+    uint8 a_digit = before[i];
+    uint8 b_digit = after[i];
+
+    if (b_digit - a_digit >= 2) {
+      mid.push_back(a_digit + (b_digit - a_digit)/2);
+      return mid;
+    } else if (a_digit == b_digit) {
+      mid.push_back(a_digit);
+
+      // Both strings are equal so far.  Will appending the suffix at this point
+      // give us the comparison we're looking for?
+      if (before.substr(i+1) < suffix && suffix < after.substr(i+1)) {

[akalin, 2012/12/29 10:17:13]

pretty sure you can rewrite this one to something similar to
FindLessThanWithSuffix, etc.  Or something like before (+)
FindLessThanWithSuffix(after (-) before, suffix (-) before) (where (+) and (-)
treat the strings like binary numbers), right?)

But I'm fine with leaving this as a TODO or something

[rlarocque, 2013/01/07 23:22:12]

I'm not sure I follow you.  How would the binary subtraction "suffix (-) before"
work if before > suffix?

I'm not sure how relevant this is, since I don't understand what you're
proposing.  However, here's the corner case that haunts all my attempts to
optimize this function:

a == "ABCDE..."
b == "ABCDG..."
suffix == "CDF..."

The ideal result would be "AB" + "CDF..." (ie. return with mid == "AB"), because
"ABCDE..." < "AB" + "CDF..." < "ABCDG...".  Any attempt to skip past the
matching common prefix of 'a' and 'b' would skip past this result, because the
common prefix is "ABCD".  That's why I think any common prefix approach in the
style of FindLessThanWithSuffix() won't work.  

There's probably some fast algorithm out there that handles this case properly. 
I don't think it's possible to do it with a single pass, though.  

I've added a comment suggesting we should take another look at this function at
some point.  

+        return mid;
+      }
+    } else {
+      DCHECK_EQ(b_digit - a_digit, 1);  // Implied by above if branches.
+
+      // The two options are off by one digit.  The choice of whether to round
+      // up or down here will have consequences on what we do with the remaining
+      // digits.  Exploring both options is an optimization and is not required
+      // for the correctness of this algorithm.
+
+      // Option A: Round down the current digit.  This makes our |mid| <
+      // |after|, no matter what we append afterwards.  We then focus on
+      // appending digits until |mid| > |before|.
+      std::string mid_a = mid;
+      mid_a.push_back(a_digit);
+      mid_a.append(FindGreaterWithSuffix(before.substr(i+1), suffix));
+
+      // Option B: Round up the current digit.  This makes our |mid| > |before|,
+      // no matter what we append afterwards.  We then focus on appending digits
+      // until |mid| < |after|.  Note that this option may not be viable if the
+      // current digit is the last one in |after|, so we skip the option in that
+      // case.
+      if (after.length() > i+1) {
+        std::string mid_b = mid;
+        mid_b.push_back(b_digit);
+        mid_b.append(FindSmallerWithSuffix(after.substr(i+1), suffix));
+
+        // Does this give us a shorter position value?  If so, use it.
+        if (mid_b.length() < mid_a.length()) {
+          return mid_b;
+        }
+      }
+      return mid_a;
+    }
+  }
+
+  // If we haven't found a midpoint yet, the following must be true:
+  DCHECK_EQ(before.substr(0, i), after.substr(0, i));
+  DCHECK_EQ(before, mid);
+  DCHECK_LT(before.length(), after.length());
+
+  // We know that we'll need to append at least one more byte to |mid| in the
+  // process of making it < |after|.  Appending any digit, regardless of the
+  // value, will make |before| < |mid|.  Therefore, the following will get us a
+  // valid position.
+
+  mid.append(FindSmallerWithSuffix(after.substr(i), suffix));
+  return mid;
+}
+
+UniquePosition::UniquePosition(const std::string& internal_rep)
+    : bytes_(internal_rep),
+      is_valid_(IsValidBytes(bytes_)) {
+}
+
+UniquePosition::UniquePosition(
+    const std::string& prefix,
+    const std::string& suffix)
+  : bytes_(prefix + suffix),
+    is_valid_(IsValidBytes(bytes_)) {
+  DCHECK(IsValidSuffix(suffix));
+  DCHECK(IsValid());
+}
+
+}  // namespace syncer