Implement merge for Merkle search tree, not yet working over network

4 files changed, 108 insertions, 8 deletions

diff --git a/lib/app/blockstore.ex b/lib/app/blockstore.ex
index 1523a44..4bef0b5 100644
--- a/lib/app/blockstore.ex
+++ b/lib/app/blockstore.ex
@@ -125,7 +125,11 @@ defmodule SApp.BlockStore do
     end
 
     def get(store, hash) do
-      GenServer.call(store.pid, {:get, hash, store.prefer_ask})
+      try do
+        GenServer.call(store.pid, {:get, hash, store.prefer_ask})
+      catch
+        :exit, {:timeout, _} -> nil
+      end
     end
 
     def copy(store, other_store, hash) do
diff --git a/lib/data/merklesearchtree.ex b/lib/data/merklesearchtree.ex
index b751a08..3676244 100644
--- a/lib/data/merklesearchtree.ex
+++ b/lib/data/merklesearchtree.ex
@@ -96,7 +96,7 @@ defmodule SData.MerkleSearchTree do
     if hash == nil do
       {nil, nil, store}
     else
-      %Page{ level: level, low: low, list: lst } = Store.get(store, hash)
+      %Page{ level: level, low: low, list: lst } = Store.get(store, hash) || Store.get(s.store, hash)
       store = Store.free(store, hash)
       [ { k0, _, _} | _ ] = lst
       case s.cmp.(key, k0) do
@@ -189,7 +189,11 @@ defmodule SData.MerkleSearchTree do
   end
 
   @doc"""
-    Merge values from another MST in this MST
+  Merge values from another MST in this MST.
+
+  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)
@@ -198,6 +202,7 @@ defmodule SData.MerkleSearchTree do
 
   defp merge_aux(s1, s2, store, r1, r2, callback) do
     case {r1, r2} do
+      _ when r1 == r2 -> { store, r1 }
       {_, nil} ->
         { store, r1 }
       {nil, _} ->
@@ -205,9 +210,54 @@ defmodule SData.MerkleSearchTree do
         rec_callback(store, r2, callback)
         { store, r2 }
       _ ->
-        IO.puts("not implemented: complex merge step")
-        #TODO
-        { store, r1 }
+        %Page{ level: level1, low: low1, list: lst1 } = Store.get(store, r1)
+        %Page{ level: level2, low: low2, list: lst2 } = Store.get(store, r2) || Store.get(s2.store, r2)
+        { level, low1, lst1, low2, lst2 } = cond do
+          level1 == level2 -> {level1, low1, lst1, low2, lst2}
+          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)
+        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
+    case {lst1, lst2} do
+      { [], [] } ->
+        {store, hash} = merge_aux(s1, s2, store, low1, low2, callback)
+        {store, hash, []}
+      { [], [ {k, v, r} | rst2 ] } ->
+        {low1l, low1h, store} = split(s1, store, low1, k)
+        {store, newlow} = merge_aux(s1, s2, store, low1l, low2, callback)
+        {store, newr, newrst} = merge_aux_rec(s1, s2, store, low1h, [], r, rst2, callback)
+        {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, [ {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, [ {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, [ {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, [ {k1, newv, newr} | newrst ]}
+        end
     end
   end
 
diff --git a/lib/manager.ex b/lib/manager.ex
index f6910e4..7c6c849 100644
--- a/lib/manager.ex
+++ b/lib/manager.ex
@@ -94,7 +94,7 @@ defmodule Shard.Manager do
   end
 
   def handle_cast({:peer_up, pk, pid, ip, port}, state) do
-    for {pk2, _, _, _} <- :ets.match(:peer_db, {:_, :_, ip, port}) do
+    for [pk2] <- :ets.match(:peer_db, {:'$1', :_, ip, port}) do
       if pk2 != pk do
         # obsolete peer information
         :ets.delete(:peer_db, pk2)
diff --git a/test/mst_test.exs b/test/mst_test.exs
index 7fe340e..cf5a898 100644
--- a/test/mst_test.exs
+++ b/test/mst_test.exs
@@ -111,7 +111,7 @@ defmodule ShardTest.MST do
     MST.last(y, nil, 10)
   end
 
-  test "merkle search tree 5" do
+  test "merkle search tree 5: MST.last" do
     y = Enum.reduce(0..1000, %MST{},
                     fn i, acc -> MST.insert(acc, i) end)
 
@@ -128,4 +128,50 @@ defmodule ShardTest.MST do
     assert MST.last(y, 500, 300) == stuff
   end
 
+  test "merkle search tree 6: MST.merge" do
+    y = Enum.reduce([1, 2, 42], %MST{}, fn i, acc -> MST.insert(acc, i) end)
+    z = Enum.reduce([2, 12], %MST{}, fn i, acc -> MST.insert(acc, i) end)
+
+    IO.puts "y: "
+    MST.dump y
+    IO.puts "z: "
+    MST.dump z
+
+    mg1 = MST.merge(y, z)
+    IO.puts "mg1: "
+    MST.dump mg1
+
+    mg2 = MST.merge(y, z)
+    IO.puts "mg2: "
+    MST.dump mg2
+    assert mg1.root == mg2.root
+  end
+
+  test "merkle search tree 7: MST.merge" do
+    items1 = (for i <- 1..1000, do: i*2+40)
+    items2 = (for i <- 1..1000, do: i*3)
+
+    y = Enum.reduce(items1, %MST{}, fn i, acc -> MST.insert(acc, i) end)
+    z = Enum.reduce(items2, %MST{}, fn i, acc -> MST.insert(acc, i) end)
+
+    IO.puts "(merge) y.root: #{y.root|>Base.encode16}"
+    IO.puts "(merge) z.root: #{z.root|>Base.encode16}"
+
+    mg1 = MST.merge(y, z)
+    IO.puts "(merge) mg1.root: #{mg1.root|>Base.encode16}"
+
+    mg2 = MST.merge(y, z)
+    IO.puts "(merge) mg2.root: #{mg2.root|>Base.encode16}"
+
+    assert mg1.root == mg2.root
+
+    items1t = (for i <- items1, do: {i, true})
+    items2t = (for i <- items2, do: {i, true})
+    assert MST.last(y, nil, 1000) == items1t
+    assert MST.last(z, nil, 1000) == items2t
+
+    all_items = (items1t ++ items2t) |> Enum.sort |> Enum.uniq
+    assert MST.last(mg1, nil, 2000) == all_items
+  end
+
 end