chore: stream generations mutations (#1444)

aeternity · Jul 5, 2023 · b46803d · b46803d
1 parent 6a1fc14
commit b46803d
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 101 deletions.
diff --git a/lib/ae_mdw/db/sync/block.ex b/lib/ae_mdw/db/sync/block.ex
@@ -49,11 +49,11 @@ defmodule AeMdw.Db.Sync.Block do
           Txs.txi(),
           Blocks.height() | Blocks.block_hash()
         ) ::
-          {[height_mutations()], Txs.txi()}
+          Enumerable.t()
   def blocks_mutations(from_height, from_mbi, from_txi, to_height_or_hash) do
     from_height
     |> Db.get_blocks_per_height(to_height_or_hash)
-    |> Enum.flat_map_reduce(from_txi, fn {key_block, micro_blocks, next_kb_hash}, txi ->
+    |> Stream.transform(from_txi, fn {key_block, micro_blocks, next_kb_hash}, txi ->
       height = :aec_blocks.height(key_block)
       kb_header = :aec_blocks.to_key_header(key_block)
       {:ok, kb_hash} = :aec_headers.hash_header(kb_header)

diff --git a/lib/ae_mdw/sync/server.ex b/lib/ae_mdw/sync/server.ex
@@ -245,17 +245,14 @@ defmodule AeMdw.Sync.Server do
       end
 
     spawn_task(fn ->
-      {mutations_time, {gens_mutations, _next_txi}} =
-        :timer.tc(fn ->
-          Block.blocks_mutations(from_height, from_mbi, from_txi, to_height)
-        end)
-
-      MutationsCache.clear()
+      gens_mutations = Block.blocks_mutations(from_height, from_mbi, from_txi, to_height)
 
       {exec_time, new_state} =
         :timer.tc(fn -> exec_db_mutations(gens_mutations, db_state, clear_mem?) end)
 
-      gens_per_min = (to_height + 1 - from_height) * 60_000_000 / (mutations_time + exec_time)
+      MutationsCache.clear()
+
+      gens_per_min = (to_height + 1 - from_height) * 60_000_000 / exec_time
       Status.set_gens_per_min(gens_per_min)
 
       new_state
@@ -273,15 +270,8 @@ defmodule AeMdw.Sync.Server do
           :none -> -1
         end
 
-      {mutations_time, {gens_mutations, _next_txi}} =
-        :timer.tc(fn ->
-          Block.blocks_mutations(from_height, from_mbi, from_txi, last_hash)
-        end)
-
-      {exec_time, _new_state} = :timer.tc(fn -> exec_mem_mutations(gens_mutations, mem_state) end)
-
-      gens_per_min = length(gens_mutations) * 60_000_000 / (mutations_time + exec_time)
-      Status.set_gens_per_min(gens_per_min)
+      gens_mutations = Block.blocks_mutations(from_height, from_mbi, from_txi, last_hash)
+      _new_state = exec_mem_mutations(gens_mutations, mem_state)
 
       last_hash
     end)

diff --git a/test/ae_mdw/db/sync/block_test.exs b/test/ae_mdw/db/sync/block_test.exs
@@ -70,8 +70,8 @@ defmodule AeMdw.Db.Sync.BlockTest do
         last_mb_hash = AeMdw.Validate.id!(last_mb_hash)
 
         assert mem1 = :ets.info(:sync_hashes, :memory)
-        assert {height_blocks, txi} = Block.blocks_mutations(0, 0, 0, last_mb_hash)
-        assert {^height_blocks, ^txi} = Block.blocks_mutations(0, 0, 0, last_mb_hash)
+        assert height_blocks = 0 |> Block.blocks_mutations(0, 0, last_mb_hash) |> Enum.to_list()
+        assert ^height_blocks = 0 |> Block.blocks_mutations(0, 0, last_mb_hash) |> Enum.to_list()
         assert length(height_blocks) == 11
         assert mem2 = :ets.info(:sync_hashes, :memory)
         assert mem2 > mem1

diff --git a/test/ae_mdw/sync/block_test.exs b/test/ae_mdw/sync/block_test.exs