use all cores for eventgen

Author: vans163

ex/config/runtime.exs

@@ -20,7 +20,7 @@ config :ama, :snapshot_height, (System.get_env("SNAPSHOT_HEIGHT") || "43401193")
 # https://snapshots.amadeus.bot/000041960861.zip
 # zip -0 -r 000037454455.zip db/
 # aws s3 cp --checksum-algorithm=CRC32 --endpoint-url https://20bf2f5d11d26a322e389687896a6601.r2.cloudflarestorage.com 000039434469.zip s3://ama-snapshot
-# aria2c -x 2 https://snapshots.amadeus.bot/000041960861.zip
+# aria2c -x 2 https://snapshots.amadeus.bot/000043401193.zip
 
 # tar -C /tmp/000037454455 --xform 's@^\./@@' -cf - . | zstd -T0 -1 -o /tmp/000037454455.tar.zst
 # zstd -T0 -d --stdout /tmp/000037454455.tar.zst | tar -C /tmp/restore -xf -
@@ -38,6 +38,7 @@ config :ama, :udp_ipv4_tuple, udp_ipv4_iface
 config :ama, :udp_port, (System.get_env("UDP_PORT") || "36969") |> :erlang.binary_to_integer()
 
 config :ama, :rpc_url, (System.get_env("RPC_URL") || "https://nodes.amadeus.bot")
+config :ama, :rpc_events, ((System.get_env("RPC_EVENTS") || "true") == "true")
 
 #Nodes
 if !Util.verify_time_sync() do

ex/lib/api/api_tx.ex

@@ -157,6 +157,12 @@ defmodule API.TX do
 
         tx = put_in(tx, [:tx, :action], action)
 
+        tx = if !tx[:receipt] do tx else
+          logs = Enum.map(tx.receipt.logs, fn(line)-> RocksDB.ascii_dump(line) end)
+          receipt = Map.merge(tx.receipt, %{result: RocksDB.ascii_dump(tx.receipt.result), logs: logs})
+          Map.put(tx, :receipt, receipt)
+        end
+
         if !Map.has_key?(tx, :metadata) do tx else
             put_in(tx, [:metadata, :entry_hash], Base58.encode(tx.metadata.entry_hash))
         end

ex/lib/api/db_chain.ex

@@ -55,12 +55,7 @@ defmodule DB.Chain do
           entry = DB.Entry.by_hash(map.entry_hash, db_opts)
           tx_bytes = binary_part(entry_bytes, map.index_start, map.index_size)
 
-          receipt = if map[:result] do map.result else
-            receipt = map.receipt
-            result = RocksDB.ascii_dump(receipt.result)
-            logs = Enum.map(receipt.logs, & RocksDB.ascii_dump(&1))
-            Map.merge(receipt, %{result: result, logs: logs})
-          end
+          receipt = if map[:result] do map.result else map.receipt end
 
           TX.unpack(tx_bytes)
           |> Map.put(:receipt, receipt)

ex/lib/consensus/coordination/fabric_event_gen.ex

@@ -51,10 +51,14 @@ defmodule FabricEventGen do
     end
 
     #IO.inspect API.Chain.format_entry_for_client(entry), limit: 11111
-    entry_b58 = API.Chain.format_entry_for_client(entry)
-    txs = Enum.map(entry.txs, fn(txu)->
-      API.TX.format_tx_for_client(txu)
+    txs_task = Task.async(fn ->
+      entry.txs
+      |> Task.async_stream(fn txu -> API.TX.format_tx_for_client(txu) end)
+      |> Enum.map(fn {:ok, res} -> res end)
     end)
+    entry_b58 = API.Chain.format_entry_for_client(entry)
+    txs = Task.await(txs_task, :infinity)
+
     broadcast({:update_stats_entry_tx, API.Chain.stats(), entry_b58, txs})
 
     {:noreply, state}

ex/lib/consensus/fabric_gen.ex

@@ -131,7 +131,7 @@ defmodule FabricGen do
                 :erlang.halt()
 
               true ->
-                FabricEventGen.event_rooted(best_entry, muts_hash)
+                Application.fetch_env!(:ama, :rpc_events) && FabricEventGen.event_rooted(best_entry, muts_hash)
                 %{db: db, cf: cf} = :persistent_term.get({:rocksdb, Fabric})
                 RocksDB.put("rooted_tip", best_entry.hash, %{db: db, cf: cf.sysconf})
                 proc_consensus()
@@ -187,7 +187,7 @@ defmodule FabricGen do
           result = %{error: :ok} -> result
         end
 
-        FabricEventGen.event_applied(entry, m_hash, m, r)
+        Application.fetch_env!(:ama, :rpc_events) && FabricEventGen.event_applied(entry, m_hash, m, r)
         TXPool.delete_packed(entry.txs)
 
         proc_entries()