From 9695231c327e052c5d5bc61197cc8222599fb91a Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 11 Sep 2018 17:21:14 +0200 Subject: Block store dependency management & caching (NOT TESTED) --- shard/lib/app/blockstore.ex | 180 +++++++++++++++++++++++++++++-------- shard/lib/app/chat.ex | 13 ++- shard/lib/cli/cli.ex | 18 ++-- shard/lib/data/merklesearchtree.ex | 51 ++++------- shard/lib/manager.ex | 2 +- 5 files changed, 188 insertions(+), 76 deletions(-) (limited to 'shard/lib') diff --git a/shard/lib/app/blockstore.ex b/shard/lib/app/blockstore.ex index 5e93135..f1140b2 100644 --- a/shard/lib/app/blockstore.ex +++ b/shard/lib/app/blockstore.ex @@ -5,7 +5,14 @@ defmodule SApp.BlockStore do This is not a shard, it is a side process that a shard may use to store its data. - TODO: WIP + Uses an ETS table of: + + { block_id, why_have_it } -- waiting for data + { block_id, why_have_it, data } -- once we have the data + + why_have_it := :root + | {:req_by, some_other_block_id} + | {:cached, expiry_date} """ use GenServer @@ -13,6 +20,8 @@ defmodule SApp.BlockStore do @enforce_keys [:pid] defstruct [:pid, :prefer_ask] + @cache_ttl 600 # Ten minutes + @clean_cache_every 60 # One minute defmodule State do defstruct [:shard_id, :path, :store, :reqs, :retries] @@ -26,47 +35,86 @@ defmodule SApp.BlockStore do def init([shard_id, path]) do Shard.Manager.dispatch_to(shard_id, path, self()) - {:ok, %State{shard_id: shard_id, path: path, store: %{}, reqs: %{}, retries: %{}}} + store_path = String.to_atom "#{Base.encode16 shard_id}/#{Atom.to_string path}" + store = :ets.new store_path, [:set, :protected] + + Process.send_after(self(), :clean_cache, @clean_cache_every * 1000) + + {:ok, %State{shard_id: shard_id, path: path, store: store, reqs: %{}, retries: %{}}} end def handle_call({:get, key, prefer_ask}, from, state) do - case state.store[key] do - nil -> - case prefer_ask do - [_ | _] -> - for peer <- prefer_ask do - Shard.Manager.send(peer, {state.shard_id, state.path, {:get, key}}) - end - _ -> - ask_random_peers(state, key) - end - reqs_key = case state.reqs[key] do - nil -> - MapSet.put(MapSet.new(), from) - ms -> - MapSet.put(ms, from) - end - state = put_in(state.reqs[key], reqs_key) - {:noreply, state} - v -> + case :ets.lookup state.store, key do + [{_, _, v}] -> {:reply, v, state} + [{_, _}] -> + state = add_request(state, key, from) + {:noreply, state} + [] -> + why = {:cached, System.os_time(:seconds) + @cache_ttl} + init_rec_pull(state, key, why, prefer_ask) + state = add_request(state, key, from) + {:noreply, state} end end def handle_call({:put, val}, _from, state) do hash = SData.term_hash val - state = %{state | store: Map.put(state.store, hash, val)} + store_put(state, hash, val) {:reply, hash, state} end + defp add_request(state, key, from) do + reqs_key = case state.reqs[key] do + nil -> + MapSet.put(MapSet.new(), from) + ms -> + MapSet.put(ms, from) + end + put_in(state.reqs[key], reqs_key) + end + + defp store_put(state, hash, val) do + case :ets.lookup state.store, hash do + [] -> + :ets.insert state.store, {hash, {:cached, System.os_time(:seconds) + @cache_ttl}, val} + nil + [{_, why}] -> + :ets.insert state.store, {hash, why, val} + why + [{_, _, _}] -> + nil + end + end + + defp init_rec_pull(state, key, why, prefer_ask) do + case prefer_ask do + [_ | _] -> + for peer <- prefer_ask do + Shard.Manager.send(peer, {state.shard_id, state.path, {:get, key}}) + end + _ -> + ask_random_peers(state, key) + end + :ets.insert state.store, {key, why} + end + + def handle_cast({:rec_pull, hash, ask_to}, state) do + if :ets.lookup state.store, hash == [] do + why = {:cached, System.os_time(:seconds) + @cache_ttl} + init_rec_pull(state, hash, why, ask_to) + end + {:noreply, state} + end + def handle_cast({:msg, peer_id, _shard_id, _path, msg}, state) do state = case msg do {:get, key} -> - case state.store[key] do - nil -> - Shard.Manager.send(peer_id, {state.shard_id, state.path, {:not_found, key}}) - v -> + case :ets.lookup state.store, key do + [{_, _, v}] -> Shard.Manager.send(peer_id, {state.shard_id, state.path, {:info, key, v}}) + _ -> + Shard.Manager.send(peer_id, {state.shard_id, state.path, {:not_found, key}}) end state {:info, hash, value} -> @@ -80,12 +128,22 @@ defmodule SApp.BlockStore do Map.delete(state.reqs, hash) end state = %{state | retries: Map.delete(state.retries, hash)} - %{state | store: Map.put(state.store, hash, value), reqs: reqs} + rec_why = store_put state, hash, value + if rec_why != nil do + sub_why = case rec_why do + {:cached, ttl} -> {:cached, ttl} + _ -> {:req_by, hash} + end + for dep <- SData.Page.refs value do + init_rec_pull(state, dep, sub_why, [peer_id]) + end + end + %{state | reqs: reqs} else state end {:not_found, key} -> - if state.reqs[key] != nil and state.store[key] == nil do + if state.reqs[key] != nil and :ets.lookup state.store, key == [] do nretry = case state.retries[key] do nil -> 1 n -> n+1 @@ -107,6 +165,64 @@ defmodule SApp.BlockStore do end {:noreply, state} end + + def handle_cast({:set_roots, roots}, state) do + cached_why = {:cached, System.os_time(:seconds) + @cache_ttl} + + # Set old roots as cached + for [id, val] <- :ets.match state.store, {:"$1", :root, :"$2"} do + :ets.insert state.store, {id, cached_why, val} + end + for [id] <- :ets.match state.store, {:"$1", :root} do + :ets.insert state.store, {id, cached_why} + end + + # Set old deps as cached + for [id, val] <- :ets.match state.store, {:"$1", {:req_by, :_}, :"$2"} do + :ets.insert state.store, {id, cached_why, val} + end + for [id] <- :ets.match state.store, {:"$1", {:req_by, :_}} do + :ets.insert state.store, {id, cached_why} + end + + # Set new roots as roots + for root <- roots do + case :ets.lookup state.store, root do + [{^root, _, val}] -> + :ets.insert state.store, {root, :root, val} + rec_set_dep state.store, root, val + [{^root, _}] -> + :ets.insert state.store, {root, :root} + [] -> + init_rec_pull state, root, :root, [] + end + end + {:noreply, state} + end + + defp rec_set_dep(store, hash, val0) do + for dep <- SData.Page.refs val0 do + case :ets.lookup store, dep do + [{^dep, _, val}] -> + :ets.insert store, {dep, {:req_by, hash}, val} + rec_set_dep(store, dep, val) + _ -> + :ets.insert store, {dep, {:req_by, hash}} + end + end + end + + def handle_info(:clean_cache, state) do + currtime = System.os_time :seconds + + cache_cleanup_1 = [ {{:_, {:cached, :'$1'}, :_}, [{:<, :'$1', currtime}], [:'$1']} ] + cache_cleanup_2 = [ {{:_, {:cached, :'$1'}}, [{:<, :'$1', currtime}], [:'$1']} ] + :ets.select_delete(state.store, cache_cleanup_1) + :ets.select_delete(state.store, cache_cleanup_2) + + Process.send_after(self(), :clean_cache, @clean_cache_every * 1000) + {:noreply, state} + end def ask_random_peers(state, key) do peers = Shard.Manager.get_shard_peers(state.shard_id) @@ -117,7 +233,6 @@ defmodule SApp.BlockStore do end end - defimpl SData.PageStore do def put(store, page) do hash = GenServer.call(store.pid, {:put, page}) @@ -133,12 +248,7 @@ defmodule SApp.BlockStore do end def copy(store, other_store, hash) do - page = SData.PageStore.get(other_store, hash) - refs = SData.Page.refs(page) - for ref <- refs do - copy(store, other_store, ref) - end - GenServer.call(store.pid, {:put, page}) + GenServer.cast(store.pid, {:rec_pull, hash, other_store.prefer_ask}) store end diff --git a/shard/lib/app/chat.ex b/shard/lib/app/chat.ex index 471d8f7..4752dc6 100644 --- a/shard/lib/app/chat.ex +++ b/shard/lib/app/chat.ex @@ -160,6 +160,7 @@ defmodule SApp.Chat do if mst2.root == new_root do # This was the only message missing, we are happy! state = %{state | mst: mst2} + GenServer.cast(state.block_store, {:set_roots, [mst2.root]}) msg_callback(state, msgitem) state else @@ -186,9 +187,19 @@ defmodule SApp.Chat do defp init_merge(state, new_root, source_peer) do # TODO: make the merge asynchronous + prev_last = for {x, true} <- MST.last(state.mst, nil, 100), into: MapSet.new, do: x + mgmst = %{state.mst | root: new_root} mgmst = put_in(mgmst.store.prefer_ask, [source_peer]) - mst = MST.merge(state.mst, mgmst, fn msgitem, true -> msg_callback(state, msgitem) end) + mst = MST.merge(state.mst, mgmst) + + for {x, true} <- MST.last(mst, nil, 100) do + if not MapSet.member? prev_last, x do + msg_callback(state, x) + end + end + + GenServer.cast(state.block_store, {:set_roots, [mst.root]}) %{state | mst: mst} end diff --git a/shard/lib/cli/cli.ex b/shard/lib/cli/cli.ex index c3afe8f..bf3a555 100644 --- a/shard/lib/cli/cli.ex +++ b/shard/lib/cli/cli.ex @@ -4,6 +4,10 @@ defmodule SCLI do """ def run() do + for {_chid, _manifest, chpid} <- Shard.Manager.list_shards do + GenServer.cast(chpid, {:subscribe, self()}) + end + run(nil) end @@ -64,11 +68,15 @@ defmodule SCLI do end defp handle_command(pid, ["hist"]) do - GenServer.call(pid, {:read_history, nil, 25}) - |> Enum.each(fn {{ts, nick, msg}, true} -> - IO.puts "#{ts |> DateTime.from_unix! |> DateTime.to_iso8601} <#{nick}> #{msg}" - end) - pid + if pid == nil do + IO.puts "Not currently on a channel!" + else + GenServer.call(pid, {:read_history, nil, 25}) + |> Enum.each(fn {{ts, nick, msg}, true} -> + IO.puts "#{ts |> DateTime.from_unix! |> DateTime.to_iso8601} <#{nick}> #{msg}" + end) + pid + end end defp handle_command(_pid, ["join", qchan]) do diff --git a/shard/lib/data/merklesearchtree.ex b/shard/lib/data/merklesearchtree.ex index 039f6ce..49d54a5 100644 --- a/shard/lib/data/merklesearchtree.ex +++ b/shard/lib/data/merklesearchtree.ex @@ -193,21 +193,19 @@ defmodule SData.MerkleSearchTree do The merge is not symmetrical in the sense that: - new pages are added in the store of the first argument - - the callback is called for all items found in the second argument and not the first """ - def merge(to, from, callback \\ fn _, _ -> nil end) do - { store, root } = merge_aux(to, from, to.store, to.root, from.root, callback) + def merge(to, from) do + { store, root } = merge_aux(to, from, to.store, to.root, from.root) %{ to | store: store, root: root } end - defp merge_aux(s1, s2, store, r1, r2, callback) do + defp merge_aux(s1, s2, store, r1, r2) do case {r1, r2} do _ when r1 == r2 -> { store, r1 } {_, nil} -> { store, r1 } {nil, _} -> store = Store.copy(store, s2.store, r2) - rec_callback(store, r2, callback) { store, r2 } _ -> %Page{ level: level1, low: low1, list: lst1 } = Store.get(store, r1) @@ -217,64 +215,49 @@ defmodule SData.MerkleSearchTree do level1 > level2 -> {level1, low1, lst1, r2, []} level2 > level1 -> {level2, r1, [], low2, lst2} end - { store, low, lst } = merge_aux_rec(s1, s2, store, low1, lst1, low2, lst2, callback) + { store, low, lst } = merge_aux_rec(s1, s2, store, low1, lst1, low2, lst2) page = %Page{ level: level, low: low, list: lst } {hash, store} = Store.put(store, page) {store, hash} end end - defp merge_aux_rec(s1, s2, store, low1, lst1, low2, lst2, callback) do + defp merge_aux_rec(s1, s2, store, low1, lst1, low2, lst2) do case {lst1, lst2} do { [], [] } -> - {store, hash} = merge_aux(s1, s2, store, low1, low2, callback) + {store, hash} = merge_aux(s1, s2, store, low1, low2) {store, hash, []} { [], [ {k, v, r} | rst2 ] } -> {low1l, low1h, store} = split(s1, store, low1, k) - {store, newlow} = merge_aux(s1, s2, store, low1l, low2, callback) - callback.(k, v) - {store, newr, newrst} = merge_aux_rec(s1, s2, store, low1h, [], r, rst2, callback) + {store, newlow} = merge_aux(s1, s2, store, low1l, low2) + {store, newr, newrst} = merge_aux_rec(s1, s2, store, low1h, [], r, rst2) {store, newlow, [ {k, v, newr} | newrst ]} { [ {k, v, r} | rst1 ], [] } -> {low2l, low2h, store} = split(s2, store, low2, k) - {store, newlow} = merge_aux(s1, s2, store, low1, low2l, callback) - {store, newr, newrst} = merge_aux_rec(s1, s2, store, r, rst1, low2h, [], callback) + {store, newlow} = merge_aux(s1, s2, store, low1, low2l) + {store, newr, newrst} = merge_aux_rec(s1, s2, store, r, rst1, low2h, []) {store, newlow, [ {k, v, newr} | newrst ]} { [ {k1, v1, r1} | rst1 ], [ {k2, v2, r2} | rst2 ] } -> case s1.cmp.(k1, k2) do :before -> {low2l, low2h, store} = split(s2, store, low2, k1) - {store, newlow} = merge_aux(s1, s2, store, low1, low2l, callback) - {store, newr, newrst} = merge_aux_rec(s1, s2, store, r1, rst1, low2h, lst2, callback) + {store, newlow} = merge_aux(s1, s2, store, low1, low2l) + {store, newr, newrst} = merge_aux_rec(s1, s2, store, r1, rst1, low2h, lst2) {store, newlow, [ {k1, v1, newr} | newrst ]} :after -> {low1l, low1h, store} = split(s1, store, low1, k2) - {store, newlow} = merge_aux(s1, s2, store, low1l, low2, callback) - callback.(k2, v2) - {store, newr, newrst} = merge_aux_rec(s1, s2, store, low1h, lst1, r2, rst2, callback) + {store, newlow} = merge_aux(s1, s2, store, low1l, low2) + {store, newr, newrst} = merge_aux_rec(s1, s2, store, low1h, lst1, r2, rst2) {store, newlow, [ {k2, v2, newr} | newrst ]} :duplicate -> - {store, newlow} = merge_aux(s1, s2, store, low1, low2, callback) - newv = s1.merge.(v1, v2) ## TODO: callback here ?? - {store, newr, newrst} = merge_aux_rec(s1, s2, store, r1, rst1, r2, rst2, callback) + {store, newlow} = merge_aux(s1, s2, store, low1, low2) + newv = s1.merge.(v1, v2) + {store, newr, newrst} = merge_aux_rec(s1, s2, store, r1, rst1, r2, rst2) {store, newlow, [ {k1, newv, newr} | newrst ]} end end end - defp rec_callback(store, root, callback) do - case root do - nil -> nil - _ -> - %Page{ level: _, low: low, list: lst } = Store.get(store, root) - rec_callback(store, low, callback) - for {k, v, rst} <- lst do - callback.(k, v) - rec_callback(store, rst, callback) - end - end - end - @doc""" Get value for a specific key in search tree. """ diff --git a/shard/lib/manager.ex b/shard/lib/manager.ex index 3f2bddb..8cedce2 100644 --- a/shard/lib/manager.ex +++ b/shard/lib/manager.ex @@ -33,7 +33,7 @@ defmodule Shard.Manager do - :outbox Multi-list of - { peer_id, message } + { peer_id, message, time_inserted } """ -- cgit v1.2.3