4 files changed, 73 insertions, 57 deletions

diff --git a/Announcer.hs b/Announcer.hs
index a9b30940..4f4eba8f 100644
--- a/Announcer.hs
+++ b/Announcer.hs
@@ -16,7 +16,7 @@ module Announcer
 
 import Data.Wrapper.PSQ as PSQ
 import Network.Kademlia.Search
-import Ticker
+import InterruptibleDelay
 
 import Control.Concurrent.Lifted.Instrument
 import Control.Concurrent.STM
@@ -42,10 +42,9 @@ data ScheduledItem
     | StopAnnouncer
 
 data Announcer = Announcer
-    { scheduled :: TVar (PSQ' AnnounceKey POSIXTime ScheduledItem)
+    { scheduled       :: TVar (PSQ' AnnounceKey POSIXTime ScheduledItem)
     , announcerActive :: TVar Bool
-    , lastTick :: TVar POSIXTime
-    , announceTicker :: Ticker
+    , interrutible    :: InterruptibleDelay
     }
 
 scheduleImmediately :: Announcer -> ScheduledItem -> STM ()
@@ -62,25 +61,27 @@ data AnnounceMethod r = forall nid ni addr r tok a. AnnounceMethod
     , aPublish :: r -> tok -> Maybe ni -> IO (Maybe a)
     }
 
+-- startDelay :: InterruptibleDelay -> Microseconds -> IO Bool
+-- interruptDelay :: InterruptibleDelay -> IO ()
+
 schedule :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
-schedule _ _ _ _ = do
+schedule announcer _ _ _ = do
     -- fork the search
     -- add it to the priority queue of announce methods.
-    -- update ticker
-    return ()
+    interruptDelay (interrutible announcer)
 
 cancel :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
-cancel _ _ _ _ = return ()
+cancel announcer _ _ _ = do
+    -- cancel search/announce
+    interruptDelay (interrutible announcer)
 
 
 forkAnnouncer :: IO Announcer
 forkAnnouncer = do
-    tickvar <- atomically $ newTVar 0
-    ticker <- forkTicker $ writeTVar tickvar
+    delay <- interruptibleDelay
     announcer <- atomically $ Announcer <$> newTVar PSQ.empty
                                         <*> newTVar True
-                                        <*> pure tickvar
-                                        <*> pure ticker
+                                        <*> pure delay
     fork $ announceThread announcer
     return announcer
 
@@ -89,14 +90,24 @@ announceThread :: Announcer -> IO ()
 announceThread announcer = do
     myThreadId >>= flip labelThread "announcer"
     fix $ \loop -> do
-    action <- atomically $ do
-        now <- readTVar $ lastTick announcer
-        (item,q) <- readTVar (scheduled announcer)
-                    >>= maybe retry return . PSQ.minView
-        when (prio item > now) retry        -- Is it time to do something?
-        writeTVar (scheduled announcer) q   -- Remove the event from the queue.
-        performScheduledItem announcer item -- Go for it!
-    mapM_ (>> loop) action
+    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 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 -> Binding' AnnounceKey POSIXTime ScheduledItem -> STM (Maybe (IO ()))
diff --git a/InterruptibleDelay.hs b/InterruptibleDelay.hs
new file mode 100644
index 00000000..d59ec8ef
--- /dev/null
+++ b/InterruptibleDelay.hs
@@ -0,0 +1,41 @@
+module InterruptibleDelay where
+
+import Control.Concurrent
+import Control.Monad
+import Control.Exception ({-evaluate,-}handle,ErrorCall(..))
+import Data.Time.Clock (NominalDiffTime)
+
+type Microseconds = Int
+
+microseconds :: NominalDiffTime -> Microseconds
+microseconds d = round $ 1000000 * d
+
+data InterruptibleDelay = InterruptibleDelay
+        { delayThread :: MVar ThreadId
+        }
+
+interruptibleDelay :: IO InterruptibleDelay
+interruptibleDelay = do
+    fmap InterruptibleDelay newEmptyMVar
+
+startDelay :: InterruptibleDelay -> Microseconds -> IO Bool
+startDelay d interval = do
+    thread <- myThreadId
+    handle (\(ErrorCall _)-> do
+                debugNoise $ "delay interrupted"
+                return False) $ do 
+        putMVar (delayThread d) thread
+        threadDelay interval
+        void $ takeMVar (delayThread d)
+        return True
+
+  where debugNoise str = return ()
+
+
+interruptDelay :: InterruptibleDelay -> IO ()
+interruptDelay d = do
+    mthread <- do
+        tryTakeMVar (delayThread d)
+    flip (maybe $ return ()) mthread $ \thread -> do
+    throwTo thread (ErrorCall "Interrupted delay")
+
diff --git a/Ticker.hs b/Ticker.hs
deleted file mode 100644
index ba423def..00000000
--- a/Ticker.hs
+++ /dev/null
@@ -1,36 +0,0 @@
-module Ticker where
-
-import Control.Concurrent
-import Control.Concurrent.STM
-import Control.Exception
-import Control.Monad
-import Data.Function
-import Data.Time.Clock.POSIX
-import Data.Typeable
-
-data ScheduleTick = ScheduleTick Int deriving (Show, Typeable)
-instance Exception ScheduleTick
-
-newtype Ticker = Ticker ThreadId
-
--- | Fork a thread that will invoke an STM transaction whenever a tick event
--- occurs.  The first tick happens immediately, later ticks must be scheduled
--- with 'scheduleTick'.
-forkTicker :: (POSIXTime -> STM ()) -> IO Ticker
-forkTicker tick = do
-    getPOSIXTime >>= atomically . tick
-    tid <- forkIO $ fix $ \loop -> do
-        getPOSIXTime >>= atomically . tick
-        catch loop $ \(ScheduleTick interval) -> do
-                        threadDelay interval
-                        when (interval >= 0) loop
-    return $ Ticker tid
-
--- | Schedule the next tick.  If you supply a negative number, the ticker
--- thread will terminate. Otherwise, the next tick will be scheduled for the
--- given number of microseconds from now.
---
--- Note: If a tick was scheduled to happen sooner, that tick will be canceled
--- in favor of this one.  Only one tick is scheduled at a time.
-scheduleTick :: Ticker -> Int -> IO ()
-scheduleTick (Ticker tid) usecs = throwTo tid (ScheduleTick usecs)
diff --git a/dht-client.cabal b/dht-client.cabal
index 243f8a75..81d96427 100644
--- a/dht-client.cabal
+++ b/dht-client.cabal
@@ -95,7 +95,7 @@ library
                     Text.XXD
                     Roster
                     Announcer
-                    Ticker
+                    InterruptibleDelay
 
   build-depends:      base
                     , containers