Merge branch 'toxManager' of blackbird:bittorrent into avoid-exposing-psq

1 files changed, 71 insertions, 31 deletions

diff --git a/Announcer/Tox.hs b/Announcer/Tox.hs
index 03cd0871..60a5a8b1 100644
--- a/Announcer/Tox.hs
+++ b/Announcer/Tox.hs
@@ -50,23 +50,16 @@ data AnnounceMethod r = forall nid ni sr addr tok a.
                 -- already in progress at announce time.  Repeated searches are
                 -- likely to finish faster than the first since nearby nodes
                 -- are not discarded.
-    , aPublish  :: Either (r -> sr  -> IO ())
-                          (r -> tok -> Maybe ni -> IO (Maybe a))
+    , aPublish  :: r -> tok -> Maybe ni -> IO (Maybe a)
                 -- ^ The action to perform when we find nearby nodes.  The
                 -- destination node is given as a Maybe so that methods that
                 -- treat 'Nothing' as loop-back address can be passed here,
                 -- however 'Nothing' will not be passed by the announcer
                 -- thread.
                 --
-                -- There are two cases:
-                --
-                --  [Left] The action to perform requires a search result.
-                --  This was implemented to support Tox's DHTKey and
-                --  Friend-Request messages.
-                --
-                --  [Right] The action requires a "token" from the destination
-                --  node.  This is the more typical "announce" semantics for
-                --  Kademlia.
+                -- The action requires a "token" from the destination
+                -- node.  This is the more typical "announce" semantics for
+                -- Kademlia.
     , aNearestNodes :: nid -> STM [ni]
                 -- ^ Method to obtain starting nodes from an iterative Kademlia search.
     , aTarget   :: nid
@@ -80,6 +73,38 @@ data AnnounceMethod r = forall nid ni sr addr tok a.
                 -- use the closest nodes found so far.
     }
 
+-- | This type specifies a Kademlia search and an action to perform upon the result.
+data SearchMethod r = forall nid ni sr addr tok a.
+                        ( Show nid
+                        , Hashable nid
+                        , Hashable ni
+                        , Ord addr
+                        , Ord nid
+                        , Ord ni
+                        ) => SearchMethod
+    { sSearch   :: Search nid addr tok ni sr
+                -- ^ This is the Kademlia search to run repeatedly to find the
+                -- nearby nodes.  A new search is started whenever one is not
+                -- already in progress at announce time.  Repeated searches are
+                -- likely to finish faster than the first since nearby nodes
+                -- are not discarded.
+                --
+                -- XXX: Currently, "repeatedly" is wrong.
+    , sWithResult :: r -> sr  -> IO ()
+                -- ^
+                -- The action to perform upon a search result.  This was
+                -- implemented to support Tox's DHTKey and Friend-Request
+                -- messages.
+    , sNearestNodes :: nid -> STM [ni]
+                -- ^ Method to obtain starting nodes from an iterative Kademlia search.
+    , sTarget   :: nid
+                -- ^ This is the Kademlia node-id of the item being announced.
+    , sInterval :: POSIXTime
+                -- ^ The time between searches.
+                --
+                -- XXX: Currently, search results will stop any repetition.
+    }
+
 
 -- announcement started:
 newAnnouncement :: STM (IO a)
@@ -113,17 +138,15 @@ reAnnounce checkFin announce interval = \announcer k now -> do
             announce
 
 -- | Schedule a recurring Search + Announce sequence.
-schedule :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
-schedule announcer k AnnounceMethod{aSearch,aPublish,aNearestNodes,aTarget,aInterval} r = do
+scheduleAnnounce :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
+scheduleAnnounce  announcer k AnnounceMethod{aSearch,aPublish,aNearestNodes,aTarget,aInterval} r = do
     st <- atomically $ newSearch aSearch aTarget []
     ns <- atomically $ newTVar MM.empty
     let astate = AnnounceState st ns
-        publishToNodes is
-          | Left  _       <- aPublish = return ()
-          | Right publish <- aPublish = do
+        publishToNodes is = do
             forM_ is $ \(Binding ni mtok _) -> do
             forM_ mtok $ \tok -> do
-                got <- publish r tok (Just ni)
+                got <- aPublish r tok (Just ni)
                 now <- getPOSIXTime
                 forM_ got $ \_ -> do
                     atomically $ modifyTVar ns $ MM.insertTake announceK ni (Down now)
@@ -132,20 +155,7 @@ schedule announcer k AnnounceMethod{aSearch,aPublish,aNearestNodes,aTarget,aInte
                 bs <- readTVar (searchInformant st {- :: TVar (MinMaxPSQ' ni nid tok -})
                 return $ MM.toList bs
             publishToNodes is
-        onResult sr
-            | Right _     <- aPublish = return True
-            | Left sendit <- aPublish = do
-                scheduleImmediately announcer k $ ScheduledItem $ \_ _ _ -> return $ do
-                    got <- sendit r sr
-                    -- If we had a way to get the source of a search result, we might want to
-                    -- treat it similarly to an announcing node and remember it in the 'aStoringNodes'
-                    -- MinMaxPSQ.  For now, I'm just letting the nodes for which we've already sent
-                    -- a message be forgotten.
-                    --
-                    -- forM_ got $ \_ -> do
-                    --    atomically $ modifyTVar ns $ MM.insertTake announceK ni (Down now)
-                    return ()
-                return True -- True to keep searching.
+        onResult sr = return True
         searchAgain = do
             -- Canceling a pending search here seems to make announcements more reliable.
             searchCancel st
@@ -170,3 +180,33 @@ schedule announcer k AnnounceMethod{aSearch,aPublish,aNearestNodes,aTarget,aInte
     atomically $ scheduleImmediately announcer k $ ScheduledItem (newAnnouncement searchAgain search announce aInterval)
     interruptDelay (interrutible announcer)
 
+-- | Schedule a recurring Search + Publish sequence.
+scheduleSearch :: Announcer -> AnnounceKey -> SearchMethod r -> r -> IO ()
+scheduleSearch announcer k SearchMethod{sSearch,sWithResult,sNearestNodes,sTarget,sInterval} r = do
+    st <- atomically $ newSearch sSearch sTarget []
+    ns <- atomically $ newTVar MM.empty
+    let astate = AnnounceState st ns
+        onResult sr = do
+                -- XXX: Using /k/ here as the announce key is causing the search not to repeat.
+                scheduleImmediately announcer k $ ScheduledItem $ \_ _ _ -> return $ do
+                    got <- sWithResult r sr
+                    -- If we had a way to get the source of a search result, we might want to
+                    -- treat it similarly to an announcing node and remember it in the 'aStoringNodes'
+                    -- MinMaxPSQ.  For now, I'm just letting the nodes for which we've already sent
+                    -- a message be forgotten.
+                    --
+                    -- forM_ got $ \_ -> do
+                    --    atomically $ modifyTVar ns $ MM.insertTake announceK ni (Down now)
+                    return ()
+                return True -- True to keep searching.
+        searchAgain = do
+            -- Canceling a pending search here seems to make announcements more reliable.
+            searchCancel st
+            isfin <- searchIsFinished st -- Always True, since we canceled.
+            return $ when isfin $ void $ fork search
+        search  = do -- thread to fork
+            atomically $ reset sNearestNodes sSearch sTarget st
+            searchLoop sSearch sTarget onResult st
+    atomically $ scheduleImmediately announcer k $ ScheduledItem (newAnnouncement searchAgain search (return ()) sInterval)
+    interruptDelay (interrutible announcer)
+