Added PacketQueue for handling inbound netcrypto packets.

2 files changed, 77 insertions, 2 deletions

diff --git a/src/Data/PacketQueue.hs b/src/Data/PacketQueue.hs
new file mode 100644
index 00000000..a617d502
--- /dev/null
+++ b/src/Data/PacketQueue.hs
@@ -0,0 +1,65 @@
+-- | This module is useful for implementing a lossess protocol on top of a
+-- lossy datagram style protocol.  It implements a buffer in which packets may
+-- be stored out of order, but from which they are extracted in the proper
+-- sequence.
+{-# LANGUAGE NamedFieldPuns #-}
+module Data.PacketQueue
+    ( PacketQueue
+    , new
+    , dequeue
+    , enqueue
+    ) where
+
+import Control.Concurrent.STM
+import Control.Concurrent.STM.TArray
+import Control.Monad
+import Data.Word
+import Data.Array.MArray
+
+data PacketQueue a = PacketQueue
+    { pktq    :: TArray Word32 (Maybe a)
+    , seqno   :: TVar Word32
+    , qsize   :: Word32
+    }
+
+-- | Create a new PacketQueue.
+new :: Word32 -- ^ Capacity of queue.
+    -> Word32 -- ^ Initial sequence number.
+    -> STM (PacketQueue a)
+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
+    return PacketQueue
+        { pktq   = q
+        , seqno  = seqv
+        , qsize  = cap
+        }
+
+-- | Retry until the next expected packet is enqueued.  Then return it.
+dequeue :: PacketQueue a -> STM a
+dequeue PacketQueue { pktq, seqno, qsize } = do
+    i0 <- readTVar seqno
+    let i = i0 `mod` qsize
+    x <- maybe retry return =<< readArray pktq i
+    writeArray pktq i Nothing
+    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.
+enqueue :: PacketQueue a -- ^ The packet queue.
+        -> Word32        -- ^ Sequence number of the packet.
+        -> a             -- ^ The packet.
+        -> STM ()
+enqueue PacketQueue{ pktq, seqno, qsize } no x = do
+    low <- readTVar seqno
+    let proj = no - low
+    -- Ignore packet if out of range.
+    when ( proj < qsize) $ do
+    let i = no `mod` qsize
+    writeArray pktq i (Just x)
+
+-- lookup :: PacketQueue a -> Word32 -> STM (Maybe a)
+-- lookup PacketQueue{ pktq, seqno, qsize } no = _todo
diff --git a/src/Network/Tox/Crypto/Handlers.hs b/src/Network/Tox/Crypto/Handlers.hs
index 13db02f3..787c69c2 100644
--- a/src/Network/Tox/Crypto/Handlers.hs
+++ b/src/Network/Tox/Crypto/Handlers.hs
@@ -18,6 +18,8 @@ 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.Serialize             as S
 import Data.Word
 import GHC.Conc (unsafeIOToSTM)
@@ -63,7 +65,7 @@ data NetCryptoSession = NCrypto { ncState         :: TVar NetCryptoSessionStatus
                                 , ncTheirBaseNonce :: TVar Nonce24 -- base nonce + packet number
                                 , ncMyPacketNonce :: TVar Nonce24 -- base nonce + packet number
                                 , ncHandShake     :: TVar (Maybe (Handshake Encrypted))
-                                , ncCookie        :: TVar (Maybe Cookie)
+                                , ncCookie        :: TVar (Maybe Cookie) -- ^ Cookie issued by remote peer
                                 , ncTheirDHTKey   :: PublicKey
                                 , ncTheirSessionPublic :: Maybe PublicKey
                                 , ncSessionSecret :: SecretKey
@@ -80,6 +82,8 @@ data NetCryptoSession = NCrypto { ncState         :: TVar NetCryptoSessionStatus
                                                                      --   needs to possibly start another, as is
                                                                      --   the case in group chats
                                 , ncView :: TVar SessionView
+                                , ncPacketQueue :: PacketQueue CryptoMessage
+                                , ncBufferStart :: TVar Word32
                                 }
 
 data NetCryptoSessions = NCSessions { netCryptoSessions :: TVar (Map.Map SockAddr NetCryptoSession)
@@ -88,6 +92,7 @@ data NetCryptoSessions = NCSessions { netCryptoSessions :: TVar (Map.Map SockAdd
                                     , defaultUnrecognizedHook :: MessageType -> NetCryptoHook
                                     , sessionView :: SessionView
                                     , msgTypeArray :: MsgTypeArray
+                                    , inboundQueueCapacity :: Word32
                                     }
 
 newSessionsState :: TransportCrypto
@@ -120,6 +125,7 @@ newSessionsState crypto unrechook hooks = do
                                                   , svDownloadDir = svDownloadDir0
                                                   }
                       , msgTypeArray = allMsgTypes -- todo make this a parameter
+                      , inboundQueueCapacity = 200
                       }
 
 data HandshakeParams
@@ -195,6 +201,8 @@ freshCryptoSession  sessions
     ncUnrecognizedHook0 <- atomically $ newTVar (defaultUnrecognizedHook sessions)
     ncMessageTypes0 <- atomically $ newTVar (msgTypeArray sessions)
     ncView0 <- atomically $ newTVar (sessionView sessions)
+    pktq <- atomically $ PQ.new (inboundQueueCapacity sessions) 0
+    bufstart <- atomically $ newTVar 0
     let netCryptoSession =
             NCrypto { ncState         = ncState0
                     , ncTheirBaseNonce= ncTheirBaseNonce0
@@ -210,6 +218,8 @@ freshCryptoSession  sessions
                     , ncAllSessions = sessions
                     , ncMessageTypes = ncMessageTypes0
                     , ncView = ncView0
+                    , ncPacketQueue        = pktq
+                    , ncBufferStart        = bufstart
                     }
     atomically $ modifyTVar allsessions (Map.insert addr netCryptoSession)
 
@@ -224,7 +234,7 @@ updateCryptoSession sessions addr hp session = do
         -- duplicate handshake packet, otherwise disregard everything, and
         -- refresh all state.
         --
-     then when (  Just ncTheirBaseNonce0 /= hpTheirBaseNonce hp
+     then when (  Just ncTheirBaseNonce0 /= hpTheirBaseNonce hp -- XXX: Do we really want to compare base nonce here?
                || ncTheirDHTKey session /= hpCookieRemoteDhtkey hp
                ) $ freshCryptoSession sessions addr hp
      else if ( Just ncTheirBaseNonce0 /= hpTheirBaseNonce hp)