From 0b5d43e1857e541af58575a7e1c2bbe729436b15 Mon Sep 17 00:00:00 2001
From: Alex Auvolat <alex@adnab.me>
Date: Sat, 1 Sep 2018 15:04:53 +0200
Subject: Implement merge for Merkle search tree, not yet working over network

---
 lib/app/blockstore.ex        |  6 ++++-
 lib/data/merklesearchtree.ex | 60 ++++++++++++++++++++++++++++++++++++++++----
 lib/manager.ex               |  2 +-
 3 files changed, 61 insertions(+), 7 deletions(-)

(limited to 'lib')

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)
-- 
cgit v1.2.3