extremely early WIP beginnings of state management

(committing only because i'm leaving for the most of the next two weeks)

reference/Transport/Chunks.hs

@@ -54,13 +54,14 @@ chunkDataSize = netPayloadSize - chunkMetaSize
 -- partition into pieces, and prepare each chunk's metadata
 chunkMsgPayload :: SessionId -> MsgIdx -> ByteString -> Array Int Chunk
 chunkMsgPayload sessionId msgIdx msgPayload = arr where
-  padded = buildLazyByteString (putMsgPayload msgPayload)
-  m      = fromIntegral (L.length padded)
-  (k,0)  = divMod m chunkDataSize
-  nOrigs = fromIntegral k
-  pieces = partitionLazyByteString chunkDataSize padded
-  arr    = listArray (0,k-1) 
-             [ MkChunk (MkChunkMeta sessionId msgIdx nOrigs) i (L.toStrict dat) | (i,dat) <- zip [0..] pieces ]
+  padded  = buildLazyByteString (putMsgPayload msgPayload)
+  mSize   = fromIntegral (L.length padded)
+  (k,0)   = divMod mSize chunkDataSize
+  nOrigs  = fromIntegral k
+  nTotals = 0     -- hackety hack :'(
+  pieces  = partitionLazyByteString chunkDataSize padded
+  arr     = listArray (0,k-1) 
+              [ MkChunk (MkChunkMeta sessionId (fromIntegral msgIdx) nOrigs nTotals) i (L.toStrict dat) | (i,dat) <- zip [0..] pieces ]
 
 -- | Before chunking, we encoded message payloads by prepending their length (8 bytes), 
 -- and then padding with zero bytes. We need to reverse this after recovering from EC chunks

reference/Transport/EC.hs

@@ -56,7 +56,7 @@ testEC = withLeopard $ do
 
   bad <- randomRIO (0,ecM+1)
   received <- catMaybes <$> maskListRandomly bad allChunks
-
+  
   -- putStrLn $ "chunks received:"
   -- forM_ received $ \chunk -> do
   --   putStrLn $ " - " ++ show chunk
@@ -175,14 +175,17 @@ decodeFromChunks chunks =
           Left  leoRes -> Left (LeopardError leoRes)
           Right arr    -> case parseMsgPayload (B.concat (elems arr)) of
             Nothing      -> Left  $ CannotParseMsg
-            Just bs      -> Right $ MkDecodedMessage sessionId msgIdx bs            
+            Just bs      -> Right $ MkDecodedMessage sessionId (fromIntegral msgIdx) bs            
 
       where
-        MkChunkMeta sessionId msgIdx ecK = meta
+        MkChunkMeta sessionId msgIdx ecK ecN = meta
         ecK_int  = fromIntegral ecK                 :: Int
-        ecp      = deterministicECParams ecK_int    :: ECParams
-        ecM_int  = _ecM ecp                         :: Int
-        ecN_int  = _ecN ecp                         :: Int
+        -- ecp      = deterministicECParams ecK_int    :: ECParams
+        -- ecM_int  = _ecM ecp                         :: Int
+        -- ecN_int  = _ecN ecp                         :: Int
+        ecN_int  = fromIntegral ecN                 :: Int
+        ecM_int  = ecN_int - ecK_int                :: Int
+        ecp      = ECParams { _ecK = ecK_int , _ecN = ecN_int }
         haveIdxs = map fst ibss                     :: [ChunkIdx]
         haveCnt  = length  haveIdxs                 :: Int
         minIdx   = minimum haveIdxs                 :: ChunkIdx

reference/Transport/Heuristics.hs

@@ -0,0 +1,51 @@
+
+-- | Heuristics involved in the transport layer
+
+{-# LANGUAGE StrictData, RecordWildCards, MultiWayIf #-}
+module Transport.Heuristics where
+
+--------------------------------------------------------------------------------
+
+import Leopard.Types
+
+import Transport.Protocol
+import Transport.Chunks
+
+--------------------------------------------------------------------------------
+-- * redundancy settings
+
+data Redundancy 
+  = NoRedundancy
+  | LowRedundancy
+  | MediumRedundancy
+  | HighRedundancy
+  deriving (Eq,Show)
+
+calculateECParams :: Redundancy -> Int -> ECParams
+calculateECParams redundancy k = ECParams k (calculateReservedNChunks redundancy k)
+
+-- | We send out this amount of chunks without waiting for any feedback
+calculateInitialNChunks :: Redundancy -> Int -> Int
+calculateInitialNChunks redundancy k = 
+  case redundancy of
+    NoRedundancy       -> k
+    LowRedundancy      -> max (k+1) (extra 1.01)
+    MediumRedundancy   -> max (k+2) (extra 1.04)
+    HighRedundancy     -> max (k+3) (extra 1.10)
+  where
+    extra s = ceiling (s * fromIntegral k)
+
+-- | However, we generate this many of total chunks initially,
+-- so that if a request for more chunks come in, we can just
+-- use one of the remaining, without having to retransmit everything
+calculateReservedNChunks :: Redundancy -> Int -> Int
+calculateReservedNChunks redundancy k = 
+  case redundancy of
+    NoRedundancy       -> min (2*k) $ max (k+ 5) (extra 1.15)
+    LowRedundancy      -> min (2*k) $ max (k+ 8) (extra 1.25)
+    MediumRedundancy   -> min (2*k) $ max (k+11) (extra 1.35)
+    HighRedundancy     -> min (2*k) $ max (k+15) (extra 1.50)
+  where
+    extra s = ceiling (s * fromIntegral k)
+
+--------------------------------------------------------------------------------

reference/Transport/Logic.hs

@@ -0,0 +1,153 @@
+
+-- | State machine logic
+
+{-# LANGUAGE StrictData, RecordWildCards, MultiWayIf #-}
+module Transport.Logic where
+
+--------------------------------------------------------------------------------
+
+import Data.IORef
+import System.IO.Unsafe
+
+import Control.Applicative
+import Control.Monad
+import Control.Monad.Except
+
+import Data.ByteString as ByteString
+import qualified Data.ByteString as B
+
+import Data.Map (Map)  
+import qualified Data.Map as Map
+
+import Transport.Protocol
+import Transport.Heuristics
+import Transport.Chunks
+import Transport.State
+import Transport.Result
+import Transport.Types
+
+--------------------------------------------------------------------------------
+
+-- | Seconds since the start of the protocol. TODO: more proper time representation?
+type Time         = Double
+type TimeInterval = Double
+
+-- temporary
+type TimerInfo = String
+
+type AddrOrSURB = Either SURB NetworkAddress 
+
+data NewConnection = MkNewConnection
+  { targetAddress  :: AddrOrSURB                   -- ^ how can we reach the other party
+  , targetAppProto :: ApplicationProtocol          -- ^ application level protocol (so that they can understand the request)
+  , initialMessage :: Maybe ByteString             -- ^ initial message
+  , replyHint      :: Maybe Int                    -- ^ expected reply size, in bytes
+  }
+  deriving (Eq,Show)
+
+data Event
+  = PacketReceived Chunk               -- ^ we received a Mix packet
+  | TimerTriggered TimerInfo           -- ^ self-set timer event
+  deriving (Eq,Show)
+
+{-
+data UserAction
+  = UserInitConn  NewConnection        -- ^ the user want to open a new connection
+  | UserSend      Message              -- ^ the user wants to send a message
+  | UserCloseConn SessionId            -- ^ the user want the close an existing connection
+  deriving (Eq,Show)
+-}
+
+data Action
+  = SendPacket  AddrOrSURB Chunk       -- ^ send a mix packet
+  | SignalError ErrMsg                 -- ^ signal some failure toward the user (TODO: proper failure type) 
+  | SetTimer    Time TimerInfo         -- ^ set a timer for ourselves 
+  deriving (Eq,Show)
+
+--------------------------------------------------------------------------------
+-- * open a new connection
+
+data ConnectionOpts = MkConnectionOpts
+  { sessionTimeout    :: TimeInterval
+  , messageTimeout    :: TimeInterval
+  , messageRedundancy :: Redundancy
+  }
+
+openConnection :: ConnectionOpts -> NewConnection -> IOResult SessionId
+openConnection opts newConn = do
+  throwError "not yet implemented"
+
+--------------------------------------------------------------------------------
+-- * global state
+
+data LogicState = MkLogicState 
+  { openConns :: Map SessionId Connection
+  }
+  deriving Show
+
+initialLogicState :: LogicState
+initialLogicState = MkLogicState
+  { openConns = Map.empty
+  }
+
+{-# NOINLINE theLogicState #-}
+theLogicState :: IORef LogicState
+theLogicState = unsafePerformIO $ newIORef initialLogicState
+
+--------------------------------------------------------------------------------
+
+eventHandler :: Event -> IO [Action]
+eventHandler event = do
+  state <- readIORef theLogicState
+  case event of 
+    PacketReceived chunk -> packetReceived state chunk
+    TimerTriggered timer -> timerTriggered state timer
+
+-- TODO: implement this
+timerTriggered :: LogicState -> TimerInfo -> IO [Action]
+timerTriggered _ _ = return []
+
+--------------------------------------------------------------------------------
+-- * reciving packets
+
+packetReceived :: LogicState -> Chunk -> IO [Action]
+packetReceived state@(MkLogicState openConns) chunk@(MkChunk chunkMeta@(MkChunkMeta{..}) chunkIdx chunkData) = do
+  case Map.lookup _cmSessionId openConns of
+
+    -- implicitly opened new connection
+    Nothing -> do
+      newConn <- newRecvConnection _cmSessionId
+      let openConns' = Map.insert _cmSessionId newConn openConns
+      let state'     = state { openConns = openConns' }
+      writeIORef theLogicState state'
+      -- now that exists, we can simply reuse the normal handler
+      packetReceived state' chunk
+
+    -- existing connection
+    Just oldConn@(MkConnection{..}) -> do
+      (newConn,actions) <- packetReceived' oldConn chunk 
+      let openConns' = Map.insert _cmSessionId newConn openConns
+      let state'     = state { openConns = openConns' }
+      writeIORef theLogicState state'
+      return actions
+
+packetReceived' :: Connection -> Chunk -> IO (Connection, [Action])
+packetReceived' conn@(MkConnection{..}) chunk = do
+  -- TODO
+  undefined
+
+packetReceived'' :: ConnMeta -> ConnectionRecvState -> Chunk -> IO (ConnectionRecvState, [Action])
+packetReceived'' connMeta connState chunk@(MkChunk{..}) = do
+  -- TODO
+  undefined
+
+{-
+  let AllCounters ns nr ls lr = _connCounters
+
+  if 
+    -- this message was already fully received; we just ignore it
+    | msgIdx <= lr   -> return connState
+
+    | Map.notMember msgIdx _connCounters
+-}
+

reference/Transport/Protocol.hs

@@ -19,7 +19,7 @@ data SessionControl
   = InitialMessage AppProtocol          -- ^ the first message estabilishing a connection
   | Continue                            -- ^ proceed normally
   | CloseSession                        -- ^ closing down the session
-  deriving (Eq,Show,Enum)
+  deriving (Eq,Show)
 
 -- | Application layer protocol
 data AppProtocol = MkAppProtocol
@@ -31,29 +31,35 @@ data AppProtocol = MkAppProtocol
 data AckEntry
   = MsgReceived    MsgIdx               -- ^ message received intact
   | MsgFailed      MsgIdx FailReason    -- ^ message failed (eg. decoding, checksum or timeout)
-  | NeedMoreChunks MsgIdx Int           -- ^ we need more chunks 
-  | ResendChunks   MsgIdx [Int]         -- ^ re-transmit particular chunks
+  | NeedMoreChunks MsgIdx ChunkIdx      -- ^ we need more chunks 
+  | ResendChunks   MsgIdx [ChunkIdx]    -- ^ re-transmit particular chunks
   deriving (Eq,Show)
 
 data FailReason
   = FailTimeout             -- ^ couldn't decode within the expected time
   | FailDecode              -- ^ erasure code decoding failed
   | FailChecksum            -- ^ checksum failed
+  | NotYetImplemented       -- ^ we haven't implemented the feature
   deriving (Eq,Show,Enum)
 
 --------------------------------------------------------------------------------
 
-data MsgMeta = MkMsgMeta
-  { _msgControl   :: SessionControl   -- ^ where we are in the session
-  , _msgReqSURBs  :: Int              -- ^ how many more SURBs we need 
-  , _msgAcks      :: [AckEntry]       -- ^ status of previous messages
+data TransportVersion
+  = V1
+  deriving (Eq,Show,Enum)
+
+data TransportMeta = MkTransportMeta
+  { _transVersion   :: TransportVersion   -- ^ protocol version 
+  , _transControl   :: SessionControl     -- ^ where we are in the session
+  , _transReqSURBs  :: Int                -- ^ how many more SURBs we need 
+  , _transAcks      :: [AckEntry]         -- ^ status of previous messages
   }
   deriving (Eq,Show)
 
 data Message = MkMessage
-  { _msgMeta      :: MsgMeta          -- ^ message metadata
-  , _msgSURBs     :: [SURB]           -- ^ the new SURBs we send
-  , _msgPayload   :: ByteString       -- ^ the actual payload
+  { _msgMeta      :: TransportMeta     -- ^ transport layer metadata
+  , _msgSURBs     :: [SURB]            -- ^ the new SURBs we send
+  , _msgPayload   :: ByteString        -- ^ the actual payload
   }
   deriving Eq
 

reference/Transport/Result.hs

@@ -0,0 +1,54 @@
+
+-- TODO: move this module somewhere else?
+
+{-# LANGUAGE StrictData, FlexibleInstances, MultiParamTypeClasses, FunctionalDependencies #-}
+module Transport.Result where
+
+--------------------------------------------------------------------------------
+
+import Control.Applicative
+import Control.Monad
+import Control.Monad.Except
+
+--------------------------------------------------------------------------------
+
+type ErrMsg = String
+
+data Result a
+  = Ok  a
+  | Err ErrMsg
+  deriving (Eq,Show)
+
+newtype IOResult a 
+  = IOR { runIOResult :: IO (Result a) }
+
+--------------------------------------------------------------------------------
+
+instance Functor IOResult where
+  fmap f (IOR action) = IOR $ do
+    r <- action
+    return $ case r of
+      Ok  x -> Ok (f x)
+      Err m -> Err m
+
+instance Applicative IOResult where
+  pure x = IOR $ pure (Ok x)
+  (<*>)  = ap
+
+instance Monad IOResult where
+  IOR a >>= u = IOR $ do
+    r <- a
+    case r of
+      Ok  x -> runIOResult (u x)
+      Err m -> return (Err m)
+
+instance MonadError ErrMsg IOResult where
+  throwError e = IOR $ pure (Err e)
+  catchError (IOR action) handler = IOR $ do
+    r <- action
+    case r of
+      Ok  x -> return r
+      Err m -> runIOResult (handler m)
+
+--------------------------------------------------------------------------------
+

reference/Transport/State.hs

@@ -1,7 +1,7 @@
 
 -- | State of a connection
 
-{-# LANGUAGE StrictData #-}
+{-# LANGUAGE StrictData, PatternSynonyms, FlexibleInstances #-}
 module Transport.State where
 
 --------------------------------------------------------------------------------
@@ -22,29 +22,116 @@ data Role
   | Server                      -- ^ the other party in the connection 
   deriving (Eq,Show)
 
+-- | Network addresses here we encode as strings, for simplicity
+type NetworkAddress = String
+
 -- | The address of the other party
 data ConnAddress 
-  = NetworkAddress String       -- ^ we know some kind of address for the other party
-  | ReplyToSURB                 -- ^ we don't know their address, instead we can use SURBs to send
+  = NetworkAddr NetworkAddress      -- ^ we know some kind of address for the other party
+  | ReplyToSURB                     -- ^ we don't know their address, instead we can use SURBs to send
   deriving (Eq,Show)
 
+newtype ApplicationProtocol 
+  = MkAppProto String
+  deriving (Eq,Show)
+
+data ConnMeta = MkConnMeta
+  { _connRole        :: Role                        -- ^ our role in the connection (client or server)
+  , _connAppProto    :: Maybe ApplicationProtocol   -- ^ application layer protocol
+  , _connDestAddr    :: ConnAddress                 -- ^ the address of the other party
+  }
+  deriving (Eq,Show)
+
+defaultServerConnMeta :: ConnMeta
+defaultServerConnMeta = MkConnMeta
+  { _connRole     = Server
+  , _connAppProto = Nothing
+  , _connDestAddr = ReplyToSURB
+  }
+
 data Connection = MkConnection
   { _connSessionId   :: SessionId                   -- ^ the session id
-  , _connRole        :: Role                        -- ^ our role in the connection (client or server)
-  , _connDestAddr    :: ConnAddress                 -- ^ the address of the other party
-  , _connState       :: IORef ConnectionState       -- ^ current state of the connection
+  , _connMeta        :: ConnMeta                    -- ^ connection metadata
+  , _connStateRef    :: IORef ConnectionState       -- ^ current state of the connection
   }
+  deriving (Show)
 
-data ConnectionState = MkConnState
-  { _connNextSendIdx  :: MsgIdx                      -- ^ index of the next message to send
-  , _connNextRecvIdx  :: MsgIdx                      -- ^ index of the next message to receive
-  , _connUnusedSURBs  :: [SURB]                      -- ^ set of unused SURBs
-  , _connSentMsgState :: Map MsgIdx SentMsgState     -- ^ state of messages we sent
-  , _connRecvMsgState :: Map MsgIdx RecvMsgState     -- ^ state of messages we are receiving
-  , _connExpectedData :: Maybe Int                   -- ^ number of bytes we are expecting to receive, and haven't sent SURBs for yet
+instance Show (IORef ConnectionState) where
+  show _ = "<<IORef ConnectionState>>"
+
+newRecvConnection :: SessionId -> IO Connection
+newRecvConnection sessionId = do
+  stateRef <- newIORef emptyConnectionState
+  return $ MkConnection
+    { _connSessionId = sessionId
+    , _connMeta      = defaultServerConnMeta
+    , _connStateRef  = stateRef
+    }
+
+-- | Message index counters for sent messages
+data SentCounters = MkSentCounters
+  { _ctrLastFullyAckIdx  :: MsgIdx       -- ^ index of the last message fully sent and acknowledged
+  , _ctrNextSendIdx      :: MsgIdx       -- ^ index of the next message to send
+  }
+  deriving (Eq,Show)
+
+-- | Message index counters for received messages
+data RecvCounters = MkRecvCounters
+  { _ctrLastFullyRecvIdx :: MsgIdx       -- ^ index of the last message fully received
+  , _ctrNextRecvIdx      :: MsgIdx       -- ^ index of the next message to receive
+  }
+  deriving (Eq,Show)
+
+emptySentCounters :: SentCounters
+emptySentCounters = MkSentCounters (-1) 0
+
+emptyRecvCounters :: RecvCounters
+emptyRecvCounters = MkRecvCounters (-1) 0
+
+data ConnectionSentState = MkConnSentState
+  { sentCounters       :: SentCounters               -- ^ message index counters
+  , sentUnusedFwdSURBs :: [SURB]                     -- ^ set of unused SURBs we can use
+  , sentMsgState       :: Map MsgIdx SentMsgState    -- ^ state of messages we sent
+  , sentExpectedBytes  :: Maybe Int                  -- ^ number of bytes we are expecting to send, and haven't yet asked fwdSURBs for 
+  }
+  deriving (Eq,Show)
+
+data ConnectionRecvState = MkConnRecvState
+  { recvCounters       :: RecvCounters               -- ^ message index counters
+  , recvUnusedBwdSURBs :: [SURB]                     -- ^ set of unused SURBs we can send (this can be always refreshed)
+  , recvMsgState       :: Map MsgIdx RecvMsgState    -- ^ state of messages we are receiving
+  , recvExpectedBytes  :: Maybe Int                  -- ^ number of bytes we are expecting to receive, and haven't yet sent bwdSURBs for
   }  
   deriving (Eq,Show)
 
+data ConnectionState = MkConnState
+  { connSentState :: ConnectionSentState
+  , connRecvState :: ConnectionRecvState
+  }  
+  deriving (Eq,Show)
+
+emptyConnectionSentState :: ConnectionSentState
+emptyConnectionSentState = MkConnSentState
+  { sentCounters       = emptySentCounters
+  , sentUnusedFwdSURBs = []
+  , sentMsgState       = Map.empty
+  , sentExpectedBytes  = Nothing
+  }  
+
+emptyConnectionRecvState :: ConnectionRecvState
+emptyConnectionRecvState = MkConnRecvState
+  { recvCounters       = emptyRecvCounters
+  , recvUnusedBwdSURBs = []
+  , recvMsgState       = Map.empty
+  , recvExpectedBytes  = Nothing
+  }  
+
+emptyConnectionState :: ConnectionState
+emptyConnectionState = MkConnState
+  { connSentState = emptyConnectionSentState
+  , connRecvState = emptyConnectionRecvState
+  }  
+
 -- | State of a message we are sending
 data SentMsgState = SentMsgState
   { _origChunks   :: [Chunk]
@@ -53,22 +140,23 @@ data SentMsgState = SentMsgState
   }
   deriving (Eq,Show)
 
-data SentCounter = MkSentCounter
+data SentChunkCounter = MkSentChunkCounter
   { _alreadySent  :: Int        -- ^ we already sent this many chunks (0..K-1 means orig; >= K means parity)
   , _needsToSend  :: Int        -- ^ we were asked for this many more chunks
   }        
   deriving (Eq,Show)
 
 data SentMsgStatus
-  = SentMsgFailed                           -- ^ the message failed completely; we should probably retransmit  
-  | SentMsgSingleton                        -- ^ the message was only a single chunk, so we don't do EC; we can just retransmit
-  | SentMsgNeedsMoreChunks SentCounter      -- ^ the other party asked for more chunks
+  = SentMsgFailed                               -- ^ the message failed completely; we should probably retransmit  
+  | SentMsgSingleton                            -- ^ the message was only a single chunk, so we don't do EC; we can just retransmit
+  | SentMsgNeedsMoreChunks SentChunkCounter     -- ^ the other party asked for more chunks
   deriving (Eq,Show)
 
 -- | State of a message we are receiving
 data RecvMsgState = RecvMsgState
-  { _receivedChunks :: Map ChunkIdx Chunk   -- ^ chunks we received so far
-  , _msgNOrigCHunks :: Int                  -- ^ number of original chunks
+  { _receivedChunks  :: Map ChunkIdx Chunk   -- ^ chunks we received so far
+  , _msgNOrigChunks  :: Int                  -- ^ number of original chunks
+  , _msgNTotalChunks :: Int                  -- ^ number of total chunks (original + parity)
   }
   deriving (Eq,Show)
   

reference/Transport/Types.hs

@@ -16,7 +16,7 @@ import Data.Octets
 --------------------------------------------------------------------------------
 
 -- | Message index within a session
-type MsgIdx = Word32
+type MsgIdx = Int -- Word32
 
 -- | Chunk index within a chunked messages
 type ChunkIdx = Word16
@@ -26,7 +26,7 @@ type ChunkIdx = Word16
 -- | A (random) session identifier
 newtype SessionId
   = MkSessionId [Word8]
-  deriving (Eq)
+  deriving (Eq,Ord)
 
 instance Show SessionId where
   show (MkSessionId xs) = "<session_id = " ++ showHexBytes xs ++ ">"

transport-over-mix.cabal

@@ -37,6 +37,7 @@ Library
                         array >= 0.5 && < 0.6,
                         containers >= 0.7 && < 0.8,
                         bytestring >= 0.12 && < 0.14,
+                        mtl >= 2.2.2 && < 2.5,
                         random >= 1.3 && < 1.4,
                         binary >= 0.8 && < 0.9,
                         pqueue >= 1.7.0 && < 2.0.0,
@@ -45,9 +46,12 @@ Library
   Exposed-Modules:      Sphinx.Header
                         Transport.Protocol
                         Transport.State
+                        Transport.Logic
+                        Transport.Heuristics
                         Transport.EC
                         Transport.Chunks
                         Transport.Types
+                        Transport.Result
                         Transport.Misc
                         Simulate.Mix
                         Simulate.DiscreteEvents