Simplified PacketQueue/PacketBuffer interface.

2 files changed, 201 insertions, 213 deletions

diff --git a/src/Data/PacketBuffer.hs b/src/Data/PacketBuffer.hs
new file mode 100644
index 00000000..c5ede50d
--- /dev/null
+++ b/src/Data/PacketBuffer.hs
@@ -0,0 +1,102 @@
+{-# LANGUAGE TupleSections #-}
+module Data.PacketBuffer
+    ( PacketBuffer
+    , newPacketBuffer
+    , PacketEvent(..)
+    , grokPacket
+    , awaitReadyPacket
+    , packetNumbersToRequest
+    , expectingSequenceNumber
+    , nextToSendSequenceNumber
+    , retrieveForResend
+    , decompressSequenceNumbers
+    ) where
+
+import Data.PacketQueue as Q
+
+import Control.Concurrent.STM
+import Control.Monad
+import Data.Maybe
+import Data.Word
+
+data PacketBuffer a b = PacketBuffer
+    { inQueue   :: PacketQueue a
+    , outBuffer :: PacketQueue b
+    }
+
+-- | Initialize the packet buffers.  Note, the capacity of the inbound packet
+-- queue is currently hardcoded to 200 packets and the capacity of the outbound
+-- buffer is hardcoded to 400 packets.
+newPacketBuffer :: STM (PacketBuffer a b)
+newPacketBuffer = PacketBuffer <$> Q.new 200 0
+                               <*> Q.new 400 0
+
+-- | Input for 'grokPacket'.
+data PacketEvent a b
+    = PacketSent { peSeqNum      :: Word32 -- ^ Sequence number for payload.
+                 , peSentPayload :: b      -- ^ Payload packet we sent to them.
+                 }
+    | PacketReceived { peSeqNum          :: Word32 -- ^ Sequence number for payload.
+                     , peReceivedPayload :: a      -- ^ Payload packet they sent to us.
+                     , peAcknowledgedNum :: Word32 -- ^ Earliest sequence number they've seen from us.
+                     }
+    | PacketReceivedLossy { peSeqNum          :: Word32 -- ^ Sequence number for lossy packet.
+                          , peAcknowledgedNum :: Word32 -- ^ Earliest sequence number they've seen from us.
+                          }
+
+-- | Whenever a packet is received or sent (but not resent) from the network,
+-- this function should be called to update the relevant buffers.
+--
+-- On outgoing packets, if the outbound buffer is full, this will block
+-- indefinitely until it is called in another thread with an inbound
+-- acknowledgement.
+grokPacket :: PacketBuffer a b -> PacketEvent a b -> STM ()
+grokPacket (PacketBuffer _   outb) (PacketSent seqno a)
+    = do (n,_) <- Q.enqueue outb seqno a
+         when (n/=0) retry
+grokPacket (PacketBuffer inb outb) (PacketReceived seqno a ack)
+    = do Q.enqueue inb seqno a
+         Q.dropPacketsBefore outb ack
+grokPacket (PacketBuffer inb outb) (PacketReceivedLossy seqno ack)
+    = do Q.observeOutOfBand inb seqno
+         Q.dropPacketsBefore outb ack
+
+-- | Wait until an inbound packet is ready to process.  Any necessary
+-- re-ordering will have been done.
+awaitReadyPacket :: PacketBuffer a b -> STM a
+awaitReadyPacket (PacketBuffer inb _) = Q.dequeue inb
+
+-- | Obtain a list of sequence numbers that may have been dropped.  This would
+-- be any number not yet received that is prior to the maxium sequence number
+-- we have received.  For convenience, a lowerbound for the missing squence numbers
+-- is also returned as the second item of the pair.
+packetNumbersToRequest :: PacketBuffer a b -> STM ([Word32],Word32)
+packetNumbersToRequest (PacketBuffer inb _) = do
+    ns <- Q.getMissing inb
+    lb <- Q.getLastDequeuedPlus1 inb
+    return (ns,lb)
+
+expectingSequenceNumber :: PacketBuffer a b -> STM Word32
+expectingSequenceNumber (PacketBuffer inb _ ) = Q.getLastDequeuedPlus1 inb
+
+nextToSendSequenceNumber :: PacketBuffer a b -> STM Word32
+nextToSendSequenceNumber (PacketBuffer _ outb) = Q.getLastEnqueuedPlus1 outb
+
+-- | Retrieve already-sent packets by their sequence numbers.  See
+-- 'decompressSequenceNumbers' to obtain the sequence number list from a
+-- compressed encoding.  There is no need to call 'grokPacket' when sending the
+-- packets returned from this call.
+retrieveForResend :: PacketBuffer a b -> [Word32] -> STM [(Word32,b)]
+retrieveForResend (PacketBuffer _ outb) seqnos =
+    catMaybes <$> forM seqnos (\no -> fmap (no,) <$> Q.lookup outb no)
+
+-- | Expand a compressed set of sequence numbers.  The first squence number is
+-- given directly and the rest are computed using 8-bit offsets.  This is
+-- normally used to obtain input for 'retrieveForResend'.
+decompressSequenceNumbers :: Word32 -> [Word8] -> [Word32]
+decompressSequenceNumbers baseno ns = foldr doOne (const []) ns (baseno-1)
+    where
+        doOne :: Word8 -> (Word32 -> [Word32]) -> Word32 -> [Word32]
+        doOne 0 f addend = f (addend+255)
+        doOne x f addend = let y = fromIntegral x + addend
+                           in y : f y
diff --git a/src/Data/PacketQueue.hs b/src/Data/PacketQueue.hs
index 6f997ac0..c5d5ac09 100644
--- a/src/Data/PacketQueue.hs
+++ b/src/Data/PacketQueue.hs
@@ -3,30 +3,22 @@
 -- be stored out of order, but from which they are extracted in the proper
 -- sequence.
 {-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE FlexibleContexts #-}
 module Data.PacketQueue
     ( PacketQueue
     , getCapacity
     , getLastDequeuedPlus1
+    , getLastEnqueuedPlus1
     , new
     , dequeue
+    , dropPacketsBefore
     , getMissing
-    , dequeueOrGetMissing
-    , markButNotDequeue
+    -- , dequeueOrGetMissing
+    -- , markButNotDequeue
     , enqueue
     , observeOutOfBand
-    , PacketOutQueue
     , packetQueueViewList
-    , newOutGoing
-    , readyOutGoing
-    , toPNums
-    , getRequested
-    , peekPacket
-    , tryAppendQueueOutgoing
-    , dequeueOutgoing
-    , getHighestHandledPacketPlus1
-    , mapOutGoing
-    , OutGoingResult(..)
+    -- , mapQ
+    , Data.PacketQueue.lookup
     ) where
 
 import Control.Concurrent.STM
@@ -34,24 +26,35 @@ import Control.Monad
 import Data.Word
 import Data.Array.MArray
 import Data.Maybe
-import DPut
 
 data PacketQueue a = PacketQueue
     { pktq    :: TArray Word32 (Maybe a)
     , seqno   :: TVar Word32 -- (buffer_start)
     , qsize   :: Word32
     , buffend :: TVar Word32 -- on incoming, next packet they'll send + 1
+                             -- i.e. one more than the largest seen sequence number.
     }
 
+-- | Obtain a list of non-empty slots in the 'PacketQueue'.  The numeric value
+-- is an index into the underlying array, not a sequence number.
 packetQueueViewList :: PacketQueue a -> STM [(Word32,a)]
 packetQueueViewList p = do
     let f (n,Nothing) = Nothing
         f (n,Just x)  = Just (n,x)
     catMaybes . map f <$> getAssocs (pktq p)
 
+-- | This returns the earliest sequence number with a slot in the queue.
 getLastDequeuedPlus1 :: PacketQueue a -> STM Word32
 getLastDequeuedPlus1 PacketQueue {seqno} = readTVar seqno
 
+-- | This returns the least upper bound of sequence numbers that have been
+-- enqueued.
+getLastEnqueuedPlus1 :: PacketQueue a -> STM Word32
+getLastEnqueuedPlus1 PacketQueue {buffend} = readTVar buffend
+
+
+-- | This is the number of consequetive sequence numbers, starting at
+-- 'getLastDequeuedPlus1' that can be stored in the queue
 getCapacity :: Applicative m => PacketQueue t -> m Word32
 getCapacity (PacketQueue { qsize }) = pure qsize
 
@@ -63,7 +66,7 @@ new capacity seqstart = do
     let cap = if capacity `mod` 2 == 0 then capacity else capacity + 1
     q <- newArray (0,cap - 1) Nothing
     seqv <- newTVar seqstart
-    bufe <- newTVar 0
+    bufe <- newTVar seqstart
     return PacketQueue
         { pktq    = q
         , seqno   = seqv
@@ -72,8 +75,13 @@ new capacity seqstart = do
         }
 
 -- | Update the packet queue given:
+--
 --      * packet queue
+--
 --      * the number of next lossless packet they intend to send you
+--
+-- This behaves exactly like 'enqueue' except that no packet data is written to
+-- the queue.
 observeOutOfBand :: PacketQueue a -> Word32-> STM ()
 observeOutOfBand PacketQueue { seqno, qsize, buffend } numberOfNextLosslessPacketThatTheyWillSend = do
     low <- readTVar seqno
@@ -85,37 +93,35 @@ observeOutOfBand PacketQueue { seqno, qsize, buffend } numberOfNextLosslessPacke
 -- | If seqno < buffend then return expected packet numbers for all
 --   the Nothings in the array between them.
 --   Otherwise, return empty list.
-getMissing :: Show a => PacketQueue a -> STM [Word32]
+getMissing :: PacketQueue a -> STM [Word32]
 getMissing PacketQueue { pktq, seqno, qsize, buffend } = do
     seqno0 <- readTVar seqno
     buffend0 <- readTVar buffend
     -- note relying on fact that [ b .. a ] is null when a < b
     let indices = take (fromIntegral qsize) $ map (`mod` qsize) [ seqno0 .. buffend0 - 1]
-    -- tput XNetCrypto $ "(netCRYPTO getMissing indices: " ++ show indices
     maybes <- mapM (\i -> do {x <- readArray pktq i; return (i,x)}) indices
-    tput XNetCrypto $ "(netCRYPTO) getMissing: (" ++ show seqno0 ++ " " ++ show buffend0 ++") => " ++ show maybes
     let nums = map fst . filter (isNothing . snd) $ maybes
     return nums
 
--- | If seqno < buffend then return expected packet numbers for all
---   the Nothings in the array between them.
---   Otherwise, behave as 'dequeue' would.
---   TODO: Do we need this function? Delete it if not.
-dequeueOrGetMissing :: PacketQueue a -> STM (Either [Word32] a)
-dequeueOrGetMissing PacketQueue { pktq, seqno, qsize, buffend } = do
-    seqno0 <- readTVar seqno
-    buffend0 <- readTVar buffend
-    if seqno0 < buffend0
-      then do
-        maybes <- mapM (readArray pktq) (take (fromIntegral qsize) $ map (`mod` qsize) [ seqno0 .. buffend0 ])
-        let nums = map fst . filter (isNothing . snd) $ zip [buffend0 ..]  maybes
-        return (Left nums)
-      else do
-        let i = seqno0 `mod` qsize
-        x <- maybe retry return =<< readArray pktq i
-        writeArray pktq i Nothing
-        modifyTVar' seqno   succ
-        return (Right x)
+-- -- | If seqno < buffend then return expected packet numbers for all
+-- --   the Nothings in the array between them.
+-- --   Otherwise, behave as 'dequeue' would.
+-- --   TODO: Do we need this function? Delete it if not.
+-- dequeueOrGetMissing :: PacketQueue a -> STM (Either [Word32] a)
+-- dequeueOrGetMissing PacketQueue { pktq, seqno, qsize, buffend } = do
+--     seqno0 <- readTVar seqno
+--     buffend0 <- readTVar buffend
+--     if seqno0 < buffend0
+--       then do
+--         maybes <- mapM (readArray pktq) (take (fromIntegral qsize) $ map (`mod` qsize) [ seqno0 .. buffend0 ])
+--         let nums = map fst . filter (isNothing . snd) $ zip [buffend0 ..]  maybes
+--         return (Left nums)
+--       else do
+--         let i = seqno0 `mod` qsize
+--         x <- maybe retry return =<< readArray pktq i
+--         writeArray pktq i Nothing
+--         modifyTVar' seqno   succ
+--         return (Right x)
 
 -- | Retry until the next expected packet is enqueued.  Then return it.
 dequeue :: PacketQueue a -> STM a
@@ -127,20 +133,46 @@ dequeue PacketQueue { pktq, seqno, qsize } = do
     modifyTVar' seqno   succ
     return x
 
--- | Like dequeue, but marks as viewed rather than removing
-markButNotDequeue :: PacketQueue (Bool,a) -> STM a
-markButNotDequeue PacketQueue { pktq, seqno, qsize } = do
-    i0 <- readTVar seqno
-    let i = i0 `mod` qsize
-    (b,x) <- maybe retry return =<< readArray pktq i
-    writeArray pktq i (Just (True,x))
-    modifyTVar' seqno   succ
-    return x
+-- | Drop all packets preceding the given packet number.
+dropPacketsBefore :: PacketQueue a -> Word32 -> STM ()
+dropPacketsBefore PacketQueue{ pktq, seqno, qsize, buffend } no0 = do
+    low <- readTVar seqno
+    let no = no0 - 1 -- possibly negative
+        proj = no - low -- possibly negative
+    if (proj < qsize)
+        then do
+            -- Clear some, but not all, slots.
+            let ilow = low  `mod` qsize
+                i = no `mod` qsize
+                ranges = if ilow <= i then [[ilow .. i]]
+                                      else [[0 .. i],[ilow .. qsize-1]]
+            writeTVar seqno no
+            forM_ ranges $ mapM_ $ \i -> writeArray pktq i Nothing
+        else do
+            -- Reset to empty queue.
+            writeTVar seqno no
+            writeTVar buffend no
+            (z,n) <- getBounds pktq
+            forM_ [z .. n] $ \i -> writeArray pktq i Nothing
+
+-- -- | Like dequeue, but marks as viewed rather than removing
+-- markButNotDequeue :: PacketQueue (Bool,a) -> STM a
+-- markButNotDequeue PacketQueue { pktq, seqno, qsize } = do
+--     i0 <- readTVar seqno
+--     let i = i0 `mod` qsize
+--     (b,x) <- maybe retry return =<< readArray pktq i
+--     writeArray pktq i (Just (True,x))
+--     modifyTVar' seqno   succ
+--     return x
 
 -- | Enqueue a packet.  Packets need not be enqueued in order as long as there
 -- is spare capacity in the queue.  If there is not, the packet will be
--- silently discarded without blocking. (Unless this is an Overwrite-queue,
--- in which case, the packets will simply wrap around overwriting the old ones.)
+-- silently discarded without blocking. (Unless this is an Overwrite-queue, in
+-- which case, the packets will simply wrap around overwriting the old ones.)
+--
+-- If the packet was enqueued, (0,i) will be retuned where /i/ is the index at
+-- which the new packet was stored in the buffer.  If the queue was full, the
+-- first of the returned pair will be non-zero.
 enqueue :: PacketQueue a -- ^ The packet queue.
         -> Word32        -- ^ Sequence number of the packet.
         -> a             -- ^ The packet.
@@ -155,167 +187,21 @@ enqueue PacketQueue{ pktq, seqno, qsize, buffend} no x = do
         modifyTVar' buffend (\be -> if be - low <= proj then no + 1 else be)
     return (proj `divMod` qsize)
 
--- lookup :: PacketQueue a -> Word32 -> STM (Maybe a)
--- lookup PacketQueue{ pktq, seqno, qsize } no = _todo
-
------------------------------------------------------
--- * PacketOutQueue
---
-
-data PacketOutQueue extra msg toWire fromWire = PacketOutQueue
-    { pktoInPQ   :: PacketQueue fromWire -- ^ reference to the incoming 'PacketQueue'
-    , pktoOutPQ  :: PacketQueue (Word32,toWire)
-    , pktoPacketNo :: TVar Word32
-    , pktoToWireIO :: IO (STM extra)
-    , pktoToWire :: STM extra
-                 -> Word32{-packet number we expect to recieve-}
-                 -> Word32{- buffer_end -}
-                 -> Word32{- packet number -}
-                 -> msg
-                 -> STM (Maybe (toWire,Word32{-next packet no-}))
-    }
-
-mapOutGoing :: ((Word32,towire) -> Maybe (Word32,towire)) -> PacketOutQueue extra msg towire fromwire -> STM ()
-mapOutGoing f q@(PacketOutQueue { pktoOutPQ=PacketQueue{ pktq } }) = do
-    (z,n) <- getBounds pktq
-    let ff i = do
-        e <- readArray pktq i
-        writeArray pktq i (e>>=f)
-    mapM_ ff [z .. n]
-
-newOutGoing :: PacketQueue fromwire
-                      -- ^ Incoming queue
-            -> (STM io -> Word32 {-packet number we expect to recieve-} -> Word32{-buffer_end-} -> Word32{-packet number-} -> msg -> STM (Maybe (wire,Word32{-next packet no-})))
-                      -- ^ toWire callback
-            -> IO (STM io)
-                      -- ^ io action to get extra parameter
-            -> Word32 -- ^ packet number of first outgoing packet
-            -> Word32 -- ^ Capacity of queue.
-            -> Word32 -- ^ Initial sequence number.
-            -> STM (PacketOutQueue io msg wire fromwire)
-newOutGoing inq towire toWireIO num capacity seqstart = do
-    outq <- new capacity seqstart
-    numVar <- newTVar num
-    return $ PacketOutQueue
-                { pktoInPQ = inq
-                , pktoOutPQ = outq
-                , pktoPacketNo = numVar
-                , pktoToWireIO = toWireIO
-                , pktoToWire = towire
-                }
-
-data OutGoingResult a = OGSuccess a | OGFull | OGEncodeFail
-    deriving (Show)
-
-instance Eq (OutGoingResult a) where
-    OGSuccess _ == OGSuccess _   = True
-    OGFull == OGFull             = True
-    OGEncodeFail == OGEncodeFail = True
-    _ == _ = False
-
-
--- | do something in IO before appending to the queue
-readyOutGoing :: PacketOutQueue extra msg wire fromwire -> IO (STM extra)
-readyOutGoing (PacketOutQueue {pktoToWireIO }) = pktoToWireIO
-
-getRequested :: Show wire => STM extra -> PacketOutQueue extra msg wire fromwire -> Word32 -> [Word8] -> STM [Maybe (Word32,wire)]
-getRequested _        _     _    [] = return []
-getRequested getExtra pktoq snum ns = do
-    let pnums = toPNums snum ns
-        indices = map toIndex pnums
-    tput XNetCrypto $ "getRequested: snum = " ++ show snum
-    tput XNetCrypto $ "getRequested: pnums = " ++ show pnums ++ " indices = " ++ show indices
-    xs <- packetQueueViewList (pktoOutPQ pktoq)
-    tput XNetCrypto $ "getRequested viewList -> "
-    mapM_ (tput XNetCrypto . show) xs
-    forM indices $ \i -> readArray (pktq $ pktoOutPQ pktoq) i
-  where
-    toIndex :: Word32 -> Word32
-    toIndex = (`mod` qsize (pktoOutPQ pktoq))
-
-toPNums :: Word32 -> [Word8] -> [Word32]
-toPNums snum ns = reverse . snd $ foldl doOne (snum-1,[]) ns
-    where
-        doOne :: (Word32,[Word32]) -> Word8 -> (Word32,[Word32])
-        doOne (addend,as) 0 = (addend+255,as)
-        doOne (addend,as) x = let y = fromIntegral x + addend
-                                  in (y,y:as)
-
--- | Makes no modifications to the PacketOutQueue state, useful for lossy packets
-peekPacket :: STM extra -> PacketOutQueue extra msg wire fromwire -> msg -> STM (Maybe (wire,Word32))
-peekPacket getExtra q@(PacketOutQueue { pktoInPQ, pktoOutPQ, pktoPacketNo, pktoToWireIO, pktoToWire }) msg
- = do
-    be <- readTVar (buffend pktoOutPQ)
-    let i = be `mod` (qsize pktoOutPQ)
-    let arrayEmpty :: MArray a e m => a Word32 e -> m Bool
-        arrayEmpty ar = do (lowB,highB) <- getBounds ar
-                           let result= lowB > highB
-                           tput XNetCrypto
-                                        ("arrayEmpty result=" ++ show result
-                                        ++ " lowB="    ++ show lowB
-                                        ++ " highB = " ++ show highB
-                                        ++ " i = " ++ show i)
-                           return result
-    mbPkt <- do emp <- arrayEmpty (pktq pktoOutPQ)
-                if emp then tput XNetCrypto "(peekPacket empty)" >> return Nothing
-                       else do  tput XNetCrypto "(peekPacket nonempty)"
-                                result <- readArray (pktq pktoOutPQ) i
-                                tput XNetCrypto ("readArray (isJust result)=="  ++ show (isJust result))
-                                return result
-    pktno <- readTVar pktoPacketNo
-    nextno <- readTVar (seqno pktoInPQ)
-    pktoToWire getExtra nextno be pktno msg
-
--- | Convert a message to packet format and append it to the front of a queue
---   used for outgoing messages. (Note that ‘front‛ usually means the higher
---   index in this implementation.)
-tryAppendQueueOutgoing :: STM extra -> PacketOutQueue extra msg wire fromwire -> msg -> STM (OutGoingResult wire)
-tryAppendQueueOutgoing getExtra q@(PacketOutQueue { pktoInPQ, pktoOutPQ, pktoPacketNo, pktoToWireIO, pktoToWire }) msg
-  = do
-    be <- readTVar (buffend pktoOutPQ)
-    let i = be `mod` (qsize pktoOutPQ)
-    let arrayEmpty :: MArray a e m => a Word32 e -> m Bool
-        arrayEmpty ar = do (lowB,highB) <- getBounds ar
-                           return $ lowB > highB
-    mbPkt <- do emp <- arrayEmpty (pktq pktoOutPQ)
-                if emp then return Nothing
-                       else do  readArray (pktq pktoOutPQ) i
-    pktno <- readTVar pktoPacketNo
-    nextno <- readTVar (seqno pktoInPQ)
-    mbWire <- pktoToWire getExtra nextno be pktno msg
-    -- TODO all the above lines ^^ can be replaced with call to peekPacket
-    case mbWire of
-        Just (pkt,pktno')
-            -> case mbPkt of
-                    -- slot is free, insert element
-                    Nothing -> do
-                        modifyTVar' (buffend pktoOutPQ) (+1)
-                        writeTVar pktoPacketNo $! pktno'
-                        writeArray (pktq pktoOutPQ) i (Just (pktno,pkt))
-                        return (OGSuccess pkt)
-                    -- queue is full
-                    Just (n,_) -> do
-                        nn <- getHighestHandledPacketPlus1 q
-                        if (n < nn)
-                          -- but we can overwrite an old packet
-                          then do
-                            modifyTVar' (buffend pktoOutPQ) (+1)
-                            writeTVar pktoPacketNo $! pktno'
-                            writeArray (pktq pktoOutPQ) i (Just (pktno,pkt))
-                            return (OGSuccess pkt)
-                          --  uh oh this packet is still needed...
-                          else return OGFull
-        -- don't know how to send this message
-        Nothing -> return OGEncodeFail
-
-dequeueOutgoing :: PacketOutQueue extra msg wire fromwire -> STM (Word32,wire)
-dequeueOutgoing (PacketOutQueue {pktoOutPQ=PacketQueue { pktq, seqno, qsize }}) = do
-    i0 <- readTVar seqno
-    let i = i0 `mod` qsize
-    x <- maybe retry return =<< readArray pktq i
-    -- writeArray pktq i Nothing -- not cleaning
-    modifyTVar' seqno   succ
-    return x
-
-getHighestHandledPacketPlus1 :: PacketOutQueue extra msg wire fromwire -> STM Word32
-getHighestHandledPacketPlus1 (PacketOutQueue { pktoInPQ }) = readTVar (buffend pktoInPQ)
+-- | Obtain the packet with the given sequence number if it is stored in the
+-- queue, otherwise /Nothing/ is returned without blocking.
+lookup :: PacketQueue a -> Word32 -> STM (Maybe a)
+lookup PacketQueue{ pktq, seqno, qsize } no = do
+    low <- readTVar seqno
+    let proj = no - low
+    if proj < qsize
+        then let i = no `mod` qsize
+             in readArray pktq i
+        else return Nothing
+
+-- -- | For each item in the queue, modify or delete it.
+-- mapQ :: (a -> Maybe a) -> PacketQueue a -> STM ()
+-- mapQ f PacketQueue{pktq} = do
+--     (z,n) <- getBounds pktq
+--     forM_ [z .. n] $ \i -> do
+--         e <- readArray pktq i
+--         writeArray pktq i (e>>=f)