diff --git a/hans.cabal b/hans.cabal --- a/hans.cabal +++ b/hans.cabal @@ -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 diff --git a/src/Hans/Buffer/Datagram.hs b/src/Hans/Buffer/Datagram.hs --- a/src/Hans/Buffer/Datagram.hs +++ b/src/Hans/Buffer/Datagram.hs @@ -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 ---------------------------------------------------------------- diff --git a/src/Hans/Config.hs b/src/Hans/Config.hs --- a/src/Hans/Config.hs +++ b/src/Hans/Config.hs @@ -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 diff --git a/src/Hans/IP4/Packet.hs b/src/Hans/IP4/Packet.hs --- a/src/Hans/IP4/Packet.hs +++ b/src/Hans/IP4/Packet.hs @@ -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 #-} diff --git a/src/Hans/Socket/Handle.hs b/src/Hans/Socket/Handle.hs new file mode 100644 --- /dev/null +++ b/src/Hans/Socket/Handle.hs @@ -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 "" 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 () diff --git a/src/Hans/Socket/Tcp.hs b/src/Hans/Socket/Tcp.hs --- a/src/Hans/Socket/Tcp.hs +++ b/src/Hans/Socket/Tcp.hs @@ -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) diff --git a/src/Hans/Socket/Types.hs b/src/Hans/Socket/Types.hs --- a/src/Hans/Socket/Types.hs +++ b/src/Hans/Socket/Types.hs @@ -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. diff --git a/src/Hans/Socket/Udp.hs b/src/Hans/Socket/Udp.hs --- a/src/Hans/Socket/Udp.hs +++ b/src/Hans/Socket/Udp.hs @@ -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 diff --git a/src/Hans/Tcp/Output.hs b/src/Hans/Tcp/Output.hs --- a/src/Hans/Tcp/Output.hs +++ b/src/Hans/Tcp/Output.hs @@ -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. diff --git a/src/Hans/Tcp/Packet.hs b/src/Hans/Tcp/Packet.hs --- a/src/Hans/Tcp/Packet.hs +++ b/src/Hans/Tcp/Packet.hs @@ -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 diff --git a/src/Hans/Tcp/SendWindow.hs b/src/Hans/Tcp/SendWindow.hs --- a/src/Hans/Tcp/SendWindow.hs +++ b/src/Hans/Tcp/SendWindow.hs @@ -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. -- diff --git a/src/Hans/Tcp/State.hs b/src/Hans/Tcp/State.hs --- a/src/Hans/Tcp/State.hs +++ b/src/Hans/Tcp/State.hs @@ -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 diff --git a/src/Hans/Tcp/Tcb.hs b/src/Hans/Tcp/Tcb.hs --- a/src/Hans/Tcp/Tcb.hs +++ b/src/Hans/Tcp/Tcb.hs @@ -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 diff --git a/tests/Tests/Network.hs b/tests/Tests/Network.hs new file mode 100644 --- /dev/null +++ b/tests/Tests/Network.hs @@ -0,0 +1,9 @@ +module Tests.Network where + +import Hans.Network.Types + +import Test.QuickCheck (Gen,arbitraryBoundedRandom) + + +arbitraryProtocol :: Gen NetworkProtocol +arbitraryProtocol = arbitraryBoundedRandom