Simplified PacketQueue/PacketBuffer interface.

5 files changed, 305 insertions, 255 deletions

diff --git a/src/DPut.hs b/src/DPut.hs
index e02f8ce6..f3760645 100644
--- a/src/DPut.hs
+++ b/src/DPut.hs
@@ -19,6 +19,7 @@ data DebugTag
   | XLan
   | XMan
   | XNetCrypto
+  | XNetCryptoOut
   | XOnion
   | XRoutes
   | XPing
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)
diff --git a/src/Network/Tox/Crypto/Handlers.hs b/src/Network/Tox/Crypto/Handlers.hs
index 6f440ea5..db7543f7 100644
--- a/src/Network/Tox/Crypto/Handlers.hs
+++ b/src/Network/Tox/Crypto/Handlers.hs
@@ -22,8 +22,7 @@ import qualified Data.ByteString  as B
 import Data.ByteString (ByteString)
 import Control.Lens
 import Data.Function
-import qualified Data.PacketQueue as PQ
-         ;import Data.PacketQueue (PacketQueue)
+import Data.PacketBuffer as PB
 import qualified Data.CyclicBuffer as CB
          ;import Data.CyclicBuffer (CyclicBuffer)
 import Data.Serialize             as S
@@ -354,7 +353,7 @@ data NetCryptoSession = NCrypto
      --   the case in group chats
     , ncView               :: TVar SessionView
     -- ^ contains your nick, status etc
-    , ncPacketQueue        :: PacketQueue CryptoData
+    , ncPacketBuffer       :: PacketBuffer CryptoData (CryptoPacket Encrypted)
     -- ^ a buffer in which incoming packets may be stored out of order
     -- but from which they may be extracted in sequence,
     -- helps ensure lossless packets are processed in order
@@ -383,13 +382,13 @@ data NetCryptoSession = NCrypto
     , ncPingThread         :: TVar (Maybe ThreadId)
     -- ^ thread which actually queues outgoing pings
     , ncIdleEventThread    :: TVar (Maybe ThreadId)
-    , ncOutgoingQueue      :: TVar
-                                (UponHandshake
+    , ncOutgoingQueue      :: TVar (UponHandshake NetCryptoOutQueue)
+                                    {-
                                     (PQ.PacketOutQueue
                                           (State,Nonce24,U.RangeMap TArray Word8 TVar)
                                           CryptoMessage
                                           (CryptoPacket Encrypted)
-                                          CryptoData))
+                                          CryptoData)) -}
     -- ^ To send a message add it to this queue, by calling 'tryAppendQueueOutgoing'
     --   but remember to call 'readyOutGoing' first, because the shared secret cache
     --   presently requires the IO monad.
@@ -643,7 +642,7 @@ freshCryptoSession  sessions
         insertArrayAt outHooks 0 (A.array (0,64) (map assignHook [0..64]))
         return (idmap,lossyEsc,losslessEsc,outHooks)
     ncView0 <- newTVar (sessionView sessions)
-    pktq <- PQ.new (inboundQueueCapacity sessions) 0
+    pktq <- PB.newPacketBuffer
     bufstart <- newTVar 0
     mbpktoq
      <- case mbtheirSessionKey of
@@ -696,7 +695,7 @@ freshCryptoSession  sessions
                     , ncOutgoingIdMapEscapedLossy    = lossyEscapeIdMap
                     , ncOutgoingIdMapEscapedLossless = losslessEscapeIdMap
                     , ncView               = ncView0
-                    , ncPacketQueue        = pktq
+                    , ncPacketBuffer       = pktq
                     , ncStoredRequests     = ncStoredRequests0
                     , ncRequestInterval    = ncRequestInterval0
                     , ncAliveInterval      = ncAliveInterval0
@@ -720,12 +719,25 @@ freshCryptoSession  sessions
             HaveHandshake pktoq -> return (runUponHandshake netCryptoSession0 addr pktoq)
     return (myhandshake,maybeLaunchMissles)
 
+{-
 type NetCryptoOutQueue = PQ.PacketOutQueue (State,Nonce24,U.RangeMap TArray Word8 TVar)
                                            CryptoMessage
                                            (CryptoPacket Encrypted)
                                            CryptoData
+-}
+data NetCryptoOutQueue = NetCryptoOutQueue
+    { nqPacketBuffer :: PacketBuffer CryptoData (CryptoPacket Encrypted)
+    , nqToWire :: STM (State, Nonce24, U.RangeMap TArray Word8 TVar)
+                     -> Word32
+                     -> Word32
+                     -> Word32
+                     -> XMessage
+                     -> STM (Maybe (CryptoPacket Encrypted, Word32))
+    , nqToWireIO :: IO (STM (State, Nonce24, U.RangeMap TArray Word8 TVar))
+    , nqPacketNo :: TVar Word32
+    }
 
-createNetCryptoOutQueue :: NetCryptoSessions -> SecretKey -> PublicKey -> PacketQueue CryptoData
+createNetCryptoOutQueue :: NetCryptoSessions -> SecretKey -> PublicKey -> PacketBuffer CryptoData (CryptoPacket Encrypted)
                         -> TVar Nonce24    -> U.RangeMap TArray Word8 TVar     -> STM (UponHandshake NetCryptoOutQueue)
 createNetCryptoOutQueue sessions newsession theirSessionKey pktq ncMyPacketNonce0 ncOutgoingIdMap0 = do
             let crypto = transportCrypto sessions
@@ -739,8 +751,13 @@ createNetCryptoOutQueue sessions newsession theirSessionKey pktq ncMyPacketNonce
                             ++ "\n toWireIO: my public session key = " ++ show (key2id (toPublic newsession))
                               ) $ writeTVar ncMyPacketNonce0 n24plus1
                         return (return (f n24, n24, ncOutgoingIdMap0))
-            pktoq <- PQ.newOutGoing pktq ncToWire toWireIO 0 (outboundQueueCapacity sessions) 0
-            return (HaveHandshake pktoq)
+            seqnoVar <- newTVar 0
+            return (HaveHandshake NetCryptoOutQueue
+                        { nqPacketBuffer = pktq
+                        , nqToWire       = ncToWire
+                        , nqToWireIO     = toWireIO
+                        , nqPacketNo     = seqnoVar
+                        })
 
 -- | add new session to the lookup maps
 addSessionToMap :: NetCryptoSessions -> SockAddr -> NetCryptoSession -> STM ()
@@ -788,11 +805,39 @@ addSessionToMapIfNotThere sessions addrRaw netCryptoSession = do
             -- in case we're using the same long term key on different IPs ...
             modifyTVar allsessionsByKey (Map.insert remotePublicKey (netCryptoSession:xs))
 
+data OutGoingResult a = OGSuccess a | OGFull | OGEncodeFail
+    deriving (Show)
+
+-- | 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.)
+--
+-- Called from 'runUponHandshake' and 'sendCrypto'.
+--
+-- Whenever this is called, you should also send the resulting packet out on
+-- the network.
+
+tryAppendQueueOutgoing :: STM (State, Nonce24, U.RangeMap TArray Word8 TVar)
+                            -> NetCryptoOutQueue
+                            -> CryptoMessage
+                            -> STM (OutGoingResult (CryptoPacket Encrypted))
+tryAppendQueueOutgoing getExtra outq msg = do
+    pktno <- readTVar (nqPacketNo outq)
+    nextno <- PB.expectingSequenceNumber (nqPacketBuffer outq)
+    be <- PB.nextToSendSequenceNumber (nqPacketBuffer outq)
+    mbWire <- nqToWire outq getExtra nextno be pktno msg
+    case mbWire of
+        Just (payload,seqno) -> do
+            PB.grokPacket (nqPacketBuffer outq) (PacketSent seqno payload)
+            return $ OGSuccess payload
+        Nothing -> return OGEncodeFail
+
+
 runUponHandshake :: NetCryptoSession -> SockAddr -> NetCryptoOutQueue -> IO ()
 runUponHandshake netCryptoSession0 addr pktoq = do
     dput XNetCrypto "(((((((runUponHandshake))))))) Launching threads"
     let sessions = ncAllSessions netCryptoSession0
-        pktq = ncPacketQueue netCryptoSession0
+        pktq = ncPacketBuffer netCryptoSession0
         remotePublicKey = ncTheirPublicKey netCryptoSession0
         crypto = transportCrypto sessions
         allsessions = netCryptoSessions sessions
@@ -805,7 +850,7 @@ runUponHandshake netCryptoSession0 addr pktoq = do
         atomically $ writeTVar (ncDequeueThread netCryptoSession0) (Just tid)
         labelThread tid ("NetCryptoDequeue." ++ show (key2id remotePublicKey) ++ sidStr)
         fix $ \loop -> do
-            cd <- atomically $ PQ.dequeue pktq
+            cd <- atomically $ PB.awaitReadyPacket pktq
             if msgID (bufferData cd) == PacketRequest
               then do
                 dput XNetCrypto $ "Dequeued::PacketRequest seqno=" ++ show (bufferStart cd) ++ " " ++ show (bufferData cd)
@@ -824,19 +869,21 @@ runUponHandshake netCryptoSession0 addr pktoq = do
         labelThread tid ("NetCryptoRequest." ++ show (key2id remotePublicKey) ++ sidStr)
         fix $ \loop -> do
             atomically (readTVar (ncRequestInterval netCryptoSession0)) >>= threadDelay . (* 1000)
-            nums <- atomically $ PQ.getMissing pktq
+            (nums,seqno) <- atomically $ PB.packetNumbersToRequest pktq
             dput XNetCrypto $ "(Request Thread) Missing Packets detected:" ++ show nums
-            getOutGoingParam <- PQ.readyOutGoing pktoq
-            atomically $ do 
-                seqno <- PQ.getLastDequeuedPlus1 pktq
-                ogresult <- PQ.tryAppendQueueOutgoing getOutGoingParam pktoq (createRequestPacket seqno nums)
+            getOutGoingParam <- nqToWireIO pktoq
+            x <- atomically $ do
+                ogresult <- tryAppendQueueOutgoing getOutGoingParam pktoq (createRequestPacket seqno nums)
                 case ogresult of
-                    PQ.OGSuccess _ -> return ()
+                    OGSuccess x -> return x
                     _ -> retry
+            sendSessionPacket sessions addr x
             loop
     dput XNetCrypto $ "runUponHandshake: " ++ show threadid ++ " = NetCryptoRequest." ++ show (key2id remotePublicKey) ++ sidStr
 
     -- launch dequeueOutgoing thread
+    {-
+    -- TODO
     threadidOutgoing <- forkIO $ do
         tid <- myThreadId
         atomically $ writeTVar (ncDequeueOutGoingThread netCryptoSession0) (Just tid)
@@ -846,7 +893,8 @@ runUponHandshake netCryptoSession0 addr pktoq = do
             dput XNetCrypto "NetCryptoDequeueOutgoing thread... Sending encrypted Packet"
             sendSessionPacket sessions addr pkt
             loop
-    dput XNetCrypto $ "runUponHandshake: " ++ show threadidOutgoing ++ " = NetCryptoDequeueOutgoing." ++ show (key2id remotePublicKey) ++ sidStr
+    -}
+    dput XNetCrypto $ "runUponHandshake: " ++ show "threadidOutgoing" ++ " = NetCryptoDequeueOutgoing." ++ show (key2id remotePublicKey) ++ sidStr
 
     -- launch ping Machine thread
     pingMachine <- forkPingMachineDynamic ("NetCrypto." ++ show (key2id remotePublicKey) ++ sidStr) (ncIdleEvent netCryptoSession0) (ncTimeOut netCryptoSession0)
@@ -859,10 +907,10 @@ runUponHandshake netCryptoSession0 addr pktoq = do
         labelThread tid ("NetCryptoPingSender." ++ show (key2id remotePublicKey) ++ sidStr)
         fix $ \loop -> do
             atomically (readTVar (ncAliveInterval netCryptoSession0)) >>= threadDelay . (* 1000)
-            dput XNetCrypto $ "pingThread (session: " ++ show (ncSessionId netCryptoSession0) ++ ") Sending Alive(PING) Packet"
+            dput XNetCryptoOut $ "pingThread (session: " ++ show (ncSessionId netCryptoSession0) ++ ") Sending Alive(PING) Packet"
             lr <- sendPing crypto netCryptoSession0
             case lr of
-                Left s -> dput XNetCrypto $ "(pingThread session: " ++ show (ncSessionId netCryptoSession0) ++ ") " ++ s
+                Left s -> dput XNetCryptoOut $ "(pingThread session: " ++ show (ncSessionId netCryptoSession0) ++ ") " ++ s
                 Right _ -> return ()
             loop
 
@@ -972,7 +1020,7 @@ updateCryptoSession sessions addr newsession timestamp hp session handshake = do
                                                 sessions
                                                 newsession
                                                 theirSessionPublic
-                                                (ncPacketQueue session)
+                                                (ncPacketBuffer session)
                                                 (ncMyPacketNonce session)
                                                 (ncOutgoingIdMap session)
                         writeTVar (ncOutgoingQueue session) mbpktoq
@@ -1055,7 +1103,7 @@ sessionPacketH sessions addrRaw (CryptoPacket nonce16 encrypted) = do
         Nothing -> do
             dput XNetCrypto "Dropping packet.. no session"
             return Nothing -- drop packet, we have no session
-        Just session@(NCrypto { ncIncomingTypeArray, ncState, ncPacketQueue, ncHooks,
+        Just session@(NCrypto { ncIncomingTypeArray, ncState, ncPacketBuffer, ncHooks,
                                 ncSessionSecret, ncTheirSessionPublic, ncTheirBaseNonce,
                                 ncPingMachine, ncSessionId, ncStoredRequests}) -> do
          -- Unrecognized packets, try them thrice so as to give
@@ -1130,9 +1178,10 @@ sessionPacketH sessions addrRaw (CryptoPacket nonce16 encrypted) = do
                         isLossy (GrpMsg KnownLossy _) = True
                         isLossy (Msg mid) | lossyness mid == Lossy = True
                         isLossy _ = False
+                        ack = bufferStart -- Earliest sequence number they've seen from us.
                     if isLossy msgTypMapped
                         then do dput XNetCrypto $ "enqueue ncPacketQueue Lossy " ++ show cm
-                                atomically $ PQ.observeOutOfBand ncPacketQueue bufferEnd
+                                atomically $ PB.grokPacket ncPacketBuffer (PacketReceivedLossy bufferEnd ack)
                                 runCryptoHook session (bufferData cd)
                         else do dput XNetCrypto $ "enqueue ncPacketQueue Lossless " ++ show cm
                                 when (msgID cm == PING) $
@@ -1143,7 +1192,7 @@ sessionPacketH sessions addrRaw (CryptoPacket nonce16 encrypted) = do
                                   --    num <- CB.getNextSequenceNum ncStoredRequests
                                   --    CB.enqueue ncStoredRequests num cd
                                      handlePacketRequest session cd
-                                atomically $ PQ.enqueue ncPacketQueue bufferEnd cd
+                                atomically $ PB.grokPacket ncPacketBuffer (PacketReceived bufferEnd cd ack)
                                 return Nothing
     where
         last2Bytes :: Nonce24 -> Word16
@@ -1262,14 +1311,18 @@ sendCrypto crypto session updateLocal cm = do
     HaveHandshake outq <- atomically $ readTVar (ncOutgoingQueue session)
     -- XXX: potential race? if shared secret comes out of sync with cache?
     dput XNetCrypto "sendCrypto: enter "
-    getOutGoingParam <- PQ.readyOutGoing outq
+    getOutGoingParam <- nqToWireIO outq
     dput XNetCrypto "sendCrypto: got the io extra stuff"
-    atomically $ do
-        result <- PQ.tryAppendQueueOutgoing getOutGoingParam outq cm
+    r <- atomically $ do
+        result <- tryAppendQueueOutgoing getOutGoingParam outq cm
         case result of
-            PQ.OGSuccess x  -> updateLocal >> return (Right x)
-            PQ.OGFull       -> return (Left "Outgoing packet buffer is full")
-            PQ.OGEncodeFail -> return (Left "Failed to encode outgoing packet")
+            OGSuccess x  -> updateLocal >> return (Right x)
+            OGFull       -> return (Left "Outgoing packet buffer is full")
+            OGEncodeFail -> return (Left "Failed to encode outgoing packet")
+    case ncSockAddr session of
+        HaveDHTKey saddr -> mapM_ (sendSessionPacket (ncAllSessions session) saddr) r
+        _                -> return ()
+    return r
 
 sendPing :: TransportCrypto -> NetCryptoSession -> IO (Either String (CryptoPacket Encrypted))
 sendPing crypto session = do
@@ -1327,8 +1380,16 @@ sendCryptoLossy crypto session updateLocal cm = do
             updateLocal
             return (Left errmsg)
         HaveHandshake outq -> do
-            getOutGoingParam <- PQ.readyOutGoing outq
-            mbPkt <- atomically $ PQ.peekPacket getOutGoingParam outq cm
+            getOutGoingParam <- nqToWireIO outq
+            mbPkt <- atomically $ do
+                    pktno <- readTVar (nqPacketNo outq)
+                    nextno <- PB.expectingSequenceNumber (nqPacketBuffer outq)
+                    be <- PB.nextToSendSequenceNumber (nqPacketBuffer outq)
+                    nqToWire outq getOutGoingParam -- See 'ncToWire'
+                        nextno -- packet number we expect to recieve
+                        be     -- buffer_end  (for lossy)
+                        pktno  -- packet number (for lossless)
+                        cm
             case mbPkt of
                 Nothing  -> do
                     let errmsg = "Error sending lossy packet! (sessionid: " ++ show (ncSessionId session) ++ ") " ++ show cm
@@ -1466,11 +1527,10 @@ handlePacketRequest session (CryptoData { bufferStart=num
     mbOutQ <- atomically $ readTVar (ncOutgoingQueue session)
     case mbOutQ of
         HaveHandshake pktoq -> do
-            getOutGoingParam <-PQ.readyOutGoing pktoq
-            ps <- atomically $ PQ.getRequested getOutGoingParam pktoq num bs
-            let resend (Just (n,pkt)) = sendSessionPacket (ncAllSessions session) addr pkt
-                resend _ = return ()
-            dput XNetCrypto ("Re-Sending Packets: " ++ show (map fst $ catMaybes ps))
+            getOutGoingParam <-nqToWireIO pktoq
+            ps <- atomically $ PB.retrieveForResend (nqPacketBuffer pktoq) $ PB.decompressSequenceNumbers num bs
+            let resend (n,pkt) = sendSessionPacket (ncAllSessions session) addr pkt
+            dput XNetCrypto ("Re-Sending Packets: " ++ show (map fst ps))
             mapM_ resend ps
         _ -> return ()
 
diff --git a/src/Network/Tox/Crypto/Transport.hs b/src/Network/Tox/Crypto/Transport.hs
index a453a13c..8bb41254 100644
--- a/src/Network/Tox/Crypto/Transport.hs
+++ b/src/Network/Tox/Crypto/Transport.hs
@@ -82,12 +82,13 @@ import Data.Text.Encoding as T
 import Data.Serialize as S
 import Control.Arrow
 import DPut
-import Data.PacketQueue (toPNums)
+import Data.PacketBuffer as PB
 import Data.Function
 
 showCryptoMsg :: Word32 -> CryptoMessage -> [Char]
-showCryptoMsg seqno (UpToN PacketRequest bytes) =  "UpToN PacketRequest --> "  ++ show (toPNums seqno $ B.unpack bytes)
-showCryptoMsg _ msg = show msg
+showCryptoMsg seqno (UpToN PacketRequest bytes) = "UpToN PacketRequest --> "
+                                                  ++ show (PB.decompressSequenceNumbers seqno $ B.unpack bytes)
+showCryptoMsg _     msg                         = show msg
 
 parseCrypto :: (ByteString, SockAddr) -> Either (CryptoPacket Encrypted, SockAddr) (ByteString, SockAddr)
 parseCrypto ((B.uncons -> Just (0x1b,pkt)),saddr) = either (\_ -> Right (pkt,saddr))