Block store dependency management & caching (NOT TESTED)

2 files changed, 157 insertions, 36 deletions

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