Current section

34 Versions

Jump to

Compare versions

3 files changed
+100 additions
-132 deletions
  @@ -4,13 +4,14 @@
4 4 <<"A fast, multi-master, distributed global process registry, with automatic distribution of worker processes.">>}.
5 5 {<<"elixir">>,<<"~> 1.3">>}.
6 6 {<<"files">>,
7 - [<<"lib/swarm.ex">>,<<"lib/swarm/app.ex">>,
8 - <<"lib/swarm/distribution/ring.ex">>,
7 + [<<"lib">>,<<"lib/swarm">>,<<"lib/swarm.ex">>,<<"lib/swarm/app.ex">>,
8 + <<"lib/swarm/distribution">>,<<"lib/swarm/distribution/ring.ex">>,
9 9 <<"lib/swarm/distribution/static_quorum_ring.ex">>,
10 10 <<"lib/swarm/distribution/strategy.ex">>,<<"lib/swarm/logger.ex">>,
11 - <<"lib/swarm/registry.ex">>,<<"lib/swarm/tracker/crdt.ex">>,
12 - <<"lib/swarm/tracker/entry.ex">>,<<"lib/swarm/tracker/tracker.ex">>,
13 - <<"src/swarm.erl">>,<<"mix.exs">>,<<"README.md">>,<<"LICENSE.md">>]}.
11 + <<"lib/swarm/registry.ex">>,<<"lib/swarm/tracker">>,
12 + <<"lib/swarm/tracker/crdt.ex">>,<<"lib/swarm/tracker/entry.ex">>,
13 + <<"lib/swarm/tracker/tracker.ex">>,<<"src">>,<<"src/swarm.erl">>,
14 + <<"mix.exs">>,<<"README.md">>,<<"LICENSE.md">>]}.
14 15 {<<"licenses">>,[<<"MIT">>]}.
15 16 {<<"links">>,[{<<"Github">>,<<"https://github.com/bitwalker/swarm">>}]}.
16 17 {<<"maintainers">>,[<<"Paul Schoenfelder">>]}.
  @@ -26,4 +27,4 @@
26 27 {<<"optional">>,false},
27 28 {<<"repository">>,<<"hexpm">>},
28 29 {<<"requirement">>,<<"~> 1.0">>}]]}.
29 - {<<"version">>,<<"3.2.0">>}.
30 + {<<"version">>,<<"3.2.1">>}.
  @@ -183,10 +183,10 @@ defmodule Swarm.Tracker do
183 183 info "selected sync node: #{sync_node}"
184 184 # Send sync request
185 185 ref = Process.monitor({__MODULE__, sync_node})
186 - GenStateMachine.cast({__MODULE__, sync_node}, {:sync, self()})
186 + GenStateMachine.cast({__MODULE__, sync_node}, {:sync, self(), state.clock})
187 187 {:next_state, :syncing, %{state | sync_node: sync_node, sync_ref: ref}}
188 188 end
189 - def cluster_wait(:cast, {:sync, from}, %TrackerState{nodes: [from_node]} = state) when node(from) == from_node do
189 + def cluster_wait(:cast, {:sync, from, _rclock}, %TrackerState{nodes: [from_node]} = state) when node(from) == from_node do
190 190 info "joining cluster.."
191 191 sync_node = node(from)
192 192 info "syncing with #{sync_node}"
  @@ -195,7 +195,7 @@ defmodule Swarm.Tracker do
195 195 GenStateMachine.cast(from, {:sync_recv, self(), clock, Registry.snapshot()})
196 196 {:next_state, :awaiting_sync_ack, %{state | clock: clock, sync_node: sync_node, sync_ref: ref}}
197 197 end
198 - def cluster_wait(:cast, {:sync, from}, %TrackerState{} = state) do
198 + def cluster_wait(:cast, {:sync, from, _rclock}, %TrackerState{} = state) do
199 199 info "pending sync request from #{node(from)}"
200 200 {:keep_state, %{state | pending_sync_reqs: [from|state.pending_sync_reqs]}}
201 201 end
  @@ -229,7 +229,7 @@ defmodule Swarm.Tracker do
229 229 info "selected sync node: #{new_sync_node}"
230 230 # Send sync request
231 231 ref = Process.monitor({__MODULE__, new_sync_node})
232 - GenStateMachine.cast({__MODULE__, new_sync_node}, {:sync, self()})
232 + GenStateMachine.cast({__MODULE__, new_sync_node}, {:sync, self(), state.clock})
233 233 new_state = %{state | sync_node: new_sync_node, sync_ref: ref}
234 234 {:keep_state, new_state}
235 235 end
  @@ -260,7 +260,7 @@ defmodule Swarm.Tracker do
260 260 info "selected sync node: #{new_sync_node}"
261 261 # Send sync request
262 262 ref = Process.monitor({__MODULE__, new_sync_node})
263 - GenStateMachine.cast({__MODULE__, new_sync_node}, {:sync, self()})
263 + GenStateMachine.cast({__MODULE__, new_sync_node}, {:sync, self(), state.clock})
264 264 new_state = %{state | nodes: new_nodes,
265 265 strategy: Strategy.remove_node(strategy, node),
266 266 sync_node: new_sync_node,
  @@ -275,91 +275,114 @@ defmodule Swarm.Tracker do
275 275 end
276 276 {:keep_state, new_state}
277 277 end
278 - def syncing(:cast, {:sync_begin_tiebreaker, from, rdie}, %TrackerState{sync_node: sync_node} = state) when node(from) == sync_node do
279 - # Break the tie and continue
280 - ldie = :rand.uniform(20)
281 - info "there is a tie between syncing nodes, breaking with die roll (#{ldie}).."
282 - GenStateMachine.cast(from, {:sync_end_tiebreaker, self(), rdie, ldie})
283 - cond do
284 - ldie > rdie or (rdie == ldie and Node.self > node(from)) ->
285 - info "we won the die roll (#{ldie} vs #{rdie}), sending registry.."
286 - # This is the new seed node
287 - {clock, rclock} = Clock.fork(state.clock || Clock.seed())
288 - GenStateMachine.cast(from, {:sync_recv, self(), rclock, Registry.snapshot()})
289 - {:next_state, :awaiting_sync_ack, %{state | clock: clock}}
290 - :else ->
291 - info "#{node(from)} won the die roll (#{rdie} vs #{ldie}), "
292 - # the other node wins the roll
293 - :keep_state_and_data
294 - end
295 - end
296 - def syncing(:cast, {:sync_end_tiebreaker, from, ldie, rdie}, %TrackerState{sync_node: sync_node} = state) when node(from) == sync_node do
297 - cond do
298 - rdie > ldie or (rdie == ldie and node(from) > Node.self) ->
299 - info "#{node(from)} won the die roll (#{rdie} vs #{ldie}), waiting for payload.."
300 - # The other node won the die roll, either by a greater die roll, or the absolute
301 - # tie breaker of node ordering
302 - :keep_state_and_data
303 - :else ->
304 - info "we won the die roll (#{ldie} vs #{rdie}), sending payload.."
305 - # This is the new seed node
306 - {clock, rclock} = Clock.fork(state.clock || Clock.seed())
307 - GenStateMachine.cast(from, {:sync_recv, self(), rclock, Registry.snapshot()})
308 - {:next_state, :awaiting_sync_ack, %{state | clock: clock}}
309 - end
310 - end
311 278 # Successful first sync
312 279 def syncing(:cast, {:sync_recv, from, clock, registry}, %TrackerState{clock: nil, sync_node: sync_node} = state)
313 280 when node(from) == sync_node do
314 281 info "received registry from #{node(from)}, loading.."
315 - # let remote node know we've got the registry
316 - GenStateMachine.cast(from, {:sync_ack, Node.self})
317 282 # load target registry
318 - for entry(name: name, pid: pid, meta: meta, clock: clock) <- registry do
283 + for entry(name: name, pid: pid, meta: meta, clock: entry_clock) <- registry do
319 284 case Registry.get_by_name(name) do
320 285 :undefined ->
321 286 ref = Process.monitor(pid)
322 - Registry.new!(entry(name: name, pid: pid, ref: ref, meta: meta, clock: clock))
287 + Registry.new!(entry(name: name, pid: pid, ref: ref, meta: meta, clock: entry_clock))
323 288 entry(pid: ^pid, meta: ^meta) ->
324 289 :ok
325 290 entry(pid: ^pid, meta: old_meta) ->
326 291 # Merge meta
327 292 new_meta = Map.merge(old_meta, meta)
328 - Registry.update(name, meta: new_meta)
293 + Registry.update(name, meta: new_meta, clock: Clock.event(entry_clock))
329 294 entry(pid: _other_pid) ->
330 295 # the local registration already exists, but for a differnt pid
331 - warn "conflicting registration for #{inspect name}!"
296 + warn "conflicting registration for #{inspect name}, it will not be replicated!"
332 297 end
333 298 end
334 299 # update our local clock to match remote clock
335 300 state = %{state | clock: clock}
301 + # let remote node know we've got the registry
302 + GenStateMachine.cast(from, {:sync_ack, self(), clock, Registry.snapshot()})
336 303 info "finished sync, acknowledgement sent to #{node(from)}"
337 304 # clear any pending sync requests to prevent blocking
338 305 resolve_pending_sync_requests(state)
339 306 end
340 307 # Successful anti-entropy sync
341 - def syncing(:cast, {:sync_recv, from, _clock, registry}, %TrackerState{sync_node: sync_node} = state) when node(from) == sync_node do
342 - info "received registry from #{sync_node}, merging.."
343 - # let remote node know we've got the registry
344 - GenStateMachine.cast(from, {:sync_ack, Node.self})
308 + def syncing(:cast, {:sync_recv, from, _sync_clock, registry}, %TrackerState{sync_node: sync_node} = state)
309 + when node(from) == sync_node do
310 + info "received registry from #{sync_node}, merging.."
311 + new_state = sync_registry(from, registry, state)
312 + # let remote node know we've got the registry
313 + GenStateMachine.cast(from, {:sync_ack, self(), new_state.clock, Registry.snapshot()})
314 + info "local synchronization with #{sync_node} complete!"
315 + resolve_pending_sync_requests(new_state)
316 + end
317 + def syncing(:cast, {:sync_err, from}, %TrackerState{nodes: nodes, sync_node: sync_node} = state) when node(from) == sync_node do
318 + Process.demonitor(state.sync_ref, [:flush])
319 + cond do
320 + # Something weird happened during sync, so try a different node,
321 + # with this implementation, we *could* end up selecting the same node
322 + # again, but that's fine as this is effectively a retry
323 + length(nodes) > 0 ->
324 + warn "a problem occurred during sync, choosing a new node to sync with"
325 + # we need to choose a different node to sync with and try again
326 + new_sync_node = Enum.random(nodes)
327 + ref = Process.monitor({__MODULE__, new_sync_node})
328 + GenStateMachine.cast({__MODULE__, new_sync_node}, {:sync, self(), state.clock})
329 + {:keep_state, %{state | sync_node: new_sync_node, sync_ref: ref}}
330 + # Something went wrong during sync, but there are no other nodes to sync with,
331 + # not even the original sync node (which probably implies it shutdown or crashed),
332 + # so we're the sync node now
333 + :else ->
334 + warn "a problem occurred during sync, but no other available sync targets, becoming seed node"
335 + {:next_state, :tracking, %{state | pending_sync_reqs: [], sync_node: nil, sync_ref: nil, clock: Clock.seed()}}
336 + end
337 + end
338 + def syncing(:cast, {:sync, from, rclock}, %TrackerState{sync_node: sync_node} = state) when node(from) == sync_node do
339 + # We're trying to sync with another node while it is trying to sync with us, deterministically
340 + # choose the node which will coordinate the synchronization.
341 + local_node = Node.self
342 + case Clock.compare(state.clock, rclock) do
343 + :lt ->
344 + # The local clock is dominated by the remote clock, so the remote node will begin the sync
345 + info "syncing from #{sync_node} based on tracker clock"
346 + :keep_state_and_data
347 + :gt ->
348 + # The local clock dominates the remote clock, so the local node will begin the sync
349 + info "syncing to #{sync_node} based on tracker clock"
350 + {lclock, rclock} = Clock.fork(state.clock)
351 + GenStateMachine.cast(from, {:sync_recv, self(), rclock, Registry.snapshot()})
352 + {:next_state, :awaiting_sync_ack, %{state | clock: lclock}}
353 + result when result in [:eq, :concurrent] and sync_node > local_node ->
354 + # The remote node will begin the sync
355 + info "syncing from #{sync_node} based on node precedence"
356 + :keep_state_and_data
357 + result when result in [:eq, :concurrent] ->
358 + # The local node begins the sync
359 + info "syncing to #{sync_node} based on node precedence"
360 + {lclock, rclock} = Clock.fork(state.clock)
361 + GenStateMachine.cast(from, {:sync_recv, self(), rclock, Registry.snapshot()})
362 + {:next_state, :awaiting_sync_ack, %{state | clock: lclock}}
363 + end
364 + end
365 + def syncing(:cast, {:sync, from, _rclock}, %TrackerState{} = state) do
366 + info "pending sync request from #{node(from)}"
367 + new_pending_reqs = Enum.uniq([from|state.pending_sync_reqs])
368 + {:keep_state, %{state | pending_sync_reqs: new_pending_reqs}}
369 + end
370 + def syncing(_event_type, _event_data, _state) do
371 + {:keep_state_and_data, :postpone}
372 + end
373 +
374 + defp sync_registry(from, registry, %TrackerState{} = state) when is_pid(from) do
375 + sync_node = node(from)
345 376 # map over the registry and check that all local entries are correct
346 - new_state = Enum.reduce(registry, state, fn
377 + Enum.reduce(registry, state, fn
347 378 entry(name: rname, pid: rpid, meta: rmeta, clock: rclock) = rreg, %TrackerState{clock: clock} = state ->
348 379 case Registry.get_by_name(rname) do
349 380 :undefined ->
350 381 # missing local registration
351 - cond do
352 - Clock.leq(rclock, clock) ->
353 - # our clock is ahead, so we'll do nothing
354 - debug "local is missing #{inspect rname}, but local clock is dominant, ignoring"
355 - state
356 - :else ->
357 - # our clock is behind, so add the registration
358 - debug "local is missing #{inspect rname}, adding.."
359 - ref = Process.monitor(rpid)
360 - Registry.new!(entry(name: rname, pid: rpid, ref: ref, meta: rmeta, clock: rclock))
361 - %{state | clock: Clock.event(clock)}
362 - end
382 + debug "local tracker is missing #{inspect rname}, adding to registry"
383 + ref = Process.monitor(rpid)
384 + Registry.new!(entry(name: rname, pid: rpid, ref: ref, meta: rmeta, clock: rclock))
385 + %{state | clock: Clock.event(clock)}
363 386 entry(pid: ^rpid, meta: ^rmeta, clock: ^rclock) ->
364 387 # this entry matches, nothing to do
365 388 state
  @@ -426,52 +449,6 @@ defmodule Swarm.Tracker do
426 449 end
427 450 end
428 451 end)
429 - info "synchronization complete!"
430 - resolve_pending_sync_requests(new_state)
431 - end
432 - def syncing(:cast, {:sync_recv, from, _clock, _registry}, _state) do
433 - # somebody is sending us a thing which expects an ack, but we no longer care about it
434 - # we should reply even though we're dropping this message
435 - GenStateMachine.cast(from, {:sync_ack, Node.self})
436 - :keep_state_and_data
437 - end
438 - def syncing(:cast, {:sync_err, from}, %TrackerState{nodes: nodes, sync_node: sync_node} = state) when node(from) == sync_node do
439 - Process.demonitor(state.sync_ref, [:flush])
440 - cond do
441 - # Something weird happened during sync, so try a different node,
442 - # with this implementation, we *could* end up selecting the same node
443 - # again, but that's fine as this is effectively a retry
444 - length(nodes) > 0 ->
445 - warn "a problem occurred during sync, choosing a new node to sync with"
446 - # we need to choose a different node to sync with and try again
447 - new_sync_node = Enum.random(nodes)
448 - ref = Process.monitor({__MODULE__, new_sync_node})
449 - GenStateMachine.cast({__MODULE__, new_sync_node}, {:sync, self()})
450 - {:keep_state, %{state | sync_node: new_sync_node, sync_ref: ref}}
451 - # Something went wrong during sync, but there are no other nodes to sync with,
452 - # not even the original sync node (which probably implies it shutdown or crashed),
453 - # so we're the sync node now
454 - :else ->
455 - warn "a problem occurred during sync, but no other available sync targets, becoming seed node"
456 - {:next_state, :tracking, %{state | pending_sync_reqs: [], sync_node: nil, sync_ref: nil, clock: Clock.seed()}}
457 - end
458 - end
459 - def syncing(:cast, {:sync, from}, %TrackerState{sync_node: sync_node}) when node(from) == sync_node do
460 - # Incoming sync request from our target node, in other words, A is trying to sync with B,
461 - # and B is trying to sync with A - we need to break the deadlock
462 - # 20d, I mean why not?
463 - die = :rand.uniform(20)
464 - info "there is a tie between syncing nodes, breaking with die roll (#{die}).."
465 - GenStateMachine.cast(from, {:sync_begin_tiebreaker, self(), die})
466 - :keep_state_and_data
467 - end
468 - def syncing(:cast, {:sync, from}, %TrackerState{} = state) do
469 - info "pending sync request from #{node(from)}"
470 - new_pending_reqs = Enum.uniq([from|state.pending_sync_reqs])
471 - {:keep_state, %{state | pending_sync_reqs: new_pending_reqs}}
472 - end
473 - def syncing(_event_type, _event_data, _state) do
474 - {:keep_state_and_data, :postpone}
475 452 end
476 453
477 454 defp resolve_pending_sync_requests(%TrackerState{pending_sync_reqs: []} = state) do
  @@ -505,10 +482,12 @@ defmodule Swarm.Tracker do
505 482 end
506 483 end
507 484
508 - def awaiting_sync_ack(:cast, {:sync_ack, sync_node}, %TrackerState{sync_node: sync_node} = state) do
509 - info "received sync acknowledgement from #{sync_node}"
510 - Process.demonitor(state.sync_ref, [:flush])
511 - resolve_pending_sync_requests(%{state | sync_node: nil, sync_ref: nil})
485 + def awaiting_sync_ack(:cast, {:sync_ack, from, _sync_clock, registry}, %TrackerState{sync_node: sync_node} = state)
486 + when sync_node == node(from) do
487 + info "received sync acknowledgement from #{node(from)}, syncing with remote registry"
488 + new_state = sync_registry(from, registry, state)
489 + info "local synchronization with #{node(from)} complete!"
490 + resolve_pending_sync_requests(new_state)
512 491 end
513 492 def awaiting_sync_ack(:info, {:DOWN, ref, _type, _pid, _info}, %TrackerState{sync_ref: ref} = state) do
514 493 warn "wait for acknowledgement from #{state.sync_node} cancelled, tracker down"
  @@ -536,12 +515,6 @@ defmodule Swarm.Tracker do
536 515 end
537 516 {:keep_state, new_state}
538 517 end
539 - def awaiting_sync_ack(:cast, {:sync_recv, from, _clock, _registry}, _state) do
540 - # somebody is sending us a thing which expects an ack,
541 - # we should reply even though we're dropping this message
542 - GenStateMachine.cast(from, {:sync_ack, Node.self})
543 - :keep_state_and_data
544 - end
545 518 def awaiting_sync_ack(_event_type, _event_data, _state) do
546 519 {:keep_state_and_data, :postpone}
547 520 end
  @@ -619,7 +592,7 @@ defmodule Swarm.Tracker do
619 592 sync_node = Enum.random(nodes)
620 593 info "syncing with #{sync_node}"
621 594 ref = Process.monitor({__MODULE__, sync_node})
622 - GenStateMachine.cast({__MODULE__, sync_node}, {:sync, self()})
595 + GenStateMachine.cast({__MODULE__, sync_node}, {:sync, self(), state.clock})
623 596 new_state = %{state | sync_node: sync_node, sync_ref: ref}
624 597 interval = Application.get_env(:swarm, :anti_entropy_interval, @default_anti_entropy_interval)
625 598 Process.send_after(self(), :anti_entropy, interval)
  @@ -1029,7 +1002,7 @@ defmodule Swarm.Tracker do
1029 1002 end
1030 1003
1031 1004 # This is the handler for local operations on the tracker which are asynchronous
1032 - defp handle_cast({:sync, from}, %TrackerState{clock: clock} = state) do
1005 + defp handle_cast({:sync, from, _rclock}, %TrackerState{clock: clock} = state) do
1033 1006 debug "received sync request from #{node(from)}"
1034 1007 {lclock, rclock} = Clock.fork(clock)
1035 1008 sync_node = node(from)
  @@ -1037,12 +1010,6 @@ defmodule Swarm.Tracker do
1037 1010 GenStateMachine.cast(from, {:sync_recv, self(), rclock, Registry.snapshot()})
1038 1011 {:next_state, :awaiting_sync_ack, %{state | clock: lclock, sync_node: sync_node, sync_ref: ref}}
1039 1012 end
1040 - defp handle_cast({:sync_recv, from, _clock, _registry}, _state) do
1041 - # somebody is sending us a thing which expects an ack, but we no longer care about it
1042 - # we should reply even though we're dropping this message
1043 - GenStateMachine.cast(from, {:sync_ack, Node.self})
1044 - :keep_state_and_data
1045 - end
1046 1013 defp handle_cast(msg, _state) do
1047 1014 warn "unrecognized cast: #{inspect msg}"
1048 1015 :keep_state_and_data
  @@ -13,7 +13,7 @@ defmodule Swarm.Mixfile do
13 13
14 14 def project do
15 15 [app: :swarm,
16 - version: "3.2.0",
16 + version: "3.2.1",
17 17 elixir: "~> 1.3",
18 18 elixirc_paths: elixirc_paths(Mix.env),
19 19 build_embedded: Mix.env == :prod,