From e7e255682a81f4212171051bb59d0fedd0e88d3e Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Fri, 31 Aug 2018 22:30:20 +0200 Subject: Chat using Merkle search tree & block store, not yet 100% complete --- lib/app/blockstore.ex | 145 +++++++++++++++++++++++++++++++++++++++++++ lib/app/blockstore.ex_ | 78 ----------------------- lib/app/chat.ex | 86 ++++++++++++++++--------- lib/cli/cli.ex | 11 +--- lib/data/merklesearchtree.ex | 109 +++++++++++++++++++++++++++----- 5 files changed, 298 insertions(+), 131 deletions(-) create mode 100644 lib/app/blockstore.ex delete mode 100644 lib/app/blockstore.ex_ (limited to 'lib') diff --git a/lib/app/blockstore.ex b/lib/app/blockstore.ex new file mode 100644 index 0000000..1523a44 --- /dev/null +++ b/lib/app/blockstore.ex @@ -0,0 +1,145 @@ +defmodule SApp.BlockStore do + @moduledoc """ + A module that implements a content-adressable storage (blocks, or pages, + identified by the hash of their contents). + + This is not a shard, it is a side process that a shard may use to store its data. + + TODO: WIP + """ + + use GenServer + + @enforce_keys [:pid] + defstruct [:pid, :prefer_ask] + + + defmodule State do + defstruct [:shard_id, :path, :store, :reqs, :retries] + end + + + def start_link(shard_id, path) do + GenServer.start_link(__MODULE__, [shard_id, path]) + end + + 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: %{}}} + 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 -> + {:reply, v, 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)} + {:reply, hash, 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 -> + Shard.Manager.send(peer_id, {state.shard_id, state.path, {:info, key, v}}) + end + state + {:info, hash, value} -> + if SData.term_hash value == hash do + reqs = case state.reqs[hash] do + nil -> state.reqs + pids -> + for pid <- pids do + GenServer.reply(pid, value) + end + Map.delete(state.reqs, hash) + end + state = %{state | retries: Map.delete(state.retries, hash)} + %{state | store: Map.put(state.store, hash, value), reqs: reqs} + else + state + end + {:not_found, key} -> + if state.reqs[key] != nil and state.store[key] == nil do + nretry = case state.retries[key] do + nil -> 1 + n -> n+1 + end + if nretry < 3 do + ask_random_peers(state, key) + %{state | retries: Map.put(state.retries, key, nretry)} + else + for pid <- state.reqs[key] do + GenServer.reply(pid, nil) + end + state = %{state | reqs: Map.delete(state.reqs, key)} + state = %{state | retries: Map.delete(state.retries, key)} + state + end + else + state + end + end + {:noreply, state} + end + + def ask_random_peers(state, key) do + peers = :ets.lookup(:shard_peer_db, state.shard_id) + |> Enum.shuffle + |> Enum.take(3) + for peer <- peers do + Shard.Manager.send(peer, {state.shard_id, state.path, {:get, key}}) + end + end + + + defimpl SData.PageStore do + def put(store, page) do + hash = GenServer.call(store.pid, {:put, page}) + { hash, store } + end + + def get(store, hash) do + GenServer.call(store.pid, {:get, hash, store.prefer_ask}) + 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}) + store + end + + def free(store, _hash) do + store ## DO SOMETHING??? + end + end +end diff --git a/lib/app/blockstore.ex_ b/lib/app/blockstore.ex_ deleted file mode 100644 index 2854161..0000000 --- a/lib/app/blockstore.ex_ +++ /dev/null @@ -1,78 +0,0 @@ -defmodule SApp.BlockStore do - @moduledoc """ - A module that implements a content-adressable storage (blocks, or pages, - identified by the hash of their contents). - - Establishes full node connectivity and uses rendez-vous hashing to select - which nodes are responsible of a given hash. - - TODO: WIP - """ - - use GenServer - - defmodule State do - defstruct [:name, :id, :manifest, - :ncopies, - :store, :peers] - end - - - def start_link(name) do - GenServer.start_link(__MODULE__, name) - end - - def init(name) do - manifest = {:blockstore, name} - id = SData.term_hash manifest - - GenServer.cast(Shard.Manager, {:register, id, self()}) - GenServer.cast(self(), :init_pull) - - {:ok, %State{name: name, id: id, manifest: manifest, - ncopies: 3, - store: %{}, peers: %{}}} - end - - def handle_call(:manifest, _from, state) do - {:reply, state.manifest, state} - end - - def handle_call({:get, key}, from, state) do - # TODO - end - - def handle_call({:put, val}, state) do - # TODO - end - - def handle_cast({:redundant, _}, _state) do - exit :normal - end - - def handle_cast(:init_pull, state) do - GenServer.call(SNet.Manager, :get_all) - |> Enum.each(&(GenServer.cast(&1, {:send_msg, {:interested, [state.id]}}))) - {:noreply, state} - end - - def handle_cast({:interested, peer_id, peer_pid}, state) do - new_peers = Map.put(state.peers, peer_id, peer_pid) - new_state = %{ state | peers: new_peers } - initial_sync(new_state, peer_id, peer_pid) - {:noreply, new_state} - end - - def handle_cast({:msg, peer_id, peer_pid, msg}, state) do - # TODO - {:noreply, state} - end - - defp initial_sync(state, peer_id, peer_pid) do - # TODO - end - - defp send(state, to, msg) do - GenServer.cast(to, {:send_msg, {state.id, msg}}) - end -end diff --git a/lib/app/chat.ex b/lib/app/chat.ex index e28e896..051fab6 100644 --- a/lib/app/chat.ex +++ b/lib/app/chat.ex @@ -19,7 +19,8 @@ defmodule SApp.Chat do use GenServer - alias SData.MerkleList, as: ML + require Logger + alias SData.MerkleSearchTree, as: MST @doc """ Start a process that connects to a given channel @@ -32,19 +33,22 @@ defmodule SApp.Chat do Initialize channel process. """ def init(channel) do - store = ML.new(&msg_cmp/2) manifest = {:chat, channel} id = SData.term_hash manifest case Shard.Manager.register(id, manifest, self()) do :ok -> Shard.Manager.dispatch_to(id, nil, self()) + {:ok, block_store} = SApp.BlockStore.start_link(id, :block_store) + mst = %MST{store: %SApp.BlockStore{pid: block_store}, + cmp: &msg_cmp/2} GenServer.cast(self(), :init_pull) {:ok, %{channel: channel, id: id, manifest: manifest, - store: store, + block_store: block_store, + mst: mst, subs: MapSet.new, } } @@ -60,8 +64,8 @@ defmodule SApp.Chat do {:reply, state.manifest, state} end - def handle_call({:read_history, start, num}, _from, state) do - ret = ML.read(state.store, start, num) + def handle_call({:read_history, top_bound, num}, _from, state) do + ret = MST.last(state.mst, top_bound, num) {:reply, ret, state} end @@ -86,7 +90,9 @@ defmodule SApp.Chat do msgitem = {(System.os_time :seconds), Shard.Identity.get_nickname(), msg} - new_state = %{state | store: ML.insert(state.store, msgitem)} + prev_root = state.mst.root + mst = MST.insert(state.mst, msgitem) + state = %{state | mst: mst} for pid <- state.subs do if Process.alive?(pid) do @@ -94,11 +100,12 @@ defmodule SApp.Chat do end end + notif = {:append, prev_root, msgitem, mst.root} for {_, peer_id} <- :ets.lookup(:shard_peer_db, state.id) do - push_messages(new_state, peer_id, nil, 5) + Shard.Manager.send(peer_id, {state.id, nil, notif}) end - {:noreply, new_state} + {:noreply, state} end @doc """ @@ -106,7 +113,7 @@ defmodule SApp.Chat do connected to asks to recieve data for this channel. """ def handle_cast({:interested, peer_id}, state) do - push_messages(state, peer_id, nil, 10) + Shard.Manager.send(peer_id, {state.id, nil, {:root, state.mst.root}}) {:noreply, state} end @@ -126,27 +133,52 @@ defmodule SApp.Chat do Merkle hash of the store of older messages. """ def handle_cast({:msg, peer_id, _shard_id, nil, msg}, state) do - case msg do + state = case msg do {:get_manifest} -> Shard.Manager.send(peer_id, {state.id, nil, {:manifest, state.manifest}}) - {:get, start} -> push_messages(state, peer_id, start, 20) - {:info, _start, list, rest} -> - if rest != nil and not ML.has(state.store, rest) do - Shard.Manager.send(peer_id, {state.id, nil, {:get, rest}}) + state + {:append, prev_root, msgitem, new_root} -> + # Append message: one single mesage has arrived + if new_root == state.mst.root do + # We already have the message, do nothing + state + else + # Try adding the message + if prev_root == state.mst.root do + mst2 = MST.insert(state.mst, msgitem) + if mst2.root == new_root do + # This was the only message missing, we are happy! + state = %{state | mst: mst2} + msg_callback(state, msgitem) + state + else + # More messages are missing, start a full merge + init_merge(state, new_root, peer_id) + end + else + init_merge(state, new_root, peer_id) + end + end + {:root, new_root} -> + if new_root == state.mst.root do + # already up to date, ignore + state + else + init_merge(state, new_root, peer_id) end - who = self() - spawn_link(fn -> - Process.sleep 1000 - GenServer.cast(who, {:deferred_insert, list}) - end) - _ -> nil + x -> + Logger.info("Unhandled message: #{inspect x}") + state end {:noreply, state} end - def handle_cast({:deferred_insert, list}, state) do - new_store = ML.insert_many(state.store, list, (fn msg -> msg_callback(state, msg) end)) - {:noreply, %{state | store: new_store}} + defp init_merge(state, new_root, source_peer) do + # TODO: make the merge asynchronous + 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) + %{state | mst: mst} end def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do @@ -154,14 +186,6 @@ defmodule SApp.Chat do {:noreply, %{ state | subs: new_subs }} end - defp push_messages(state, to, start, num) do - case ML.read(state.store, start, num) do - {:ok, list, rest} -> - Shard.Manager.send(to, {state.id, nil, {:info, start, list, rest}}) - _ -> nil - end - end - defp msg_callback(state, {ts, nick, msg}) do for pid <- state.subs do if Process.alive?(pid) do diff --git a/lib/cli/cli.ex b/lib/cli/cli.ex index 8928040..2fbf8c2 100644 --- a/lib/cli/cli.ex +++ b/lib/cli/cli.ex @@ -64,15 +64,10 @@ defmodule SCLI do end defp handle_command(pid, ["hist"]) do - case GenServer.call(pid, {:read_history, nil, 100}) do - {:ok, list, _rest} -> - list - |> Enum.reverse - |> Enum.each(fn {ts, nick, msg} -> - IO.puts "#{ts |> DateTime.from_unix! |> DateTime.to_iso8601} <#{nick}> #{msg}" + 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) - _ -> nil - end pid end diff --git a/lib/data/merklesearchtree.ex b/lib/data/merklesearchtree.ex index 0554f82..b751a08 100644 --- a/lib/data/merklesearchtree.ex +++ b/lib/data/merklesearchtree.ex @@ -48,17 +48,15 @@ defmodule SData.MerkleSearchTree do end - + @doc""" + Insert an item into the search tree. + """ def insert(state, key, value \\ true) do level = calc_level(key) {hash, store} = insert_at(state, state.store, state.root, key, level, value) %{ state | root: hash, store: store } end - def remove(_state, _key) do - # TODO - end - defp insert_at(s, store, root, key, level, value) do {new_page, store} = if root == nil do { %Page{ level: level, low: nil, list: [ { key, value, nil } ] }, store } @@ -190,34 +188,118 @@ defmodule SData.MerkleSearchTree do end end - def get(state, key) do - get(state, state.store, state.root, key) + @doc""" + Merge values from another MST in this MST + """ + def merge(to, from, callback \\ fn _, _ -> nil end) do + { store, root } = merge_aux(to, from, to.store, to.root, from.root, callback) + %{ to | store: store, root: root } + end + + defp merge_aux(s1, s2, store, r1, r2, callback) do + case {r1, r2} do + {_, nil} -> + { store, r1 } + {nil, _} -> + store = Store.copy(store, s2.store, r2) + rec_callback(store, r2, callback) + { store, r2 } + _ -> + IO.puts("not implemented: complex merge step") + #TODO + { store, r1 } + end end - defp get(s, store, root, key) do + defp rec_callback(store, root, callback) do case root do nil -> nil _ -> %Page{ level: _, low: low, list: lst } = Store.get(store, root) - get_aux(s, store, low, lst, key) + 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. + """ + def get(state, key) do + get(state, state.root, key) + end + + defp get(s, root, key) do + case root do + nil -> nil + _ -> + %Page{ level: _, low: low, list: lst } = Store.get(s.store, root) + get_aux(s, low, lst, key) end end - defp get_aux(s, store, low, lst, key) do + defp get_aux(s, low, lst, key) do case lst do [] -> - get(s, store, low, key) + get(s, low, key) [ {k, v, low2} | rst ] -> case s.cmp.(key, k) do :duplicate -> v :before -> - get(s, store, low, key) + get(s, low, key) :after -> - get_aux(s, store, low2, rst, key) + get_aux(s, low2, rst, key) end end end + @doc""" + Get the last n items of the tree, or the last n items + strictly before given upper bound if non nil + """ + def last(state, top_bound, num) do + last(state, state.root, top_bound, num) + end + + defp last(s, root, top_bound, num) do + case root do + nil -> [] + _ -> + %Page{ level: _, low: low, list: lst } = Store.get(s.store, root) + last_aux(s, low, lst, top_bound, num) + end + end + + defp last_aux(s, low, lst, top_bound, num) do + case lst do + [] -> + last(s, low, top_bound, num) + [ {k, v, low2} | rst ] -> + if top_bound == nil or s.cmp.(top_bound, k) == :after do + items = last_aux(s, low2, rst, top_bound, num) + items = if Enum.count(items) < num do + [ {k, v} | items ] + else + items + end + cnt = Enum.count items + if cnt < num do + last(s, low, top_bound, num - cnt) ++ items + else + items + end + else + last(s, low, top_bound, num) + end + end + end + + + @doc""" + Dump Merkle search tree structure. + """ def dump(state) do dump(state.store, state.root, "") end @@ -251,5 +333,4 @@ defmodule SData.MerkleSearchTree do defp count_leading_zeroes(_) do 0 end - end -- cgit v1.2.3