Add LwwRegister CRDT

blimp/helium/lww_register.h

Index: blimp/helium/lww_register.h
diff --git a/blimp/helium/lww_register.h b/blimp/helium/lww_register.h
new file mode 100644
index 0000000000000000000000000000000000000000..fe04fbf43c124fc22a513ef8b96dcbf35334c3ff
--- /dev/null
+++ b/blimp/helium/lww_register.h
@@ -0,0 +1,115 @@
+// Copyright 2016 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.
+
+#ifndef BLIMP_HELIUM_LWW_REGISTER_H_
+#define BLIMP_HELIUM_LWW_REGISTER_H_
+
+#include "base/memory/ptr_util.h"
+#include "blimp/helium/blimp_helium_export.h"
+#include "blimp/helium/syncable.h"
+#include "blimp/helium/syncable_common.h"
+#include "blimp/helium/syncable_primitive_serializer.h"
+#include "blimp/helium/version_vector.h"
+#include "blimp/helium/version_vector_generator.h"
+#include "third_party/protobuf/src/google/protobuf/io/coded_stream.h"
+
+namespace blimp {
+namespace helium {
+
+// Provides a simple syncable and atomically-writable "register" holding
+// contents of type |RegisterType|. When there is a write conflict, it is
+// resolved by assuming the writer indicated by |bias| has the correct value.
+template <class RegisterType>
+class BLIMP_HELIUM_EXPORT LwwRegister : public Syncable {
+ public:
+  LwwRegister(VersionVectorGenerator* version_gen, Peer bias, Peer running_as);

[steimel, 2016/10/19 16:42:49]

Since I end up just storing whether bias == running_as, would it make sense to
just take a bool?

[Kevin M, 2016/10/19 17:49:11]

That's up to you, but IMO explicitly breaking out the parameters makes things
more explicit?

[steimel, 2016/10/20 00:30:07]

Agreed. I'll leave it

+  ~LwwRegister() = default;
+
+  void Set(const RegisterType& value);
+
+  const RegisterType& Get() const;
+
+  // Syncable implementation.
+  bool ModifiedSince(const VersionVector& from) const override;
+  void CreateChangesetToCurrent(
+      const VersionVector& from,
+      google::protobuf::io::CodedOutputStream* output_stream) override;
+  void ApplyChangeset(
+      const VersionVector& from,
+      const VersionVector& to,
+      google::protobuf::io::CodedInputStream* input_stream) override;
+  void ReleaseCheckpointsBefore(const VersionVector& checkpoint) override;
+
+ private:
+  VersionVectorGenerator* version_gen_;

[CJ, 2016/10/20 00:20:16]

Does the LWW_Register take ownership over the VersionVectorGenerator? If so,
unique_ptr for this.

[steimel, 2016/10/20 00:30:06]

I believe a single, global generator will be used with all syncables, so I won't
own it

+  VersionVector last_modified_;
+  bool locally_owned_;
+  RegisterType value_;
+  bool value_set_ = false;
+
+  DISALLOW_COPY_AND_ASSIGN(LwwRegister);
+};
+
+template <class RegisterType>
+LwwRegister<RegisterType>::LwwRegister(VersionVectorGenerator* version_gen,
+                                       Peer bias,
+                                       Peer running_as)
+    : version_gen_(version_gen),
+      last_modified_(version_gen->current()),
+      locally_owned_(bias == running_as) {
+  DCHECK(version_gen_);
+}
+
+template <class RegisterType>
+void LwwRegister<RegisterType>::Set(const RegisterType& value) {
+  value_ = value;
+  value_set_ = true;
+  version_gen_->Increment();
+  last_modified_ = last_modified_.MergeWith(version_gen_->current());
+}
+
+template <class RegisterType>
+const RegisterType& LwwRegister<RegisterType>::Get() const {
+  DCHECK(value_set_);
+  return value_;
+}
+
+template <class RegisterType>
+bool LwwRegister<RegisterType>::ModifiedSince(const VersionVector& from) const {
+  return from.local_revision() < last_modified_.local_revision();
+}
+
+template <class RegisterType>
+void LwwRegister<RegisterType>::CreateChangesetToCurrent(
+    const VersionVector& from,
+    google::protobuf::io::CodedOutputStream* output_stream) {
+  SyncablePrimitiveSerializer::Serialize(value_, output_stream);
+}
+
+template <class RegisterType>
+void LwwRegister<RegisterType>::ApplyChangeset(
+    const VersionVector& from,
+    const VersionVector& to,
+    google::protobuf::io::CodedInputStream* input_stream) {
+  VersionVector::Comparison cmp = last_modified_.CompareTo(to);
+  if (cmp == VersionVector::Comparison::LessThan ||
+      (cmp == VersionVector::Comparison::Conflict && !locally_owned_)) {
+    // TODO(steimel): deal with Deserialize returning false once ApplyChangeset
+    // returns a HeliumResult.
+    SyncablePrimitiveSerializer::Deserialize(input_stream, &value_);
+    value_set_ = true;
+  }
+  last_modified_ = last_modified_.MergeWith(to);
+}
+
+template <class RegisterType>
+void LwwRegister<RegisterType>::ReleaseCheckpointsBefore(
+    const VersionVector& checkpoint) {
+  // no-op
+}
+
+}  // namespace helium
+}  // namespace blimp
+
+#endif  // BLIMP_HELIUM_LWW_REGISTER_H_