Bootstrapping rework in progress.

1 files changed, 46 insertions, 24 deletions

diff --git a/src/Network/BitTorrent/MainlineDHT.hs b/src/Network/BitTorrent/MainlineDHT.hs
index c7ef560c..268cacfb 100644
--- a/src/Network/BitTorrent/MainlineDHT.hs
+++ b/src/Network/BitTorrent/MainlineDHT.hs
@@ -6,6 +6,7 @@
 {-# LANGUAGE FlexibleInstances          #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE LambdaCase                 #-}
+{-# LANGUAGE NamedFieldPuns             #-}
 {-# LANGUAGE PatternSynonyms            #-}
 {-# LANGUAGE StandaloneDeriving         #-}
 {-# LANGUAGE TupleSections              #-}
@@ -502,14 +503,24 @@ newSwarmsDatabase = do
 
 data Routing = Routing
     { tentativeId :: NodeInfo
-    , sched4      :: !( TVar (Int.PSQ POSIXTime) )
-    , routing4    :: !( TVar (R.BucketList NodeInfo) )
     , committee4  :: TriadCommittee NodeId SockAddr
-    , sched6      :: !( TVar (Int.PSQ POSIXTime) )
-    , routing6    :: !( TVar (R.BucketList NodeInfo) )
     , committee6  :: TriadCommittee NodeId SockAddr
+    , refresher4  :: BucketRefresher NodeId NodeInfo
+    , refresher6  :: BucketRefresher NodeId NodeInfo
     }
 
+sched4 :: Routing -> TVar (Int.PSQ POSIXTime)
+sched4 Routing { refresher4 = BucketRefresher { refreshQueue } } = refreshQueue
+
+sched6 :: Routing -> TVar (Int.PSQ POSIXTime)
+sched6 Routing { refresher6 = BucketRefresher { refreshQueue } } = refreshQueue
+
+routing4 :: Routing -> TVar (R.BucketList NodeInfo)
+routing4 Routing { refresher4 = BucketRefresher { refreshBuckets } } = refreshBuckets
+
+routing6 :: Routing -> TVar (R.BucketList NodeInfo)
+routing6 Routing { refresher6 = BucketRefresher { refreshBuckets } } = refreshBuckets
+
 traced :: Show tid => TableMethods t tid -> TableMethods t tid
 traced (TableMethods ins del lkup)
             = TableMethods (\tid mvar t -> trace ("insert "++show tid) $ ins tid mvar t)
@@ -537,7 +548,7 @@ newClient swarms addr = do
             <$> global6
     addr4 <- atomically $ newTChan
     addr6 <- atomically $ newTChan
-    routing <- atomically $ do
+    mkrouting <- atomically $ do
         let nobkts = R.defaultBucketCount :: Int
         tbl4 <- newTVar $ R.nullTable (comparing nodeId) (\s -> hashWithSalt s . nodeId) tentative_info nobkts
         tbl6 <- newTVar $ R.nullTable (comparing nodeId) (\s -> hashWithSalt s . nodeId) tentative_info6 nobkts
@@ -558,9 +569,22 @@ newClient swarms addr = do
         committee6 <- newTriadCommittee $ updateIPVote tbl6 addr6
         sched4 <- newTVar Int.empty
         sched6 <- newTVar Int.empty
-        return $ Routing tentative_info sched4 tbl4 committee4 sched6 tbl6 committee6
+        return $ \client ->
+            let refresher sched bkts = BucketRefresher
+                    { refreshInterval = 15 * 60
+                    , refreshQueue    = sched
+                    , refreshSearch   = nodeSearch client
+                    , refreshBuckets  = bkts
+                    , refreshPing     = ping client
+                    }
+                refresher4 = refresher sched4 tbl4
+                refresher6 = refresher sched6 tbl6
+            in Routing tentative_info committee4 committee6 refresher4 refresher6
     map_var <- atomically $ newTVar (0, mempty)
-    let net = onInbound (updateRouting outgoingClient routing)
+
+    let routing = mkrouting outgoingClient
+
+        net = onInbound (updateRouting outgoingClient routing)
               $ layerTransport parsePacket encodePacket
               $ udp
 
@@ -625,14 +649,9 @@ newClient swarms addr = do
         -- TODO: trigger bootstrap ipv6
         again
 
-    refresh_thread4 <- forkPollForRefresh
-                            (15*60)
-                            (sched4 routing)
-                            (refreshBucket (nodeSearch client) (routing4 routing))
-    refresh_thread6 <- forkPollForRefresh
-                            (15*60)
-                            (sched6 routing)
-                            (refreshBucket (nodeSearch client) (routing6 routing))
+
+    refresh_thread4 <- forkPollForRefresh $ refresher4 routing
+    refresh_thread6 <- forkPollForRefresh $ refresher6 routing
 
     return (client, routing)
 
@@ -676,14 +695,17 @@ defaultHandler meth = MethodHandler decodePayload errorPayload returnError
     returnError :: NodeInfo -> BValue -> IO Error
     returnError _ _ = return $ Error MethodUnknown ("Unknown method " <> meth)
 
-mainlineKademlia :: MainlineClient -> TriadCommittee NodeId SockAddr -> TVar (R.BucketList NodeInfo) -> TVar (Int.PSQ POSIXTime) -> Kademlia NodeId NodeInfo
-mainlineKademlia client committee var sched
+mainlineKademlia :: MainlineClient
+                    -> TriadCommittee NodeId SockAddr
+                    -> BucketRefresher NodeId NodeInfo
+                    -> Kademlia NodeId NodeInfo
+mainlineKademlia client committee refresher
     = Kademlia quietInsertions
                mainlineSpace
-               (vanillaIO var $ ping client)
+               (vanillaIO (refreshBuckets refresher) $ ping client)
                    { tblTransition = \tr -> do
                         io1 <- transitionCommittee committee tr
-                        io2 <- touchBucket mainlineSpace (15*60) var sched tr
+                        io2 <- touchBucket refresher tr
                         return $ do
                             io1 >> io2
                             {-  noisy (timestamp updates are currently reported as transitions to Accepted)
@@ -712,11 +734,11 @@ transitionCommittee committee _ = return $ return ()
 updateRouting :: MainlineClient -> Routing -> NodeInfo -> Message BValue -> IO ()
 updateRouting client routing naddr msg = do
     case prefer4or6 naddr Nothing of
-        Want_IP4 -> go (routing4 routing) (committee4 routing) (sched4 routing)
-        Want_IP6 -> go (routing6 routing) (committee6 routing) (sched6 routing)
+        Want_IP4 -> go (committee4 routing) (refresher4 routing)
+        Want_IP6 -> go (committee6 routing) (refresher6 routing)
  where
-    go tbl committee sched = do
-        self <- atomically $ R.thisNode <$> readTVar tbl
+    go committee refresher = do
+        self <- atomically $ R.thisNode <$> readTVar (refreshBuckets refresher)
         when (nodeIP self /= nodeIP naddr) $ do
             case msg of
                 R { rspReflectedIP = Just sockaddr }
@@ -724,7 +746,7 @@ updateRouting client routing naddr msg = do
                         -- hPutStrLn stderr $ "External: "++show (nodeId naddr,sockaddr)
                         atomically $ addVote committee (nodeId naddr) sockaddr
                 _   -> return ()
-            insertNode (mainlineKademlia client committee tbl sched) naddr
+            insertNode (mainlineKademlia client committee refresher) naddr
 
 data Ping = Ping deriving Show