this repo has no description
Compact on coordinator
+22
+22
lib/servers/coordinator.ex
+22
lib/servers/coordinator.ex
···
288
288
# Server
289
289
290
290
@state_partition 0
291
+
@work_interval_ms 1000
291
292
292
293
def init(%{id: id, replicas: replicas}) do
293
294
SimProcess.flag(:trap_exit, true)
···
366
367
state = start_manager_if_primary(state)
367
368
368
369
SimProcess.send_after(self(), :tick, @tick_interval_ms)
370
+
SimProcess.send_after(self(), :work, @work_interval_ms)
369
371
370
372
SimLogger.debug "Coordinator (id=#{state.id}) initialized"
371
373
{:ok, state}
···
467
469
end
468
470
end
469
471
472
+
def handle_info(:work, %State{} = state) do
473
+
state = work(state)
474
+
SimProcess.send_after self(), :work, @work_interval_ms
475
+
{:noreply, state}
476
+
end
477
+
478
+
def handle_info({:xks, _nonce, _msg} = message, %State{} = state) do
479
+
:ok = XKS.handle_message(state.xks, message)
480
+
{:noreply, state}
481
+
end
482
+
470
483
# Tick
471
484
472
485
def handle_info(:tick, %State{} = state) do
···
496
509
defp tick(_role, _status, %State{} = state), do: state
497
510
498
511
# KV
512
+
513
+
defp work(%State{} = state) do
514
+
%{xks: xks} = state
515
+
516
+
XKS.maybe_rotate_memtable(xks, @state_partition)
517
+
XKS.perform_work_async(xks)
518
+
519
+
state
520
+
end
499
521
500
522
defp get(%State{xks: xks, version: version}, key) when is_binary(key) do
501
523
XKS.Reader.get(xks, @state_partition, version, key)