{-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE NondecreasingIndentation #-} module Announcer ( Announcer , AnnounceKey , packAnnounceKey , unpackAnnounceKey , AnnounceMethod(..) , forkAnnouncer , stopAnnouncer , schedule , cancel ) where import qualified Data.MinMaxPSQ as MM import Data.Wrapper.PSQ as PSQ import InterruptibleDelay import Network.Kademlia.Routing as R import Network.Kademlia.Search import Control.Concurrent.Lifted.Instrument import Control.Concurrent.STM import Control.Monad import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as Char8 import Data.Function import Data.Hashable import Data.Maybe import Data.Time.Clock.POSIX newtype AnnounceKey = AnnounceKey ByteString deriving (Hashable,Ord,Eq) packAnnounceKey :: Announcer -> String -> STM AnnounceKey packAnnounceKey _ = return . AnnounceKey . Char8.pack unpackAnnounceKey :: AnnounceKey -> AnnounceKey -> STM String unpackAnnounceKey _ (AnnounceKey bs) = return $ Char8.unpack bs data ScheduledItem = forall r. ScheduledItem (AnnounceMethod r) | StopAnnouncer | NewAnnouncement (IO ()) (IO ()) POSIXTime | SearchFinished (IO ()) (IO ()) POSIXTime | Announce (IO ()) POSIXTime | DeleteAnnouncement data Announcer = Announcer { scheduled :: TVar (PSQ' AnnounceKey POSIXTime ScheduledItem) , announcerActive :: TVar Bool , interrutible :: InterruptibleDelay } scheduleImmediately :: Announcer -> AnnounceKey -> ScheduledItem -> STM () scheduleImmediately announcer k item = modifyTVar' (scheduled announcer) (PSQ.insert' k item 0) stopAnnouncer :: Announcer -> IO () stopAnnouncer announcer = do atomically $ scheduleImmediately announcer (AnnounceKey "*stop*") StopAnnouncer interruptDelay (interrutible announcer) atomically $ readTVar (announcerActive announcer) >>= check . not data AnnounceMethod r = forall nid ni addr tok a. ( Show nid , Hashable nid , Hashable ni , Ord addr , Ord nid , Ord ni ) => AnnounceMethod { aSearch :: Search nid addr tok ni r , aPublish :: r -> tok -> Maybe ni -> IO (Maybe a) , aBuckets :: TVar (R.BucketList ni) , aTarget :: nid , aInterval :: POSIXTime } schedule :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO () schedule announcer k AnnounceMethod{aSearch,aPublish,aBuckets,aTarget,aInterval} r = do st <- atomically $ newSearch aSearch aTarget [] let announce = do -- publish to current search results is <- atomically $ do bs <- readTVar (searchInformant st {- :: TVar (MinMaxPSQ' ni nid tok -}) return $ MM.toList bs forM_ is $ \(Binding ni tok _) -> do aPublish r tok (Just ni) return () onResult _ = return True -- action for each search-hit (True = keep searching) search = do -- thread to fork atomically $ reset aBuckets aSearch aTarget st searchLoop aSearch aTarget onResult st atomically $ scheduleImmediately announcer k $ SearchFinished {- st -} search announce aInterval interruptDelay (interrutible announcer) atomically $ scheduleImmediately announcer k $ NewAnnouncement search announce aInterval interruptDelay (interrutible announcer) cancel :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO () cancel announcer k _ _ = do atomically $ scheduleImmediately announcer k $ DeleteAnnouncement interruptDelay (interrutible announcer) forkAnnouncer :: IO Announcer forkAnnouncer = do delay <- interruptibleDelay announcer <- atomically $ Announcer <$> newTVar PSQ.empty <*> newTVar True <*> pure delay fork $ announceThread announcer return announcer announceThread :: Announcer -> IO () announceThread announcer = do myThreadId >>= flip labelThread "announcer" fix $ \loop -> do join $ atomically $ do item <- maybe retry return =<< findMin <$> readTVar (scheduled announcer) return $ do now <- getPOSIXTime -- Is it time to do something? if (prio item < now) then do -- Yes. Dequeue and handle this event. action <- atomically $ do modifyTVar' (scheduled announcer) (PSQ.delete (key item)) performScheduledItem announcer now item -- Are we finished? mapM_ (>> loop) -- No? Okay, perform scheduled op and loop. action else do -- No. Wait a bit. startDelay (interrutible announcer) (microseconds $ prio item - now) loop -- We're done. Let 'stopAnnouncer' know that it can stop blocking. atomically $ writeTVar (announcerActive announcer) False performScheduledItem :: Announcer -> POSIXTime -> Binding' AnnounceKey POSIXTime ScheduledItem -> STM (Maybe (IO ())) performScheduledItem announcer now = \case (Binding _ StopAnnouncer _) -> return Nothing -- announcement started: (Binding k (NewAnnouncement search announce interval) _) -> do modifyTVar (scheduled announcer) (PSQ.insert' k (Announce announce interval) (now + interval)) return $ Just $ do fork search return () -- announcement removed: (Binding k DeleteAnnouncement _) -> return $ Just $ return () -- time for periodic announce: -- (re-)announce to the current known set of storing-nodes. -- TODO: If the search is finished, restart the search. (Binding k (Announce announce interval) _) -> do modifyTVar (scheduled announcer) (PSQ.insert' k (Announce announce interval) (now + interval)) return $ Just announce -- search finished: -- if any of the current storing-nodes set have not been -- announced to, announce to them. (Binding _ (SearchFinished {- st -} search announce interval) _) -> return $ Just $ return ()