Implementation of strong consistency using locks

.gitignore

@@ -32,5 +32,6 @@ test/.rebar3/erlcinfo
 .idea/
 antidote.iml
 edoc/
+_checkouts/
 #docker local builds uses a tmp dir to fetch build scripts
 docker-tmpdir*

.travis.yml

@@ -10,6 +10,7 @@ script:
   - make test
   - make reltest
   - make systests
+  - make propertytests
   - make coverage
   - rebar3 as test coveralls send
   - make dialyzer

Makefile

@@ -77,9 +77,18 @@ multidc: compile-utils rel
 	rm -f test/multidc/*.beam
 	mkdir -p logs
 ifdef SUITE
-	ct_run -pa ./_build/default/lib/*/ebin test/utils/ -logdir logs -suite test/multidc/${SUITE} -cover test/antidote.coverspec
+	ct_run -pa ./_build/default/lib/*/ebin test/utils/ -logdir logs -suite test/multidc/${SUITE} -cover test/antidote.coverspec -erl_args -hidden
 else
-	ct_run -pa ./_build/default/lib/*/ebin test/utils/ -logdir logs -dir test/multidc -cover test/antidote.coverspec
+	ct_run -pa ./_build/default/lib/*/ebin test/utils/ -logdir logs -dir test/multidc -cover test/antidote.coverspec -erl_args -hidden
+endif
+
+propertytests: compile-utils rel
+	rm -f test/propertytests/*.beam
+	mkdir -p logs
+ifdef SUITE
+	ct_run -pa ./_build/test/lib/*/ebin test/utils/ -logdir logs -suite test/propertytests/${SUITE} -cover test/antidote.coverspec -erl_args -hidden
+else
+	ct_run -pa ./_build/test/lib/*/ebin test/utils/ -logdir logs -dir test/propertytests -cover test/antidote.coverspec -erl_args -hidden
 endif
 
 systests: singledc multidc

include/antidote.hrl

@@ -160,7 +160,13 @@
 
 -define(CLOCKSI_TIMEOUT, 1000).
 
--type txn_properties() :: [{update_clock, boolean()} | {certify, use_default | certify | dont_certify}].
+-type txn_property() ::
+      {update_clock, boolean()}
+    | {certify, use_default | certify | dont_certify}
+    | {shared_locks, [binary()]}
+    | {exclusive_locks, [binary()]}
+.
+-type txn_properties() :: [txn_property()].
 
 -record(transaction, {
     snapshot_time_local :: clock_time(),

include/antidote_locks.hrl

@@ -0,0 +1,58 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright <2013-2018> <
+%%  Technische Universität Kaiserslautern, Germany
+%%  Université Pierre et Marie Curie / Sorbonne-Université, France
+%%  Universidade NOVA de Lisboa, Portugal
+%%  Université catholique de Louvain (UCL), Belgique
+%%  INESC TEC, Portugal
+%% >
+%%
+%% 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 expressed or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.
+%%
+%% List of the contributors to the development of Antidote: see AUTHORS file.
+%% Description and complete License: see LICENSE file.
+%% -------------------------------------------------------------------
+
+-record(read_crdt_state, {
+    snapshot_time :: snapshot_time(),
+    objects :: [bound_object()],
+    data :: antidote_lock_server_state:read_crdt_state_onresponse_data()
+}).
+
+-record(update_crdt_state, {
+    snapshot_time :: snapshot_time(),
+    updates :: [{bound_object(), op_name(), op_param()}],
+    data :: antidote_lock_server_state:update_crdt_state_onresponse_data()
+}).
+
+-record(send_inter_dc_message, {
+    receiver :: dcid(),
+    message :: antidote_lock_server_state:inter_dc_message()
+}).
+
+-record(accept_request, {
+    requester :: requester(),
+    clock :: snapshot_time()
+}).
+
+-record(abort_request, {
+    requester :: requester()
+}).
+
+-record(set_timeout, {
+    timeout :: antidote_lock_server_state:milliseconds(),
+    message :: any()
+}).

include/antidote_message_types.hrl

@@ -12,6 +12,7 @@
 -define(OK_MSG,4).
 -define(ERROR_MSG,5).
 -define(BCOUNTER_REQUEST,6).
+-define(LOCK_SERVER_REQUEST, 7).
 
 %% The number of bytes a parition id is in a message
 -define(PARTITION_BYTE_LENGTH, 20).
@@ -22,4 +23,4 @@
 %% Needed for dialyzer, must be the size of the request id bits plus the version bits
 -define(MESSAGE_HEADER_BIT_LENGTH, 32).
 
--type inter_dc_message_type() :: ?CHECK_UP_MSG | ?LOG_READ_MSG | ?OK_MSG | ?ERROR_MSG | ?BCOUNTER_REQUEST.
+-type inter_dc_message_type() :: ?CHECK_UP_MSG | ?LOG_READ_MSG | ?OK_MSG | ?ERROR_MSG | ?BCOUNTER_REQUEST | ?LOCK_SERVER_REQUEST.

rebar.config

@@ -86,7 +86,7 @@
         {extra_src_dirs, [{"test", [{recursive, true}]}]},
         {erl_opts, [warnings_as_errors, debug_info, no_inline_list_funcs]},
         {plugins, [{coveralls, {git, "https://github.com/markusn/coveralls-erl", {branch, "master"}}}]},
-        {deps, [meck]}]}
+        {deps, [meck, dorer]}]}
 ]}.
 
 {cover_enabled, true}.

src/antidote_list_utils.erl

@@ -0,0 +1,145 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright <2013-2018> <
+%%  Technische Universität Kaiserslautern, Germany
+%%  Université Pierre et Marie Curie / Sorbonne-Université, France
+%%  Universidade NOVA de Lisboa, Portugal
+%%  Université catholique de Louvain (UCL), Belgique
+%%  INESC TEC, Portugal
+%% >
+%%
+%% 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 expressed or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.
+%%
+%% List of the contributors to the development of Antidote: see AUTHORS file.
+%% Description and complete License: see LICENSE file.
+%% -------------------------------------------------------------------
+
+
+-module(antidote_list_utils).
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+-endif.
+
+
+-export([group_by_first/1, group_by/2, group_by/4, reduce/2, topsort/2, find_first/2, orddict_remove_keys/2, pmap/2, pmap_err/2]).
+
+
+%% groups a list of key-value pairs by key
+-spec group_by_first([{K, V}]) -> #{K => [V]}.
+group_by_first(List) ->
+    group_by(
+        fun({K, _}) -> K end,
+        fun({_, V}) -> [V] end,
+        fun({_, V}, Xs) -> [V | Xs] end,
+        List
+    ).
+
+% groups
+-spec group_by(fun((E) -> K), [E]) -> #{K => [E]}.
+group_by(F, List) ->
+    group_by(F, fun(X) -> [X] end, fun(X, Xs) -> [X | Xs] end, List).
+
+-spec group_by(fun((E) -> K), fun((E) -> V), fun((E, V) -> V), [E]) -> #{K => V}.
+group_by(F, Init, Merge, List) ->
+    lists:foldr(fun(X, M) ->
+        K = F(X),
+        maps:update_with(K, fun(L) -> Merge(X, L) end, Init(X), M)
+    end, maps:new(), List).
+
+
+-spec reduce(fun((E, E) -> E), [E]) -> E.
+reduce(_, []) -> throw('cannot reduce empty list');
+reduce(_, [X]) -> X;
+reduce(M, [X, Y | Xs]) -> reduce(M, [M(X, Y) | Xs]).
+
+
+topsort(_Cmp, []) -> [];
+topsort(Cmp, Xs) ->
+    % Min are all elements X from Xs, such that for all elements Y from Xs: not Y < X
+    {Min, NotMin} = lists:partition(
+        fun(X) ->
+            lists:all(fun(Y) -> not Cmp(Y, X) end, Xs)
+        end,
+        Xs
+    ),
+    Min ++ topsort(Cmp, NotMin).
+
+
+-spec find_first(fun((T) -> boolean()), [T]) -> error | {ok, T}.
+find_first(Pred, []) when is_function(Pred) -> error;
+find_first(Pred, [X | Xs]) ->
+    case Pred(X) of
+        true -> {ok, X};
+        false -> find_first(Pred, Xs)
+    end.
+
+orddict_remove_keys(Dict, Keys) ->
+    lists:foldl(fun(K, Acc) -> orddict:erase(K, Acc) end, Dict, Keys).
+
+%% like lists:map, but each element is computed in its own process
+-spec pmap(fun((A) -> B), [A]) -> [B].
+pmap(F, List) ->
+    Self = self(),
+    Pids = [spawn_link(fun() -> Self ! {self(), F(X)} end) || X <- List],
+    [receive {Pid, X} -> X end || Pid <- Pids].
+
+
+%% like pmap, but the function may fail
+%% Returns the first received error if any or a list of results.
+-spec pmap_err(fun((A) -> {ok, B} | {error, Reason}), [A]) -> {ok, [B]} | {error, Reason}.
+pmap_err(F, List) ->
+    Self = self(),
+    Pids = [spawn_link(fun() -> Self ! {self(), F(X)} end) || X <- List],
+    Receive = fun
+        Receive(0, Acc) -> {ok, Acc};
+        Receive(N, Acc) ->
+            receive
+                {Pid, Msg} ->
+                    case Msg of
+                        {ok, Val} ->
+                            Receive(N - 1, maps:put(Pid, Val, Acc));
+                        {error, Reason} ->
+                            {error, Reason};
+                        Other ->
+                            {error, {unhandled_message, Other}}
+                    end
+            end
+    end,
+    case Receive(length(Pids), #{}) of
+        {error, Reason} ->
+            {error, Reason};
+        {ok, Acc} ->
+            {ok, [maps:get(Pid, Acc) || Pid <- Pids]}
+    end.
+
+
+-ifdef(TEST).
+group_by_first_test() ->
+    M = group_by_first([{a, 1}, {b, 2}, {a, 3}, {a, 4}]),
+    ?assertEqual(#{a => [1, 3, 4], b => [2]}, M).
+
+
+pmap_test() ->
+    ?assertEqual([1, 4, 9, 16], pmap(fun(X) -> X * X end, [1, 2, 3, 4])).
+
+
+pmap_err1_test() ->
+    ?assertEqual({ok, [1, 4, 9, 16]}, pmap_err(fun(X) -> {ok, X * X} end, [1, 2, 3, 4])).
+
+pmap_err2_test() ->
+    ?assertEqual({error, blub}, pmap_err(fun(3) -> {error, blub}; (X) -> {ok, X * X} end, [1, 2, 3, 4])).
+
+-endif.

src/antidote_lock_crdt.erl

@@ -0,0 +1,95 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright <2013-2018> <
+%%  Technische Universität Kaiserslautern, Germany
+%%  Université Pierre et Marie Curie / Sorbonne-Université, France
+%%  Universidade NOVA de Lisboa, Portugal
+%%  Université catholique de Louvain (UCL), Belgique
+%%  INESC TEC, Portugal
+%% >
+%%
+%% 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 expressed or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.
+%%
+%% List of the contributors to the development of Antidote: see AUTHORS file.
+%% Description and complete License: see LICENSE file.
+%% -------------------------------------------------------------------
+
+%% @doc Functions to work with the CRDT that stores lock values
+%% Locks are  stored in a CRDT of type antidote_crdt_map_rr, where the keys are
+%% DC-ids and the values are MV-registers containing DC-ids.
+
+-module(antidote_lock_crdt).
+
+-include("antidote.hrl").
+
+-define(LOCK_BUCKET, '__antidote_lock_bucket').
+
+% there is one lock-part per datacenter.
+% the map stores lock-part to current owner
+-export_type([value/0]).
+
+-export([get_lock_objects/1, get_lock_object/1, parse_lock_value/1, make_lock_updates/2, parse_lock_values/1, get_lock_objects_from_spec/1]).
+
+-type value() :: #{dcid() => dcid()}.
+
+
+%% Lock CRDT, stored under Lock, antidote_crdt_map_rr, ?LOCK_BUCKET}
+%% In the map: Lock-part to current lock holder
+%%   keys: {DcId, antidote_crdt_register_mv}
+%%   values: DcId
+
+-spec get_lock_objects_from_spec(antidote_locks:lock_spec()) -> list(bound_object()).
+get_lock_objects_from_spec(Locks) ->
+    lists:map(fun({Lock, _Kind}) -> get_lock_object(Lock) end, Locks).
+
+-spec get_lock_objects([antidote_locks:lock()]) -> list(bound_object()).
+get_lock_objects(Locks) ->
+    lists:map(fun get_lock_object/1, Locks).
+
+% Dialyzer bug: does not accept this type signature
+%-spec get_lock_object(antidote_locks:lock()) -> bound_object().
+get_lock_object(Lock) when is_binary(Lock) orelse is_atom(Lock) ->
+    {Lock, antidote_crdt_map_rr, ?LOCK_BUCKET}.
+
+-spec parse_lock_value(antidote_crdt_map_rr:value()) -> value().
+parse_lock_value(RawV) ->
+    maps:from_list([{K, read_mv(V)} || {{K, _}, V} <- RawV]).
+
+-spec parse_lock_values([antidote_crdt_map_rr:value()]) -> [value()].
+parse_lock_values(RawVs) ->
+    [parse_lock_value(V) || V <- RawVs].
+
+
+read_mv([V]) -> V;
+read_mv(Vs) ->
+    % TODO this should never happen, probably a bug when restoring Antidote state from log
+    case lists:usort(Vs) of
+        [V] ->
+            logger:warning("antidote_lock_crdt contains the same value multiple times: ~p", [Vs]),
+            V;
+        [V|_] ->
+            logger:error("antidote_lock_crdt contains multiple values: ~p", [Vs]),
+            V
+    end.
+
+
+-spec make_lock_updates(antidote_locks:lock(), [{dcid(), dcid()}]) -> [{bound_object(), op_name(), op_param()}].
+make_lock_updates(_Lock, []) -> [];
+make_lock_updates(Lock, Updates) ->
+    [{get_lock_object(Lock), update,
+        [{{K, antidote_crdt_register_mv}, {assign, V}} || {K, V} <- Updates]}].
+
+
+