diff --git a/reference/Transport/Chunks.hs b/reference/Transport/Chunks.hs index 4ac3f11..76c10c0 100644 --- a/reference/Transport/Chunks.hs +++ b/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 diff --git a/reference/Transport/EC.hs b/reference/Transport/EC.hs index 0c47fbf..709bbc3 100644 --- a/reference/Transport/EC.hs +++ b/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 diff --git a/reference/Transport/Heuristics.hs b/reference/Transport/Heuristics.hs new file mode 100644 index 0000000..04001bb --- /dev/null +++ b/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) + +-------------------------------------------------------------------------------- diff --git a/reference/Transport/Logic.hs b/reference/Transport/Logic.hs new file mode 100644 index 0000000..4a832d0 --- /dev/null +++ b/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 +-} + diff --git a/reference/Transport/Protocol.hs b/reference/Transport/Protocol.hs index 65d39c8..0dd41d8 100644 --- a/reference/Transport/Protocol.hs +++ b/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 diff --git a/reference/Transport/Result.hs b/reference/Transport/Result.hs new file mode 100644 index 0000000..985c73c --- /dev/null +++ b/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) + +-------------------------------------------------------------------------------- + diff --git a/reference/Transport/State.hs b/reference/Transport/State.hs index 0d5ea9b..465ee06 100644 --- a/reference/Transport/State.hs +++ b/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 _ = "<>" + +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) diff --git a/reference/Transport/Types.hs b/reference/Transport/Types.hs index b48f52c..b451ead 100644 --- a/reference/Transport/Types.hs +++ b/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) = "" diff --git a/transport-over-mix.cabal b/transport-over-mix.cabal index fda1d15..cbd064f 100644 --- a/transport-over-mix.cabal +++ b/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