hans 3.0.0.1 → 3.0.1
raw patch · 14 files changed
+244/−40 lines, 14 files
Files
- hans.cabal +3/−1
- src/Hans/Buffer/Datagram.hs +6/−1
- src/Hans/Config.hs +3/−1
- src/Hans/IP4/Packet.hs +2/−1
- src/Hans/Socket/Handle.hs +105/−0
- src/Hans/Socket/Tcp.hs +16/−5
- src/Hans/Socket/Types.hs +10/−0
- src/Hans/Socket/Udp.hs +9/−0
- src/Hans/Tcp/Output.hs +36/−8
- src/Hans/Tcp/Packet.hs +26/−17
- src/Hans/Tcp/SendWindow.hs +5/−0
- src/Hans/Tcp/State.hs +5/−4
- src/Hans/Tcp/Tcb.hs +9/−2
- tests/Tests/Network.hs +9/−0
hans.cabal view
@@ -1,5 +1,5 @@ name: hans-version: 3.0.0.1+version: 3.0.1 cabal-version: >= 1.18 license: BSD3 license-file: LICENSE@@ -92,6 +92,7 @@ Hans.Network.Types Hans.Serialize Hans.Socket+ Hans.Socket.Handle Hans.Tcp.Input Hans.Tcp.Message Hans.Tcp.Output@@ -155,4 +156,5 @@ Tests.IP4.Fragmentation Tests.IP4.Icmp4 Tests.IP4.Packet+ Tests.Network Tests.Utils
src/Hans/Buffer/Datagram.hs view
@@ -6,13 +6,14 @@ writeChunk, readChunk, tryReadChunk,+ isEmptyBuffer ) where import Hans.Buffer.Signal import Control.Monad (when) import qualified Data.ByteString as S-import Data.IORef (IORef,newIORef,atomicModifyIORef')+import Data.IORef (IORef,newIORef,atomicModifyIORef',readIORef) import qualified Data.Sequence as Seq @@ -68,6 +69,10 @@ Nothing -> return Nothing +-- | See if the buffer is empty.+isEmptyBuffer :: Buffer a -> IO Bool+isEmptyBuffer Buffer { .. } =+ (Seq.null . bufChunks) `fmap` readIORef bufContents -- Buffer State ----------------------------------------------------------------
src/Hans/Config.hs view
@@ -91,7 +91,9 @@ , cfgTcpListenTableSize = 5 , cfgTcpActiveTableSize = 67 , cfgTcpTimeoutTimeWait = 60.0 -- one minute- , cfgTcpInitialMSS = 512+ , cfgTcpInitialMSS = 1380+ -- ^ 1500 - (max (20 for IPv4) (60 for IPv6)+ -- + 60 for maximum TCP header) , cfgTcpMaxSynBacklog = 128 , cfgTcpInitialWindow = 14600 , cfgTcpMSL = 60 -- one minute
src/Hans/IP4/Packet.hs view
@@ -216,7 +216,8 @@ ip4FragmentOffset = ip4Fragment . lens f g where f frag = (frag .&. 0x1fff) `shiftL` 3- g frag len = frag .|. ((len `shiftR` 3) .&. 0x1fff)+ g frag len = (frag .&. complement 0x1fff)+ .|. ((len `shiftR` 3) .&. 0x1fff) {-# INLINE ip4FragmentOffset #-}
+ src/Hans/Socket/Handle.hs view
@@ -0,0 +1,105 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiWayIf #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Hans.Socket.Handle(makeHansHandle) where++import Control.Concurrent(threadDelay)+import Control.Exception(throwIO)+import qualified Data.ByteString as BSS+import qualified Data.ByteString.Lazy as BS+import Data.Typeable(Typeable)+import Foreign.Ptr(Ptr, castPtr, plusPtr)+import GHC.IO.Buffer(newByteBuffer)+import GHC.IO.BufferedIO(BufferedIO(..),+ readBuf, readBufNonBlocking,+ writeBuf, writeBufNonBlocking)+import GHC.IO.Device(IODevice(..), RawIO(..), IODeviceType(..))+import GHC.IO.Handle(mkFileHandle, noNewlineTranslation)+import Hans.Network(Network(..))+import Hans.Socket(Socket(..), DataSocket(..))+import Prelude hiding (read)+import System.IO(Handle, IOMode)+++instance (Socket sock, DataSocket sock, Network addr) =>+ IODevice (sock addr) where+ ready dev forWrite msecs =+ do let tester = if forWrite then sCanWrite else sCanRead+ canDo <- tester dev+ if | canDo -> return True+ | msecs <= 0 -> return False+ | otherwise -> do let delay = min msecs 100+ threadDelay (delay * 1000)+ ready dev forWrite (msecs - delay)+ close bs = sClose bs+ isTerminal _ = return False+ isSeekable _ = return False+ seek _ _ _ = throwIO (userError "Seek on HaNS socket.")+ tell _ = throwIO (userError "Tell on HaNS socket.")+ getSize _ = throwIO (userError "getSize on HaNS socket.")+ setSize _ _ = throwIO (userError "setSize on HaNS socket.")+ setEcho _ _ = throwIO (userError "setEcho on HaNS socket.")+ getEcho _ = throwIO (userError "getEcho on HaNS socket.")+ setRaw _ _ = return ()+ devType _ = return Stream+ dup _ = throwIO (userError "dup on HaNS socket.")+ dup2 _ _ = throwIO (userError "dup2 on HaNS socket.")++instance (Socket sock, DataSocket sock, Network addr) =>+ RawIO (sock addr) where+ read sock dptr sz =+ do bstr <- sRead sock (fromIntegral sz)+ copyToPtr dptr sz bstr+ readNonBlocking sock dptr sz =+ do mbstr <- sTryRead sock (fromIntegral sz)+ case mbstr of+ Nothing -> return Nothing+ Just bstr -> Just `fmap` copyToPtr dptr sz bstr+ write sock ptr sz =+ do bstr <- BSS.packCStringLen (castPtr ptr, sz)+ sendAll (BS.fromStrict bstr)+ where+ sendAll bstr+ | BS.null bstr = return ()+ | otherwise = do num <- sWrite sock bstr+ sendAll (BS.drop (fromIntegral num) bstr)+ writeNonBlocking sock ptr sz =+ do bstr <- BSS.packCStringLen (castPtr ptr, sz)+ num <- sWrite sock (BS.fromStrict bstr)+ return (fromIntegral num)++instance (Socket sock, DataSocket sock, Network addr) =>+ BufferedIO (sock addr) where+ newBuffer _ = newByteBuffer (64 * 1024)+ fillReadBuffer = readBuf+ fillReadBuffer0 = readBufNonBlocking+ flushWriteBuffer = writeBuf+ flushWriteBuffer0 = writeBufNonBlocking++-- |Make a GHC Handle from a Hans handle.+makeHansHandle :: (Socket sock, DataSocket sock, Network addr, Typeable sock) =>+ sock addr -> IOMode -> IO Handle+makeHansHandle socket mode =+ mkFileHandle socket "<socket>" mode Nothing noNewlineTranslation++copyToPtr :: Num a => Ptr b -> Int -> BS.ByteString -> IO a+copyToPtr ptr sz bstr+ | BS.length bstr == 0 = return 0+ | BS.length bstr > fromIntegral sz = fail "Too big a chunk for copy!"+ | otherwise =+ do copyBS (BS.toChunks bstr) ptr sz+ return (fromIntegral (BS.length bstr))++copyBS :: [BSS.ByteString] -> Ptr a -> Int -> IO ()+copyBS [] _ _ = return ()+copyBS (f:rest) sptr szLeft+ | BSS.null f = copyBS rest sptr szLeft+ | szLeft <= 0 = return ()+ | otherwise =+ do let (chunk1, chunk2) = BSS.splitAt szLeft f+ amt = fromIntegral (BSS.length chunk1)+ BSS.useAsCString chunk1 $ \ dptr -> memcpy dptr sptr amt+ copyBS (chunk2 : rest) (sptr `plusPtr` amt) (szLeft - amt)++foreign import ccall unsafe "string.h memcpy"+ memcpy :: Ptr a -> Ptr b -> Int -> IO ()
src/Hans/Socket/Tcp.hs view
@@ -1,5 +1,6 @@-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeFamilies #-} module Hans.Socket.Tcp where @@ -15,12 +16,13 @@ import qualified Hans.Tcp.SendWindow as Send import Hans.Types -import Control.Concurrent (newEmptyMVar,tryPutMVar,takeMVar)-import Control.Exception (throwIO)-import Control.Monad (unless)+import Control.Concurrent (newEmptyMVar,tryPutMVar,takeMVar,yield)+import Control.Exception (throwIO, handle)+import Control.Monad (unless,when) import qualified Data.ByteString.Lazy as L import Data.IORef (readIORef) import Data.Time.Clock (getCurrentTime)+import Data.Typeable(Typeable) import System.CPUTime (getCPUTime) @@ -29,6 +31,7 @@ data TcpSocket addr = TcpSocket { tcpNS :: !NetworkStack , tcpTcb :: !Tcb }+ deriving (Typeable) instance HasNetworkStack (TcpSocket addr) where networkStack = to tcpNS@@ -189,12 +192,20 @@ tcpTcb <- activeOpen tcpNS ri srcPort dst dstPort return TcpSocket { .. } + sCanWrite TcpSocket { .. } =+ handle ((\ _ -> return False) :: (ConnectionException -> IO Bool)) $+ guardSend tcpTcb (canSend tcpTcb)+ -- segmentize the bytes, and return to the user the number of bytes that have -- been moved into the send window sWrite TcpSocket { .. } bytes = guardSend tcpTcb $ do len <- sendData tcpNS tcpTcb bytes++ when (len < L.length bytes) yield+ return $! fromIntegral len + sCanRead TcpSocket { .. } = guardRecv tcpTcb (haveBytesAvail tcpTcb) sRead TcpSocket { .. } len = guardRecv tcpTcb (receiveBytes len tcpTcb) sTryRead TcpSocket { .. } len = guardRecv tcpTcb (tryReceiveBytes len tcpTcb)
src/Hans/Socket/Types.hs view
@@ -59,8 +59,18 @@ -> SockPort -- ^ Remote port -> IO (sock addr) + -- | Returns True iff there is currently space in the buffer to accept a+ -- write. Note, this is probably a bad thing to count on in a concurrent+ -- system ...+ sCanWrite :: Network addr => sock addr -> IO Bool+ -- | Send a chunk of data on a socket. sWrite :: Network addr => sock addr -> L.ByteString -> IO Int++ -- | Returns True iff there is data in the buffer that can be read.+ -- Note, this is probably a bad thing to count on in a concurrent+ -- system ...+ sCanRead :: Network addr => sock addr -> IO Bool -- | Read a chunk of data from a socket. Reading an empty result indicates -- that the socket has closed.
src/Hans/Socket/Udp.hs view
@@ -103,6 +103,12 @@ return UdpSocket { .. } + sCanWrite UdpSocket { .. } =+ do path <- readIORef udpSockState+ case path of+ KnownRoute _ _ _ _ -> return True+ KnownSource{} -> return False+ sWrite UdpSocket { .. } bytes = do path <- readIORef udpSockState case path of@@ -115,6 +121,9 @@ KnownSource{} -> throwIO NoConnection++ sCanRead UdpSocket { .. } =+ not `fmap` DGram.isEmptyBuffer udpBuffer sRead UdpSocket { .. } len = do (_,buf) <- DGram.readChunk udpBuffer
src/Hans/Tcp/Output.hs view
@@ -11,6 +11,7 @@ sendAck, sendFin, sendData,+ canSend, -- ** From the fast-path queueTcp,@@ -62,15 +63,42 @@ _ <- sendWithTcb ns tcb hdr L.empty return () --- | Send a data segment. When the remote window is full, this returns 0.+-- | Send a data segment, potentially sending multiple packets if the send+-- window allows, and the payload is larger than MSS. When the remote window is+-- full, this returns 0. sendData :: NetworkStack -> Tcb -> L.ByteString -> IO Int64-sendData ns tcb bytes =- do let hdr = set tcpAck True- $ set tcpPsh True emptyTcpHeader- mb <- sendWithTcb ns tcb hdr bytes- case mb of- Just len -> return len- Nothing -> return 0+sendData ns tcb = go 0+ where+ -- technically the MSS could change between sends, so this ensures that we+ -- would pick up that change.+ go acc bytes+ | L.null bytes =+ return acc++ | otherwise =+ do mss <- fromIntegral `fmap` readIORef (tcbMss tcb)+ mb <- sendWithTcb ns tcb hdr (L.take mss bytes)+ case mb of++ -- when the amount sent was less than mss, we've filled the send+ -- window, and won't be able to send any more.+ Just len | len < mss -> return $! acc + len+ | otherwise -> let acc' = acc + len+ in acc' `seq` go acc' (L.drop len bytes)++ -- the send window is full, return the accumulator+ Nothing -> return acc++ hdr = set tcpAck True+ $ set tcpPsh True+ emptyTcpHeader+++-- | Determine if there is any room in the remote window for us to send+-- data.+canSend :: Tcb -> IO Bool+canSend Tcb { .. } =+ (not . Send.fullWindow) `fmap` readIORef tcbSendWindow -- | Send a segment and queue it in the remote window. The number of bytes that -- were sent is returned.
src/Hans/Tcp/Packet.hs view
@@ -10,6 +10,7 @@ TcpHeader(..), TcpPort, putTcpPort, emptyTcpHeader,+ tcpHeaderSize, -- ** Sequence Numbers TcpSeqNum, TcpAckNum,@@ -31,6 +32,8 @@ TcpOption(..), TcpOptionTag(..), tcpOptionTag, SackBlock(..),+ tcpOptionsSize,+ tcpOptionSize, -- * Segment Operations tcpSegLen,@@ -181,9 +184,15 @@ -- | The length of the fixed part of the TcpHeader, in 4-byte octets.-tcpFixedHeaderLength :: Int-tcpFixedHeaderLength = 5+tcpFixedHeaderSize :: Int+tcpFixedHeaderSize = 5 +-- | The encoded size of a header.+tcpHeaderSize :: TcpHeader -> Int+tcpHeaderSize TcpHeader { .. } =+ let (size,padding) = tcpOptionsSize tcpOptions_+ in size + padding + tcpFixedHeaderSize+ -- | Render a TcpHeader. The checksum value is never rendered, as it is -- expected to be calculated and poked in afterwords. putTcpHeader :: Putter TcpHeader@@ -192,8 +201,8 @@ putTcpPort tcpDestPort putTcpSeqNum tcpSeqNum putTcpAckNum tcpAckNum- let (optLen,padding) = tcpOptionsLength tcpOptions_- putTcpControl (tcpFixedHeaderLength + optLen) tcpFlags_+ let (optLen,padding) = tcpOptionsSize tcpOptions_+ putTcpControl (tcpFixedHeaderSize + optLen) tcpFlags_ putWord16be tcpWindow putWord16be 0 putWord16be tcpUrgentPointer@@ -217,7 +226,7 @@ tcpUrgentPointer <- getWord16be -- options, not including the end-of-list option- let optsLen = dataOff - tcpFixedHeaderLength+ let optsLen = dataOff - tcpFixedHeaderSize opts <- label "options" (isolate (optsLen `shiftL` 2) getTcpOptions) let tcpOptions_ = filter (/= OptEndOfOptions) opts @@ -322,23 +331,23 @@ -- | Get the rendered length of a list of TcpOptions, in 4-byte words, and the -- number of padding bytes required. This rounds up to the nearest 4-byte word.-tcpOptionsLength :: [TcpOption] -> (Int,Int)-tcpOptionsLength opts+tcpOptionsSize :: [TcpOption] -> (Int,Int)+tcpOptionsSize opts | left == 0 = (len,0) | otherwise = (len + 1,4 - left) where- (len,left) = F.foldl' (\acc o -> acc + tcpOptionLength o) 0 opts `quotRem` 4+ (len,left) = F.foldl' (\acc o -> acc + tcpOptionSize o) 0 opts `quotRem` 4 -tcpOptionLength :: TcpOption -> Int-tcpOptionLength OptEndOfOptions{} = 1-tcpOptionLength OptNoOption{} = 1-tcpOptionLength OptMaxSegmentSize{} = 4-tcpOptionLength OptWindowScaling{} = 3-tcpOptionLength OptSackPermitted{} = 2-tcpOptionLength (OptSack bs) = sackLength bs-tcpOptionLength OptTimestamp{} = 10-tcpOptionLength (OptUnknown _ len _) = fromIntegral len+tcpOptionSize :: TcpOption -> Int+tcpOptionSize OptEndOfOptions{} = 1+tcpOptionSize OptNoOption{} = 1+tcpOptionSize OptMaxSegmentSize{} = 4+tcpOptionSize OptWindowScaling{} = 3+tcpOptionSize OptSackPermitted{} = 2+tcpOptionSize (OptSack bs) = sackLength bs+tcpOptionSize OptTimestamp{} = 10+tcpOptionSize (OptUnknown _ len _) = fromIntegral len putTcpOption :: Putter TcpOption
src/Hans/Tcp/SendWindow.hs view
@@ -9,6 +9,7 @@ sndUna, sndWnd, nullWindow,+ fullWindow, flushWindow, -- ** Timestamp Clock@@ -171,6 +172,10 @@ -- | True when the window is empty. nullWindow :: Window -> Bool nullWindow Window { .. } = null wRetransmitQueue++-- | True when the window is full.+fullWindow :: Window -> Bool+fullWindow Window { .. } = wSndAvail == 0 -- | The value of SND.NXT. --
src/Hans/Tcp/State.hs view
@@ -2,6 +2,7 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE LambdaCase #-}+{-# LANGUAGE BangPatterns #-} module Hans.Tcp.State ( -- * Tcp State@@ -79,10 +80,10 @@ {-# INLINE listenKey #-} -data Key = Key !Addr -- ^ Remote address- !TcpPort -- ^ Remote port- !Addr -- ^ Local address- !TcpPort -- ^ Local port+data Key = Key !Addr -- Remote address+ !TcpPort -- Remote port+ !Addr -- Local address+ !TcpPort -- Local port deriving (Show,Eq,Ord,Generic) tcbKey :: Getting r Tcb Key
src/Hans/Tcp/Tcb.hs view
@@ -54,6 +54,7 @@ -- ** Windowing queueBytes,+ haveBytesAvail, receiveBytes, tryReceiveBytes, -- * TimeWait TCBs@@ -77,6 +78,7 @@ import Data.IORef (IORef,newIORef,atomicModifyIORef',readIORef ,atomicWriteIORef)+import Data.Int (Int64) import qualified Data.Sequence as Seq import Data.Time.Clock (NominalDiffTime,getCurrentTime) import Data.Word (Word16,Word32)@@ -403,7 +405,7 @@ , tcbRemote :: !Addr -- ^ Remote host -- Fragmentation information- , tcbMss :: !(IORef Int) -- ^ Maximum segment size+ , tcbMss :: !(IORef Int64) -- ^ Maximum segment size -- Timers , tcbTimers :: !(IORef TcpTimers)@@ -446,7 +448,7 @@ tcbRcvUp <- newIORef 0 tcbNeedsDelayedAck <- newIORef False tcbIrs <- newIORef 0- tcbMss <- newIORef cfgTcpInitialMSS+ tcbMss <- newIORef (fromIntegral cfgTcpInitialMSS) tcbTimers <- newIORef emptyTcpTimers tcbTSRecent <- newIORef 0@@ -482,6 +484,11 @@ -- | Queue bytes in the receive buffer. queueBytes :: S.ByteString -> Tcb -> IO () queueBytes bytes Tcb { .. } = Stream.putBytes bytes tcbRecvBuffer++-- | Determine if there are bytes in the receive buffer that can be read.+haveBytesAvail :: Tcb -> IO Bool+haveBytesAvail Tcb { .. } =+ Stream.bytesAvailable tcbRecvBuffer -- | Remove data from the receive buffer, and move the right-side of the receive -- window. Reading 0 bytes indicates that the remote side has closed the
+ tests/Tests/Network.hs view
@@ -0,0 +1,9 @@+module Tests.Network where++import Hans.Network.Types++import Test.QuickCheck (Gen,arbitraryBoundedRandom)+++arbitraryProtocol :: Gen NetworkProtocol+arbitraryProtocol = arbitraryBoundedRandom