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. 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 @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] 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()) 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 :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 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 :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} -> 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)} 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 :ets.lookup state.store, key == [] 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 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 = [ {{:_, {:cached, :'$1'}, :_}, [{:<, :'$1', currtime}], [true]}, {{:_, {:cached, :'$1'}}, [{:<, :'$1', currtime}], [true]} ] :ets.select_delete(state.store, cache_cleanup) 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) |> 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 try do GenServer.call(store.pid, {:get, hash, store.prefer_ask}) catch :exit, {:timeout, _} -> nil end end def copy(store, other_store, hash) do GenServer.cast(store.pid, {:rec_pull, hash, other_store.prefer_ask}) store end def free(store, _hash) do store ## DO SOMETHING??? end end end