Make the clock a ticker

Author: rodrigo-o

lib/lambda_ethereum_consensus/beacon/beacon_node.ex

@@ -24,12 +24,12 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
 
     Cache.initialize_cache()
 
-    libp2p_args = get_libp2p_args()
-
     time = :os.system_time(:second)
 
     ForkChoice.init_store(store, time)
 
+    libp2p_args = get_libp2p_args(store.genesis_time)
+
     validator_manager =
       get_validator_manager(
         deposit_tree_snapshot,
@@ -39,8 +39,8 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
 
     children =
       [
-        {LambdaEthereumConsensus.Beacon.Clock, {store.genesis_time, time}},
         {LambdaEthereumConsensus.Libp2pPort, libp2p_args},
+        {LambdaEthereumConsensus.Beacon.Ticker, [LambdaEthereumConsensus.Libp2pPort]},
         {Task.Supervisor, name: PruneStatesSupervisor},
         {Task.Supervisor, name: PruneBlocksSupervisor},
         {Task.Supervisor, name: PruneBlobsSupervisor}
@@ -63,7 +63,7 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
     ]
   end
 
-  defp get_libp2p_args() do
+  defp get_libp2p_args(genesis_time) do
     config = Application.fetch_env!(:lambda_ethereum_consensus, :libp2p)
     port = Keyword.fetch!(config, :port)
     bootnodes = Keyword.fetch!(config, :bootnodes)
@@ -75,6 +75,7 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
     end
 
     [
+      genesis_time: genesis_time,
       listen_addr: listen_addr,
       enable_discovery: true,
       discovery_addr: "0.0.0.0:#{port}",

lib/lambda_ethereum_consensus/beacon/clock.ex

@@ -1,87 +1,52 @@
-defmodule LambdaEthereumConsensus.Beacon.Clock do
+defmodule LambdaEthereumConsensus.Beacon.Ticker do
   @moduledoc false
 
   use GenServer
 
-  alias LambdaEthereumConsensus.Libp2pPort
-
   require Logger
 
-  @type state :: %{
-          genesis_time: Types.uint64(),
-          time: Types.uint64()
-        }
-
-  @spec start_link({Types.uint64(), Types.uint64()}) :: :ignore | {:error, any} | {:ok, pid}
+  @spec start_link([atom()]) :: :ignore | {:error, any} | {:ok, pid}
   def start_link(opts) do
     GenServer.start_link(__MODULE__, opts, name: __MODULE__)
   end
 
+  @spec register_to_tick(atom() | [atom()]) :: :ok
+  def register_to_tick(to_tick) when is_atom(to_tick), do: register_to_tick([to_tick])
+  def register_to_tick(to_tick) when is_list(to_tick) do
+    GenServer.cast(__MODULE__, {:register_to_tick, to_tick})
+  end
+
   ##########################
   ### GenServer Callbacks
   ##########################
 
-  @impl GenServer
-  @spec init({Types.uint64(), Types.uint64()}) ::
-          {:ok, state()} | {:stop, any}
-  def init({genesis_time, time}) do
+  @impl true
+  @spec init([atom()]) :: {:ok, [atom()]} | {:stop, any}
+  def init(to_tick) when is_list(to_tick) do
     schedule_next_tick()
 
-    {:ok,
-     %{
-       genesis_time: genesis_time,
-       time: time
-     }}
+    {:ok, to_tick}
   end
 
   @impl true
-  def handle_info(:on_tick, state) do
+  def handle_cast({:register_to_tick, to_tick_additions}, to_tick) do
+    new_to_tick = Enum.uniq(to_tick ++ to_tick_additions)
+    {:noreply, new_to_tick}
+  end
+
+  @impl true
+  def handle_info(:on_tick, to_tick) do
     schedule_next_tick()
     time = :os.system_time(:second)
-    new_state = %{state | time: time}
-
-    if time >= state.genesis_time do
-
-      # TODO: reduce time between ticks to account for gnosis' 5s slot time.
-      old_logical_time = compute_logical_time(state)
-      new_logical_time = compute_logical_time(new_state)
 
-      Libp2pPort.on_tick({time, new_logical_time, new_logical_time != old_logical_time})
-    end
+    Enum.each(to_tick, & &1.on_tick(time))
 
-    {:noreply, new_state}
+    {:noreply, to_tick}
   end
 
   def schedule_next_tick() do
     # For millisecond precision
     time_to_next_tick = 1000 - rem(:os.system_time(:millisecond), 1000)
     Process.send_after(__MODULE__, :on_tick, time_to_next_tick)
   end
-
-  @type slot_third :: :first_third | :second_third | :last_third
-  @type logical_time :: {Types.slot(), slot_third()}
-
-  @spec compute_logical_time(state()) :: logical_time()
-  defp compute_logical_time(state) do
-    elapsed_time = state.time - state.genesis_time
-
-    slot_thirds = div(elapsed_time * 3, ChainSpec.get("SECONDS_PER_SLOT"))
-    slot = div(slot_thirds, 3)
-
-    slot_third =
-      case rem(slot_thirds, 3) do
-        0 -> :first_third
-        1 -> :second_third
-        2 -> :last_third
-      end
-
-    {slot, slot_third}
-  end
-
-  defp log_new_slot({slot, :first_third}) do
-    :telemetry.execute([:sync, :store], %{slot: slot})
-    Logger.info("[Clock] Slot transition", slot: slot)
-  end
-
-  defp log_new_slot(_), do: :ok
 end

lib/libp2p_port.ex

@@ -62,13 +62,16 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   ]
 
   @type init_arg ::
-          {:listen_addr, [String.t()]}
+          {:genesis_time, Types.uint64()}
+          | {:listen_addr, [String.t()]}
           | {:enable_discovery, boolean()}
           | {:discovery_addr, String.t()}
           | {:bootnodes, [String.t()]}
           | {:join_init_topics, boolean()}
           | {:enable_request_handlers, boolean()}
 
+  @type slot_data() :: {Types.uint64(), :first_third | :second_third | :last_third}
+
   @type node_identity() :: %{
           peer_id: binary(),
           # Pretty-printed version of the peer ID
@@ -350,6 +353,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
 
   @impl GenServer
   def init(args) do
+    {genesis_time, args} = Keyword.pop!(args, :genesis_time)
     {join_init_topics, args} = Keyword.pop(args, :join_init_topics, false)
     {enable_request_handlers, args} = Keyword.pop(args, :enable_request_handlers, false)
 
@@ -374,6 +378,8 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
 
     {:ok,
      %{
+       genesis_time: genesis_time,
+       slot_data: {0, :first_third},
        port: port,
        subscribers: %{},
        requests: Requests.new(),
@@ -393,18 +399,22 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   end
 
   @impl GenServer
-  def handle_cast({:on_tick, {time, slot_data, changed_slot_data}}, state) do
+  def handle_cast({:on_tick, time}, %{genesis_time: genesis_time} = state) when time <= genesis_time, do: {:noreply, state}
+  def handle_cast({:on_tick, time}, %{genesis_time: genesis_time, slot_data: slot_data} = state) do
     # TODO: we probably want to remove this from here, but we keep it here to have this serialized
     # with respect to the other fork choice store modifications.
+
     ForkChoice.on_tick(time)
 
-    # For testing that calling it from the libp2p works, and its just a matter of the notify new block,
-    # not the clock being the one who calls the notify tick.
-    if changed_slot_data do
+    new_slot_data = compute_slot(genesis_time, time)
+
+    if slot_data != new_slot_data do
       ValidatorManager.notify_tick(slot_data)
     end
 
-    {:noreply, state}
+    log_new_slot(new_slot_data)
+
+    {:noreply, %{state | slot_data: new_slot_data}}
   end
 
   def handle_cast(
@@ -682,4 +692,32 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
       add_subscriber(state, topic, module)
     end)
   end
+
+  @spec compute_slot(Types.uint64(), Types.uint64()) :: slot_data()
+  defp compute_slot(genesis_time, time) do
+    # TODO: This was copied as it is from the Clock to convert it into just a Ticker,
+    # slot calculations are spread across modules, we should probably centralize them.
+    elapsed_time = time - genesis_time
+
+    slot_thirds = div(elapsed_time * 3, ChainSpec.get("SECONDS_PER_SLOT"))
+    slot = div(slot_thirds, 3)
+
+    slot_third =
+      case rem(slot_thirds, 3) do
+        0 -> :first_third
+        1 -> :second_third
+        2 -> :last_third
+      end
+
+    {slot, slot_third}
+  end
+
+  defp log_new_slot({slot, :first_third}) do
+    # TODO: as with the previous function, this was copied from the Clock module.
+    # is use :sync, :store as the slot event, probably something to look into.
+    :telemetry.execute([:sync, :store], %{slot: slot})
+    Logger.info("[Libp2p] Slot transition", slot: slot)
+  end
+
+  defp log_new_slot(_), do: :ok
 end