[WIP] Partisan support for Riak Core.

.gitignore

@@ -22,3 +22,5 @@ sim.out
 .eqc
 compile_commands.json
 rebar3.crashdump
+tags
+.rebar3/erlcinfo

rebar.config

@@ -14,7 +14,8 @@
 {deps, [
   %% Stuff we need to work with rebar3
   {goldrush, "~>0.1.8"},
-  {cuttlefish, "~>2.0.11"},
+  {cuttlefish, ".*", {git, "https://github.com/lasp-lang/cuttlefish",
+                      {branch, "master"}}},
   {clique, "~>0.3.9"},
   %% We now include folsom
   folsom,
@@ -23,14 +24,17 @@
   {poolboy, "0.8.2", {pkg, basho_poolboy}},
   {basho_stats, "~>1.0.3"},
   {riak_sysmon, "~>2.1.5"},
-  %% {riak_ensemble, ".*", {git, "https://github.com/basho/riak_ensemble", {tag, "2.1.8"}}},
-  {riak_ensemble, "~>2.4.0", {pkg, riak_ensemble_ng}},
+  {riak_ensemble, ".*", {git, "https://github.com/lasp-lang/riak_ensemble",
+                         {branch, "master"}}},
   {pbkdf2, "~>2.0.0"},
   {jam, "~>1.0.0"},
   {blume, "~>0.1.0"},
   {chash, "~>0.1.1"},
   {eleveldb, "~>2.2.20"},
-  {exometer_core, "~>1.0.0", {pkg, basho_exometer_core}}
+  {exometer_core, "~>1.0.0", {pkg, basho_exometer_core}},
+
+  {partisan, ".*", {git, "https://github.com/lasp-lang/partisan", {branch, "master"}}},
+  {riak_core_partisan_utils, ".*", {git, "https://github.com/lasp-lang/riak_core_partisan_utils", {branch, "master"}}}
 ]}.
 
 

rebar.lock

@@ -1,44 +1,70 @@
 {"1.1.0",
-[{<<"basho_stats">>,{pkg,<<"basho_stats">>,<<"1.0.3">>},0},
+[{<<"acceptor_pool">>,{pkg,<<"acceptor_pool">>,<<"1.0.0-rc.0">>},1},
+ {<<"basho_stats">>,{pkg,<<"basho_stats">>,<<"1.0.3">>},0},
  {<<"bear">>,{pkg,<<"bear">>,<<"0.8.5">>},1},
  {<<"blume">>,{pkg,<<"blume">>,<<"0.1.0">>},0},
  {<<"chash">>,{pkg,<<"chash">>,<<"0.1.2">>},0},
  {<<"clique">>,{pkg,<<"clique">>,<<"0.3.9">>},0},
- {<<"cuttlefish">>,{pkg,<<"cuttlefish">>,<<"2.0.11">>},0},
+ {<<"cuttlefish">>,
+  {git,"https://github.com/lasp-lang/cuttlefish",
+       {ref,"42d0aa275deeb8ae6640fa65053ec626f9cbff64"}},
+  0},
  {<<"edown">>,{pkg,<<"edown">>,<<"0.8.1">>},2},
  {<<"eleveldb">>,{pkg,<<"eleveldb">>,<<"2.2.20">>},0},
  {<<"exometer_core">>,{pkg,<<"basho_exometer_core">>,<<"1.0.2">>},0},
  {<<"folsom">>,{pkg,<<"folsom">>,<<"0.8.5">>},0},
- {<<"getopt">>,{pkg,<<"getopt">>,<<"0.8.2">>},1},
+ {<<"getopt">>,
+  {git,"git://github.com/jcomellas/getopt.git",
+       {ref,"659a28f4145bc9843598972854299dc4ea77e4cb"}},
+  1},
  {<<"goldrush">>,{pkg,<<"goldrush">>,<<"0.1.9">>},0},
  {<<"jam">>,{pkg,<<"jam">>,<<"1.0.0">>},0},
  {<<"lager">>,{pkg,<<"lager">>,<<"3.2.4">>},0},
+ {<<"neotoma">>,
+  {git,"git://github.com/seancribbs/neotoma.git",
+       {ref,"760928ec8870da02eb11bccb501e2700925d06c6"}},
+  1},
  {<<"parse_trans">>,{pkg,<<"parse_trans">>,<<"2.9.0">>},1},
+ {<<"partisan">>,
+  {git,"https://github.com/lasp-lang/partisan",
+       {ref,"a053dd793a52a0d3e6cd450a409d99ed19f9a00a"}},
+  0},
  {<<"pbkdf2">>,{pkg,<<"pbkdf2">>,<<"2.0.0">>},0},
  {<<"poolboy">>,{pkg,<<"basho_poolboy">>,<<"0.8.2">>},0},
- {<<"riak_ensemble">>,{pkg,<<"riak_ensemble_ng">>,<<"2.4.0">>},0},
+ {<<"rand_compat">>,{pkg,<<"rand_compat">>,<<"0.0.3">>},1},
+ {<<"riak_core_partisan_utils">>,
+  {git,"https://github.com/lasp-lang/riak_core_partisan_utils",
+       {ref,"2f93659f295124fe901576d8cd72ea028b4e87b1"}},
+  0},
+ {<<"riak_ensemble">>,
+  {git,"https://github.com/lasp-lang/riak_ensemble",
+       {ref,"1ebdae3c26f7a33a7223b07dd822e83ca8c733f1"}},
+  0},
  {<<"riak_sysmon">>,{pkg,<<"riak_sysmon">>,<<"2.1.5">>},0},
- {<<"setup">>,{pkg,<<"setup">>,<<"1.7.0">>},1}]}.
+ {<<"setup">>,{pkg,<<"setup">>,<<"1.7.0">>},1},
+ {<<"time_compat">>,{pkg,<<"time_compat">>,<<"0.0.1">>},1},
+ {<<"types">>,{pkg,<<"types">>,<<"0.1.6">>},1}]}.
 [
 {pkg_hash,[
+ {<<"acceptor_pool">>, <<"679D741DF87FC13599B1AEF2DF8F78F1F880449A6BEFAB7C44FB6FAE0E92A2DE">>},
  {<<"basho_stats">>, <<"7E1174151509C64FCC1934120ED32295E14F84DAAE7F84926BA2C8D3700D146C">>},
  {<<"bear">>, <<"E95FCA1627CD9E15BAF93CE0A52AFF16917BAF325F0EE65B88CD715376CD2344">>},
  {<<"blume">>, <<"62DDA46B43E24D4B8D5407B8F6DDB78EBD2A8CDC84D9BC822CDDAEF6F402F3EB">>},
  {<<"chash">>, <<"AF02484F2640C653C4B9A8557A14CA0704989DBEDB27E7CCBC442F1903A3BCA7">>},
  {<<"clique">>, <<"A56298B7582C1AC2C278F21D3C779CE3ECCE02AE06751DAA107441ED90DC5E69">>},
- {<<"cuttlefish">>, <<"D266AC68BA801803ACE4EF3DEC0FC419F184092E1882E19F7AFC29A6D4334E6A">>},
  {<<"edown">>, <<"7333B6F6B7BBC736C263E9CEB8261667D7C8F7D92E251829F1A9FE8F24E7B694">>},
  {<<"eleveldb">>, <<"1FFF63A5055BBF4BF821F797EF76065882B193F5E8095F95FCD9287187773B58">>},
  {<<"exometer_core">>, <<"DC4BC9B0B47EDEE0C053B16AC5B9C692E662C7E1513FFC96F4D928BC3328D65F">>},
  {<<"folsom">>, <<"94A027B56FE84FEED264F9B33CB4C6AC9A801FAD84B87DBDA0836CE83C3B8D69">>},
- {<<"getopt">>, <<"B17556DB683000BA50370B16C0619DF1337E7AF7ECBF7D64FBF8D1D6BCE3109B">>},
  {<<"goldrush">>, <<"F06E5D5F1277DA5C413E84D5A2924174182FB108DABB39D5EC548B27424CD106">>},
  {<<"jam">>, <<"ED9B180F2F3A775E6A47AC490954976802F0638C19A393F3E86D4BA4CF890582">>},
  {<<"lager">>, <<"A6DEB74DAE7927F46BD13255268308EF03EB206EC784A94EAF7C1C0F3B811615">>},
  {<<"parse_trans">>, <<"3F5F7B402928FB9FD200C891E635DE909045D1EFAC40CE3F924D3892898F85EB">>},
  {<<"pbkdf2">>, <<"11C23279FDED5C0027AB3996CFAE77805521D7EF4BABDE2BD7EC04A9086CF499">>},
  {<<"poolboy">>, <<"AD5524BF4B8D30EEE0E5386847B74DAD739BB6AE1893F3AF10F54D427ACC96E4">>},
- {<<"riak_ensemble">>, <<"6570463BF09F1200C003DE298C5D6615C418D3BF0A2DF5CAF469A9DF2558059A">>},
+ {<<"rand_compat">>, <<"011646BC1F0B0C432FE101B816F25B9BBB74A085713CEE1DAFD2D62E9415EAD3">>},
  {<<"riak_sysmon">>, <<"2331BBCAEBE73EA7B2449F4C765827142476CC73D7D86999B4209EE005694D6B">>},
- {<<"setup">>, <<"15DF8E57C6DF9755E22BFB1AEF0C640BD97E9889396FDFB2A85A5536A9043674">>}]}
+ {<<"setup">>, <<"15DF8E57C6DF9755E22BFB1AEF0C640BD97E9889396FDFB2A85A5536A9043674">>},
+ {<<"time_compat">>, <<"23FE0AD1FDF3B5B88821B2D04B4B5E865BF587AE66056D671FE0F53514ED8139">>},
+ {<<"types">>, <<"03BB7140016C896D3441A77CB0B7D6ACAA583D6D6E9C4A3E1FD3C25123710290">>}]}
 ].

src/riak_core.app.src

@@ -6,7 +6,7 @@
      {included_applications,[riak_ensemble,chash]},
      {applications,
          [kernel,stdlib,lager,sasl,crypto,riak_sysmon,os_mon,basho_stats,
-          eleveldb,pbkdf2,poolboy,exometer_core,clique,folsom]},
+          eleveldb,pbkdf2,poolboy,exometer_core,clique,folsom,partisan]},
      {mod,{riak_core_app,[]}},
      {env,
          [{cluster_name,"default"},

src/riak_core.erl

@@ -88,6 +88,9 @@ get_other_ring(Node) ->
 standard_join(Node, Rejoin, Auto) when is_atom(Node) ->
     case net_adm:ping(Node) of
         pong ->
+            %% Initiate the partisan connections.
+            riak_core_partisan_utils:join(Node),
+
             case get_other_ring(Node) of
                 {ok, Ring} ->
                     standard_join(Node, Ring, Rejoin, Auto);
@@ -115,7 +118,8 @@ standard_join(Node, Ring, Rejoin, Auto) ->
 
     SameSize = (riak_core_ring:num_partitions(MyRing) =:=
                 riak_core_ring:num_partitions(Ring)),
-    Singleton = ([node()] =:= riak_core_ring:all_members(MyRing)),
+    RemoteMembers = riak_core_ring:all_members(MyRing),
+    Singleton = ([node()] =:= RemoteMembers),
     case {InitComplete, Rejoin or Singleton, SameSize} of
         {false, _, _} ->
             {error, node_still_starting};
@@ -124,6 +128,9 @@ standard_join(Node, Ring, Rejoin, Auto) ->
         {_, _, false} ->
             {error, different_ring_sizes};
         _ ->
+            %% Connect all incoming members via partisan.
+            riak_core_partisan_utils:join(RemoteMembers),
+
             GossipVsn = riak_core_gossip:gossip_version(),
             Ring2 = riak_core_ring:add_member(node(), Ring,
                                               node()),
@@ -205,6 +212,10 @@ leave() ->
     end.
 
 standard_leave(Node) ->
+    %% Force leave from partisan.
+    riak_core_partisan_utils:leave(Node),
+
+    %% Perform ring transition.
     riak_core_ring_manager:ring_trans(
       fun(Ring2, _) ->
               Ring3 = riak_core_ring:leave_member(Node, Ring2, Node),

src/riak_core_apl.erl

@@ -476,12 +476,15 @@ chbin_test_() ->
     {timeout, 180, fun chbin_test_scenario/0}.
 
 chbin_test_scenario() ->
+    meck:new(riak_core_partisan_utils, [unstick]),
+
     [chbin_test_scenario(Size, NumNodes)
      || Size     <- [32, 64, 128],
         NumNodes <- [1, 2, 3, 4, 5, 8, Size div 4]],
     ok.
 
 chbin_test_scenario(Size, NumNodes) ->
+    meck:expect(riak_core_partisan_utils, update, fun(_) -> ok end),
     RingTop = 1 bsl 160,
     Ring = riak_core_test_util:fake_ring(Size, NumNodes),
     Nodes = riak_core_ring:all_members(Ring),

src/riak_core_broadcast.erl

@@ -568,8 +568,7 @@ all_peers(Root, Sets, Default) ->
 send(Msg, Peers) when is_list(Peers) ->
     [send(Msg, P) || P <- Peers];
 send(Msg, P) ->
-    %% TODO: add debug logging
-    gen_server:cast({?SERVER, P}, Msg).
+    riak_core_partisan_utils:cast(broadcast, {?SERVER, P}, Msg).
 
 schedule_lazy_tick() ->
     schedule_tick(lazy_tick, broadcast_lazy_timer, 1000).

src/riak_core_console.erl

@@ -60,6 +60,9 @@ print_member_status(Ring) ->
     io:format("~33..=s Membership ~34..=s~n", ["", ""]),
     io:format("Status     Ring    Pending    Node~n"),
     io:format("~79..-s~n", [""]),
+    lager:info("~33..=s Membership ~34..=s~n", ["", ""]),
+    lager:info("Status     Ring    Pending    Node~n"),
+    lager:info("~79..-s~n", [""]),
     AllStatus = lists:keysort(2, riak_core_ring:all_member_status(Ring)),
     IsPending = ([] /= riak_core_ring:pending_changes(Ring)),
 
@@ -71,10 +74,15 @@ print_member_status(Ring) ->
 
                             case IsPending of
                                 true ->
+                                    lager:info("~-8s  ~5.1f%    ~5.1f%    ~p~n",
+                                               [Status, RingPercent,
+                                                NextPercent, Node]),
                                     io:format("~-8s  ~5.1f%    ~5.1f%    ~p~n",
                                               [Status, RingPercent,
                                                NextPercent, Node]);
                                 false ->
+                                    lager:info("~-8s  ~5.1f%      --      ~p~n",
+                                               [Status, RingPercent, Node]),
                                     io:format("~-8s  ~5.1f%      --      ~p~n",
                                               [Status, RingPercent, Node])
                             end,
@@ -91,6 +99,9 @@ print_member_status(Ring) ->
                                     {Joining0, Valid0, Down0, Leaving0, Exiting0 + 1}
                             end
                     end, {0,0,0,0,0}, AllStatus),
+    lager:info("~79..-s~n", [""]),
+    lager:info("Valid:~b / Leaving:~b / Exiting:~b / Joining:~b / Down:~b~n",
+               [Valid, Leaving, Exiting, Joining, Down]),
     io:format("~79..-s~n", [""]),
     io:format("Valid:~b / Leaving:~b / Exiting:~b / Joining:~b / Down:~b~n",
               [Valid, Leaving, Exiting, Joining, Down]),

src/riak_core_gossip.erl

@@ -197,7 +197,12 @@ handle_cast({send_ring_to, Node}, State) ->
     riak_core_ring:check_tainted(RingOut,
                                  "Error: riak_core_gossip/send_ring_to :: "
                                  "Sending tainted ring over gossip"),
-    gen_server:cast({?MODULE, Node}, {reconcile_ring, RingOut}),
+
+    %% This will force partisan connections to initialize, bootstrapping
+    %% off of disterl -- this connection may be pruned by partisan but
+    %% once the member is valid in the ring -- it will persist.
+    %%
+    riak_core_partisan_utils:cast(gossip, {?MODULE, Node}, {reconcile_ring, RingOut}),
     Tokens = State#state.gossip_tokens - 1,
     {noreply, State#state{gossip_tokens=Tokens}};
 
@@ -309,6 +314,7 @@ reconcile(Ring0, [OtherRing0]) ->
         {_, _, new_ring} ->
             Ring3 = riak_core_ring:ring_changed(Node, Ring2),
             riak_core_stat:update(rings_reconciled),
+            riak_core_ring:notify_external_membership(Ring3),
             log_membership_changes(Ring, Ring3),
             {reconciled_ring, Ring3};
         {_, _, _} ->

src/riak_core_partisan_proxy_service.erl

@@ -0,0 +1,100 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright (c) 2017 Christopher S. Meiklejohn.  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(riak_core_partisan_proxy_service).
+-author("Christopher S. Meiklejohn <[email protected]>").
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0,
+         start_link/1]).
+
+%% gen_server callbacks
+-export([init/1,
+         handle_call/3,
+         handle_cast/2,
+         handle_info/2,
+         terminate/2,
+         code_change/3]).
+
+%% State record.
+-record(state, {}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+%% @doc Same as start_link([]).
+-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
+start_link() ->
+    start_link([]).
+
+%% @doc Start and link to calling process.
+-spec start_link(list())-> {ok, pid()} | ignore | {error, term()}.
+start_link(Opts) ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, Opts, []).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+%% @private
+-spec init([]) -> {ok, #state{}}.
+init([]) ->
+    {ok, #state{}}.
+
+%% @private
+-spec handle_call(term(), {pid(), term()}, #state{}) ->
+    {reply, term(), #state{}}.
+
+%% @private
+handle_call(Msg, _From, State) ->
+    _ = lager:warning("Unhandled messages: ~p", [Msg]),
+    {reply, ok, State}.
+
+-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
+%% @private
+handle_cast(Msg, State) ->
+    _ = lager:warning("Unhandled messages: ~p", [Msg]),
+    {noreply, State}.
+
+%% @private
+handle_info({forward, Pid, Message}, State) ->
+    erlang:send(Pid, Message),
+    {noreply, State};
+handle_info(Msg, State) ->
+    _ = lager:warning("Unhandled messages: ~p", [Msg]),
+    {noreply, State}.
+
+%% @private
+-spec terminate(term(), #state{}) -> term().
+terminate(_Reason, _State) ->
+    ok.
+
+%% @private
+-spec code_change(term() | {down, term()}, #state{}, term()) ->
+    {ok, #state{}}.
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================