Basic infrastructure for dependency between shards

1 files changed, 150 insertions, 24 deletions

diff --git a/shard/lib/manager.ex b/shard/lib/manager.ex
index 3a0e21c..990bcea 100644
--- a/shard/lib/manager.ex
+++ b/shard/lib/manager.ex
@@ -26,7 +26,11 @@ defmodule Shard.Manager do
     - :shard_db (persistent with DETS)
     
       List of
-        { id, manifest, state }
+        { id, manifest, why_have_it, state }
+
+      why_have_it := {:pinned, %MapSet{who requires it...}, %MapSet{who it requires...}}
+                   | {:req, %MapSet{who requires it...}, %MapSet{who it requires...}}
+                   | {:cached, expiry_date}
 
     - :peer_db (persistent with DETS)
 
@@ -46,6 +50,9 @@ defmodule Shard.Manager do
 
   require Logger
 
+  @cache_ttl 3600*24      # 24 hours
+  @clean_cache_every 60   # one minute
+
   @shard_db [Application.get_env(:shard, :data_path), "shard_db"] |> Path.join |> String.to_atom
   @peer_db [Application.get_env(:shard, :data_path), "peer_db"] |> Path.join |> String.to_atom
 
@@ -61,24 +68,13 @@ defmodule Shard.Manager do
 
     :ets.new(:shard_procs, [:set, :protected, :named_table])
 
+    Process.send_after(self(), :clean_cache, 1000)
     {:ok, %{}}
   end
 
   def handle_call({:find_or_start, shard_id, manifest}, _from, state) do
-    case :dets.lookup(@shard_db, shard_id) do
-      [] -> :dets.insert(@shard_db, {shard_id, manifest, nil})
-      _ -> nil
-    end
-
-    case :ets.lookup(:shard_procs, {shard_id, nil}) do
-      [] ->
-        {:ok, pid} = apply(Shard.Manifest.module(manifest), :start_link, [manifest])
-        :ets.insert(:shard_procs, {{shard_id, nil}, pid})
-        state = Map.put(state, pid, {shard_id, nil})
-        {:reply, pid, state}
-      pid ->
-        {:reply, pid, state}
-    end
+    {pid, state} = find_or_start(state, shard_id, manifest)
+    {:reply, pid, state}
   end
 
   def handle_cast({:dispatch_to, shard_id, path, pid}, state) do
@@ -100,6 +96,84 @@ defmodule Shard.Manager do
     {:noreply, state}
   end
 
+  def handle_cast({:save_state, shard_id, shst}, state) do
+    case :dets.lookup(@shard_db, shard_id) do
+      [{^shard_id, manifest, why_have_it, _old_state}] -> 
+        :dets.insert(@shard_db, {shard_id, manifest, why_have_it, shst})
+    end
+    {:noreply, state}
+  end
+
+  def handle_cast({:pin, shard_id}, state) do
+    case :dets.lookup(@shard_db, shard_id) do
+      [{^shard_id, manifest, {:cached, _}, shst}] ->
+        :dets.insert(@shard_db, {shard_id, manifest, {:pinned, %MapSet{}, %MapSet{}}, shst})
+        {pid, state} = find_or_start(state, shard_id, manifest)
+        GenServer.cast(pid, :send_deps)
+        {:noreply, state}
+      [{^shard_id, manifest, {:req, a, b}, shst}] ->
+        :dets.insert(@shard_db, {shard_id, manifest, {:pinned, a, b}, shst})
+        {:noreply, state}
+      _ ->
+        {:noreply, state}
+    end
+  end
+
+  def handle_cast({:unpin, shard_id}, state) do
+    case :dets.lookup(@shard_db, shard_id) do
+      [{^shard_id, manifest, {:pinned, a, b}, shst}] ->
+        if MapSet.size(a) > 0 do
+          :dets.insert(@shard_db, {shard_id, manifest, {:req, a, b}, shst})
+        else
+          for dep <- b do
+            rm_dep_link(shard_id, dep)
+          end
+          :dets.insert(@shard_db, {shard_id, manifest, cached(), shst})
+        end
+      _ -> nil
+    end
+    {:noreply, state}
+  end
+
+  def handle_cast({:dep_list, shard_id, manifests}, state) do
+    case :dets.lookup(@shard_db, shard_id) do
+      [{^shard_id, manifest, {reason, a, b}, shst}] when reason == :pinned or reason == :req ->
+        bnew_pairs = Enum.map(manifests, fn m -> {SData.term_hash(m), m} end)
+        bnew_map = Enum.reduce(bnew_pairs, %{}, fn {id, m}, map -> Map.put(map, id, m) end)
+        bnew_set = Enum.reduce(bnew_pairs, %MapSet{}, fn {id, _m}, ms -> MapSet.put(ms, id) end)
+        state = MapSet.difference(bnew_set, b)
+                |> Enum.reduce(state, fn idadd, state ->
+                    add_dep_link(state, shard_id, idadd, bnew_map[idadd])
+                  end)
+        for idrm <- MapSet.difference(b, bnew_set) do
+          rm_dep_link(shard_id, idrm)
+        end
+        :dets.insert(@shard_db, {shard_id, manifest, {reason, a, bnew_set}, shst})
+        {:noreply, state}
+      _ ->
+        {:noreply, state}
+    end
+  end
+
+  def handle_info(:clean_cache, state) do
+    currtime = System.os_time :seconds
+
+    shards = :dets.select(@shard_db, [{
+      {:'$1', :_, {:cached, :'$2'}, :_}, [{:<, :'$2', currtime}], [:'$1']}
+    ])
+    for [id] <- shards do
+      case :ets.lookup(:shard_procs, {id, nil}) do
+        [{{^id, nil}, pid}] ->
+          GenServer.cast(pid, :delete_shard)
+        _ -> nil
+      end
+      :dets.delete(@shard_db, id)
+    end
+
+    Process.send_after(self(), :clean_cache, @clean_cache_every * 1000)
+    {:noreply, state}
+  end
+
   def handle_info({:DOWN, _, :process, pid, reason}, state) do
     handle_info({:EXIT, pid, reason}, state)
   end
@@ -114,6 +188,59 @@ defmodule Shard.Manager do
     end
   end
 
+  defp find_or_start(state, shard_id, manifest) do
+    case :dets.lookup(@shard_db, shard_id) do
+      [] ->
+        :dets.insert(@shard_db, {shard_id, manifest, cached(), nil})
+      [{^shard_id, ^manifest, {:cached, _}, shst}] ->
+        :dets.insert(@shard_db, {shard_id, manifest, cached(), shst})
+      _ -> nil
+    end
+
+    case :ets.lookup(:shard_procs, {shard_id, nil}) do
+      [] ->
+        {:ok, pid} = apply(Shard.Manifest.module(manifest), :start_link, [manifest])
+        :ets.insert(:shard_procs, {{shard_id, nil}, pid})
+        state = Map.put(state, pid, {shard_id, nil})
+        {pid, state}
+      [{{^shard_id, nil}, pid}] ->
+        {pid, state}
+    end
+  end
+
+  defp cached() do
+    {:cached, System.os_time(:seconds) + @cache_ttl}
+  end
+
+  defp add_dep_link(state, shard_id, id2, m2) do
+    case :dets.lookup(@shard_db, id2) do
+      [{^id2, ^m2, {reason, a, b}, shst}] when reason == :pinned or reason == :req ->
+        :dets.insert(@shard_db, {id2, m2, {reason, MapSet.put(a, shard_id), b}, shst})
+        state
+      _ ->
+        a = MapSet.new() |> MapSet.put(shard_id)
+        :dets.insert(@shard_db, {id2, m2, {:req, a, %MapSet{}}, nil})
+        {pid, state} = find_or_start(state, id2, m2)
+        GenServer.cast(pid, :send_deps)
+        state
+    end
+  end
+
+  defp rm_dep_link(shard_id, id2) do
+    case :dets.lookup(@shard_db, id2) do
+      [{^id2, m2, {reason, a, b}, shst}] when reason == :pinned or reason == :req ->
+        a2 = MapSet.delete(a, shard_id)
+        if reason == :req and MapSet.size(a2) == 0 do
+          :dets.insert(@shard_db, {id2, m2, cached(), shst})
+          for dep <- b do
+            rm_dep_link(id2, dep)
+          end
+        else
+          :dets.insert(@shard_db, {id2, m2, {reason, a2, b}, shst})
+        end
+    end
+  end
+
 
   # ======================
   # CALLED BY SNet.TcpConn
@@ -125,7 +252,7 @@ defmodule Shard.Manager do
   def incoming(conn_pid, peer_info, auth, {:interested, shards}) do
     for shard_id <- shards do
       case :dets.lookup(@shard_db, shard_id) do
-        [{ ^shard_id, manifest, _ }] ->
+        [{ ^shard_id, manifest, _, _ }] ->
           GenServer.cast(__MODULE__, {:peer_db_insert, shard_id, peer_info})
           pid = case :ets.lookup(:shard_procs, {shard_id, nil}) do
             [] ->
@@ -146,7 +273,7 @@ defmodule Shard.Manager do
     case :dets.lookup(@shard_db, shard_id) do
       [] ->
         GenServer.cast(conn_pid, {:send_msg, {:not_interested, shard_id}})
-      [{ ^shard_id, manifest, _}] ->
+      [{ ^shard_id, manifest, _, _}] ->
         GenServer.cast(__MODULE__, {:peer_db_insert, shard_id, peer_info})
         pid = case :ets.lookup(:shard_procs, {shard_id, path}) do
           [] ->
@@ -190,7 +317,7 @@ defmodule Shard.Manager do
   """
   def load_state(shard_id) do
     case :dets.lookup(@shard_db, shard_id) do
-      [{^shard_id, _, state}] -> state
+      [{^shard_id, _, _, state}] -> state
       _ -> nil
     end
   end
@@ -199,10 +326,7 @@ defmodule Shard.Manager do
   Save a state value for a shard
   """
   def save_state(shard_id, state) do
-    case :dets.lookup(@shard_db, shard_id) do
-      [{^shard_id, manifest, _old_state}] -> 
-        :dets.insert(@shard_db, {shard_id, manifest, state})
-    end
+    GenServer.cast(__MODULE__, {:save_state, shard_id, state})
   end
 
 
@@ -234,9 +358,11 @@ defmodule Shard.Manager do
   end
 
   @doc"""
-  Return the list of all shards.
+  Return the list of all shards. Returns a list of tuples:
+
+      {id, manifest, why_have_it}
   """
   def list_shards() do
-    for [x] <- :dets.match(@shard_db, :"$1"), do: x
+    for [{id, m, why, _}] <- :dets.match(@shard_db, :"$1"), do: {id, m, why}
   end
 end