Implementing a Distributed Process Registry on Riak Core

NYC Erlang Factory Lite 2013

Transcript

1. Implementing a Distributed Process Registry on Riak Core Christopher Meiklejohn @cmeik Saturday, September 14, 13
2. cmeiklejohn / @cmeik Saturday, September 14, 13
3. Saturday, September 14, 13
4. Saturday, September 14, 13
5. Saturday, September 14, 13
6. Saturday, September 14, 13
7. The Goal Saturday, September 14, 13
8. Build a highly-available, fault-tolerant registry. The Goal Saturday, September 14, 13
9. Understand the tradeo!s. The Goal Saturday, September 14, 13
10. Riak Stream The Goal Saturday, September 14, 13
11. The Problem Saturday, September 14, 13
12. Highly-available distributed process groups. The Problem Saturday, September 14, 13
13. Examples: pg2, gproc The Problem Saturday, September 14, 13
14. Reappearing groups; synchronous global writes. The pg2 Problem Saturday, September 14, 13
15. Election deadlocks; con"icts; dynamic clusters. The gproc Problem Saturday, September 14, 13
16. The Challenges Saturday, September 14, 13
17. Dynamic addition and removal of nodes. The 3 Challenges Saturday, September 14, 13
18. Coordination of state mutation. The 3 Challenges Saturday, September 14, 13
19. Resolution of con"icting values. The 3 Challenges Saturday, September 14, 13
20. Riak PG Saturday, September 14, 13
21. Dynamic membership through virtual nodes. Riak PG Saturday, September 14, 13
22. Replicated state; quorum reads and writes. Riak PG Saturday, September 14, 13
23. Con"ict-free resolution with CRDTs. Riak PG Saturday, September 14, 13
24. Eventually consistent; harvest vs. yield tradeo!. Riak PG Saturday, September 14, 13
25. Eventual consistency is a consistency model used in distributed computing that informally guarantees that, if no new updates are made to a given data item, eventually all accesses to that item will return the last updated value. “Eventual Consistency”, Wikipedia Saturday, September 14, 13
26. Our approaches tolerate partial failures by emphasizing simple composition mechanisms that promote fault containment, and by translating possible partial failure modes into engineering mechanisms that provide smoothly degrading functionality rather than lack of availability of the service as a whole. “Harvest, Yield, and Scalable Tolerant Systems”, Fox and Brewer Saturday, September 14, 13
27. The Requirements Saturday, September 14, 13
28. Structured names. The Requirements Saturday, September 14, 13
29. Multiple non-unique names per process. The Requirements Saturday, September 14, 13
30. Dynamic cluster membership. The Requirements Saturday, September 14, 13
31. Partition tolerance and con"ict resolution. The Requirements Saturday, September 14, 13
32. The Applications Saturday, September 14, 13
33. Service lookup pattern; publish and subscribe. The Applications Saturday, September 14, 13
34. Trade consistency for availability. The Applications Saturday, September 14, 13
35. Riak Core; CRDTs The Background Saturday, September 14, 13
36. Riak Core Saturday, September 14, 13
37. Erlang implementation of Dynamo. Riak Core Saturday, September 14, 13
38. Consistent hashing. Riak Core Saturday, September 14, 13
39. Hash-space partitioning. Riak Core Saturday, September 14, 13
40. Dynamic membership. Riak Core Saturday, September 14, 13
41. Replication factor. Riak Core Saturday, September 14, 13
42. Observed-Removed Set Saturday, September 14, 13
43. CvRDT; bounded join-semilattice. Observed-Removed Set Saturday, September 14, 13
44. Set; with merge function computing a LUB. Observed-Removed Set Saturday, September 14, 13
45. Two G-Sets; preserves monotonicity. Observed-Removed Set Saturday, September 14, 13
46. [ [{1, a}], [] ] Saturday, September 14, 13
47. [ [{1, a}], [] ] [ [{1, a}, {2, b}], [] ] Saturday, September 14, 13
48. [ [{1, a}], [] ] [ [{1, a}, {2, b}], [] ] [ [{1, a}, {2, b}], [{1, a}] ] Saturday, September 14, 13
49. [ [{1, a}], [] ] [ [{1, a}, {2, b}], [] ] [ [{1, a}, {2, b}], [{1, a}] ] [ [{1, a}, {2, b}], [{1, a}] ] Saturday, September 14, 13
50. [ [{1, a}], [] ] Saturday, September 14, 13
51. [ [{1, a}], [] ] [ [{1, a}], [] ] Saturday, September 14, 13
52. [ [{1, a}], [] ] [ [{1, a}], [] ] [ [{1, a}, {2, b}], [] ] Saturday, September 14, 13
53. [ [{1, a}], [] ] [ [{1, a}], [] ] [ [{1, a}, {2, b}], [] ] [ [{1, a}], [{1, a}] ] Saturday, September 14, 13
54. [ [{1, a}], [] ] [ [{1, a}], [] ] [ [{1, a}, {2, b}], [] ] [ [{1, a}], [{1, a}] ] [ [{1, a}, {2, b}], [{1, a}] ] Saturday, September 14, 13
55. [ [{1, a}], [] ] [ [{1, a}], [] ] Saturday, September 14, 13
56. [ [{1, a}], [] ] [ [{1, a}], [] ] [ [{1, a}], [{1, a}] ] [ [{1, a}], [{1, a}] ] Saturday, September 14, 13
57. [ [{1, a}], [] ] [ [{1, a}], [] ] [ [{1, a}], [{1, a}] ] [ [{1, a}], [{1, a}] ] [ [{1, a}, {2, a}], [{1, a}] ] Saturday, September 14, 13
58. [ [{1, a}], [] ] [ [{1, a}], [] ] [ [{1, a}], [{1, a}] ] [ [{1, a}], [{1, a}] ] [ [{1, a}, {2, a}], [{1, a}] ] [ [{1, a}, {2, a}], [{1, a}] ] Saturday, September 14, 13
59. The Implementation Saturday, September 14, 13
60. Same as pg2; create, join, leave, and members. The Implementation Saturday, September 14, 13
61. Extended with local and connected members. The Implementation Saturday, September 14, 13
62. Membership vnode stores registrations. The Implementation Saturday, September 14, 13
63. Con"ict-free resolution with OR-set. The Implementation Saturday, September 14, 13
64. Process pruning; lack of monitors. The Implementation Saturday, September 14, 13
65. Code examples. The Implementation Saturday, September 14, 13
66. The Virtual Node Saturday, September 14, 13
67. %% @doc Respond to a join request. handle_command({join, {ReqId, _}, Group, Pid}, _Sender, #state{groups=Groups0, partition=Partition}=State) -> %% Find existing list of Pids, and add object to it. Pids0 = pids(Groups0, Group, riak_dt_vvorset:new()), Pids = riak_dt_vvorset:update({add, Pid}, Partition, Pids0), %% Store back into the dict. Groups = dict:store(Group, Pids, Groups0), %% Return updated groups. {reply, {ok, ReqId}, State#state{groups=Groups}}; %% @doc Return pids from the dict. -spec pids(dict(), atom(), term()) -> term(). pids(Groups, Group, Default) -> case dict:find(Group, Groups) of {ok, Object} -> Object; _ -> Default end. riak_pg/src/riak_pg_memberships_vnode.erl Saturday, September 14, 13
68. %% @doc Respond to a leave request. handle_command({leave, {ReqId, _}, Group, Pid}, _Sender, #state{groups=Groups0, partition=Partition}=State) -> %% Find existing list of Pids, and remove object from it. Pids0 = pids(Groups0, Group, riak_dt_vvorset:new()), Pids = riak_dt_vvorset:update({remove, Pid}, Partition, Pids0), %% Store back into the dict. Groups = dict:store(Group, Pids, Groups0), %% Return updated groups. {reply, {ok, ReqId}, State#state{groups=Groups}}; %% @doc Return pids from the dict. -spec pids(dict(), atom(), term()) -> term(). pids(Groups, Group, Default) -> case dict:find(Group, Groups) of {ok, Object} -> Object; _ -> Default end. riak_pg/src/riak_pg_memberships_vnode.erl Saturday, September 14, 13
69. The Write Coordinator Saturday, September 14, 13
70. %% @doc Execute the request. execute(timeout, #state{preflist=Preflist, req_id=ReqId, coordinator=Coordinator, group=Group, pid=Pid}=State) -> riak_pg_memberships_vnode:join(Preflist, {ReqId, Coordinator}, Group, Pid), {next_state, waiting, State}. %% @doc Attempt to write to every single node responsible for this %% group. waiting({ok, ReqId}, #state{responses=Responses0, from=From}=State0) -> Responses = Responses0 + 1, State = State0#state{responses=Responses}, case Responses =:= ?W of true -> From ! {ReqId, ok}, {stop, normal, State}; false -> {next_state, waiting, State} end. riak_pg/src/riak_pg_memberships_vnode.erl Saturday, September 14, 13
71. The Read Coordinator Saturday, September 14, 13
72. %% @doc Pull a unique list of memberships from replicas, and %% relay the message to it. waiting({ok, _ReqId, IndexNode, Reply}, #state{from=From, req_id=ReqId, num_responses=NumResponses0, replies=Replies0}=State0) -> NumResponses = NumResponses0 + 1, Replies = [{IndexNode, Reply}|Replies0], State = State0#state{num_responses=NumResponses, replies=Replies}, case NumResponses =:= ?R of true -> Pids = riak_dt_vvorset:value(merge(Replies)), From ! {ReqId, ok, Pids}, case NumResponses =:= ?N of true -> {next_state, finalize, State, 0}; false -> {next_state, waiting_n, State} end; false -> {next_state, waiting, State} end. riak_pg/src/riak_pg_members_fsm.erl Saturday, September 14, 13
73. %% @doc Perform merge of replicas. merge(Replies) -> lists:foldl(fun({_, Pids}, Acc) -> riak_dt_vvorset:merge(Pids, Acc) end, riak_dt_vvorset:new(), Replies). riak_pg/src/riak_pg_members_fsm.erl Saturday, September 14, 13
74. %% @doc Wait for the remainder of responses from replicas. waiting_n({ok, _ReqId, IndexNode, Reply}, #state{num_responses=NumResponses0, replies=Replies0}=State0) -> NumResponses = NumResponses0 + 1, Replies = [{IndexNode, Reply}|Replies0], State = State0#state{num_responses=NumResponses, replies=Replies}, case NumResponses =:= ?N of true -> {next_state, finalize, State, 0}; false -> {next_state, waiting_n, State} end. riak_pg/src/riak_pg_members_fsm.erl Saturday, September 14, 13
75. %% @doc Perform read repair. finalize(timeout, #state{replies=Replies}=State) -> Merged = merge(Replies), Pruned = prune(Merged), ok = repair(Replies, State#state{pids=Pruned}), {stop, normal, State}. %% @doc If the node is connected, and the process is not alive, prune %% it. prune_pid(Pid) when is_pid(Pid) -> lists:member(node(Pid), nodes()) andalso (is_process_alive(Pid) =:= false). %% @doc Based on connected nodes, prune out processes that no longer %% exist. prune(Set) -> Pids0 = riak_dt_vvorset:value(Set), lists:foldl(fun(Pid, Pids) -> case prune_pid(Pid) of true -> riak_dt_vvorset:update({remove, Pid}, none, Pids); false -> Pids end end, Set, Pids0). riak_pg/src/riak_pg_members_fsm.erl Saturday, September 14, 13
76. The Evaluation Saturday, September 14, 13
77. pg2 members vs riak_pg connected members The Evaluation Saturday, September 14, 13
78. Partitions heal without con"icts. The Evaluation Saturday, September 14, 13
79. Howl; CloudI Process Groups; Riak Pipe The Related Work Saturday, September 14, 13
80. The Future Work Saturday, September 14, 13
81. CRDT garbage collection. The Future Work Saturday, September 14, 13
82. Active anti-entropy mechanism. The Future Work Saturday, September 14, 13
83. The Conclusion Saturday, September 14, 13
84. http://github.com/cmeiklejohn/riak_pg Thanks! Questions? Saturday, September 14, 13