1 files changed, 102 insertions, 8 deletions

diff --git a/Kademlia.hs b/Kademlia.hs
index c8eb2b93..9316f135 100644
--- a/Kademlia.hs
+++ b/Kademlia.hs
@@ -3,12 +3,14 @@
 {-# LANGUAGE DeriveFunctor, DeriveTraversable #-}
 -- {-# LANGUAGE TypeFamilies #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE PatternSynonyms #-}
 module Kademlia where
 
+import Data.Function
 import Data.Maybe
 import Network.DHT.Routing as R
 import Data.Time.Clock       (getCurrentTime)
-import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)
+import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds, getPOSIXTime)
 #ifdef THREAD_DEBUG
 import Control.Concurrent.Lifted.Instrument
 #else
@@ -24,7 +26,8 @@ import Control.Concurrent.STM
 import Control.Monad
 import Data.Monoid
 import Data.Time.Clock.POSIX (POSIXTime)
-
+import Data.Wrapper.PSQInt ( pattern (:->) )
+import qualified Data.Wrapper.PSQInt as Int
 
 -- | The status of a given node with respect to a given routint table.
 data RoutingStatus
@@ -64,7 +67,7 @@ contramapIR f ir = InsertionReporter
     }
 
 -- | All the IO operations neccessary to maintain a Kademlia routing table.
-data TableStateIO nid ni = TableStateIO
+data TableStateIO ni = TableStateIO
     { -- | Write the routing table.  Typically 'writeTVar'.
       tblWrite :: R.BucketList ni -> STM ()
 
@@ -88,7 +91,7 @@ data TableStateIO nid ni = TableStateIO
     , tblTransition :: RoutingTransition ni -> STM (IO ())
     }
 
-vanillaIO :: TVar (BucketList ni) -> (ni -> IO Bool) -> TableStateIO nid ni
+vanillaIO :: TVar (BucketList ni) -> (ni -> IO Bool) -> TableStateIO ni
 vanillaIO var ping = TableStateIO
     { tblRead    = readTVar var
     , tblWrite   = writeTVar var
@@ -100,7 +103,7 @@ vanillaIO var ping = TableStateIO
 -- information) entries.
 data Kademlia nid ni = Kademlia (InsertionReporter ni)
                                 (KademliaSpace nid ni)
-                                (TableStateIO nid ni)
+                                (TableStateIO ni)
 
 
 -- Helper to 'insertNode'.
@@ -132,8 +135,8 @@ insertNode (Kademlia reporter space io) node = do
     tblWrite io t'
     reaction <- case ps of
         _ | inserted -> -- Node transition: Stranger --> Accepted
-                        tblTransition io $ RoutingTransition node Accepted 
-        (_:_)        ->  -- Node transition: Stranger --> Applicant
+                        tblTransition io $ RoutingTransition node Accepted
+        (_:_)        -> -- Node transition: Stranger --> Applicant
                         tblTransition io $ RoutingTransition node Applicant
         _            -> return $ return ()
     return (ps, reaction)
@@ -145,7 +148,7 @@ insertNode (Kademlia reporter space io) node = do
     myThreadId >>= flip labelThread "pingResults"
     forM_ ps $ \n -> do
         b <- tblPing io n
-        reportPingResult reporter tm n b
+        reportPingResult reporter tm n b -- XXX: tm is timestamp of original triggering packet, not result
         join $ atomically $ do
             tbl <- tblRead io
             let (replacements, t') = R.updateForPingResult space n b tbl
@@ -156,3 +159,94 @@ insertNode (Kademlia reporter space io) node = do
             return $ sequence_ ios
 
   return ()
+
+
+-- TODO: Bootstrap/Refresh
+--
+-- From BEP 05:
+--
+-- Each bucket should maintain a "last changed" property to indicate how
+-- "fresh" the contents are.
+--
+-- Note: We will use a "time to next refresh" property instead and store it in
+-- a priority search queue.
+--
+-- When...
+--
+-- >>> data BucketTouchEvent = RoutingStatus :--> RoutingStatus
+-- >>> bucketEvents =
+-- >>>  [ Applicant :--> Stranger -- a node in a bucket is pinged and it responds,
+-- >>>
+-- >>>  , Stranger  :--> Accepted -- or a node is added to a bucket,
+-- >>>
+-- >>>  , Accepted  :--> Stranger -- or a node in a bucket is replaced
+-- >>>  , Applicant :--> Accepted -- with another node,
+-- >>>  ]
+--
+-- the bucket's last changed property should be updated. Buckets
+-- that have not been changed in 15 minutes should be "refreshed." This is done
+-- by picking a random ID in the range of the bucket and performing a
+-- find_nodes search on it.
+--
+-- The only other possible BucketTouchEvents are as follows:
+--
+-- >>> not_handled =
+-- >>>   , Stranger :--> Applicant -- A ping is pending, it's result is covered:
+-- >>>                             --     (Applicant :--> Stranger)
+-- >>>                             --     (Applicant :--> Accepted)
+-- >>>   , Accepted :--> Applicant -- Never happens
+-- >>>   ]
+--
+
+-- XXX: This will be redundantly triggered twice upon every node replacement
+-- because we do not currently distinguish between standalone
+-- insertion/deletion events and an insertion/deletion pair constituting
+-- replacement.
+--
+-- It might also be better to pass the timestamp of the transition here and
+-- keep the refresh queue in better sync with the routing table by updating it
+-- within the STM monad.
+touchBucket :: POSIXTime -> (ni -> Int) -> TVar (Int.PSQ POSIXTime) -> RoutingTransition ni -> STM (IO ())
+touchBucket interval bktnum psq tr
+    | (transitionedTo tr == Applicant)
+                = return $ return ()
+    | otherwise = return $ do
+         now <- getPOSIXTime
+         atomically $ modifyTVar' psq
+                    $ Int.insert (bktnum $ transitioningNode tr)
+                                 (now + interval)
+
+-- | > pollForRefresh interval queue refresh
+--
+-- Fork a refresh loop.  Kill the returned thread to terminate it.  The
+-- arguments are: a staleness threshold (if a bucket goes this long without
+-- being touched, a refresh will be triggered), a TVar with the time-to-refresh
+-- schedule for each bucket, and a refresh action to be forked when a bucket
+-- excedes the staleness threshold.
+--
+-- TO "touch" a bucket and prevent it from being refreshed, reschedule it's
+-- refresh time to some time into the future by modifying the 'Int.PSQ' in the
+-- TVar.
+pollForRefresh :: POSIXTime -> TVar (Int.PSQ POSIXTime) -> (Int -> IO ()) -> IO ThreadId
+pollForRefresh interval psq refresh = do
+    fork $ do
+      myThreadId >>= flip labelThread "pollForRefresh"
+      fix $ \again -> do
+        join $ atomically $ do
+            nextup <- Int.findMin <$> readTVar psq
+            maybe retry (return . go again) nextup
+ where
+    go again ( bktnum :-> refresh_time ) = do
+        now <- getPOSIXTime
+        case fromEnum (refresh_time - now) of
+            x | x <= 0 -> do -- Refresh time!
+                             -- Move it to the back of the refresh queue.
+                             atomically $ modifyTVar' psq
+                                        $ Int.insert bktnum (now + interval)
+                             -- Now fork the refresh operation.
+                             -- TODO: We should probably propogate the kill signal to this thread.
+                             fork $ do myThreadId >>= flip labelThread ("refresh."++show bktnum)
+                                       refresh bktnum
+                             return ()
+            seconds    -> threadDelay ( seconds * 1000000 )
+        again