announcer scheduling redux

2 files changed, 36 insertions, 63 deletions

diff --git a/Announcer.hs b/Announcer.hs
index e8288756..c9c4b60b 100644
--- a/Announcer.hs
+++ b/Announcer.hs
@@ -21,12 +21,10 @@ module Announcer
     -- lower level, Announcer.Tox needs these.
     , scheduleImmediately
     , ScheduledItem(ScheduledItem)
-    , interrutible
     ) 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
 
@@ -86,25 +84,20 @@ emptySchedule :: Schedule
 emptySchedule = Schedule PSQ.empty
 
 type KPS = (AnnounceKey, POSIXTime, ScheduledItem)
-findNextScheduled :: Announcer -> STM KPS
-findNextScheduled announcer = maybe retry return =<< (findMin . unSchedule) <$> readTVar (scheduled announcer)
-
-removeFromSchedule :: Announcer -> KPS -> STM ()
-removeFromSchedule announcer item = modifyTVar' (scheduled announcer) (Schedule . PSQ.delete (key item) . unSchedule)
 
 scheduleToList :: Announcer -> STM [KPS]
 scheduleToList announcer = PSQ.toList . unSchedule <$> readTVar (scheduled announcer)
 
 data Announcer = Announcer
     { -- | The queue of scheduled events.  The priority is the time at which an
-      -- event is supposed to occur.
+      -- event is supposed to occur.  Do not write to this TVar ever.
       scheduled       :: TVar Schedule
       -- | This TVar is False when the Announcer thread has finished.
     , announcerActive :: TVar Bool
       -- | This delay is used to wait until it's time to act on the earliest
       -- scheduled item.  It will be interrupted whenever a new item is
       -- scheduled.
-    , interrutible    :: InterruptibleDelay
+    , commander :: TChan SchedulerCommand
     }
 
 -- | Schedules an event to occur long ago at the epoch (which effectively makes
@@ -112,34 +105,29 @@ data Announcer = Announcer
 -- also want to interrupt the 'interrutible' delay so that it finds this item
 -- immediately.
 scheduleImmediately :: Announcer -> AnnounceKey -> ScheduledItem -> STM ()
-scheduleImmediately announcer k item
-    = modifyTVar' (scheduled announcer) (Schedule . PSQ.insert' k item 0 . unSchedule)
+scheduleImmediately announcer k item = scheduleAbs announcer k item 0
 
 scheduleAbs :: Announcer -> AnnounceKey -> ScheduledItem -> POSIXTime -> STM ()
 scheduleAbs announcer k item absTime =
-        modifyTVar (scheduled announcer)
-                   (Schedule . PSQ.insert' k item absTime . unSchedule)
+    writeTChan (commander announcer) $ ScheduleAction (k, absTime, item)
 
 -- | Terminate the 'Announcer' thread.  Don't use the Announcer after this.
 stopAnnouncer :: Announcer -> IO ()
 stopAnnouncer announcer = do
-    atomically $ scheduleImmediately announcer (AnnounceKey "*stop*") StopAnnouncer
-    interruptDelay (interrutible announcer)
+    atomically $ writeTChan (commander announcer) ShutdownScheduler
     atomically $ readTVar (announcerActive announcer) >>= check . not
 
 cancel :: Announcer -> AnnounceKey -> IO ()
 cancel announcer k = do
-    atomically $ scheduleImmediately announcer k $ DeleteAnnouncement
-    interruptDelay (interrutible announcer)
+    atomically $ writeTChan (commander announcer) $ ScheduleAction (k, 0, (ScheduledItem (\a k p -> return $ atomically $ return ())))
 
 -- | Construct an 'Announcer' object and fork a thread in which to perform the
 -- Kademlia searches and announces.
 forkAnnouncer :: IO Announcer
 forkAnnouncer = do
-    delay <- interruptibleDelay
     announcer <- atomically $ Announcer <$> newTVar emptySchedule
                                         <*> newTVar True
-                                        <*> pure delay
+                                        <*> newTChan
     fork $ announceThread announcer
     return announcer
 
@@ -151,66 +139,54 @@ readTChanTimeout timeout pktChannel = do
     <|> pure Nothing <* (readTVar >=> check) delay
   where
     toMicroseconds :: POSIXTime -> Int
-    toMicroseconds = undefined
+    toMicroseconds = round . (* 1000) . (* 1000)
 
-data SchedulerCommand = ShutdownScheduler | ScheduledAction KPS | SendNudes (TChan [KPS])
+data SchedulerCommand = ShutdownScheduler | ScheduleAction KPS
 
-listener :: TChan SchedulerCommand -> IO ()
-listener chan = relisten PSQ.empty
+listener :: Announcer -> IO ()
+listener announcer = relisten
+  -- This function assumes it is the only writer to the (scheduled announcer :: TVar Schedule).
+  -- If that TVar is written in another thread, the changes may be overwritten here.
   where
+    chan = commander announcer
     note :: String -> IO ()
-    note =
-      if False
-        then print
-        else const (return ())
-    relisten queue = do
+    note = if False then print else const (return ())
+    relisten = do
+      queue <- fmap unSchedule $ atomically $ readTVar $ scheduled announcer
       case minView queue of
         Nothing -> do
           note "queue empty - listening indefinitely"
           atomically (readTChan chan) >>= handleCommand
         Just ((k, p, s), queue') -> do
-          note "queue full - listening with timeout"
+          -- note "queue full - listening with timeout"
           now <- getPOSIXTime
+          note $ "queue full - listening with timeout - " ++ show (p, now)
           readTChanTimeout (p - now) chan >>= \case
             Just cmd -> handleCommand cmd
             Nothing -> do
               note "timed out - executing from queue"
-              runAction s
-              mapM id =<< atomically (performScheduledItem undefined now (k, p, s))
-              relisten queue'
+              atomically $ writeTVar (scheduled announcer) (Schedule queue')
+              case s of
+                DeleteAnnouncement -> relisten
+                StopAnnouncer -> declareInactive
+                ScheduledItem f -> (fork $ join $ atomically $ f announcer k now) >> relisten
+              relisten
       where
-        handleCommand = \case
-            ShutdownScheduler         -> return ()
-            ScheduledAction (k, p, s) -> relisten $ PSQ.insert' k s p queue
-            SendNudes nudesChan       -> do atomically $ writeTChan nudesChan (PSQ.toList queue)
-                                            relisten queue
-
-runAction :: Monad m => ScheduledItem -> m ()
-runAction DeleteAnnouncement = return ()
-runAction StopAnnouncer = return ()
-runAction (ScheduledItem a) = undefined
+        declareInactive = atomically $ writeTVar (announcerActive announcer) False
+        handleCommand =
+          \case
+            ShutdownScheduler -> declareInactive
+            ScheduleAction (k, p, s) -> do
+              atomically $
+                modifyTVar
+                  (scheduled announcer)
+                  (Schedule . PSQ.insert' k s p . unSchedule)
+              relisten
 
 announceThread :: Announcer -> IO ()
 announceThread announcer = do
     myThreadId >>= flip labelThread "announcer"
-    fix $ \loop -> do
-    item <- atomically $ findNextScheduled announcer
-
-    now <- getPOSIXTime
-    -- Is it time to do something?
-    if (prio item <= now)
-        then do -- Yes.  Dequeue and handle this event.
-              action <- atomically $ do
-                  removeFromSchedule announcer 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
+    listener announcer
 
 -- | Returns 'Nothing' to stop the announcer thread (when the event is
 -- StopAnnouncer).  Otherwise, returns an action that will be performed in the
diff --git a/Announcer/Tox.hs b/Announcer/Tox.hs
index 60a5a8b1..d0fc828f 100644
--- a/Announcer/Tox.hs
+++ b/Announcer/Tox.hs
@@ -134,7 +134,6 @@ reAnnounce checkFin announce interval = \announcer k now -> do
         scheduleAbs announcer k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval)
         return $ do
             isfin
-            hPutStrLn stderr $ "This print avoids negative-time future scheduling. Weird bug. TODO: fix it. "++show now
             announce
 
 -- | Schedule a recurring Search + Announce sequence.
@@ -178,7 +177,6 @@ scheduleAnnounce  announcer k AnnounceMethod{aSearch,aPublish,aNearestNodes,aTar
             interruptDelay (interrutible announcer)
             -}
     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 ()
@@ -208,5 +206,4 @@ scheduleSearch announcer k SearchMethod{sSearch,sWithResult,sNearestNodes,sTarge
             atomically $ reset sNearestNodes sSearch sTarget st
             searchLoop sSearch sTarget onResult st
     atomically $ scheduleImmediately announcer k $ ScheduledItem (newAnnouncement searchAgain search (return ()) sInterval)
-    interruptDelay (interrutible announcer)