{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE NondecreasingIndentation #-} module Announcer.Tox where -- , AnnounceMethod(..) -- , schedule import Announcer import qualified Data.MinMaxPSQ as MM import Data.Wrapper.PSQ as PSQ import InterruptibleDelay import Network.Kademlia.Search import Control.Concurrent.Lifted.Instrument import Control.Concurrent.STM import Control.Monad import Data.Hashable import Data.Maybe import Data.Ord import Data.Time.Clock.POSIX import System.IO announceK :: Int announceK = 8 data AnnounceState = forall nid addr tok ni r. AnnounceState { aState :: SearchState nid addr tok ni r , aStoringNodes :: TVar (MM.MinMaxPSQ ni (Down POSIXTime)) } -- | This type specifies an item that can be announced on appropriate nodes in -- a Kademlia network. data AnnounceMethod r = forall nid ni sr addr tok a. ( Show nid , Hashable nid , Hashable ni , Ord addr , Ord nid , Ord ni ) => AnnounceMethod { aSearch :: 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. , 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. -- -- 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 -- ^ This is the Kademlia node-id of the item being announced. , aInterval :: POSIXTime -- ^ Assuming we have nearby nodes from the search, the item -- will be announced at this interval. -- -- Current implementation is to make the scheduled -- announcements even if the search hasn't finished. It will -- 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) -> IO () -> IO () -> POSIXTime -> Announcer -> AnnounceKey -> POSIXTime -> STM (IO ()) newAnnouncement checkFin search announce interval = \announcer k now -> do scheduleAbs announcer k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval) return $ void $ fork search -- time for periodic announce: -- (re-)announce to the current known set of storing-nodes. -- TODO: If the search is finished, restart the search. reAnnounce :: STM (IO a) -> IO () -> POSIXTime -> Announcer -> AnnounceKey -> POSIXTime -> STM (IO ()) reAnnounce checkFin announce interval = \announcer k now -> do isfin <- checkFin scheduleAbs announcer k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval) return $ do isfin announce -- | Schedule a recurring Search + Announce sequence. 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 = do forM_ is $ \(Binding ni mtok _) -> do forM_ mtok $ \tok -> do got <- aPublish r tok (Just ni) now <- getPOSIXTime forM_ got $ \_ -> do atomically $ modifyTVar ns $ MM.insertTake announceK ni (Down now) announce = do -- publish to current search results is <- atomically $ do bs <- readTVar (searchInformant st {- :: TVar (MinMaxPSQ' ni nid tok -}) return $ MM.toList bs publishToNodes is onResult sr = return True 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 aNearestNodes aSearch aTarget st searchLoop aSearch aTarget onResult st fork $ do -- Announce to any nodes we haven't already announced to. is <- atomically $ do bs <- readTVar (searchInformant st {- :: TVar (MinMaxPSQ' ni nid tok -}) nq <- readTVar ns return $ filter (\(Binding ni _ _) -> not $ isJust $ MM.lookup' ni nq) $ MM.toList bs publishToNodes is return () {- atomically $ scheduleImmediately announcer k $ SearchFinished {- st -} search announce aInterval interruptDelay (interrutible announcer) -} atomically $ scheduleImmediately announcer k $ ScheduledItem (newAnnouncement searchAgain search announce aInterval) -- | 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 runAction announcer $ 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)