Queries table: Switched MVar with callback.

1 files changed, 70 insertions, 90 deletions

diff --git a/src/Network/QueryResponse.hs b/src/Network/QueryResponse.hs
index 13160d31..01981cc8 100644
--- a/src/Network/QueryResponse.hs
+++ b/src/Network/QueryResponse.hs
@@ -27,10 +27,12 @@ import qualified Data.IntMap.Strict as IntMap
          ;import Data.IntMap.Strict (IntMap)
 import qualified Data.Map.Strict    as Map
          ;import Data.Map.Strict    (Map)
+import Data.Time.Clock.POSIX
 import qualified Data.Word64Map     as W64Map
          ;import Data.Word64Map     (Word64Map)
 import Data.Word
 import Data.Maybe
+import GHC.Event
 import Network.Socket
 import Network.Socket.ByteString    as B
 import System.Endian
@@ -39,6 +41,7 @@ import System.IO.Error
 import System.Timeout
 import DPut
 import DebugTag
+import Data.TableMethods
 
 -- | Three methods are required to implement a datagram based query\/response protocol.
 data TransportA err addr x y = Transport
@@ -203,6 +206,52 @@ forkListener name client = do
         closeTransport client
         killThread thread_id
 
+asyncQuery_ :: Client err meth tid addr x
+              -> MethodSerializer tid addr x meth a b
+              -> a
+              -> addr
+              -> (Maybe b -> IO ())
+              -> IO (tid,POSIXTime,Int)
+asyncQuery_ (Client net d err pending whoami _) meth q addr0 withResponse = do
+    now <- getPOSIXTime
+    (tid,addr,expiry) <- atomically $ do
+        tbl <- readTVar pending
+        ((tid,addr,expiry), tbl') <- dispatchRegister (tableMethods d)
+                (methodTimeout meth)
+                now
+                (withResponse . fmap (unwrapResponse meth))
+                addr0
+                tbl
+        -- (addr,expiry) <- methodTimeout meth tid addr0
+        writeTVar pending tbl'
+        return (tid,addr,expiry)
+    self <- whoami (Just addr)
+    mres <- do sendMessage net addr (wrapQuery meth tid self addr q)
+               return $ Just ()
+             `catchIOError` (\e -> return Nothing)
+    return (tid,now,expiry)
+
+asyncQuery :: Client err meth tid addr x
+              -> MethodSerializer tid addr x meth a b
+              -> a
+              -> addr
+              -> (Maybe b -> IO ())
+              -> IO ()
+asyncQuery client meth q addr withResponse0 = do
+    tm <- getSystemTimerManager
+    tidvar <- newEmptyMVar
+    timedout <- registerTimeout tm 300000000 $ do
+            withResponse0 Nothing
+            tid <- takeMVar tidvar
+            case client of
+                Client { clientDispatcher = d, clientPending = pending } -> do
+                    atomically $ readTVar pending >>= dispatchCancel (tableMethods d) tid >>= writeTVar pending
+    (tid,now,expiry) <- asyncQuery_ client meth q addr $ \x -> do
+        unregisterTimeout tm timedout
+        withResponse0 x
+    putMVar tidvar tid
+    updateTimeout tm timedout expiry
+
 -- | Send a query to a remote peer.  Note that this function will always time
 -- out if 'forkListener' was never invoked to spawn a thread to receive and
 -- dispatch the response.
@@ -213,25 +262,14 @@ sendQuery ::
         -> a                                    -- ^ The outbound query.
         -> addr                                 -- ^ Destination address of query.
         -> IO (Maybe b)                         -- ^ The response, or 'Nothing' if it timed out.
-sendQuery (Client net d err pending whoami _ enterQuery leaveQuery) meth q addr0 = do
+sendQuery c@(Client net d err pending whoami _) meth q addr0 = do
     mvar <- newEmptyMVar
-    (tid,addr,expiry) <- atomically $ do
-        tbl <- readTVar pending
-        (tid, tbl') <- dispatchRegister (tableMethods d) mvar addr0 tbl
-        (addr,expiry) <- methodTimeout meth tid addr0
-        writeTVar pending tbl'
-        return (tid,addr,expiry)
-    self <- whoami (Just addr)
-    enterQuery tid
-    mres <- do sendMessage net addr (wrapQuery meth tid self addr q)
-               timeout expiry $ takeMVar mvar
-            `catchIOError` (\e -> return Nothing)
-    leaveQuery tid (isJust mres)
+    (tid,now,expiry) <- asyncQuery_ c meth q addr0 $ mapM_ (putMVar mvar)
+    mres <- timeout expiry $ takeMVar mvar
     case mres of
-        Just x -> return $ Just $ unwrapResponse meth x
+        Just b -> return $ Just b
         Nothing -> do
             atomically $ readTVar pending >>= dispatchCancel (tableMethods d) tid >>= writeTVar pending
-            reportTimeout err (method meth) tid addr
             return Nothing
 
 -- * Implementing a query\/response 'Client'.
@@ -259,10 +297,6 @@ data Client err meth tid addr x = forall tbl. Client
       -- /tid/ includes a unique cryptographic nonce, then it should be
       -- generated here.
     , clientResponseId :: tid -> IO tid
-      -- | The enter/leave methods are no-ops by default.  They are useful for
-      -- serializing all queries for debugging purposes.
-    , clientEnterQuery :: tid -> IO ()
-    , clientLeaveQuery :: tid -> Bool -> IO ()
     }
 
 -- | An incoming message can be classified into three cases.
@@ -353,7 +387,7 @@ data TransactionMethods d tid addr x = TransactionMethods
       -- response will be written too.  The returned /tid/ is a transaction id
       -- that can be used to forget the 'MVar' if the remote peer is not
       -- responding.
-      dispatchRegister :: MVar x -> addr -> d -> STM (tid, d)
+      dispatchRegister :: (tid -> addr -> STM (addr,Int)) -> POSIXTime -> (Maybe x -> IO ()) -> addr -> d -> STM ((tid,addr,Int), d)
       -- | This method is invoked when an incoming packet /x/ indicates it is
       -- a response to the transaction with id /tid/.  The returned IO action
       -- will write the packet to the correct 'MVar' thus completing the
@@ -364,69 +398,37 @@ data TransactionMethods d tid addr x = TransactionMethods
     , dispatchCancel :: tid -> d -> STM d
     }
 
--- | The standard lookup table methods for use as input to 'transactionMethods'
--- in lieu of directly implementing 'TransactionMethods'.
-data TableMethods t tid = TableMethods
-    { -- | Insert a new /tid/ entry into the transaction table.
-      tblInsert :: forall a. tid -> a -> t a -> t a
-      -- | Delete transaction /tid/ from the transaction table.
-    , tblDelete :: forall a. tid -> t a -> t a
-       -- | Lookup the value associated with transaction /tid/.
-    , tblLookup :: forall a. tid -> t a -> Maybe a
-    }
-
--- | Methods for using 'Data.IntMap'.
-intMapMethods :: TableMethods IntMap Int
-intMapMethods = TableMethods IntMap.insert IntMap.delete IntMap.lookup
-
--- | Methods for using 'Data.Word64Map'.
-w64MapMethods :: TableMethods Word64Map Word64
-w64MapMethods = TableMethods W64Map.insert W64Map.delete W64Map.lookup
-
--- | Methods for using 'Data.Map'
-mapMethods :: Ord tid => TableMethods (Map tid) tid
-mapMethods = TableMethods Map.insert Map.delete Map.lookup
-
--- | Change the key type for a lookup table implementation.
---
--- This can be used with 'intMapMethods' or 'mapMethods' to restrict lookups to
--- only a part of the generated /tid/ value.  This is useful for /tid/ types
--- that are especially large due their use for other purposes, such as secure
--- nonces for encryption.
-instance Contravariant (TableMethods t) where
-    -- contramap :: (tid -> t1) -> TableMethods t t1 -> TableMethods t tid
-    contramap f (TableMethods ins del lookup) =
-        TableMethods (\k v t -> ins (f k) v t)
-                     (\k t   -> del (f k) t)
-                     (\k t   -> lookup (f k) t)
-
 -- | Construct 'TransactionMethods' methods out of 3 lookup table primitives and a
 -- function for generating unique transaction ids.
 transactionMethods ::
     TableMethods t tid  -- ^ Table methods to lookup values by /tid/.
     -> (g -> (tid,g))   -- ^ Generate a new unique /tid/ value and update the generator state /g/.
-    -> TransactionMethods (g,t (MVar x)) tid addr x
-transactionMethods methods generate = transactionMethods' id tryPutMVar methods generate
+    -> TransactionMethods (g,t (Maybe x -> IO ())) tid addr x
+transactionMethods methods generate = transactionMethods' id id methods generate
+
+microsecondsDiff :: Int -> POSIXTime
+microsecondsDiff us = fromIntegral us / 1000000
 
 -- | Like 'transactionMethods' but allows extra information to be stored in the
 -- table of pending transactions.  This also enables multiple 'Client's to
 -- share a single transaction table.
 transactionMethods' ::
-    (MVar x -> a)          -- ^ store MVar into table entry
-    -> (a -> x -> IO void) -- ^ load MVar from table entry
-    -> TableMethods t tid  -- ^ Table methods to lookup values by /tid/.
-    -> (g -> (tid,g))      -- ^ Generate a new unique /tid/ value and update the generator state /g/.
+    ((Maybe x -> IO ()) -> a)    -- ^ store MVar into table entry
+    -> (a -> Maybe x -> IO void) -- ^ load MVar from table entry
+    -> TableMethods t tid        -- ^ Table methods to lookup values by /tid/.
+    -> (g -> (tid,g))            -- ^ Generate a new unique /tid/ value and update the generator state /g/.
     -> TransactionMethods (g,t a) tid addr x
 transactionMethods' store load (TableMethods insert delete lookup) generate = TransactionMethods
     { dispatchCancel = \tid (g,t) -> return (g, delete tid t)
-    , dispatchRegister = \v _ (g,t) ->
+    , dispatchRegister = \getTimeout now v a0 (g,t) -> do
         let (tid,g') = generate g
-            t' = insert tid (store v) t
-        in return ( tid, (g',t') )
+        (a,expiry) <- getTimeout tid a0
+        let t' = insert tid (store v) (now + microsecondsDiff expiry) t
+        return ( (tid,a,expiry), (g',t') )
     , dispatchResponse = \tid x (g,t) ->
         case lookup tid t of
             Just v -> let t' = delete tid t
-                      in return ((g,t'),void $ load v x)
+                      in return ((g,t'),void $ load v $ Just x)
             Nothing -> return ((g,t), return ())
     }
 
@@ -459,8 +461,6 @@ data ErrorReporter addr x meth tid err = ErrorReporter
     , reportMissingHandler :: meth -> addr -> x -> IO ()
       -- | Incoming: unable to identify request.
     , reportUnknown :: addr -> x -> err -> IO ()
-      -- | Outgoing: remote peer is not responding.
-    , reportTimeout :: meth -> tid -> addr -> IO ()
     }
 
 ignoreErrors :: ErrorReporter addr x meth tid err
@@ -468,7 +468,6 @@ ignoreErrors = ErrorReporter
     { reportParseError = \_ -> return ()
     , reportMissingHandler = \_ _ _ -> return ()
     , reportUnknown = \_ _ _ -> return ()
-    , reportTimeout = \_ _ _ -> return ()
     }
 
 logErrors :: ( Show addr
@@ -478,7 +477,6 @@ logErrors = ErrorReporter
     { reportParseError = \err -> dput XMisc err
     , reportMissingHandler = \meth addr x -> dput XMisc $ show addr ++ " --> Missing handler ("++show meth++")"
     , reportUnknown = \addr x err -> dput XMisc $ show addr ++ " --> " ++ err
-    , reportTimeout = \meth tid addr -> dput XMisc $ show addr ++ " --> Timeout ("++show meth++")"
     }
 
 printErrors :: ( Show addr
@@ -488,17 +486,15 @@ printErrors h = ErrorReporter
     { reportParseError = \err -> hPutStrLn h err
     , reportMissingHandler = \meth addr x -> hPutStrLn h $ show addr ++ " --> Missing handler ("++show meth++")"
     , reportUnknown = \addr x err -> hPutStrLn h $ show addr ++ " --> " ++ err
-    , reportTimeout = \meth tid addr -> hPutStrLn h $ show addr ++ " --> Timeout ("++show meth++")"
     }
 
 -- Change the /err/ type for an 'ErrorReporter'.
 instance Contravariant (ErrorReporter addr x meth tid) where
     -- contramap :: (t5 -> t4) -> ErrorReporter t3 t2 t1 t t4 -> ErrorReporter t3 t2 t1 t t5
-    contramap f (ErrorReporter pe mh unk tim)
+    contramap f (ErrorReporter pe mh unk)
         = ErrorReporter (\e -> pe (f e))
                         mh
                         (\addr x e -> unk addr x (f e))
-                        tim
 
 -- | Handle a single inbound packet and then invoke the given continuation.
 -- The 'forkListener' function is implemented by passing this function to 'fix'
@@ -509,7 +505,7 @@ handleMessage ::
     -> addr
     -> x
     -> IO (Maybe (x -> x))
-handleMessage (Client net d err pending whoami responseID _ _) addr plain = do
+handleMessage (Client net d err pending whoami responseID) addr plain = do
     -- Just (Left e)              -> do reportParseError err e
     --                                  return $! Just id
     -- Just (Right (plain, addr)) -> do
@@ -637,19 +633,3 @@ testPairTransport = do
         b = SockAddrInet 2 2
     return ( chanTransport (const bchan) a achan aclosed
            , chanTransport (const achan) b bchan bclosed )
-
-serializeClient :: Client err meth tid addr x -> IO (Client err meth tid addr x)
-serializeClient c = do
-    mvar <- newMVar ()
-    return $ c { clientEnterQuery = \tid -> takeMVar mvar
-               , clientLeaveQuery = \tid didRespond -> putMVar mvar ()
-               }
-
-retardSend :: Int -> Client err meth tid addr x  -> IO (Client err meth tid addr x)
-retardSend micros client = do
-    mvar <- newMVar () :: IO (MVar ())
-    return client { clientEnterQuery = \tid -> do
-                        takeMVar mvar
-                        threadDelay micros
-                        putMVar mvar ()
-                  }