BlockValidationCounter is not needed once the query is switched to `COUNT(*)` and an index is added for `blocks.miner_hash`. Creating the index on eth60-test took only 12 seconds. ``` 2018-12-21T08:32:46.375 application=ecto_sql [info] == Running 20181221143000 Explorer.Repo.Migrations.CreateBlocksMinerHashIndex.change/0 forward 2018-12-21T08:32:46.375 application=ecto_sql [info] create index blocks_miner_hash_index 2018-12-21T08:32:59.211 application=ecto_sql [info] == Migrated 20181221143000 in 12.8s ``` Before the index both `COUNT(*)` and `COUNT(hash)` (the old usage beforepull/1275/head76fc8e0377in92f99cac22) take > 800ms, which would have a noticable impact on the UI. ``` sql> SELECT COUNT(*) FROM blocks WHERE blocks.miner_hash = ( SELECT blocks.miner_hash FROM blocks LIMIT 1 ) [2018-12-21 08:27:43] 1 row retrieved starting from 1 in 906 ms (execution: 889 ms, fetching: 17 ms) sql> SELECT COUNT(hash) FROM blocks WHERE blocks.miner_hash = ( SELECT blocks.miner_hash FROM blocks LIMIT 1 ) [2018-12-21 08:28:04] 1 row retrieved starting from 1 in 819 ms (execution: 811 ms, fetching: 8 ms) ``` After the index both queries benefited because `COUNT(hash)` was able to use Bitmap Index Scan, but `COUNT(*)` was much better using a normal Index Scan: ``` sql> SELECT COUNT(hash) FROM blocks WHERE blocks.miner_hash = ( SELECT blocks.miner_hash FROM blocks LIMIT 1 ) [2018-12-21 08:33:43] 1 row retrieved starting from 1 in 776 ms (execution: 768 ms, fetching: 8 ms) sql> SELECT COUNT(*) FROM blocks WHERE blocks.miner_hash = ( SELECT blocks.miner_hash FROM blocks LIMIT 1 ) [2018-12-21 08:33:55] 1 row retrieved starting from 1 in 130 ms (execution: 120 ms, fetching: 10 ms) ``` The `SELECT` to get a miner_hash takes 54ms alone, so the `COUNT(*)` takes ~70ms, but EXPLAINs take just as long so ~50ms is my latency to eth60-test, meaning the `SELECT COUNT(*)` takes about 20ms plus latency.
Luke Imhoff
6 years ago
parent
f67241cfc0
commit
903d322553
@ -1,115 +0,0 @@ |
||||
defmodule Explorer.Counters.BlockValidationCounter do |
||||
use GenServer |
||||
|
||||
@moduledoc """ |
||||
Module responsible for fetching and consolidating the number of |
||||
validations from an address. |
||||
""" |
||||
|
||||
alias Explorer.Chain |
||||
alias Explorer.Chain.Hash |
||||
|
||||
@table :block_validation_counter |
||||
|
||||
def table_name do |
||||
@table |
||||
end |
||||
|
||||
# It is undesirable to automatically start the consolidation in all environments. |
||||
# Consider the test environment: if the consolidation initiates but does not |
||||
# finish before a test ends, that test will fail. This way, hundreds of |
||||
# tests were failing before disabling the consolidation and the scheduler in |
||||
# the test env. |
||||
config = Application.get_env(:explorer, Explorer.Counters.BlockValidationCounter) |
||||
@enable_consolidation Keyword.get(config, :enable_consolidation) |
||||
|
||||
@doc """ |
||||
Creates a process to continually monitor the validation counts. |
||||
""" |
||||
@spec start_link(term()) :: GenServer.on_start() |
||||
def start_link(_) do |
||||
GenServer.start_link(__MODULE__, :ok, name: __MODULE__) |
||||
end |
||||
|
||||
## Server |
||||
@impl true |
||||
def init(args) do |
||||
create_table() |
||||
|
||||
if enable_consolidation?() do |
||||
Task.start_link(&consolidate_blocks/0) |
||||
end |
||||
|
||||
Chain.subscribe_to_events(:blocks) |
||||
|
||||
{:ok, args} |
||||
end |
||||
|
||||
def create_table do |
||||
opts = [ |
||||
:set, |
||||
:named_table, |
||||
:public, |
||||
read_concurrency: true |
||||
] |
||||
|
||||
:ets.new(table_name(), opts) |
||||
end |
||||
|
||||
@doc """ |
||||
Consolidates the number of block validations grouped by `address_hash`. |
||||
""" |
||||
def consolidate_blocks do |
||||
Chain.each_address_block_validation_count(fn {address_hash, total} -> |
||||
insert_or_update_counter(address_hash, total) |
||||
end) |
||||
end |
||||
|
||||
@doc """ |
||||
Fetches the number of validations related to an `address_hash`. |
||||
""" |
||||
@spec fetch(Hash.Address.t()) :: non_neg_integer |
||||
def fetch(addr_hash) do |
||||
do_fetch(:ets.lookup(table_name(), to_string(addr_hash))) |
||||
end |
||||
|
||||
defp do_fetch([{_, result} | _]), do: result |
||||
defp do_fetch([]), do: 0 |
||||
|
||||
@impl true |
||||
def handle_info({:chain_event, :blocks, _type, blocks}, state) do |
||||
blocks |
||||
|> Enum.map(& &1.miner_hash) |
||||
|> Enum.each(&insert_or_update_counter(&1, 1)) |
||||
|
||||
{:noreply, state} |
||||
end |
||||
|
||||
@doc """ |
||||
Inserts a new item into the `:ets` table. |
||||
|
||||
When the record exist, the counter will be incremented by one. When the |
||||
record does not exist, the counter will be inserted with a default value. |
||||
""" |
||||
@spec insert_or_update_counter(Hash.Address.t(), non_neg_integer) :: term() |
||||
def insert_or_update_counter(addr_hash, number) do |
||||
string_addr = to_string(addr_hash) |
||||
default = {string_addr, 0} |
||||
|
||||
:ets.update_counter(table_name(), string_addr, number, default) |
||||
end |
||||
|
||||
@doc """ |
||||
Returns a boolean that indicates whether consolidation is enabled |
||||
|
||||
In order to choose whether or not to enable the scheduler and the initial |
||||
consolidation, change the following Explorer config: |
||||
|
||||
`config :explorer, Explorer.Counters.BlockValidationCounter, enable_consolidation: true` |
||||
|
||||
to: |
||||
|
||||
`config :explorer, Explorer.Counters.BlockValidationCounter, enable_consolidation: false` |
||||
""" |
||||
def enable_consolidation?, do: @enable_consolidation |
||||
end |
||||
@ -0,0 +1,7 @@ |
||||
defmodule Explorer.Repo.Migrations.CreateBlocksMinerHashIndex do |
||||
use Ecto.Migration |
||||
|
||||
def change do |
||||
create(index(:blocks, [:miner_hash])) |
||||
end |
||||
end |
||||
@ -1,45 +0,0 @@ |
||||
defmodule Explorer.Counters.BlockValidationCounterTest do |
||||
use Explorer.DataCase |
||||
|
||||
alias Explorer.Counters.BlockValidationCounter |
||||
|
||||
setup do |
||||
start_supervised!(BlockValidationCounter) |
||||
|
||||
:ok |
||||
end |
||||
|
||||
describe "consolidate/0" do |
||||
test "loads the address' validations consolidated info" do |
||||
address = insert(:address) |
||||
|
||||
insert(:block, miner: address, miner_hash: address.hash) |
||||
insert(:block, miner: address, miner_hash: address.hash) |
||||
|
||||
another_address = insert(:address) |
||||
|
||||
insert(:block, miner: another_address, miner_hash: another_address.hash) |
||||
|
||||
BlockValidationCounter.consolidate_blocks() |
||||
|
||||
assert BlockValidationCounter.fetch(address.hash) == 2 |
||||
assert BlockValidationCounter.fetch(another_address.hash) == 1 |
||||
end |
||||
end |
||||
|
||||
describe "fetch/1" do |
||||
test "fetches the total block validations by a given address" do |
||||
address = insert(:address) |
||||
another_address = insert(:address) |
||||
|
||||
assert BlockValidationCounter.fetch(address.hash) == 0 |
||||
assert BlockValidationCounter.fetch(another_address.hash) == 0 |
||||
|
||||
BlockValidationCounter.insert_or_update_counter(address.hash, 1) |
||||
BlockValidationCounter.insert_or_update_counter(another_address.hash, 10) |
||||
|
||||
assert BlockValidationCounter.fetch(address.hash) == 1 |
||||
assert BlockValidationCounter.fetch(another_address.hash) == 10 |
||||
end |
||||
end |
||||
end |
||||
Loading…
Reference in new issue