Use dotted version vectors for causality tracking when running tests

lrascao · lrascao · commit 94e60bedda66 · 2017-07-04T18:59:25.000+01:00
diff --git a/rebar.config b/rebar.config
@@ -22,6 +22,12 @@
     ]},
     {docs, [
         {deps, [edown]}
+    ]},
+    {test, [
+        {deps, [
+            {dvv, ".*",
+                {git, "https://github.com/ricardobcl/Dotted-Version-Vectors.git", {tag, "1.0"}}}
+        ]}
     ]}
 ]}.
 
diff --git a/test/plumtree_SUITE.erl b/test/plumtree_SUITE.erl
@@ -161,8 +161,8 @@ membership_test(Config) ->
     lists:foreach(fun(_) ->
                     {_, Node} = plumtree_test_utils:select_random(Nodes),
                     ok = rpc:call(Node,
-                                  plumtree_broadcast, broadcast,
-                                  [{k, rand_compat:uniform()}, plumtree_test_broadcast_handler])
+                                  plumtree_test_broadcast_handler, put,
+                                  [k, rand_compat:uniform()])
                   end, lists:seq(1, BroadcastRounds1)),
     %% allow 100ms per broadcast to settle
     timer:sleep(100 * BroadcastRounds1),
@@ -177,8 +177,8 @@ membership_test(Config) ->
     lists:foreach(fun(_) ->
                     {_, Node} = plumtree_test_utils:select_random(Nodes),
                     ok = rpc:call(Node,
-                                  plumtree_broadcast, broadcast,
-                                  [{k, rand_compat:uniform()}, plumtree_test_broadcast_handler])
+                                  plumtree_test_broadcast_handler, put,
+                                  [k, rand_compat:uniform()])
                   end, lists:seq(1, BroadcastRounds2)),
     %% allow 100ms per broadcast to settle
     timer:sleep(100 * BroadcastRounds1),
@@ -225,8 +225,8 @@ broadcast_test(Config) ->
     lists:foreach(fun(_) ->
                     {_, Node} = plumtree_test_utils:select_random(Nodes),
                     ok = rpc:call(Node,
-                                  plumtree_broadcast, broadcast,
-                                  [{k, rand_compat:uniform()}, plumtree_test_broadcast_handler])
+                                  plumtree_test_broadcast_handler, put,
+                                  [k, rand_compat:uniform()])
                   end, lists:seq(1, BroadcastRounds1)),
     %% allow 500ms per broadcast to settle
     timer:sleep(200 * BroadcastRounds1),
@@ -238,13 +238,13 @@ broadcast_test(Config) ->
     Rand = rand_compat:uniform(),
     {_, RandomNode} = plumtree_test_utils:select_random(Nodes),
     ok = rpc:call(RandomNode,
-                  plumtree_broadcast, broadcast,
-                  [{k, Rand}, plumtree_test_broadcast_handler]),  
+                  plumtree_test_broadcast_handler, put,
+                  [k, Rand]),
     ct:pal("requested node ~p to broadcast {k, ~p}",
            [RandomNode, Rand]),
 
     VerifyFun = fun(Node, Rand0) ->
-                    case rpc:call(Node, plumtree_test_broadcast_handler, read, [k]) of
+                    case rpc:call(Node, plumtree_test_broadcast_handler, get, [k]) of
                         {error, not_found} ->
                             {false, not_found};
                         {ok, NodeRand} when NodeRand =:= Rand0 -> true;
diff --git a/test/plumtree_test_broadcast_handler.erl b/test/plumtree_test_broadcast_handler.erl
@@ -30,6 +30,7 @@
          is_stale/1,
          graft/1,
          exchange/1]).
+
 %% gen_server callbacks
 -export([init/1,
          handle_call/3,
@@ -40,40 +41,42 @@
 
 %% API
 -export([start_link/0,
-         read/1]).
+         get/1,
+         put/2]).
 
 -record(state, {}).
--type state()           :: #state{}.
+-type state() :: #state{}.
 
 -spec start_link() -> ok.
 start_link() ->
     {ok, _} = gen_server:start_link({local, ?SERVER}, ?MODULE,
                                     [], []),
     ok.
 
--spec read(Key :: any()) -> {ok, any()} | {error, not_found}.
-read(Key) ->
-    case ets:lookup(?MODULE, Key) of
-        [{Key, Value}] ->
-            % lager:info("read key ~p: ~p",
-            %            [Key, Value]),
-            {ok, Value};
-        _ ->
-            lager:info("unable to find key: ~p",
-                       [Key]),
-            {error, not_found}
+-spec get(Key :: any()) -> {error, not_found} | {ok, any()}.
+get(Key) ->
+    case dbread(Key) of
+        undefined -> {error, not_found};
+        Obj ->
+            {ok, plumtree_test_object:value(Obj)}
     end.
 
+-spec put(Key :: any(),
+          Value :: any()) -> ok.
+put(Key, Value) ->
+    Existing = dbread(Key),
+    UpdatedObj = plumtree_test_object:modify(Existing, Value, this_server_id()),
+    dbwrite(Key, UpdatedObj),
+    plumtree_broadcast:broadcast({Key, UpdatedObj}, plumtree_test_broadcast_handler),
+    ok.
+
 %%%===================================================================
 %%% gen_server callbacks
 %%%===================================================================
 
 %% @private
 -spec init([[any()], ...]) -> {ok, state()}.
 init([]) ->
-    msgs_seen = ets:new(msgs_seen, [named_table, set, public,
-                                    {keypos, 1},
-                                    {read_concurrency, true}]),
     ?MODULE = ets:new(?MODULE, [named_table, set, public,
                                 {keypos, 1},
                                 {read_concurrency, true}]),
@@ -111,61 +114,50 @@ code_change(_OldVsn, State, _Extra) ->
 
 %% Return a two-tuple of message id and payload from a given broadcast
 -spec broadcast_data(any()) -> {any(), any()}.
-broadcast_data({Key, _Value} = Data) ->
-    MsgId = erlang:phash2(Data),
+broadcast_data({Key, Object}) ->
+    MsgId = {Key, plumtree_test_object:context(Object)},
     lager:info("broadcast_data(~p), msg id: ~p",
-               [Data, MsgId]),
-    true = ets:insert(msgs_seen, {MsgId, Key}),
-    true = ets:insert(?MODULE, Data),
-    {MsgId, Data}.
+               [Object, MsgId]),
+    {MsgId, Object}.
 
 %% Given the message id and payload, merge the message in the local state.
 %% If the message has already been received return `false', otherwise return `true'
 -spec merge(any(), any()) -> boolean().
-merge(MsgId, {Key, _Value} = Payload) ->
-    case ets:lookup(msgs_seen, MsgId) of
-        [{MsgId, _}] ->
-            lager:info("msg with id ~p has already been seen",
-                      [MsgId]),
-            false;
-        _ ->
-            lager:info("merging(~p, ~p) in local state",
-                       [MsgId, Payload]),
-            %% insert the message in the local state
-            true = ets:insert(?MODULE, Payload),
-            %% mark this message as been seen
-            true = ets:insert_new(msgs_seen, {MsgId, Key}),
+merge({Key, _Context} = MsgId, RemoteObj) ->
+    lager:info("merge msg id ~p, object: ~p",
+              [MsgId, RemoteObj]),
+    Existing = dbread(Key),
+    case plumtree_test_object:reconcile(RemoteObj, Existing) of
+        false -> false;
+        {true, Reconciled} ->
+            dbwrite(Key, Reconciled),
             true
     end.
 
 %% Return true if the message (given the message id) has already been received.
 %% `false' otherwise
 -spec is_stale(any()) -> boolean().
-is_stale(MsgId) ->
-    case ets:lookup(msgs_seen, MsgId) of
-        [{MsgId, _}] ->
-            lager:info("is_stale(~p): ~p",
-                       [MsgId, true]),
-            true;
-        _ ->
-            lager:info("is_stale(~p): ~p",
-                       [MsgId, false]),
-            false
-    end.
+is_stale({Key, Context}) ->
+    Existing = dbread(Key),
+    plumtree_test_object:is_stale(Context, Existing).
 
 %% Return the message associated with the given message id. In some cases a message
 %% has already been sent with information that subsumes the message associated with the given
 %% message id. In this case, `stale' is returned.
 -spec graft(any()) -> stale | {ok, any()} | {error, any()}.
-graft(MsgId) ->
-    % lager:info("graft(~p)",
-    %            [MsgId]),
-    case ets:lookup(msgs_seen, MsgId) of
-        [{MsgId, Key}] ->
-            [{Key,Msg}] = ets:lookup(?MODULE, Key),
-            {ok, {Key, Msg}};
-        _ ->
-            {error, not_found}
+graft({Key, Context}) ->
+    case dbread(Key) of
+        undefined ->
+            %% this *really* should not happen
+            lager:alert("unable to graft key ~p, could not find it",
+                        [Key]),
+            {error, not_found};
+        Object ->
+            LocalContext = plumtree_test_object:context(Object),
+            case LocalContext =:= Context of
+                true -> {ok, Object};
+                false -> stale
+            end
     end.
 
 %% Trigger an exchange between the local handler and the handler on the given node.
@@ -176,3 +168,24 @@ graft(MsgId) ->
 -spec exchange(node()) -> {ok, pid()} | {error, term()}.
 exchange(_Node) ->
     {ok, self()}.
+
+%% @private
+-spec dbread(Key :: any()) -> any() | undefined.
+dbread(Key) ->
+    case ets:lookup(?MODULE, Key) of
+        [{Key, Object}] ->
+            Object;
+        _ ->
+            undefined
+    end.
+
+%% @private
+-spec dbwrite(Key :: any(),
+              Value :: any()) -> any().
+dbwrite(Key, Object) ->
+    ets:insert(?MODULE, {Key, Object}),
+    Object.
+
+%% @private
+this_server_id() -> node().
+
diff --git a/test/plumtree_test_object.erl b/test/plumtree_test_object.erl
@@ -0,0 +1,118 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright (c) 2013 Basho Technologies, Inc.  All Rights Reserved.
+%%
+%% This file is provided to you under the Apache License,
+%% Version 2.0 (the "License"); you may not use this file
+%% except in compliance with the License.  You may obtain
+%% a copy of the License at
+%%
+%%   http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing,
+%% software distributed under the License is distributed on an
+%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+%% KIND, either express or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.
+%%
+%% -------------------------------------------------------------------
+-module(plumtree_test_object).
+
+%% API
+-export([context/1,
+         value/1,
+         values/1,
+         modify/3,
+         modify/4,
+         is_stale/2,
+         reconcile/2]).
+
+-type test_object()     :: dvvset:clock().
+-type test_context()    :: dvvset:vector().
+-type test_value()      :: any().
+
+%% @doc returns a single value. if the object holds more than one value an error is generated
+%% @see values/2
+-spec value(test_object()) -> test_value().
+value(Object) ->
+    [Value] = values(Object),
+    Value.
+
+%% @doc returns a list of values held in the object
+-spec values(test_object()) -> [test_value()].
+values(Object) ->
+    [Value || {Value, _Ts} <- dvvset:values(Object)].
+
+%% @doc returns the context (opaque causal history) for the given object
+-spec context(test_object()) -> test_context().
+context(Object) ->
+    dvvset:join(Object).
+
+%% @doc modifies a potentially existing object, setting its value and updating
+%% the causual history. 
+-spec modify(test_object() | undefined,
+             test_value(),
+             term()) -> test_object().
+modify(undefined, Value, ServerId) ->
+    modify(undefined, undefined, Value, ServerId);
+modify(Object, Value, ServerId) ->
+    modify(Object, context(Object), Value, ServerId).
+
+-spec modify(test_object() | undefined,
+             test_context(),
+             test_value(),
+             term()) -> test_object().
+modify(undefined, _Context, Value, ServerId) ->
+    %% Ignore the context since we dont have a value, its invalid if not
+    %% empty anyways, so give it a valid one
+    NewRecord = dvvset:new(timestamped_value(Value)),
+    dvvset:update(NewRecord, ServerId);
+modify(Existing, Context, Value, ServerId) ->
+    InsertRec = dvvset:new(Context, timestamped_value(Value)),
+    dvvset:update(InsertRec, Existing, ServerId).
+
+%% @doc Determines if the given context (version vector) is causually newer than
+%% an existing object. If the object missing or if the context does not represent
+%% an ancestor of the current key, false is returned. Otherwise, when the context
+%% does represent an ancestor of the existing object or the existing object itself,
+%% true is returned
+%% @private
+is_stale(RemoteContext, Obj) ->
+    LocalContext = context(Obj),
+    %% returns true (stale) when local context is causally newer or equal to remote context
+    descends(LocalContext, RemoteContext).
+
+%% @doc Reconciles a remote object received during replication or anti-entropy
+%% with a local object. If the remote object is an anscestor of or is equal to the local one
+%% `false' is returned, otherwise the reconciled object is returned as the second
+%% element of the two-tuple
+reconcile(RemoteObj, undefined) ->
+    {true, RemoteObj};
+reconcile(undefined, _LocalObj) ->
+    false;
+reconcile(RemoteObj, LocalObj) ->
+    Less  = dvvset:less(RemoteObj, LocalObj),
+    Equal = dvvset:equal(RemoteObj, LocalObj),
+    case not (Equal or Less) of
+        false -> false;
+        true ->
+            LWW = fun ({_,TS1}, {_,TS2}) -> TS1 =< TS2 end,
+            {true, dvvset:lww(LWW, dvvset:sync([LocalObj, RemoteObj]))}
+    end.
+
+%% @private
+descends(_, []) ->
+    true;
+descends(Ca, Cb) ->
+    [{NodeB, CtrB} | RestB] = Cb,
+    case lists:keyfind(NodeB, 1, Ca) of
+        false -> false;
+        {_, CtrA} ->
+            (CtrA >= CtrB) andalso descends(Ca, RestB)
+    end.
+
+%% @private
+timestamped_value(Value) ->
+    {Value, os:timestamp()}.
+
