ncLastNMsgs is now CyclicBuffer type

4 files changed, 30 insertions, 28 deletions

diff --git a/dht-client.cabal b/dht-client.cabal
index 9e3c78ae..fa49aff1 100644
--- a/dht-client.cabal
+++ b/dht-client.cabal
@@ -90,6 +90,7 @@ library
                     Network.BitTorrent.MainlineDHT.Symbols
                     System.Global6
                     Data.PacketQueue
+                    Data.CyclicBuffer
                     Data.Word64Map
                     Data.Word64RangeMap
                     OnionRouter
diff --git a/examples/dhtd.hs b/examples/dhtd.hs
index 4b79b132..ce1e1b16 100644
--- a/examples/dhtd.hs
+++ b/examples/dhtd.hs
@@ -108,6 +108,7 @@ import Connection
 import ToxToXMPP
 import qualified Connection.Tcp as Tcp (ConnectionEvent(..))
 import Control.Concurrent.Supply
+import qualified Data.CyclicBuffer as CB
 
 
 showReport :: [(String,String)] -> String
@@ -708,14 +709,11 @@ clientSession s@Session{..} sock cnum h = do
                                         { ncSessionId = id
                                         , ncMyPublicKey = yourkey
                                         , ncTheirPublicKey = theirkey
-                                        , ncLastNMsgs = msgQ
+                                        , ncLastNMsgs = lastN
                                         , ncSockAddr = sockAddr
-                                        , ncMsgNumVar = msgNumVar
-                                        , ncDropCntVar = dropCntVar
                                         }) = do
-                        num <- atomically (readTVar msgNumVar)
-                        dropped <- atomically (readTVar dropCntVar)
-                        as <- atomically (packetQueueViewList msgQ)
+                        (num,dropped) <- atomically $ liftA2 (,) (CB.getTotal lastN) (CB.getDropped lastN)
+                        as <- atomically (CB.cyclicBufferViewList lastN)
                         let (h,u) = partition (fst . snd) as
                             countHandled   = length h
                             countUnhandled = length u
diff --git a/src/Data/CyclicBuffer.hs b/src/Data/CyclicBuffer.hs
index ab022e3f..0cc87459 100644
--- a/src/Data/CyclicBuffer.hs
+++ b/src/Data/CyclicBuffer.hs
@@ -1,6 +1,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE FlexibleContexts #-}
-module Data.CyclicBuffer where
+module Data.CyclicBuffer {- TODO: export list -} where
+
 
 import Control.Concurrent.STM
 import Control.Concurrent.STM.TArray
@@ -19,6 +20,7 @@ data CyclicBuffer a = CyclicBuffer
     , qsize   :: Word32
     , buffend :: TVar Word32 -- on incoming, highest packet number handled + 1
     , dropCnt :: TVar Word32
+    , totalCnt :: TVar Word32
     }
 
 cyclicBufferViewList :: CyclicBuffer a -> STM [(Word32,a)]
@@ -30,6 +32,15 @@ cyclicBufferViewList p = do
 getCapacity :: Applicative m => CyclicBuffer t -> m Word32
 getCapacity (CyclicBuffer { qsize }) = pure qsize
 
+getTotal :: CyclicBuffer t -> STM Word32
+getTotal (CyclicBuffer { totalCnt }) = readTVar totalCnt
+
+getDropped :: CyclicBuffer t -> STM Word32
+getDropped (CyclicBuffer { dropCnt }) = readTVar dropCnt
+
+getNextSequenceNum :: CyclicBuffer t -> STM Word32
+getNextSequenceNum (CyclicBuffer { seqno }) = readTVar seqno
+
 -- | Create a new CyclicBuffer with Overwrite on Wrap.
 new :: Word32 -- ^ Capacity of queue.
     -> Word32 -- ^ Initial sequence number.
@@ -41,6 +52,7 @@ new capacity seqstart = do
     seqv <- newTVar seqstart
     bufe <- newTVar 0
     dropped <- newTVar 0
+    total <- newTVar 0
     return CyclicBuffer
         { vwflgs  = flgs
         , pktq    = q
@@ -48,6 +60,7 @@ new capacity seqstart = do
         , qsize   = cap
         , buffend = bufe
         , dropCnt = dropped
+        , totalCnt = total
         }
 
 observeOutOfBand :: CyclicBuffer a -> Word32-> STM ()
@@ -79,14 +92,15 @@ markButNotDequeue CyclicBuffer { vwflgs, pktq, seqno, qsize } = do
     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 the capacity is exceeded, packets are
+-- | Enqueue a packet. If the capacity is exceeded, packets are
 -- dropped and the drop count increased accordingly.
+-- TODO: We no longer really support "out of order"
+--       So perhaps drop the num parameter
 enqueue :: CyclicBuffer a -- ^ The cyclic buffer(queue)
         -> Word32        -- ^ Sequence number of the packet.
         -> a             -- ^ The packet.
         -> STM ()
-enqueue CyclicBuffer{vwflgs, pktq, seqno, qsize, buffend, dropCnt} no x = do
+enqueue CyclicBuffer{vwflgs, pktq, seqno, qsize, buffend, dropCnt, totalCnt} no x = do
     low <- readTVar seqno
     let proj = no - low
     let i = no `mod` qsize
@@ -96,5 +110,7 @@ enqueue CyclicBuffer{vwflgs, pktq, seqno, qsize, buffend, dropCnt} no x = do
             modifyTVar' dropCnt (+1)
     writeArray pktq i (Just x)
     writeArray vwflgs i False -- mark as not viewed
+    modifyTVar' totalCnt (+1)
+    writeTVar seqno (no+1)
     modifyTVar' buffend (\be -> if be - low <= proj then no + 1 else be)
     return ()
diff --git a/src/Network/Tox/Crypto/Handlers.hs b/src/Network/Tox/Crypto/Handlers.hs
index ee8af399..92cb19b8 100644
--- a/src/Network/Tox/Crypto/Handlers.hs
+++ b/src/Network/Tox/Crypto/Handlers.hs
@@ -192,17 +192,12 @@ data NetCryptoSession = NCrypto
                                                 CryptoMessage
                                                 (CryptoPacket Encrypted)
                                                 CryptoData
-    , ncLastNMsgs :: PacketQueue (Bool{-Handled?-},(ViewSnapshot,InOrOut CryptoMessage))
+    , ncLastNMsgs :: CyclicBuffer (Bool{-Handled?-},(ViewSnapshot,InOrOut CryptoMessage))
     -- ^ cyclic buffer, holds the last N non-handshake crypto messages
     --   even if there is no attached user interface.
     , ncListeners :: TVar (IntMap.IntMap (ListenerType,TMChan CryptoMessage))
     -- ^ user interfaces may "listen" by inserting themselves into this map
     -- with a unique id and a new TChan, and then reading from the TChan
-    , ncMsgNumVar  :: TVar  Word32
-    -- ^ The number of non-handshake crypto messages written to ncLastNMsgs total
-    , ncDropCntVar :: TVar  Word32
-    -- ^ The number of crypto messages that were overwritten in the ncLastNMsgs
-    --   before anybody got to see them.
     }
 
 data NetCryptoSessions = NCSessions
@@ -450,7 +445,7 @@ freshCryptoSession  sessions
                 writeTVar ncMyPacketNonce0 n24plus1
                 return (return (f n24, n24, ncOutgoingIdMap0))
     pktoq <- atomically $ PQ.newOutGoing pktq ncToWire toWireIO 0 (outboundQueueCapacity sessions) 0
-    msgQ <- atomically (PQ.newOverwrite 10 0 :: STM (PacketQueue (Bool,(ViewSnapshot,InOrOut CryptoMessage))))
+    lastNQ <- atomically (CB.new 10 0 :: STM (CyclicBuffer (Bool,(ViewSnapshot,InOrOut CryptoMessage))))
     listeners <- atomically $ newTVar IntMap.empty
     msgNum <- atomically $ newTVar 0
     dropNum <- atomically $ newTVar 0
@@ -478,10 +473,8 @@ freshCryptoSession  sessions
                     , ncDequeueThread      = Nothing -- error "you want the NetCrypto-Dequeue thread id, but is it started?"
                     , ncPingMachine        = Nothing -- error "you want the NetCrypto-PingMachine, but is it started?"
                     , ncOutgoingQueue      = pktoq
-                    , ncLastNMsgs          = msgQ
+                    , ncLastNMsgs          = lastNQ
                     , ncListeners          = listeners
-                    , ncMsgNumVar          = msgNum
-                    , ncDropCntVar         = dropNum
                     }
     -- launch dequeue thread
     threadid <- forkIO $ do
@@ -886,17 +879,11 @@ hookHelper handledFlg typ  session  cm = do
 addMsgToLastN :: Bool -> MessageType -> NetCryptoSession -> InOrOut CryptoMessage -> IO ()
 addMsgToLastN handledFlg typ session cm = do
     let lastNQ = ncLastNMsgs session
-        msgNumVar = ncMsgNumVar session
-        dropCntVar = ncDropCntVar session
     atomically $ do
-        num <- readTVar msgNumVar
         view <- readTVar (ncView session)
         snapshot <- viewSnapshot view
-        (wraps,offset) <- PQ.enqueue lastNQ num (handledFlg,(snapshot,cm))
-        capacity <- PQ.getCapacity lastNQ
-        let dropped = wraps * capacity + offset
-        modifyTVar' msgNumVar (+1)
-        writeTVar dropCntVar dropped
+        num <- CB.getNextSequenceNum lastNQ
+        CB.enqueue lastNQ num (handledFlg,(snapshot,cm))
 
 
 -- | use to add a single hook to a specific session.