Add LwwRegister CRDT

blimp/helium/lww_register.h

+// 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,
+              Bias bias,
+              RunningAs running_as);
+  ~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_;
+  VersionVector last_modified_;
+  bool locally_owned_;
+  RegisterType value_;
+  bool value_set_;

[Kevin M, 2016/10/19 00:27:08]

Suggestion: you can use an inline initializer here

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

Done.

+
+  DISALLOW_COPY_AND_ASSIGN(LwwRegister);
+};
+
+template <class RegisterType>
+LwwRegister<RegisterType>::LwwRegister(VersionVectorGenerator* version_gen,
+                                       Bias bias,
+                                       RunningAs running_as)
+    : version_gen_(version_gen),
+      last_modified_(version_gen->current()),
+      value_set_(false) {
+  DCHECK(version_gen_);
+  locally_owned_ = ((running_as == RunningAs::Client && bias == Bias::Client) ||
+                    (running_as == RunningAs::Engine && bias == Bias::Engine));
+}
+
+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_