Scratch work, Kademlia insert-node operation.

1 files changed, 182 insertions, 0 deletions

diff --git a/Kademlia.hs b/Kademlia.hs
new file mode 100644
index 00000000..d29a3240
--- /dev/null
+++ b/Kademlia.hs
@@ -0,0 +1,182 @@
+{-# LANGUAGE CPP, ScopedTypeVariables, PartialTypeSignatures, FlexibleContexts #-}
+{-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE DeriveFunctor, DeriveTraversable #-}
+-- {-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE GADTs #-}
+module Kademlia where
+
+import Data.Maybe
+import Network.DHT.Routing as R
+import Data.Time.Clock       (getCurrentTime)
+import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)
+#ifdef THREAD_DEBUG
+import Control.Concurrent.Lifted.Instrument
+#else
+import GHC.Conc (labelThread)
+import Control.Concurrent.Lifted
+#endif
+
+import Text.PrettyPrint as PP hiding ((<>), ($$))
+import Text.PrettyPrint.HughesPJClass hiding ((<>),($$))
+import Data.IP
+import Control.Concurrent.STM
+import Control.Monad
+import Data.Monoid
+import Data.Time.Clock.POSIX (POSIXTime)
+
+{-
+insertNode1 :: forall raw dht u ip.
+               ( Address ip
+               , Default u
+               , Show u
+               , Ord (NodeId dht)
+               , FiniteBits (NodeId dht)
+               , Show (NodeId dht)
+               , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
+               , DHT.Kademlia dht
+               , Ord (TransactionID dht)
+               , WireFormat raw dht
+               , Serialize (TransactionID dht)
+               , SerializableTo raw (Response dht (Ping dht))
+               , SerializableTo raw (Query dht (Ping dht))
+               , Ord (NodeId dht)
+               , Show (NodeId dht)
+               , Show (QueryMethod dht)
+               ) => DHT raw dht u ip (NodeInfo dht ip u -> Maybe ReflectedIP -> IO ())
+-}
+{-
+insertNode1 = do
+      params = DHT.TableParameters
+        { maxBuckets = R.defaultBucketCount :: Int
+        , fallbackID = myid
+        , adjustID = dhtAdjustID Proxy (DHT.fallbackID params) :: SockAddr -> Event dht ip u -> NodeId dht
+        , logMessage = (\ _ _ -> return ()) {- TODO -} :: Char -> String -> IO ()
+        , pingProbe = error "probe" :: ni -> NodeInfo dht ip u -> IO (Bool, Maybe ReflectedIP)
+        }
+  let state = DHT.TableKeeper
+        { routingInfo = tbl
+        , grokNode    = DHT.insertNode params state
+        , grokAddress = \_ _ -> return () -- :: Maybe SockAddr -> ReflectedIP -> IO ()
+        }
+  return $ \info witnessed_ip0 -> DHT.insertNode params state info witnessed_ip0
+
+-}
+
+{-
+insertNode :: forall msg ip u.
+                ( Address ip
+                , Show u
+                , Show (NodeId msg)
+                , Ord (NodeId msg)
+                , FiniteBits (NodeId msg)
+                ) => TableParameters msg ip u -> TableKeeper msg ip u -> NodeInfo msg ip u -> Maybe ReflectedIP -> IO ()
+-}
+  {-
+  let showTable = do
+        t <- atomically $ fmap myBuckets <$> readTVar routingInfo
+        let logMsg = "Routing table: " <> pPrint t
+        logMessage 'D' (render logMsg)
+      reportPingResult tm n b  = showTable
+      reportArrival tm info ps = showTable
+  -}
+
+-- | A change occured in the kademlia routing table.
+data RoutingTableChanged ni = RoutingTableChanged
+    { nodeReplaced :: !(Maybe ni) -- Deleted entry.
+    , nodeInserted :: ni          -- New routing table entry.
+    , nodeTimestamp :: !POSIXTime -- Last-seen time for the new node.
+    }
+ deriving (Eq,Ord,Show,Functor,Foldable,Traversable)
+
+data InsertionReporter ni = InsertionReporter
+    { -- | Called on every inbound packet.
+      reportArrival :: POSIXTime
+                       -> ni   -- ^ Origin of packet.
+                       -> [ni] -- ^ These will be pinged as a result.
+                       -> IO ()
+      -- | Called on every ping probe.
+    , reportPingResult :: POSIXTime
+                          -> ni   -- ^ Who was pinged.
+                          -> Bool -- ^ True if they ponged.
+                          -> IO ()
+    }
+
+contramapIR f ir = InsertionReporter
+    { reportArrival = \tm ni nis -> reportArrival ir tm (f ni) (map f nis)
+    , reportPingResult = \tm ni b -> reportPingResult ir tm (f ni) b
+    }
+
+data KademliaSpace nid ni = KademliaSpace
+    { kademliaLocation :: ni -> nid
+    , kademliaTestBit :: nid -> Word -> Bool
+    }
+
+contramapKS f ks = KademliaSpace
+    { kademliaLocation = kademliaLocation ks . f
+    }
+
+-- insertNode param@TableParameters{..} state info witnessed_ip0 = do
+insertNode ::
+    forall ni nid.
+    (Ord ni) =>
+
+    -- reporter
+    InsertionReporter ni
+
+    -- nil
+    -> R.Info ni nid
+ 
+    -- k
+    -> KademliaSpace nid ni
+
+    -- changed
+    -> (RoutingTableChanged ni -> STM (IO ()))
+
+    -- pingProbe
+    -> (ni -> IO Bool)
+
+    -- info
+    -> ni
+
+    -- var
+    -> TVar (Maybe (R.Info ni nid))
+
+
+    -> IO ()
+
+insertNode
+    reporter
+    nil
+    k
+    changed
+    pingProbe
+    info
+    var = do
+
+  tm <- utcTimeToPOSIXSeconds <$> getCurrentTime
+
+  (ps,reaction) <- atomically $ do
+    tbl <- fromMaybe nil <$> readTVar var
+    let (inserted, ps,t') = R.updateForInbound (kademliaTestBit k) (kademliaLocation k) tm info $ myBuckets tbl
+    reaction <- if inserted
+                    then changed $ RoutingTableChanged Nothing info tm
+                    else return $ return ()
+    writeTVar var (Just $ tbl { myBuckets = t' })
+    return (ps, reaction)
+
+  reportArrival reporter tm info ps
+  reaction
+
+  _ <- fork $ do
+    myThreadId >>= flip labelThread "pingResults"
+    forM_ ps $ \n -> do
+        b <- pingProbe n
+        reportPingResult reporter tm n b
+        join $ atomically $ do
+            tbl <- fromMaybe nil <$> readTVar var
+            let (replacements, t') = R.updateForPingResult (kademliaTestBit k) (kademliaLocation k) n b $ myBuckets tbl
+            writeTVar var (Just $ tbl { myBuckets = t' })
+            sequence <$> mapM (\(x,(t,y)) -> changed $ RoutingTableChanged (Just x) y t)
+                              replacements
+
+  return ()