diff --git a/CHANGELOG.md b/CHANGELOG.md index 2109d8a8..aa4e83f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,14 @@ [#509](https://github.com/haskell/network/pull/509) * Multicast support [#614](https://github.com/haskell/network/pull/614) -* Watchdog scheme for gracefulClose. - [#620](https://github.com/haskell/network/pull/620) +* gracefulClose waits for the peer's FIN through recvBufWith, which is + one code path on every platform, and drains what arrived before it up + to 128 KiB rather than reading once: data left unread makes close send + RST. It therefore keeps to its timeout on Windows, where + System.Timeout cannot cut a receive short and a peer which stayed open + and silent used to hold the caller for as long as it did. This + replaces the watchdog scheme of + [#620](https://github.com/haskell/network/pull/620). * Support for getHostName [#621](https://github.com/haskell/network/pull/621) * Fixing the misspelled WSAEACCES error description on Windows. @@ -17,8 +23,42 @@ [#626](https://github.com/haskell/network/pull/626) * WINIO: fixing a spurious EOF when a recv completes synchronously. [#626](https://github.com/haskell/network/pull/626) +* WINIO: raising the ICMP unreachable of a connected datagram socket + rather than taking it as an empty datagram. +* Adding acceptWith, recvWith, recvFromWith, recvBufWith and recvBufFromWith, which take an STM action + beside the operation and let the caller give up on it without closing + the socket or interrupting the thread, and withReadable, which they are + built from. +* Adding isClosedSocketError and isTransientAcceptError, which name what + an accept loop has to tell apart: a socket closed under it, one queued + connection which failed, and the rest. +* On Windows, the Winsock error code is kept in ioe_errno rather than + dropped. Winsock reports every error as OtherError, so the code was + the only thing telling them apart and it was only in the description. + isClosedSocketError and isTransientAcceptError read it, so an accept + loop there now stops when its socket is closed and retries a + connection the peer gave up on. +* On Windows, a datagram socket hears an ICMP unreachable which a send + of its own caused only once it is connected, as on POSIX. Winsock + reports it to every datagram socket, so one reply bouncing off a + client which had gone ended a server's receive loop; socket turns that + off and connect turns it back on. +* The readiness based waitReadSocketSTM and waitWriteSocketSTM cannot + work on Windows, where completion ports report that an operation has + finished rather than that one could be started, and now say so + instead of blocking forever. +* Add unsafeClose + [#629](https://github.com/haskell/network/pull/629) +* Add unsafeGracefulClose and recvMsgWith +* Return the bytes sent from sendMsg on Windows +* Add withWritable, connectWith and the send functions which the caller can give up on +* Raise the error which the first read of recvWith meets, rather than dropping it and waiting +* Add recvBufMsgWith, sendBufMsgWith, the give-up functions of Network.Socket.Address and of Network.Socket.ByteString.Lazy +* On Windows, giving up on an operation no longer cancels other threads' operations on the socket +* recvBufFromWith takes Ptr a, as recvBufFrom does * Emulate socketPair with TCP over loopback on Windows [#630](https://github.com/haskell/network/pull/630) +* Add isPortUnreachableError ## Version 3.2.9.0 diff --git a/Network/Socket.hs b/Network/Socket.hs index e219a68d..f35ff4b3 100644 --- a/Network/Socket.hs +++ b/Network/Socket.hs @@ -129,7 +129,9 @@ module Network.Socket ( -- ** Closing close, close', + unsafeClose, gracefulClose, + unsafeGracefulClose, shutdown, ShutdownCmd (..), @@ -401,7 +403,27 @@ module Network.Socket ( -- * Special constants maxListenQueue, - -- * STM to check read and write + -- * Event polling + -- $eventpolling + + -- ** Giving up on a blocked operation + withReadable, + withWritable, + acceptWith, + connectWith, + recvBufWith, + recvBufFromWith, + sendBufWith, + sendBufToWith, + sendBufMsgWith, + recvBufMsgWith, + + -- ** What a given-up operation can fail with + isClosedSocketError, + isTransientAcceptError, + isPortUnreachableError, + + -- ** Waiting for readiness (POSIX only) waitReadSocketSTM, waitAndCancelReadSocketSTM, waitWriteSocketSTM, @@ -441,8 +463,68 @@ import Network.Socket.SockAddr import Network.Socket.Syscall hiding (accept, bind, connect) import Network.Socket.Types import Network.Socket.Unix +import Network.Socket.Wait ( + recvBufFromWith, + isClosedSocketError, + isTransientAcceptError, + isPortUnreachableError, + recvBufWith, + sendBufWith, + withReadable, + withWritable, + ) #if !defined(mingw32_HOST_OS) import Network.Socket.Posix.Cmsg #else import Network.Socket.Win32.Cmsg #endif + +-- $eventpolling +-- +-- A thread blocked on a socket is stopped here by being told, not by +-- having the socket closed under it and not by an asynchronous exception +-- thrown into it. Each of 'withReadable', 'acceptWith', 'recvBufWith' +-- and 'recvBufFromWith' takes an 'Control.Concurrent.STM.STM' action +-- beside the operation; whichever settles first decides, 'Left' when the +-- caller gave up and 'Right' when the operation completed, and an +-- operation which has completed wins a tie. They are for one reader at +-- a time, and 'withReadable' says what else they ask of the caller. +-- 'withWritable', 'connectWith', 'sendBufWith' and 'sendBufToWith' do the +-- same for writing, one writer at a time; a stream given up on in the +-- middle of a send is to be closed. +-- +-- The other two ways are going away or gone. A close ends a wait only +-- because the event manager tells the waiter so, which 'GHC.Conc.closeFdWith' +-- is to stop doing, and a descriptor number is reused; an asynchronous +-- exception does not reach a thread blocked in a socket call on Windows +-- at all. +-- +-- A server loop which is told to stop, rather than having its listening +-- socket closed under it: +-- +-- > stopping <- newTVarIO False +-- > -- the shutdown handler runs: atomically $ writeTVar stopping True +-- > let loop = acceptWith sock (check =<< readTVar stopping) >>= \case +-- > Left () -> return () +-- > Right (conn, _) -> forkIO (serve conn) >> loop +-- +-- A receive which gives up on a deadline, where 'System.Timeout.timeout' +-- cannot cut one short: +-- +-- > late <- registerDelay (milliseconds * 1000) +-- > r <- recvBufWith s buf size (check =<< readTVar late) +-- +-- The 'Control.Concurrent.STM.STM' action is run more than once, so it +-- has to be a test rather than something which takes what it reads. +-- +-- Readiness, which these are built from on POSIX, is not available on +-- Windows: completion ports report that an operation finished, not that +-- one could be started. The four readiness functions say so rather than +-- blocking for good, and what runs there instead is the operation itself, +-- on a thread of its own, cancelled with @CancelIoEx@ when the caller +-- gives up. Those for particular operations cancel their own alone, +-- and another thread's operation on the socket carries on; the action +-- given to 'withReadable' or 'withWritable' cannot be told apart, so its +-- cancel ends every operation pending on the socket. +-- "Network.Socket.ByteString" has @recvWith@ and @recvFromWith@ for +-- 'Data.ByteString.ByteString'. diff --git a/Network/Socket/Address.hs b/Network/Socket/Address.hs index f1d90005..b083b372 100644 --- a/Network/Socket/Address.hs +++ b/Network/Socket/Address.hs @@ -7,21 +7,30 @@ module Network.Socket.Address ( -- * Socket operations connect, + connectWith, bind, accept, + acceptWith, -- * Sending and receiving ByteString sendTo, sendAllTo, + sendToWith, recvFrom, + recvWith, + recvFromWith, -- * Sending and receiving data from a buffer sendBufTo, + sendBufToWith, recvBufFrom, + recvBufFromWith, -- * Advanced IO sendBufMsg, + sendBufMsgWith, recvBufMsg, + recvBufMsgWith, ) where import Network.Socket.Buffer @@ -29,3 +38,4 @@ import Network.Socket.ByteString.IO import Network.Socket.Name import Network.Socket.Syscall import Network.Socket.Types +import Network.Socket.Wait diff --git a/Network/Socket/Buffer.hsc b/Network/Socket/Buffer.hsc index ef95d190..94b0ef96 100644 --- a/Network/Socket/Buffer.hsc +++ b/Network/Socket/Buffer.hsc @@ -15,14 +15,19 @@ module Network.Socket.Buffer ( , recvBufFrom , recvBuf , recvBufNoWait + , recvBufNoWaitOr , sendBufMsg , recvBufMsg + , sendBufToOv + , sendBufOv + , recvBufFromOv + , recvBufOv + , sendBufMsgOv + , recvBufMsgOv ) where #if !defined(mingw32_HOST_OS) import Foreign.C.Error (getErrno, eAGAIN, eWOULDBLOCK) -#else -import Foreign.Ptr (nullPtr) #endif import Foreign.Marshal.Alloc (alloca, allocaBytes) import Foreign.Marshal.Utils (with) @@ -37,6 +42,7 @@ import Network.Socket.Win32.MsgHdr import Network.Socket.Win32.WSABuf ## if __IO_MANAGER_WINIO__ >= 2 import qualified GHC.Event.Windows as Mgr +import Network.Socket.Options (getSocketType) import GHC.IO.SubSystem (()) import Foreign.Ptr (wordPtrToPtr) ## endif @@ -89,7 +95,20 @@ sendBufTo :: SocketAddress sa => -> Int -- Data to send -> sa -> IO Int -- Number of Bytes sent -sendBufTo s ptr nbytes sa = +sendBufTo = sendBufToOv nullPtr + +-- | 'sendBufTo' issued with an OVERLAPPED on Windows, unless it is +-- 'nullPtr', so that cancelling it reaches nothing else. The +-- OVERLAPPED is ignored elsewhere. So are the others ending in @Ov@. +sendBufToOv :: SocketAddress sa => Ptr () -> Socket -> Ptr a -> Int -> sa -> IO Int +#if defined(mingw32_HOST_OS) +sendBufToOv ov s ptr nbytes sa | ov /= nullPtr = + withSocketAddress sa $ \p_sa siz -> fmap fromIntegral $ + withFdSocket s $ \fd -> + throwSocketErrorWaitWrite s "Network.Socket.sendBufTo" $ + c_ovl_sendto fd (castPtr ptr) (fromIntegral nbytes) (castPtr p_sa) (fromIntegral siz) ov +#endif +sendBufToOv _ s ptr nbytes sa = ##if __IO_MANAGER_WINIO__ >= 2 sendBufToMIO s ptr nbytes sa sendBufToWinIO s ptr nbytes sa ##else @@ -144,7 +163,17 @@ sendBuf :: Socket -- Bound/Connected Socket -> Ptr Word8 -- Pointer to the data to send -> Int -- Length of the buffer -> IO Int -- Number of Bytes sent -sendBuf s str len = +sendBuf = sendBufOv nullPtr + +-- | 'sendBuf' issued with an OVERLAPPED: see 'sendBufToOv'. +sendBufOv :: Ptr () -> Socket -> Ptr Word8 -> Int -> IO Int +#if defined(mingw32_HOST_OS) +sendBufOv ov s str len | ov /= nullPtr = + fmap fromIntegral $ withFdSocket s $ \fd -> + throwSocketErrorWaitWrite s "Network.Socket.sendBuf" $ + c_ovl_send fd (castPtr str) (fromIntegral len) ov +#endif +sendBufOv _ s str len = #if defined(mingw32_HOST_OS) ## if __IO_MANAGER_WINIO__ >= 2 sendBufMIO s str len sendBufWinIO s str len @@ -204,8 +233,23 @@ sendBufWinIO s ptr len = withFdSocket s $ \sock -> -- NOTE: blocking on Windows unless you compile with -threaded (see -- GHC ticket #1129) recvBufFrom :: SocketAddress sa => Socket -> Ptr a -> Int -> IO (Int, sa) -recvBufFrom s ptr nbytes +recvBufFrom = recvBufFromOv nullPtr + +-- | 'recvBufFrom' issued with an OVERLAPPED: see 'sendBufToOv'. +recvBufFromOv :: SocketAddress sa => Ptr () -> Socket -> Ptr a -> Int -> IO (Int, sa) +recvBufFromOv _ov s ptr nbytes | nbytes <= 0 = ioError (mkInvalidRecvArgError "Network.Socket.recvBufFrom") +#if defined(mingw32_HOST_OS) + | _ov /= nullPtr = + withNewSocketAddress $ \ptr_sa sz -> alloca $ \ptr_len -> + withFdSocket s $ \fd -> do + poke ptr_len (fromIntegral sz) + len <- throwSocketErrorWaitRead s "Network.Socket.recvBufFrom" $ + c_ovl_recvfrom fd (castPtr ptr) (fromIntegral nbytes) (castPtr ptr_sa) ptr_len _ov + sockaddr <- peekSocketAddress ptr_sa + `catchIOError` \_ -> getPeerName s + return (fromIntegral len, sockaddr) +#endif | otherwise = ##if __IO_MANAGER_WINIO__ >= 2 recvBufFromImpl s ptr nbytes recvBufFromWinIO s ptr nbytes @@ -265,7 +309,7 @@ recvBufFromWinIO s ptr nbytes = completionCB err dwBytes | err == _ERROR_SUCCESS = Mgr.ioSuccess $ fromIntegral dwBytes | err == _WSAECONNABORTED = Mgr.ioSuccess 0 - | err == _WSAECONNRESET = Mgr.ioSuccess 0 + | err == _WSAECONNRESET = connResetWinIO s (Mgr.ioFailed err) | err == _WSAEDISCON = Mgr.ioSuccess 0 | err == _ERROR_HANDLE_EOF = Mgr.ioSuccess 0 | err == _ERROR_BROKEN_PIPE = Mgr.ioSuccess 0 @@ -288,8 +332,17 @@ recvBufFromWinIO s ptr nbytes = -- EOF. Historical note: Version 2.8.x.y or earlier, -- an EOF error was thrown. This was changed in version 3.0. recvBuf :: Socket -> Ptr Word8 -> Int -> IO Int -recvBuf s ptr nbytes +recvBuf = recvBufOv nullPtr + +-- | 'recvBuf' issued with an OVERLAPPED: see 'sendBufToOv'. +recvBufOv :: Ptr () -> Socket -> Ptr Word8 -> Int -> IO Int +recvBufOv _ov s ptr nbytes | nbytes <= 0 = ioError (mkInvalidRecvArgError "Network.Socket.recvBuf") +#if defined(mingw32_HOST_OS) + | _ov /= nullPtr = withFdSocket s $ \fd -> fromIntegral <$> + throwSocketErrorWaitRead s "Network.Socket.recvBuf" + (c_ovl_recv fd (castPtr ptr) (fromIntegral nbytes) _ov) +#endif | otherwise = do #if defined(mingw32_HOST_OS) ## if __IO_MANAGER_WINIO__ >= 2 @@ -348,7 +401,7 @@ recvBufWinIO s ptr nbytes = withFdSocket s $ \sock -> completionCB err dwBytes | err == _ERROR_SUCCESS = Mgr.ioSuccess $ fromIntegral dwBytes | err == _WSAECONNABORTED = Mgr.ioSuccess 0 - | err == _WSAECONNRESET = Mgr.ioSuccess 0 + | err == _WSAECONNRESET = connResetWinIO s (Mgr.ioFailed err) | err == _WSAEDISCON = Mgr.ioSuccess 0 | err == _ERROR_HANDLE_EOF = Mgr.ioSuccess 0 | err == _ERROR_BROKEN_PIPE = Mgr.ioSuccess 0 @@ -356,6 +409,13 @@ recvBufWinIO s ptr nbytes = withFdSocket s $ \sock -> | err == _ERROR_OPERATION_ABORTED = Mgr.ioSuccess 0 | err == _ERROR_IO_INCOMPLETE = Mgr.ioSuccess 0 | otherwise = Mgr.ioFailed err + +-- WSAECONNRESET on a datagram socket reports an ICMP unreachable, which +-- POSIX gives as ECONNREFUSED: an error to raise, not the end of a stream. +connResetWinIO :: Num a => Socket -> IO (Mgr.IOResult a) -> IO (Mgr.IOResult a) +connResetWinIO s failed = do + stype <- getSocketType s + if stype == Datagram then failed else Mgr.ioSuccess 0 ## endif /* __IO_MANAGER_WINIO__ */ #endif /* mingw32_HOST_OS */ @@ -365,23 +425,34 @@ recvBufWinIO s ptr nbytes = withFdSocket s $ \sock -> -- -1 is returned in the case of EAGAIN or EWOULDBLOCK. -- -2 is returned in other error cases. recvBufNoWait :: Socket -> Ptr Word8 -> Int -> IO Int -recvBufNoWait s ptr nbytes = withFdSocket s $ \fd -> do +recvBufNoWait = recvBufNoWaitOr $ return (-2) + +-- | 'recvBufNoWait' which runs the action on an error, in place of +-- answering -2. The error is gone from the socket once read, so this +-- is where to raise it. +recvBufNoWaitOr :: IO CInt -> Socket -> Ptr Word8 -> Int -> IO Int +recvBufNoWaitOr onError s ptr nbytes = withFdSocket s $ \fd -> do #if defined(mingw32_HOST_OS) alloca $ \ptr_bytes -> do res <- c_ioctlsocket fd #{const FIONREAD} ptr_bytes avail <- peek ptr_bytes - r <- if res == #{const NO_ERROR} && avail > 0 then - c_recv fd (castPtr ptr) (fromIntegral nbytes) 0{-flags-} + let failed = do + err <- c_WSAGetLastError + if err == #{const WSAEWOULDBLOCK} + || err == #{const WSAEINPROGRESS} then + return (-1) + else + onError + r <- if res == #{const NO_ERROR} && avail > 0 then do + -- A pending error, such as an ICMP unreachable, counts as + -- something to read, and recv takes it. + n <- c_recv fd (castPtr ptr) (fromIntegral nbytes) 0{-flags-} + if n == -1 then failed else return n else if avail == 0 then -- Socket would block, could also mean socket is closed but -- can't distinguish return (-1) - else do err <- c_WSAGetLastError - if err == #{const WSAEWOULDBLOCK} - || err == #{const WSAEINPROGRESS} then - return (-1) - else - return (-2) + else failed return $ fromIntegral r #else @@ -393,7 +464,7 @@ recvBufNoWait s ptr nbytes = withFdSocket s $ \fd -> do if err == eAGAIN || err == eWOULDBLOCK then return (-1) else - return (-2) + fromIntegral <$> onError #endif mkInvalidRecvArgError :: String -> IOError @@ -409,7 +480,12 @@ sendBufMsg :: SocketAddress sa -> [Cmsg] -- ^ Control messages -> MsgFlag -- ^ Message flags -> IO Int -- ^ The length actually sent -sendBufMsg s sa bufsizs cmsgs flags = do +sendBufMsg = sendBufMsgOv nullPtr + +-- | 'sendBufMsg' issued with an OVERLAPPED: see 'sendBufToOv'. +sendBufMsgOv :: SocketAddress sa + => Ptr () -> Socket -> sa -> [(Ptr Word8,Int)] -> [Cmsg] -> MsgFlag -> IO Int +sendBufMsgOv _ov s sa bufsizs cmsgs flags = do sz <- withSocketAddress sa $ \addrPtr addrSize -> #if !defined(mingw32_HOST_OS) withIOVec bufsizs $ \(iovsPtr, iovsLen) -> do @@ -435,6 +511,7 @@ sendBufMsg s sa bufsizs cmsgs flags = do withFdSocket s $ \fd -> with msgHdr $ \msgHdrPtr -> #if defined(mingw32_HOST_OS) + if _ov /= nullPtr then sendBufMsgOvl s fd msgHdrPtr cflags _ov else ## if __IO_MANAGER_WINIO__ >= 2 sendBufMsgMIO s fd msgHdrPtr cflags sendBufMsgWinIO fd msgHdrPtr cflags @@ -461,7 +538,12 @@ recvBufMsg :: SocketAddress sa -- 'MSG_CTRUNC' is returned -> MsgFlag -- ^ Message flags -> IO (sa,Int,[Cmsg],MsgFlag) -- ^ Source address, total bytes received, control messages and message flags -recvBufMsg s bufsizs clen flags = do +recvBufMsg = recvBufMsgOv nullPtr + +-- | 'recvBufMsg' issued with an OVERLAPPED: see 'sendBufToOv'. +recvBufMsgOv :: SocketAddress sa + => Ptr () -> Socket -> [(Ptr Word8,Int)] -> Int -> MsgFlag -> IO (sa,Int,[Cmsg],MsgFlag) +recvBufMsgOv _ov s bufsizs clen flags = do withNewSocketAddress $ \addrPtr addrSize -> allocaBytes clen $ \ctrlPtr -> #if !defined(mingw32_HOST_OS) @@ -495,8 +577,9 @@ recvBufMsg s bufsizs clen flags = do throwSocketErrorWaitRead s "Network.Socket.Buffer.recvmsg" $ c_recvmsg fd msgHdrPtr _cflags #else + if _ov /= nullPtr then recvBufMsgOvl s fd msgHdrPtr _ov else ## if __IO_MANAGER_WINIO__ >= 2 - recvBufMsgMIO s fd msgHdrPtr recvBufMsgWinIO fd msgHdrPtr + recvBufMsgMIO s fd msgHdrPtr recvBufMsgWinIO s fd msgHdrPtr ## else recvBufMsgMIO s fd msgHdrPtr ## endif @@ -552,9 +635,42 @@ clearCtrlOnTruncation msgHdrPtr ret = when (ret == -1) $ do sendBufMsgMIO :: Socket -> CSocket -> Ptr (MsgHdr sa) -> CInt -> IO CInt sendBufMsgMIO s fd msgHdrPtr cflags = do sendMsg <- mkSendMsgSafe <$> getWSASendMsg fd - throwSocketErrorWaitWrite s "Network.Socket.Buffer.sendMsg" $ - alloca $ \send_ptr -> + -- WSASendMsg returns 0 and puts the bytes sent here. + alloca $ \send_ptr -> do + _ <- throwSocketErrorWaitWrite s "Network.Socket.Buffer.sendMsg" $ sendMsg fd (castPtr msgHdrPtr) (fromIntegral cflags) send_ptr nullPtr nullPtr + fromIntegral <$> peek send_ptr + +-- The operations given an OVERLAPPED, under either IO manager: issued +-- with it and waited for in C, so that cancelling it reaches this alone. +sendBufMsgOvl :: Socket -> CSocket -> Ptr (MsgHdr sa) -> CInt -> Ptr () -> IO CInt +sendBufMsgOvl s fd msgHdrPtr cflags ov = + alloca $ \send_ptr -> do + fn <- getWSASendMsg fd + _ <- throwSocketErrorWaitWrite s "Network.Socket.Buffer.sendMsg" $ + c_ovl_sendmsg fn fd msgHdrPtr (fromIntegral cflags) send_ptr ov + fromIntegral <$> peek send_ptr + +recvBufMsgOvl :: Socket -> CSocket -> Ptr (MsgHdr sa) -> Ptr () -> IO Int +recvBufMsgOvl s fd msgHdrPtr ov = + alloca $ \len_ptr -> do + fn <- getWSARecvMsg fd + _ <- throwSocketErrorWaitReadBut (== #{const WSAEMSGSIZE}) s "Network.Socket.Buffer.recvmsg" $ + c_ovl_recvmsg fn fd msgHdrPtr len_ptr ov + fromIntegral <$> peek len_ptr + +foreign import ccall safe "hsnet_ovl_recv" + c_ovl_recv :: CSocket -> Ptr CChar -> CInt -> Ptr () -> IO CInt +foreign import ccall safe "hsnet_ovl_send" + c_ovl_send :: CSocket -> Ptr CChar -> CInt -> Ptr () -> IO CInt +foreign import ccall safe "hsnet_ovl_recvfrom" + c_ovl_recvfrom :: CSocket -> Ptr CChar -> CInt -> Ptr () -> Ptr CInt -> Ptr () -> IO CInt +foreign import ccall safe "hsnet_ovl_sendto" + c_ovl_sendto :: CSocket -> Ptr CChar -> CInt -> Ptr () -> CInt -> Ptr () -> IO CInt +foreign import ccall safe "hsnet_ovl_sendmsg" + c_ovl_sendmsg :: FunPtr WSASendMsgFn -> CSocket -> Ptr (MsgHdr sa) -> DWORD -> LPDWORD -> Ptr () -> IO CInt +foreign import ccall safe "hsnet_ovl_recvmsg" + c_ovl_recvmsg :: FunPtr WSARecvMsgFn -> CSocket -> Ptr (MsgHdr sa) -> LPDWORD -> Ptr () -> IO CInt ## if __IO_MANAGER_WINIO__ >= 2 sendBufMsgWinIO :: CSocket -> Ptr (MsgHdr sa) -> CInt -> IO CInt @@ -588,8 +704,8 @@ recvBufMsgMIO s fd msgHdrPtr = do fromIntegral <$> peek len_ptr ## if __IO_MANAGER_WINIO__ >= 2 -recvBufMsgWinIO :: CSocket -> Ptr (MsgHdr sa) -> IO Int -recvBufMsgWinIO fd msgHdrPtr = do +recvBufMsgWinIO :: Socket -> CSocket -> Ptr (MsgHdr sa) -> IO Int +recvBufMsgWinIO s fd msgHdrPtr = do -- Perform async WSARecvMsg using withOverlapped -- (socket already associated in socket creation) recvMsg <- mkRecvMsgUnsafe <$> getWSARecvMsg fd @@ -621,7 +737,7 @@ recvBufMsgWinIO fd msgHdrPtr = do | err == _ERROR_SUCCESS = Mgr.ioSuccess $ fromIntegral dwBytes | err == _WSAEMSGSIZE = Mgr.ioSuccess $ fromIntegral dwBytes | err == _STATUS_BUFFER_OVERFLOW = Mgr.ioSuccess $ fromIntegral dwBytes -- truncated msg - | err == _WSAECONNRESET = Mgr.ioSuccess 0 + | err == _WSAECONNRESET = connResetWinIO s (Mgr.ioFailed err) | err == _WSAECONNABORTED = Mgr.ioSuccess 0 | err == _WSAESHUTDOWN = Mgr.ioSuccess 0 | err == _WSAEDISCON = Mgr.ioSuccess 0 diff --git a/Network/Socket/ByteString.hs b/Network/Socket/ByteString.hs index 42424e94..b42002d4 100644 --- a/Network/Socket/ByteString.hs +++ b/Network/Socket/ByteString.hs @@ -22,25 +22,40 @@ module Network.Socket.ByteString ( sendAll, sendTo, sendAllTo, + sendWith, + sendAllWith, + sendToWith, -- ** Vectored I/O -- $vectored sendMany, sendManyTo, sendManyWithFds, + sendManyWith, -- * Receive data from a socket recv, recvFrom, + recvWith, + recvFromWith, -- * Advanced send and recv sendMsg, + sendMsgWith, recvMsg, + recvMsgWith, ) where +import Control.Concurrent.STM (STM) import Data.ByteString (ByteString) -import Network.Socket.ByteString.IO hiding (recvFrom, sendAllTo, sendTo) +import Network.Socket.ByteString.IO hiding ( + recvFrom, + recvFromWith, + sendAllTo, + sendTo, + sendToWith, + ) import qualified Network.Socket.ByteString.IO as G import Network.Socket.SockAddr (annotateWithSocket) import Network.Socket.Types @@ -90,3 +105,18 @@ sendAllTo s bs sa = G.sendAllTo s bs sa `annotateWithSocket` (s, Just sa) -- 'SockAddr' representing the address of the sending socket. recvFrom :: Socket -> Int -> IO (ByteString, SockAddr) recvFrom s len = G.recvFrom s len `annotateWithSocket` (s, Nothing) + +-- | 'recvFrom' which the caller can give up on. +-- +-- @since 3.2.10.0 +recvFromWith + :: Socket -> Int -> STM a -> IO (Either a (ByteString, SockAddr)) +recvFromWith s len giveUp = + G.recvFromWith s len giveUp `annotateWithSocket` (s, Nothing) + +-- | 'sendTo' which the caller can give up on. +-- +-- @since 3.2.12.0 +sendToWith :: Socket -> ByteString -> SockAddr -> STM a -> IO (Either a Int) +sendToWith s bs sa giveUp = + G.sendToWith s bs sa giveUp `annotateWithSocket` (s, Just sa) diff --git a/Network/Socket/ByteString/IO.hsc b/Network/Socket/ByteString/IO.hsc index a600b26c..67e6dc97 100644 --- a/Network/Socket/ByteString/IO.hsc +++ b/Network/Socket/ByteString/IO.hsc @@ -19,20 +19,28 @@ module Network.Socket.ByteString.IO , sendAll , sendTo , sendAllTo + , sendWith + , sendAllWith + , sendToWith -- ** Vectored I/O -- $vectored , sendMany , sendManyTo , sendManyWithFds + , sendManyWith -- * Receive data from a socket , recv , recvFrom + , recvWith + , recvFromWith + , recvMsgWith , waitWhen0 -- * Advanced send and recv , sendMsg + , sendMsgWith , recvMsg , MsgFlag(..) , Cmsg(..) @@ -45,10 +53,13 @@ import Data.ByteString.Internal (createAndTrim) import Data.ByteString.Unsafe (unsafeUseAsCStringLen) import Foreign.Marshal.Alloc (allocaBytes) +import Control.Concurrent.STM (STM) + import Network.Socket.Buffer import Network.Socket.ByteString.Internal import Network.Socket.Imports import Network.Socket.Types +import Network.Socket.Wait import Data.ByteString.Internal (create, ByteString(..)) import Foreign.ForeignPtr (withForeignPtr) @@ -82,10 +93,15 @@ import Network.Socket.Win32.MsgHdr (MsgHdr(..)) send :: Socket -- ^ Connected socket -> ByteString -- ^ Data to send -> IO Int -- ^ Number of bytes sent -send s xs = send' `annotateWithSocket` (s, Nothing) +send = sendOv nullPtr + +-- The give-ups below issue their operations with an OVERLAPPED of their +-- own on Windows, through these: see 'Network.Socket.Buffer.sendBufToOv'. +sendOv :: Ptr () -> Socket -> ByteString -> IO Int +sendOv ov s xs = send' `annotateWithSocket` (s, Nothing) where send' = unsafeUseAsCStringLen xs $ \(str, len) -> - sendBuf s (castPtr str) len + sendBufOv ov s (castPtr str) len waitWhen0 :: Int -> Socket -> IO () waitWhen0 0 s = when rtsSupportsBoundThreads $ @@ -109,6 +125,35 @@ sendAll s bs0 = loop bs0 waitWhen0 sent s when (sent /= B.length bs) $ loop $ B.drop sent bs +-- | 'send' which the caller can give up on. See +-- 'Network.Socket.withWritable' for what is left of the stream after +-- giving up. +-- +-- @since 3.2.12.0 +sendWith :: Socket -> ByteString -> STM a -> IO (Either a Int) +sendWith s xs giveUp = withWritableOv s giveUp $ \ov -> sendOv ov s xs + +-- | 'sendAll' which the caller can give up on. Part of the data may +-- have been sent before it gives up, so a stream given up on is to be +-- closed. +-- +-- @since 3.2.12.0 +sendAllWith :: Socket -> ByteString -> STM a -> IO (Either a ()) +sendAllWith _ "" _ = return $ Right () +sendAllWith s bs giveUp = do + r <- sendWith s bs giveUp + case r of + Left a -> return $ Left a + Right sent -> sendAllWith s (B.drop sent bs) giveUp + +-- | 'sendTo' which the caller can give up on. +-- +-- @since 3.2.12.0 +sendToWith + :: SocketAddress sa + => Socket -> ByteString -> sa -> STM a -> IO (Either a Int) +sendToWith s xs sa giveUp = withWritableOv s giveUp $ \ov -> sendToOv ov s xs sa + -- | Send data to the socket. The recipient can be specified -- explicitly, so the socket need not be in a connected state. -- Returns the number of bytes sent. Applications are responsible for @@ -118,8 +163,11 @@ sendTo :: SocketAddress sa => -> ByteString -- ^ Data to send -> sa -- ^ Recipient address -> IO Int -- ^ Number of bytes sent -sendTo s xs sa = - unsafeUseAsCStringLen xs $ \(str, len) -> sendBufTo s str len sa +sendTo = sendToOv nullPtr + +sendToOv :: SocketAddress sa => Ptr () -> Socket -> ByteString -> sa -> IO Int +sendToOv ov s xs sa = + unsafeUseAsCStringLen xs $ \(str, len) -> sendBufToOv ov s str len sa -- | Send data to the socket. The recipient can be specified -- explicitly, so the socket need not be in a connected state. Unlike @@ -154,10 +202,16 @@ sendMany _ [] = return () sendMany s cs = sendMany' `annotateWithSocket` (s, Nothing) where sendMany' = do - sent <- sendManyInner + sent <- sendManyOnce s cs waitWhen0 sent s when (sent >= 0) $ sendMany s $ remainingChunks sent cs - sendManyInner = + +-- One gathering write, which may send only part of the data. +sendManyOnce :: Socket -> [ByteString] -> IO Int +sendManyOnce = sendManyOnceOv nullPtr + +sendManyOnceOv :: Ptr () -> Socket -> [ByteString] -> IO Int +sendManyOnceOv _ov s cs = #if !defined(mingw32_HOST_OS) fmap fromIntegral . withIOVecfromBS cs $ \(iovsPtr, iovsLen) -> withFdSocket s $ \fd -> do @@ -168,12 +222,27 @@ sendMany s cs = sendMany' `annotateWithSocket` (s, Nothing) fmap fromIntegral . withWSABuffromBS cs $ \(wsabsPtr, wsabsLen) -> withFdSocket s $ \fd -> do let len = fromIntegral wsabsLen - alloca $ \send_ptr -> do - _ <- throwSocketErrorWaitWrite s "Network.Socket.ByteString.sendMany" $ - c_wsasend fd wsabsPtr len send_ptr 0 nullPtr nullPtr - peek send_ptr + if _ov /= nullPtr + then throwSocketErrorWaitWrite s "Network.Socket.ByteString.sendMany" $ + c_ovl_sendv fd wsabsPtr len _ov + else alloca $ \send_ptr -> do + _ <- throwSocketErrorWaitWrite s "Network.Socket.ByteString.sendMany" $ + c_wsasend fd wsabsPtr len send_ptr 0 nullPtr nullPtr + fromIntegral <$> peek send_ptr #endif +-- | 'sendMany' which the caller can give up on. Parts may have been +-- sent before it gives up, so a stream given up on is to be closed. +-- +-- @since 3.2.12.0 +sendManyWith :: Socket -> [ByteString] -> STM a -> IO (Either a ()) +sendManyWith _ [] _ = return $ Right () +sendManyWith s cs giveUp = do + r <- withWritableOv s giveUp (\ov -> sendManyOnceOv ov s cs) `annotateWithSocket` (s, Nothing) + case r of + Left a -> return $ Left a + Right sent -> sendManyWith s (remainingChunks sent cs) giveUp + -- | Send data to the socket. The recipient can be specified -- explicitly, so the socket need not be in a connected state. The -- data is sent as if the parts have been concatenated. This function @@ -267,7 +336,10 @@ recv :: Socket -- ^ Connected socket -> IO ByteString -- ^ Data received recv s nbytes | nbytes < 0 = ioError (mkInvalidRecvArgError "Network.Socket.ByteString.recv") `annotateWithSocket` (s, Nothing) - | otherwise = createAndTrim nbytes $ \ptr -> recvBuf s ptr nbytes `annotateWithSocket` (s, Nothing) + | otherwise = recvOv nullPtr s nbytes + +recvOv :: Ptr () -> Socket -> Int -> IO ByteString +recvOv ov s nbytes = createAndTrim nbytes $ \ptr -> recvBufOv ov s ptr nbytes `annotateWithSocket` (s, Nothing) -- | Receive data from the socket. The socket need not be in a -- connected state. Returns @(bytes, address)@ where @bytes@ is a @@ -279,12 +351,59 @@ recvFrom :: SocketAddress sa => Socket -- ^ Socket -> Int -- ^ Maximum number of bytes to receive -> IO (ByteString, sa) -- ^ Data received and sender address -recvFrom sock nbytes = +recvFrom = recvFromOv nullPtr + +recvFromOv :: SocketAddress sa => Ptr () -> Socket -> Int -> IO (ByteString, sa) +recvFromOv ov sock nbytes = allocaBytes nbytes $ \ptr -> do - (len, sockaddr) <- recvBufFrom sock ptr nbytes + (len, sockaddr) <- recvBufFromOv ov sock ptr nbytes str <- B.packCStringLen (ptr, len) return (str, sockaddr) +-- | 'recv' which the caller can give up on. Whichever settles first +-- decides: 'Left' when the 'STM' action fires, 'Right' when data +-- arrives. Data already waiting is taken without blocking at all and +-- so wins over a caller which has given up, as a completed operation +-- does in 'Network.Socket.withReadable'; the give-up is read there +-- and nowhere else. +-- +-- @since 3.2.10.0 +recvWith :: Socket -> Int -> STM a -> IO (Either a ByteString) +recvWith s nbytes giveUp + | nbytes < 0 = + ioError (mkInvalidRecvArgError "Network.Socket.ByteString.recvWith") + `annotateWithSocket` (s, Nothing) + | otherwise = do + bs <- createAndTrim' nbytes $ \ptr -> + recvBufNoWaitOr (throwSocketError "Network.Socket.ByteString.recvWith") s ptr nbytes + `annotateWithSocket` (s, Nothing) + case bs of + Just b -> return $ Right b + Nothing -> withReadableOv s giveUp $ \ov -> recvOv ov s nbytes + +-- | 'recvFrom' which the caller can give up on. +-- +-- @since 3.2.10.0 +recvFromWith + :: SocketAddress sa + => Socket + -> Int + -> STM a + -> IO (Either a (ByteString, sa)) +recvFromWith s nbytes giveUp = withReadableOv s giveUp $ \ov -> recvFromOv ov s nbytes + +-- A read which never waits: 'Nothing' when nothing is there yet. +createAndTrim' :: Int -> (Ptr Word8 -> IO Int) -> IO (Maybe ByteString) +createAndTrim' nbytes f = do + (len, bs) <- allocaBytes nbytes $ \ptr -> do + len <- f ptr + if len > 0 + then do + b <- B.packCStringLen (castPtr ptr, len) + return (len, b) + else return (len, B.empty) + return $ if len >= 0 then Just bs else Nothing + -- ---------------------------------------------------------------------------- -- Not exported @@ -331,11 +450,28 @@ sendMsg :: Socket -- ^ Socket -> [Cmsg] -- ^ Control messages -> MsgFlag -- ^ Message flags -> IO Int -- ^ The length actually sent -sendMsg _ _ [] _ _ = return 0 -sendMsg s addr bss cmsgs flags = sendMsg' `annotateWithSocket` (s, Just addr) +sendMsg = sendMsgOv nullPtr + +sendMsgOv :: Ptr () -> Socket -> SockAddr -> [ByteString] -> [Cmsg] -> MsgFlag -> IO Int +sendMsgOv _ _ _ [] _ _ = return 0 +sendMsgOv ov s addr bss cmsgs flags = sendMsg' `annotateWithSocket` (s, Just addr) where sendMsg' = withBufSizs bss $ \bufsizs -> - sendBufMsg s addr bufsizs cmsgs flags + sendBufMsgOv ov s addr bufsizs cmsgs flags + +-- | 'sendMsg' which the caller can give up on. +-- +-- @since 3.2.12.0 +sendMsgWith + :: Socket + -> SockAddr + -> [ByteString] + -> [Cmsg] + -> MsgFlag + -> STM a + -> IO (Either a Int) +sendMsgWith s addr bss cmsgs flags giveUp = + withWritableOv s giveUp $ \ov -> sendMsgOv ov s addr bss cmsgs flags -- | Receive data from the socket using recvmsg(2). recvMsg :: Socket -- ^ Socket @@ -347,12 +483,29 @@ recvMsg :: Socket -- ^ Socket -- 'MSG_CTRUNC' is returned -> MsgFlag -- ^ Message flags -> IO (SockAddr, ByteString, [Cmsg], MsgFlag) -- ^ Source address, received data, control messages and message flags -recvMsg s siz clen flags = recvMsg' `annotateWithSocket` (s, Nothing) +recvMsg = recvMsgOv nullPtr + +recvMsgOv :: Ptr () -> Socket -> Int -> Int -> MsgFlag -> IO (SockAddr, ByteString, [Cmsg], MsgFlag) +recvMsgOv ov s siz clen flags = recvMsg' `annotateWithSocket` (s, Nothing) where recvMsg' = do bs@(PS fptr _ _) <- create siz $ \ptr -> zeroMemory ptr (fromIntegral siz) withForeignPtr fptr $ \ptr -> do - (addr,len,cmsgs,flags') <- recvBufMsg s [(ptr,siz)] clen flags + (addr,len,cmsgs,flags') <- recvBufMsgOv ov s [(ptr,siz)] clen flags let bs' | len < siz = PS fptr 0 len | otherwise = bs return (addr, bs', cmsgs, flags') + +-- | 'recvMsg' which the caller can give up on. +-- +-- @since 3.2.12.0 +recvMsgWith + :: Socket + -> Int + -> Int + -> MsgFlag + -> STM a + -> IO (Either a (SockAddr, ByteString, [Cmsg], MsgFlag)) +recvMsgWith s siz clen flags giveUp = + withReadableOv s giveUp $ \ov -> recvMsgOv ov s siz clen flags + diff --git a/Network/Socket/ByteString/Internal.hs b/Network/Socket/ByteString/Internal.hs index c26cd146..9dc45c78 100644 --- a/Network/Socket/ByteString/Internal.hs +++ b/Network/Socket/ByteString/Internal.hs @@ -18,6 +18,7 @@ module Network.Socket.ByteString.Internal , c_recvmsg #else , c_wsasend + , c_ovl_sendv #endif ) where @@ -63,6 +64,9 @@ foreign import ccall unsafe "recvmsg" -- fixme Handle for SOCKET, see #426 foreign import CALLCONV SAFE_ON_WIN "WSASend" c_wsasend :: CSocket -> Ptr WSABuf -> DWORD -> LPDWORD -> DWORD -> Ptr () -> Ptr () -> IO CInt +-- WSASend with an OVERLAPPED of the caller's, waited for in the call. +foreign import ccall safe "hsnet_ovl_sendv" + c_ovl_sendv :: CSocket -> Ptr WSABuf -> DWORD -> Ptr () -> IO CInt -- WSASendMsg and WSARecvMsg are extension functions and cannot be -- imported by name; see Network.Socket.Win32.Load. #endif diff --git a/Network/Socket/ByteString/Lazy.hs b/Network/Socket/ByteString/Lazy.hs index 316f0782..73251f87 100644 --- a/Network/Socket/ByteString/Lazy.hs +++ b/Network/Socket/ByteString/Lazy.hs @@ -24,12 +24,16 @@ module Network.Socket.ByteString.Lazy ( send, sendAll, sendWithFds, + sendWith, + sendAllWith, -- * Receive data from a socket getContents, recv, + recvWith, ) where +import Control.Concurrent.STM (STM) import Data.ByteString.Lazy.Internal ( ByteString (..), defaultChunkSize, @@ -51,6 +55,9 @@ import qualified Data.ByteString.Lazy as L import qualified Network.Socket.ByteString as N import Network.Socket.Imports import Network.Socket.Types +#if !defined(mingw32_HOST_OS) +import Network.Socket.Wait (withWritable) +#endif -- | Send data and file descriptors over a UNIX-domain socket in -- a single system call. This function does not work on Windows. @@ -64,6 +71,32 @@ sendWithFds -> IO () sendWithFds s lbs fds = N.sendManyWithFds s (L.toChunks lbs) fds +-- | 'send' which the caller can give up on. See +-- 'Network.Socket.withWritable' for what is left of the stream after +-- giving up. +-- +-- @since 3.2.12.0 +sendWith :: Socket -> ByteString -> STM a -> IO (Either a Int64) +#if defined(mingw32_HOST_OS) +-- 'send' sends the first chunk here, and so does this, through the +-- strict one, whose give-up cancels its own operation alone. +sendWith s lbs giveUp = fmap fromIntegral <$> N.sendWith s chunk giveUp + where + chunk = case L.toChunks lbs of + [] -> S.empty + x : _ -> x +#else +sendWith s lbs giveUp = withWritable s giveUp $ send s lbs +#endif + +-- | 'sendAll' which the caller can give up on. Part of the data may +-- have been sent before it gives up, so a stream given up on is to be +-- closed. +-- +-- @since 3.2.12.0 +sendAllWith :: Socket -> ByteString -> STM a -> IO (Either a ()) +sendAllWith s lbs = N.sendManyWith s (L.toChunks lbs) + -- ----------------------------------------------------------------------------- -- Receiving @@ -110,3 +143,13 @@ recv s nbytes = chunk <$> N.recv s (fromIntegral nbytes) chunk k | S.null k = Empty | otherwise = Chunk k Empty + +-- | 'recv' which the caller can give up on. +-- +-- @since 3.2.12.0 +recvWith :: Socket -> Int64 -> STM a -> IO (Either a ByteString) +recvWith s nbytes giveUp = fmap chunk <$> N.recvWith s (fromIntegral nbytes) giveUp + where + chunk k + | S.null k = Empty + | otherwise = Chunk k Empty diff --git a/Network/Socket/Cbits.hsc b/Network/Socket/Cbits.hsc index 7a68a9cd..91a60cad 100644 --- a/Network/Socket/Cbits.hsc +++ b/Network/Socket/Cbits.hsc @@ -13,6 +13,34 @@ maxListenQueue = #const SOMAXCONN #if defined(mingw32_HOST_OS) wsaNotInitialized :: CInt wsaNotInitialized = #const WSANOTINITIALISED +-- | Whether a datagram socket reports an ICMP unreachable which came +-- back from an earlier send. Winsock turns it on by default. +sioUDPConnReset :: Word32 +sioUDPConnReset = #const SIO_UDP_CONNRESET +-- | What closesocket gives a call already blocked on the socket. +wsaEINTR :: CInt +wsaEINTR = #const WSAEINTR +-- | What a socket operation gives once the socket has been closed. +wsaENOTSOCK :: CInt +wsaENOTSOCK = #const WSAENOTSOCK +-- | What accept() gives for a connection the peer gave up on. Winsock +-- names WSAECONNRESET where accept(2) names ECONNABORTED. +wsaECONNABORTED :: CInt +wsaECONNABORTED = #const WSAECONNABORTED +wsaECONNRESET :: CInt +wsaECONNRESET = #const WSAECONNRESET +wsaENETDOWN :: CInt +wsaENETDOWN = #const WSAENETDOWN +wsaENETRESET :: CInt +wsaENETRESET = #const WSAENETRESET +wsaENETUNREACH :: CInt +wsaENETUNREACH = #const WSAENETUNREACH +wsaEHOSTDOWN :: CInt +wsaEHOSTDOWN = #const WSAEHOSTDOWN +wsaEHOSTUNREACH :: CInt +wsaEHOSTUNREACH = #const WSAEHOSTUNREACH +wsaENOPROTOOPT :: CInt +wsaENOPROTOOPT = #const WSAENOPROTOOPT #else fGetFd :: CInt fGetFd = #const F_GETFD diff --git a/Network/Socket/Internal.hs b/Network/Socket/Internal.hs index 51120c3c..3cf78948 100644 --- a/Network/Socket/Internal.hs +++ b/Network/Socket/Internal.hs @@ -62,9 +62,9 @@ import GHC.Conc (threadWaitRead, threadWaitWrite) import qualified Control.Exception as E import System.IO.Unsafe (unsafePerformIO) # if __GLASGOW_HASKELL__ >= 707 -import GHC.IO.Exception (IOErrorType(..)) +import GHC.IO.Exception (IOErrorType(..), IOException(..)) # else -import GHC.IOBase (IOErrorType(..)) +import GHC.IOBase (IOErrorType(..), IOException(..)) # endif import System.IO.Error (ioeSetErrorString, mkIOError) #else @@ -153,13 +153,16 @@ throwSocketError = throwErrno -- | Like 'throwSocketError', but the error code is supplied as an argument. -- --- On Windows, do not use errno. Use a system error code instead. +-- On Windows the code is a Winsock error code rather than an errno. It +-- is kept in 'ioe_errno' all the same: Winsock reports every error as +-- 'OtherError', so the code is the only thing which tells them apart. throwSocketErrorCode :: String -> CInt -> IO a #if defined(mingw32_HOST_OS) throwSocketErrorCode name rc = do pstr <- c_getWSError rc str <- peekCString pstr - ioError (ioeSetErrorString (mkIOError OtherError name Nothing Nothing) str) + let ioe = (mkIOError OtherError name Nothing Nothing){ioe_errno = Just rc} + ioError (ioeSetErrorString ioe str) #else throwSocketErrorCode loc errno = ioError (errnoToIOError loc (Errno errno) Nothing Nothing) diff --git a/Network/Socket/STM.hs b/Network/Socket/STM.hs index 6f2419b4..4d3c9ae0 100644 --- a/Network/Socket/STM.hs +++ b/Network/Socket/STM.hs @@ -1,24 +1,72 @@ -module Network.Socket.STM where +{-# LANGUAGE CPP #-} -import Control.Concurrent -import Control.Concurrent.STM +-- | Waiting for a socket to become ready, through STM. +-- +-- This mirrors @select@\/@epoll@ and is __POSIX only__: Windows +-- completion ports report that an operation has finished, not that one +-- could be started, so there is no readiness to wait for. All four +-- throw there. What runs on Windows instead is +-- "Network.Socket.Win32.Cancel", through 'Network.Socket.withReadable'. +-- +module Network.Socket.STM ( + -- * Waiting for readiness (POSIX only) + waitReadSocketSTM, + waitAndCancelReadSocketSTM, + waitWriteSocketSTM, + waitAndCancelWriteSocketSTM, +) where + +import Control.Concurrent.STM (STM) import Network.Socket.Types -import System.Posix.Types +#if !defined(mingw32_HOST_OS) +import Control.Concurrent (threadWaitReadSTM, threadWaitWriteSTM) +import System.Posix.Types (Fd (..)) +#endif -- | STM action to wait until the socket is ready for reading. +-- +-- __POSIX only.__ On Windows this throws: completion ports have no +-- notion of readiness. Use 'Network.Socket.recvBufWith', or one of +-- the other operations which give up, instead. waitReadSocketSTM :: Socket -> IO (STM ()) waitReadSocketSTM s = fst <$> waitAndCancelReadSocketSTM s -- | STM action to wait until the socket is ready for reading and STM -- action to cancel the waiting. +-- +-- __POSIX only.__ See 'waitReadSocketSTM'. waitAndCancelReadSocketSTM :: Socket -> IO (STM (), IO ()) +#if defined(mingw32_HOST_OS) +waitAndCancelReadSocketSTM _ = + ioError $ + userError $ + "waitAndCancelReadSocketSTM: Windows completion ports do not " + ++ "provide readiness notification; use recvBufWith or " + ++ "Network.Socket.ByteString.recvWith instead" +#else waitAndCancelReadSocketSTM s = withFdSocket s $ threadWaitReadSTM . Fd . fromIntegral +#endif -- | STM action to wait until the socket is ready for writing. +-- +-- __POSIX only.__ On Windows this throws. Completion ports give no +-- way to ask whether a send would block, and none is needed: issue +-- the send and let it complete asynchronously. waitWriteSocketSTM :: Socket -> IO (STM ()) waitWriteSocketSTM s = fst <$> waitAndCancelWriteSocketSTM s -- | STM action to wait until the socket is ready for writing and STM -- action to cancel the waiting. +-- +-- __POSIX only.__ See 'waitWriteSocketSTM'. waitAndCancelWriteSocketSTM :: Socket -> IO (STM (), IO ()) +#if defined(mingw32_HOST_OS) +waitAndCancelWriteSocketSTM _ = + ioError $ + userError $ + "waitAndCancelWriteSocketSTM: Windows completion ports do not " + ++ "provide readiness notification, and none is needed for " + ++ "sending: issue the send instead" +#else waitAndCancelWriteSocketSTM s = withFdSocket s $ threadWaitWriteSTM . Fd . fromIntegral +#endif diff --git a/Network/Socket/Shutdown.hs b/Network/Socket/Shutdown.hs index ecbc72af..6462d138 100644 --- a/Network/Socket/Shutdown.hs +++ b/Network/Socket/Shutdown.hs @@ -7,17 +7,21 @@ module Network.Socket.Shutdown ( ShutdownCmd(..) , shutdown , gracefulClose + , unsafeGracefulClose ) where -import Control.Concurrent (forkIO, killThread, threadDelay, yield) +import Control.Concurrent (rtsSupportsBoundThreads, yield) +import Control.Concurrent.STM import qualified Control.Exception as E import Foreign.Marshal.Alloc (mallocBytes, free) import qualified System.IO.Error as E +import System.Timeout import Network.Socket.Buffer import Network.Socket.Imports import Network.Socket.Internal import Network.Socket.Types +import Network.Socket.Wait data ShutdownCmd = ShutdownReceive | ShutdownSend @@ -51,8 +55,19 @@ foreign import CALLCONV unsafe "shutdown" -- -- Since: 3.1.1.0 gracefulClose :: Socket -> Int -> IO () -gracefulClose s tmout0 = - (sendRecvFIN `E.finally` close s) `annotateIOException` show s +gracefulClose = gracefulCloseWith close + +-- | 'gracefulClose' which ends with 'unsafeClose' rather than 'close', +-- for a caller which knows that no other thread is waiting on the +-- socket. The wait for the peer's FIN is this thread's own. +-- +-- @since 3.2.12.0 +unsafeGracefulClose :: Socket -> Int -> IO () +unsafeGracefulClose = gracefulCloseWith unsafeClose + +gracefulCloseWith :: (Socket -> IO ()) -> Socket -> Int -> IO () +gracefulCloseWith closer s tmout0 = + (sendRecvFIN `E.finally` closer s) `annotateIOException` show s where sendRecvFIN = do -- Sending TCP FIN. @@ -65,66 +80,42 @@ gracefulClose s tmout0 = -- FIN arrives meanwhile. yield -- Waiting TCP FIN. - E.bracket (mallocBytes bufSize) free (recvEOFloop s tmout0) + E.bracket (mallocBytes bufSize) free (recvEOF s tmout0) + +-- Waiting for the peer's FIN, draining what arrives before it. The +-- receive is given up on rather than interrupted: an asynchronous +-- exception does not reach a thread blocked in one on Windows. +recvEOF :: Socket -> Int -> Ptr Word8 -> IO () +recvEOF s tmout0 buf + | not rtsSupportsBoundThreads = recvEOFtimeout s tmout0 buf + | otherwise = do + -- millisecond to microsecond + late <- registerDelay $ tmout0 * 1000 + loop late 0 + where + loop late n0 = do + r <- E.try $ recvBufWith s buf bufSize $ check =<< readTVar late + case r of + Left (_ :: E.IOException) -> return () + -- The deadline. + Right (Left ()) -> return () + Right (Right n1) -> when (n1 > 0) $ do + let n = n0 + n1 + when (n < drainLimit) $ loop late n + +-- registerDelay needs -threaded, and so does the give-up on Windows -- +-- where this cannot cut a receive short either, but nothing else works +-- there without it. +recvEOFtimeout :: Socket -> Int -> Ptr Word8 -> IO () +-- millisecond to microsecond +recvEOFtimeout s tmout0 buf = void $ timeout (tmout0 * 1000) $ recvBuf s buf bufSize -- Don't use 4092 here. The GHC runtime takes the global lock -- if the length is over 3276 bytes in 32bit or 3272 bytes in 64bit. bufSize :: Int bufSize = 1024 --- Maximum number of bytes to drain while waiting for the peer's FIN. +-- Data left unread makes close send RST rather than FIN, so what has +-- arrived is drained -- but only this far: nothing is going to read it. drainLimit :: Int drainLimit = 128 * 1024 - --- Draining the receive queue until EOF, bounded by 'drainLimit' bytes --- and by the millisecond deadline in the second argument. --- --- The deadline is enforced by a watchdog thread calling 'abortRecv', --- rather than by 'System.Timeout.timeout': with the MIO manager on --- Windows, 'recvBuf' blocks in a foreign 'recv' call, which the --- asynchronous exception thrown by 'timeout' cannot interrupt. The --- watchdog itself only ever blocks in 'threadDelay', which is always --- interruptible, so 'killThread' reliably reaps it once EOF is --- reached. -recvEOFloop :: Socket -> Int -> Ptr Word8 -> IO () -recvEOFloop s tmout0 buf = E.bracket watchdog killThread $ \_ -> loop 0 - where - watchdog = forkIO $ do - threadDelay (tmout0 * 1000) - abortRecv s - loop n0 = do - ex <- E.tryIOError $ recvBuf s buf bufSize - case ex of - Left _ -> return () - Right n1 -> when (n1 > 0) $ do - let n = n0 + n1 - when (n < drainLimit) $ loop n - --- Aborting the drain loop's 'recvBuf' while leaving the descriptor --- valid, so that, unlike 'close', nothing here can race with --- descriptor reuse. --- --- 'shutdown' makes every recv issued from now on fail (POSIX: EOF; --- Windows: WSAESHUTDOWN) and on POSIX it also wakes a recv that is --- already blocked in the kernel. On Windows it does not, so a --- blocked recv is aborted with CancelIoEx: sockets are created with --- WSA_FLAG_OVERLAPPED, hence even a "blocking" recv is an overlapped --- operation internally, waited on inside ws2_32, and cancellation --- makes it fail with WSA_OPERATION_ABORTED. Shutting down first --- closes the race with a recv that has not yet entered the kernel: --- in every interleaving the recv returns EOF, fails, or is --- cancelled, and each of these ends the drain loop. -abortRecv :: Socket -> IO () -abortRecv s = do - void $ E.tryIOError $ shutdown s ShutdownBoth -#if defined(mingw32_HOST_OS) - void $ withFdSocket s $ \fd -> c_CancelIoEx fd nullPtr -#endif - -#if defined(mingw32_HOST_OS) --- BOOL CancelIoEx(HANDLE hFile, LPOVERLAPPED lpOverlapped) --- A SOCKET is a kernel HANDLE. A NULL lpOverlapped cancels all --- pending I/O on the handle, whichever thread issued it. -foreign import CALLCONV unsafe "CancelIoEx" - c_CancelIoEx :: CSocket -> Ptr () -> IO CInt -#endif diff --git a/Network/Socket/SockAddr.hs b/Network/Socket/SockAddr.hs index f1c744a0..c898afef 100644 --- a/Network/Socket/SockAddr.hs +++ b/Network/Socket/SockAddr.hs @@ -4,15 +4,21 @@ module Network.Socket.SockAddr ( getPeerName, getSocketName, connect, + connectWith, bind, accept, + acceptWith, sendBufTo, + sendBufToWith, recvBufFrom, sendBufMsg, recvBufMsg, annotateWithSocket, + sendBufMsgWith, + recvBufMsgWith, ) where +import Control.Concurrent.STM (STM) import qualified Control.Exception as E import System.Directory (removeFile) import qualified System.IO.Error as E @@ -23,6 +29,7 @@ import Network.Socket.Imports import Network.Socket.Internal import qualified Network.Socket.Name as G import qualified Network.Socket.Syscall as G +import qualified Network.Socket.Wait as G #if !defined(mingw32_HOST_OS) import Network.Socket.Posix.Cmsg #else @@ -48,6 +55,14 @@ connect s sa = connect' `annotateWithSocket` (s, Just sa) G.connect s sa labelSocket s (\label -> label ++ " " ++ show sa) +-- | 'connect' which the caller can give up on. Whichever settles +-- first decides: 'Left' when the 'STM' action fires, 'Right' when the +-- connection is made. A socket given up on is to be closed. +-- +-- @since 3.2.12.0 +connectWith :: Socket -> SockAddr -> STM a -> IO (Either a ()) +connectWith s sa giveUp = G.connectWith s sa giveUp `annotateWithSocket` (s, Just sa) + -- | Bind the socket to an address. The socket must not already be -- bound. The 'Family' passed to @bind@ must be the -- same as that passed to 'socket'. If the special port number @@ -94,6 +109,23 @@ accept s = accept' `annotateWithSocket` (s, Nothing) labelSocket news (\_ -> label ++ " " ++ show sa) return r +-- | 'accept' which the caller can give up on. Whichever settles +-- first decides: 'Left' when the 'STM' action fires, 'Right' when a +-- connection arrives. +-- +-- @since 3.2.10.0 +acceptWith :: Socket -> STM a -> IO (Either a (Socket, SockAddr)) +acceptWith s giveUp = acceptWith' `annotateWithSocket` (s, Nothing) + where + acceptWith' = do + r <- G.acceptWith s giveUp + case r of + Left a -> return $ Left a + Right res@(news, sa) -> do + label <- socketLabel s + labelSocket news (\_ -> label ++ " " ++ show sa) + return $ Right res + -- | Send data to the socket. The recipient can be specified -- explicitly, so the socket need not be in a connected state. -- Returns the number of bytes sent. Applications are responsible for @@ -101,6 +133,13 @@ accept s = accept' `annotateWithSocket` (s, Nothing) sendBufTo :: Socket -> Ptr a -> Int -> SockAddr -> IO Int sendBufTo s ptr len sa = G.sendBufTo s ptr len sa `annotateWithSocket` (s, Just sa) +-- | 'sendBufTo' which the caller can give up on. +-- +-- @since 3.2.12.0 +sendBufToWith :: Socket -> Ptr a -> Int -> SockAddr -> STM b -> IO (Either b Int) +sendBufToWith s ptr len sa giveUp = + G.sendBufToWith s ptr len sa giveUp `annotateWithSocket` (s, Just sa) + -- | Receive data from the socket, writing it into buffer instead of -- creating a new string. The socket need not be in a connected -- state. Returns @(nbytes, address)@ where @nbytes@ is the number of @@ -151,6 +190,24 @@ recvBufMsg -- ^ Source address, received data, control messages and message flags recvBufMsg s bufs len flag = G.recvBufMsg s bufs len flag `annotateWithSocket` (s, Nothing) +-- | 'sendBufMsg' which the caller can give up on. +-- +-- @since 3.2.12.0 +sendBufMsgWith + :: Socket -> SockAddr -> [(Ptr Word8, Int)] -> [Cmsg] -> MsgFlag -> STM a + -> IO (Either a Int) +sendBufMsgWith s sa bufs cmsgs flag giveUp = + G.sendBufMsgWith s sa bufs cmsgs flag giveUp `annotateWithSocket` (s, Just sa) + +-- | 'recvBufMsg' which the caller can give up on. +-- +-- @since 3.2.12.0 +recvBufMsgWith + :: Socket -> [(Ptr Word8, Int)] -> Int -> MsgFlag -> STM a + -> IO (Either a (SockAddr, Int, [Cmsg], MsgFlag)) +recvBufMsgWith s bufs len flag giveUp = + G.recvBufMsgWith s bufs len flag giveUp `annotateWithSocket` (s, Nothing) + ------------------------------------------------------------------------ annotateWithSocket :: IO a -> (Socket, Maybe SockAddr) -> IO a diff --git a/Network/Socket/Syscall.hs b/Network/Socket/Syscall.hs index 1eeba551..439087c9 100644 --- a/Network/Socket/Syscall.hs +++ b/Network/Socket/Syscall.hs @@ -4,12 +4,15 @@ module Network.Socket.Syscall where +import Data.Either (isRight) import qualified Control.Exception as E import Foreign.Marshal.Utils (with) #if defined(mingw32_HOST_OS) import Foreign (FunPtr) +import Foreign.Marshal.Alloc (alloca) import GHC.Conc (asyncDoProc) +import Network.Socket.Cbits import System.IO.Error (catchIOError) # if __IO_MANAGER_WINIO__ >= 2 import qualified GHC.Event.Windows as Mgr @@ -93,6 +96,7 @@ socket family stype protocol = -- So, we don't have to call "close" which uses "closeFdWith". unsetIPv6Only s setDontFragment s + unsetConnReset s return s create = do let c_stype = modifyFlag $ packSocketType stype @@ -136,6 +140,27 @@ socket family stype protocol = return () #endif + unsetConnReset s = when (stype == Datagram) $ reportConnReset s False + +-- Whether the socket reports the ICMP unreachable an earlier send of its +-- own caused. Winsock reports it to the next receive even on a socket +-- which was never connected, where POSIX says nothing of it at all: one +-- reply bouncing off a client which has gone ends a server's receive +-- loop. 'socket' therefore turns it off and 'connect' turns it back on, +-- which is the ECONNREFUSED POSIX gives a connected datagram socket. +-- +-- It is advice for datagrams alone and does nothing to any other socket, +-- so the result is not worth looking at. +reportConnReset :: Socket -> Bool -> IO () +#if defined(mingw32_HOST_OS) +reportConnReset s report = + with (if report then 1 else 0 :: Word32) $ \val -> + alloca $ \ret -> + void $ withFdSocket s $ \fd -> + c_WSAIoctl fd sioUDPConnReset val 4 nullPtr 0 ret nullPtr nullPtr +#else +reportConnReset _ _ = return () +#endif ----------------------------------------------------------------------------- -- Binding a socket @@ -155,28 +180,51 @@ bind s sa = withSocketAddress sa $ \p_sa siz -> void $ withFdSocket s $ \fd -> d -- | Connect to a remote socket at address. connect :: SocketAddress sa => Socket -> sa -> IO () -connect s sa = withSocketsDo $ withSocketAddress sa $ \p_sa sz -> - connectLoop s p_sa (fromIntegral sz) +connect s sa = void (connectWhenWritable s sa waitWrite :: IO (Either () ())) + where +#if defined(mingw32_HOST_OS) + waitWrite act = Right <$> act +#else + waitWrite act = do + withFdSocket s $ threadWaitWrite . fromIntegral + Right <$> act +#endif + +-- | 'connect' which leaves the wait for a connection in progress to +-- the caller: the argument runs the action once the socket is writable. +connectWhenWritable + :: SocketAddress sa + => Socket -> sa -> (IO () -> IO (Either a ())) -> IO (Either a ()) +connectWhenWritable s sa whenWritable = + withSocketsDo $ withSocketAddress sa $ \p_sa sz -> do + r <- connectLoop s p_sa (fromIntegral sz) whenWritable + when (isRight r) $ reportConnReset s True + return r -connectLoop :: SocketAddress sa => Socket -> Ptr sa -> CInt -> IO () -connectLoop s p_sa sz = withFdSocket s $ \fd -> loop fd +connectLoop + :: SocketAddress sa + => Socket -> Ptr sa -> CInt -> (IO () -> IO (Either a ())) -> IO (Either a ()) +#if defined(mingw32_HOST_OS) +connectLoop s p_sa sz _ = withFdSocket s $ \fd -> loop fd +#else +connectLoop s p_sa sz whenWritable = withFdSocket s $ \fd -> loop fd +#endif where errLoc = "Network.Socket.connect: " ++ show s loop fd = do r <- c_connect fd p_sa sz - when (r == -1) $ do + if r /= -1 then return (Right ()) else do #if defined(mingw32_HOST_OS) throwSocketError errLoc #else err <- getErrno case () of _ | err == eINTR -> loop fd - _ | err == eINPROGRESS -> connectBlocked --- _ | err == eAGAIN -> connectBlocked + _ | err == eINPROGRESS -> whenWritable connected +-- _ | err == eAGAIN -> whenWritable connected _otherwise -> throwSocketError errLoc - connectBlocked = do - withFdSocket s $ threadWaitWrite . fromIntegral + connected = do err <- getSocketOption s SoError when (err /= 0) $ throwSocketErrorCode errLoc (fromIntegral err) #endif @@ -265,6 +313,11 @@ foreign import CALLCONV unsafe "accept" #endif #if defined(mingw32_HOST_OS) +-- int WSAIoctl(SOCKET, DWORD, LPVOID, DWORD, LPVOID, DWORD, LPDWORD, +-- LPWSAOVERLAPPED, LPWSAOVERLAPPED_COMPLETION_ROUTINE) +foreign import CALLCONV unsafe "WSAIoctl" + c_WSAIoctl :: CSocket -> Word32 -> Ptr Word32 -> Word32 -> Ptr () -> Word32 + -> Ptr Word32 -> Ptr () -> Ptr () -> IO CInt foreign import CALLCONV safe "accept" c_accept_safe :: CSocket -> Ptr sa -> Ptr CInt{-CSockLen???-} -> IO CSocket foreign import ccall unsafe "rtsSupportsBoundThreads" diff --git a/Network/Socket/Types.hsc b/Network/Socket/Types.hsc index 281db50c..54d9add8 100644 --- a/Network/Socket/Types.hsc +++ b/Network/Socket/Types.hsc @@ -26,6 +26,7 @@ module Network.Socket.Types ( , socketLabel , close , close' + , unsafeClose , c_close -- * Types of socket , SocketType(GeneralSocketType, UnsupportedSocketType, NoSocketType @@ -298,6 +299,23 @@ close' s = invalidateSocket s (\_ -> return ()) $ \oldfd -> do label <- socketLabel s when (ret == -1) $ throwErrno $ "Network.Socket.close' <" ++ label ++ ">" +-- | Closing a socket which no thread is waiting on, with @close(2)@ +-- itself. Exceptions are not thrown, as in 'close'. +-- +-- 'close' goes through the IO manager so that a thread waiting on the +-- socket is told. On a threaded POSIX RTS that takes a lock on every +-- capability, and does nothing at all when there is no such thread. +-- This skips it, and is for a caller which knows there is none. Where +-- one might be, the warning on 'unsafeFdSocket' applies in full. +-- +-- There is nothing to skip on Windows, where the IO manager is not in +-- the way of a close, and this is 'close'. +-- +-- @since 3.2.11.0 +unsafeClose :: Socket -> IO () +unsafeClose s = invalidateSocket s (\_ -> return ()) $ \oldfd -> + void $ c_close oldfd + #if defined(mingw32_HOST_OS) foreign import CALLCONV unsafe "closesocket" c_close :: CSocket -> IO CInt diff --git a/Network/Socket/Wait.hs b/Network/Socket/Wait.hs new file mode 100644 index 00000000..c2bd9866 --- /dev/null +++ b/Network/Socket/Wait.hs @@ -0,0 +1,377 @@ +{-# LANGUAGE CPP #-} + +#include "HsNetDef.h" + +-- | Blocking socket operations which a caller can give up on. +-- +-- Each takes an 'STM' action beside the operation. Whichever +-- settles first decides the result: 'Left' when the caller gave up, +-- 'Right' when the operation completed. Nothing has to be closed or +-- interrupted to make the thread return, so neither the delivery of +-- @evtClose@ by @closeFdWith@ nor an asynchronous exception reaching +-- a blocking call is relied upon. +-- +-- __On Windows these need the threaded runtime.__ There is no +-- readiness to wait for there, so the operation is started and then +-- cancelled, and the thread it runs in blocks in a foreign call. +-- Without @-threaded@ that blocks every other thread too, including +-- the one which would have given up, and nothing returns at all. +module Network.Socket.Wait ( + withReadable, + withWritable, + withReadableOv, + withWritableOv, + acceptWith, + recvBufWith, + recvBufFromWith, + sendBufWith, + sendBufToWith, + recvBufMsgWith, + sendBufMsgWith, + connectWith, + isClosedSocketError, + isTransientAcceptError, + isPortUnreachableError, +) where + +import Control.Concurrent.STM +import qualified Control.Exception as E +import Foreign.C.Error ( + Errno (..), + eCONNABORTED, +#if !defined(mingw32_HOST_OS) + eCONNREFUSED, +#endif + eHOSTDOWN, + eHOSTUNREACH, + eNETDOWN, + eNETUNREACH, + eNONET, + eNOPROTOOPT, + ePROTO, + ) +import GHC.IO.Exception (IOErrorType (..), ioe_errno) +import System.IO.Error (ioeGetErrorType) + +import Network.Socket.Buffer +#if defined(mingw32_HOST_OS) +import Network.Socket.Cbits +import Network.Socket.Win32.Cancel +#else +import Network.Socket.STM +#endif +import Network.Socket.Flag +import Network.Socket.Imports +import Network.Socket.Syscall +import Network.Socket.Types +#if !defined(mingw32_HOST_OS) +import Network.Socket.Posix.Cmsg +#else +import Network.Socket.Win32.Cmsg +#endif + +-- | Running an action once the socket is readable, unless the 'STM' +-- action fires first. The 'STM' action is also looked at before +-- anything else, so a caller which has already given up never waits. +-- An operation which has completed wins a tie, so nothing it has +-- already received is thrown away. +-- +-- __The action must be an operation on this socket which returns +-- without blocking once the socket is readable.__ The type does not +-- say so and nothing checks it. On POSIX the action runs on the +-- calling thread after readiness and nothing interrupts it there: the +-- give-up ends the wait, not the action. On Windows there is no +-- readiness to wait for, so the action is started at once on a thread +-- of its own and the give-up cancels it with @CancelIoEx@ on this +-- socket, which reaches nothing else -- an action blocked on anything +-- else would never end. The cancel cannot tell the action from other +-- operations pending on the socket, so it ends those too, whichever +-- thread issued them. The give-ups for particular operations, such as +-- 'recvBufWith' and 'Network.Socket.ByteString.sendAllWith', cancel +-- their own and leave the rest alone. +-- +-- __One reader at a time__ is what keeps that condition. A second +-- thread taking the data between the readiness and the action leaves +-- the action blocking on the socket, with the give-up no longer in +-- play. +-- +-- __The 'STM' action is run more than once__, so it has to be a test +-- and not something which takes what it reads: it is looked at before +-- the wait and again as part of it, and 'Network.Socket.ByteString.recvWith' +-- looks at it once more before a read which would not block. Reading +-- a 'TVar' is what it is for. +withReadable :: Socket -> STM a -> IO b -> IO (Either a b) +withReadable s giveUp act = withReady Read False s giveUp $ const act + +-- | 'withReadable' for writing: running an action once the socket is +-- writable, unless the 'STM' action fires first. The same conditions +-- hold, with one writer at a time in place of one reader. +-- +-- A send which is given up on may have sent part of its data on +-- Windows, where it is cancelled rather than never started, and the +-- socket cannot be sent on after that. So a stream given up on in +-- the middle of a send is to be closed. +-- +-- @since 3.2.12.0 +withWritable :: Socket -> STM a -> IO b -> IO (Either a b) +withWritable s giveUp act = withReady Write False s giveUp $ const act + +-- | 'withReadable' for an operation which takes an OVERLAPPED: on +-- Windows it is issued with one of its own and the give-up cancels it +-- alone. 'nullPtr' is passed elsewhere. +withReadableOv :: Socket -> STM a -> (Ptr () -> IO b) -> IO (Either a b) +withReadableOv = withReady Read True + +-- | 'withWritable' for an operation which takes an OVERLAPPED. +withWritableOv :: Socket -> STM a -> (Ptr () -> IO b) -> IO (Either a b) +withWritableOv = withReady Write True + +data Ready = Read | Write + +withReady :: Ready -> Bool -> Socket -> STM a -> (Ptr () -> IO b) -> IO (Either a b) +withReady _ready _own s giveUp act = do + -- Looking at the give-up first: on Windows 'wait' starts the + -- operation before it looks at any STM, and on POSIX this keeps a + -- caller which has given up off a descriptor close may have freed. + gaveUp <- atomically $ (Just <$> giveUp) `orElse` return Nothing + case gaveUp of + Just a -> return $ Left a + Nothing -> wait + where +#if !defined(mingw32_HOST_OS) + -- A socket which is ready already skips the IO manager, whose round + -- trip costs several times the operation: a send usually finds room. + wait = do + now <- withFdSocket s $ \fd -> + c_ready_now fd $ case _ready of + Read -> 0 + Write -> 1 + if now /= 0 then Right <$> act nullPtr else waitIO + + waitIO = do + (ready, cancelWait) <- case _ready of + Read -> waitAndCancelReadSocketSTM s + Write -> waitAndCancelWriteSocketSTM s + ev <- + atomically ((Left <$> giveUp) `orElse` (Right <$> ready)) + `E.finally` (cancelWait `E.catch` gone) + either (return . Left) (const $ Right <$> act nullPtr) ev + + -- The cancel can throw, and from a finally that would replace the + -- exception which ended the wait -- the timeout the caller is waiting + -- to see, say. What it reports is a registration which has already + -- gone, and there is nothing to do about that. + gone :: E.IOException -> IO () + gone _ = return () +#else + -- Only an operation issued with an OVERLAPPED of its own can be + -- cancelled alone; any other cancel reaches the whole socket. + wait + | _own = withOverlapped waitOn + | otherwise = waitOn nullPtr + waitOn ov = do + (completed, cancelAct) <- windowsViaSTM s ov (act ov) + -- The operation wins a tie. It has already read into the + -- caller's buffer by then, and answering Left would throw those + -- bytes away -- which for a stream loses them for good. + -- + -- onException because the call runs in a thread of its own here: + -- an asynchronous exception arriving during the wait would + -- otherwise leave it blocked in a foreign call, holding the + -- socket and writing into a buffer the caller has let go of. + ev <- + atomically ((Right <$> completed) `orElse` (Left <$> giveUp)) + `E.onException` cancelAct + case ev of + Right b -> return $ Right b + Left a -> do + cancelAct + return $ Left a +#endif + +#if !defined(mingw32_HOST_OS) +foreign import ccall unsafe "hsnet_ready_now" + c_ready_now :: CInt -> CInt -> IO CInt +#endif + +-- | 'accept' which the caller can give up on. +-- +-- @since 3.2.10.0 +acceptWith :: SocketAddress sa => Socket -> STM a -> IO (Either a (Socket, sa)) +acceptWith s giveUp = withReadable s giveUp $ accept s + +-- | 'recvBuf' which the caller can give up on. +-- +-- @since 3.2.10.0 +recvBufWith :: Socket -> Ptr Word8 -> Int -> STM a -> IO (Either a Int) +recvBufWith s ptr nbytes giveUp = + withReadableOv s giveUp $ \ov -> recvBufOv ov s ptr nbytes + +-- | 'recvBufFrom' which the caller can give up on. +-- +-- @since 3.2.10.0 +recvBufFromWith + :: SocketAddress sa + => Socket -> Ptr a -> Int -> STM b -> IO (Either b (Int, sa)) +recvBufFromWith s ptr nbytes giveUp = + withReadableOv s giveUp $ \ov -> recvBufFromOv ov s ptr nbytes + +-- | 'sendBuf' which the caller can give up on. See 'withWritable' for +-- what is left of the stream after giving up. +-- +-- @since 3.2.12.0 +sendBufWith :: Socket -> Ptr Word8 -> Int -> STM a -> IO (Either a Int) +sendBufWith s ptr nbytes giveUp = + withWritableOv s giveUp $ \ov -> sendBufOv ov s ptr nbytes + +-- | 'sendBufTo' which the caller can give up on. +-- +-- @since 3.2.12.0 +sendBufToWith + :: SocketAddress sa + => Socket -> Ptr a -> Int -> sa -> STM b -> IO (Either b Int) +sendBufToWith s ptr nbytes sa giveUp = + withWritableOv s giveUp $ \ov -> sendBufToOv ov s ptr nbytes sa + +-- | 'recvBufMsg' which the caller can give up on. +-- +-- @since 3.2.12.0 +recvBufMsgWith + :: SocketAddress sa + => Socket + -> [(Ptr Word8, Int)] + -> Int + -> MsgFlag + -> STM a + -> IO (Either a (sa, Int, [Cmsg], MsgFlag)) +recvBufMsgWith s bufsizs clen flags giveUp = + withReadableOv s giveUp $ \ov -> recvBufMsgOv ov s bufsizs clen flags + +-- | 'sendBufMsg' which the caller can give up on. +-- +-- @since 3.2.12.0 +sendBufMsgWith + :: SocketAddress sa + => Socket + -> sa + -> [(Ptr Word8, Int)] + -> [Cmsg] + -> MsgFlag + -> STM a + -> IO (Either a Int) +sendBufMsgWith s sa bufsizs cmsgs flags giveUp = + withWritableOv s giveUp $ \ov -> sendBufMsgOv ov s sa bufsizs cmsgs flags + +-- | 'connect' which the caller can give up on. A socket given up on +-- may still be connecting, so it is to be closed. +-- +-- @since 3.2.12.0 +connectWith :: SocketAddress sa => Socket -> sa -> STM a -> IO (Either a ()) +#if !defined(mingw32_HOST_OS) +connectWith s sa giveUp = do + gaveUp <- atomically $ (Just <$> giveUp) `orElse` return Nothing + case gaveUp of + Just a -> return $ Left a + Nothing -> connectWhenWritable s sa $ withWritable s giveUp +#else +connectWith s sa giveUp = withWritable s giveUp $ connect s sa +#endif + +-- | Whether this is the error a socket operation gives once the socket has +-- been closed under it. +-- +-- A caller which stops when it is told also stops when the socket it was +-- waiting on goes away, and this is how that arrives. It is not an +-- error to report: closing the socket is how a server was stopped +-- before there was anything to tell it, and callers which do that are +-- still out there. +-- +-- @since 3.2.10.0 +isClosedSocketError :: E.IOException -> Bool +isClosedSocketError e = ioeGetErrorType e == InvalidArgument || isWinsock e + where +#if defined(mingw32_HOST_OS) + -- closesocket gives WSAEINTR to a call already blocked on the socket + -- and WSAENOTSOCK to one which starts afterwards. + isWinsock x = ioe_errno x `elem` map Just [wsaEINTR, wsaENOTSOCK] +#else + isWinsock _ = False +#endif + +-- | Whether an exception from 'accept' is about the connection it was +-- accepting rather than about the listening socket. +-- +-- The peer may reset the connection before it is accepted +-- (@ECONNABORTED@), and Linux reports a new socket's already-pending +-- network errors through @accept()@ as well; accept(2) asks for all of +-- these to be retried, like @EAGAIN@. The connection has left the queue +-- by then, so a retry blocks for a new one rather than spinning on the +-- same failure. +-- +-- Two which are not here: +-- +-- * Running out of descriptors (@EMFILE@, @ENFILE@) is about the process, +-- not about one connection, and retrying at once spins: the listening +-- socket stays readable. 'System.IO.Error.isFullError' names it, and a +-- caller which retries it should wait first. +-- +-- * @EOPNOTSUPP@ is on accept(2)'s list and is left off this one on +-- purpose: it is also what a listening socket which is not +-- 'Network.Socket.Stream' answers, which is permanent, and a retry +-- would spin on it forever. +-- +-- Winsock numbers its errors apart from @errno@, so on Windows a list +-- of its own is read as well. @WSAEINTR@ is not on it: there it is +-- what @closesocket@ gives a blocked call, which 'isClosedSocketError' +-- takes. @WSAENOBUFS@ is left off for the reason @EMFILE@ is. +-- +-- @since 3.2.10.0 +isTransientAcceptError :: E.IOException -> Bool +isTransientAcceptError e = + ioeGetErrorType e == Interrupted || any is errnos || isWinsock e + where + -- An errno the platform does not have is Errno (-1), which would + -- otherwise match every error carrying that. + is (Errno n) = n /= -1 && ioe_errno e == Just n + errnos = + [ eCONNABORTED + , eHOSTDOWN + , eHOSTUNREACH + , eNETDOWN + , eNETUNREACH + , eNONET + , eNOPROTOOPT + , ePROTO + ] +#if defined(mingw32_HOST_OS) + isWinsock x = ioe_errno x `elem` map Just codes + codes = + [ wsaECONNABORTED + , wsaECONNRESET + , wsaEHOSTDOWN + , wsaEHOSTUNREACH + , wsaENETDOWN + , wsaENETRESET + , wsaENETUNREACH + , wsaENOPROTOOPT + ] +#else + isWinsock _ = False +#endif + +-- | Whether a receive on a connected datagram socket failed because an +-- earlier send of its own came back as an ICMP port unreachable. +-- +-- POSIX gives it as @ECONNREFUSED@ and Winsock as @WSAECONNRESET@, so +-- the error type differs (@NoSuchThing@ and @OtherError@); this reads +-- the code instead. Only a connected socket hears it on any platform. +-- +-- @since 3.2.12.0 +isPortUnreachableError :: E.IOException -> Bool +isPortUnreachableError e = ioe_errno e == Just code + where +#if defined(mingw32_HOST_OS) + code = wsaECONNRESET +#else + Errno code = eCONNREFUSED +#endif diff --git a/Network/Socket/Win32/Cancel.hs b/Network/Socket/Win32/Cancel.hs new file mode 100644 index 00000000..1fca60fa --- /dev/null +++ b/Network/Socket/Win32/Cancel.hs @@ -0,0 +1,128 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE LambdaCase #-} + +#include "HsNetDef.h" + +-- | Giving up on an operation on a socket, on Windows. +-- +-- There is no readiness to wait for here -- completion ports report +-- that an operation has finished, not that one could be started -- so +-- the operation is run on a thread of its own and ended with +-- @CancelIoEx@. Nothing else would end it: an asynchronous exception +-- does not reach a thread blocked in a socket call on Windows. +-- +-- __These need the threaded runtime.__ The operation blocks in a +-- foreign call, which without @-threaded@ stops every other thread, +-- including the one which would cancel it. +module Network.Socket.Win32.Cancel ( + windowsViaSTM, + windowsCancelPendingIo, + withOverlapped, +) where + +import Control.Concurrent (forkIO) +import Control.Concurrent.STM +import qualified Control.Exception as E +import Foreign.Marshal.Alloc (allocaBytes) +import System.IO.Unsafe (unsafePerformIO) +import System.IO (hPutStrLn, stderr) + +import Network.Socket.Imports +import Network.Socket.Types + +-- | Running a blocking operation on the socket in a thread of its own +-- and handing its result over through STM, together with an action +-- cancelling it. If the operation fails, the 'STM' action rethrows +-- the exception. +-- +-- Cancelling waits for the operation to end, so a buffer it reads +-- into has to stay alive until either the 'STM' action completes or +-- the cancel returns. +-- +-- The cancel reaches the operation issued with the given OVERLAPPED +-- (see 'withOverlapped') and nothing else. With 'nullPtr' it reaches +-- every operation pending on the socket, whichever thread issued it. +windowsViaSTM :: Socket -> Ptr () -> IO a -> IO (STM a, IO ()) +windowsViaSTM s ov act = do + var <- newTVarIO Nothing + -- Masked so that the var is written whatever happens: a thread + -- which died between the call returning and the write would leave + -- the waiter retrying and the cancel below looping for ever. The + -- call itself is a foreign one, which no exception reaches anyway. + tid <- forkIO $ E.mask_ $ E.try act >>= atomically . writeTVar var . Just + let wait = + readTVar var >>= \case + Nothing -> retry + Just (Left e) -> throwSTM (e :: E.SomeException) + Just (Right x) -> return x + ended = isJust <$> readTVar var + return (wait, cancel s ended tid) + where + -- An asynchronous exception does not reach a thread blocked in a + -- socket call here, so the call itself is cancelled. CancelIoEx + -- reaches an operation which is already in the kernel, and a call + -- which has not got there yet is caught by trying again. The + -- socket is never shut down to end the call: the caller is entitled + -- to keep using it, and a shutdown would take that away. + -- + -- Uninterruptible because this must not be abandoned: the operation + -- is still writing into the caller's buffer until it ends, and + -- nothing but the end of it stops the loop. + cancel sock ended _tid = E.uninterruptibleMask_ $ loop sayAfter + where + loop n = do + void $ withFdSocket sock $ \fd -> c_CancelIoEx fd ov + over <- endedWithin 1000 + unless over $ do + when (n == 1) say + loop (n - 1) + -- Wait for the call to end rather than sleep through it: + -- CancelIoEx usually works at once, and a sleep here costs a + -- whole timer tick, which Windows measures in tens of ms. + endedWithin us = do + late <- registerDelay us + atomically $ + (True <$ (check =<< ended)) + `orElse` (False <$ (check =<< readTVar late)) + -- Rounds before saying something and carrying on. A round which + -- cancels nothing costs a timer tick, which Windows measures in tens + -- of milliseconds, so this is seconds rather than the hundredth of a + -- second it reads as. Two rounds is the most ever measured. + sayAfter :: Int + sayAfter = 100 + -- Said from a thread of its own, and its failure ignored. This is + -- the path which has to finish before the caller's buffer can go, + -- and writing to stderr can block on the handle's lock or on a pipe + -- nobody is draining -- for ever, inside the mask above. + say = void $ forkIO $ void (E.try warn :: IO (Either E.SomeException ())) + warn = + hPutStrLn stderr $ + "network: a socket operation given up on is not cancelling; " + ++ "still trying. Please report this." + +-- | Cancelling whatever operation is pending on the socket, while +-- leaving the socket itself usable: unlike 'Network.Socket.shutdown' +-- this does not shut it down. An operation which has not reached the +-- kernel yet is not cancelled, so the caller has to be ready to try +-- again. +windowsCancelPendingIo :: Socket -> IO () +windowsCancelPendingIo s = void $ withFdSocket s $ \fd -> c_CancelIoEx fd nullPtr + +-- | Running an action with an OVERLAPPED of its own, to issue an +-- operation which 'windowsViaSTM' can then cancel alone. It stays +-- alive until the action returns, so the operation must end first. +withOverlapped :: (Ptr () -> IO a) -> IO a +withOverlapped = allocaBytes overlappedSize + +overlappedSize :: Int +overlappedSize = fromIntegral $ unsafePerformIO c_ovl_size +{-# NOINLINE overlappedSize #-} + +foreign import ccall unsafe "hsnet_ovl_size" + c_ovl_size :: IO CInt + +-- BOOL CancelIoEx(HANDLE hFile, LPOVERLAPPED lpOverlapped) +-- A SOCKET is a kernel HANDLE. A NULL lpOverlapped cancels all +-- pending I/O on the handle, whichever thread issued it. +foreign import CALLCONV unsafe "CancelIoEx" + c_CancelIoEx :: CSocket -> Ptr () -> IO CInt diff --git a/cbits/overlapped.c b/cbits/overlapped.c new file mode 100644 index 00000000..6ac44e5f --- /dev/null +++ b/cbits/overlapped.c @@ -0,0 +1,165 @@ +/* Socket operations with an OVERLAPPED of the caller's, waited for here, so + that CancelIoEx(s, ov) cancels this one alone. Each answers like the + call it wraps, leaving the error for WSAGetLastError. */ + +#include +#include "HsNet.h" + +int +hsnet_ovl_size(void) +{ + return sizeof(OVERLAPPED); +} + +/* The low bit of hEvent keeps the completion off a completion port the + socket may be associated with. */ +static int +ovl_begin(LPOVERLAPPED ov, HANDLE *ev) +{ + memset(ov, 0, sizeof(*ov)); + *ev = CreateEventW(NULL, TRUE, FALSE, NULL); + if (*ev == NULL) { + WSASetLastError(GetLastError()); + return SOCKET_ERROR; + } + ov->hEvent = (HANDLE)((ULONG_PTR)*ev | 1); + return 0; +} + +/* r is what the call answered. A pending operation is waited for, and + then *n and *flags are what it completed with. */ +static int +ovl_end(SOCKET s, LPOVERLAPPED ov, HANDLE ev, int r, + DWORD *n, DWORD *flags, int *waited) +{ + int err = 0; + *waited = 0; + if (r == SOCKET_ERROR) { + err = WSAGetLastError(); + if (err == WSA_IO_PENDING) { + err = 0; + *waited = 1; + if (!WSAGetOverlappedResult(s, ov, n, TRUE, flags)) + err = WSAGetLastError(); + } + } + CloseHandle(ev); + if (err != 0) { + WSASetLastError(err); + return SOCKET_ERROR; + } + return 0; +} + +int +hsnet_ovl_recv(SOCKET s, char *buf, int len, LPOVERLAPPED ov) +{ + WSABUF b; + DWORD n = 0, flags = 0; + HANDLE ev; + int r, waited; + if (ovl_begin(ov, &ev) != 0) return SOCKET_ERROR; + b.buf = buf; + b.len = len; + r = WSARecv(s, &b, 1, &n, &flags, ov, NULL); + if (ovl_end(s, ov, ev, r, &n, &flags, &waited) != 0) return SOCKET_ERROR; + return (int)n; +} + +int +hsnet_ovl_recvfrom(SOCKET s, char *buf, int len, + struct sockaddr *from, int *fromlen, LPOVERLAPPED ov) +{ + WSABUF b; + DWORD n = 0, flags = 0; + HANDLE ev; + int r, waited; + if (ovl_begin(ov, &ev) != 0) return SOCKET_ERROR; + b.buf = buf; + b.len = len; + r = WSARecvFrom(s, &b, 1, &n, &flags, from, fromlen, ov, NULL); + if (ovl_end(s, ov, ev, r, &n, &flags, &waited) != 0) return SOCKET_ERROR; + return (int)n; +} + +int +hsnet_ovl_send(SOCKET s, char *buf, int len, LPOVERLAPPED ov) +{ + WSABUF b; + DWORD n = 0, flags = 0; + HANDLE ev; + int r, waited; + if (ovl_begin(ov, &ev) != 0) return SOCKET_ERROR; + b.buf = buf; + b.len = len; + r = WSASend(s, &b, 1, &n, 0, ov, NULL); + if (ovl_end(s, ov, ev, r, &n, &flags, &waited) != 0) return SOCKET_ERROR; + return (int)n; +} + +int +hsnet_ovl_sendv(SOCKET s, LPWSABUF bufs, DWORD count, LPOVERLAPPED ov) +{ + DWORD n = 0, flags = 0; + HANDLE ev; + int r, waited; + if (ovl_begin(ov, &ev) != 0) return SOCKET_ERROR; + r = WSASend(s, bufs, count, &n, 0, ov, NULL); + if (ovl_end(s, ov, ev, r, &n, &flags, &waited) != 0) return SOCKET_ERROR; + return (int)n; +} + +int +hsnet_ovl_sendto(SOCKET s, char *buf, int len, + const struct sockaddr *to, int tolen, LPOVERLAPPED ov) +{ + WSABUF b; + DWORD n = 0, flags = 0; + HANDLE ev; + int r, waited; + if (ovl_begin(ov, &ev) != 0) return SOCKET_ERROR; + b.buf = buf; + b.len = len; + r = WSASendTo(s, &b, 1, &n, 0, to, tolen, ov, NULL); + if (ovl_end(s, ov, ev, r, &n, &flags, &waited) != 0) return SOCKET_ERROR; + return (int)n; +} + +/* Answers 0 and puts the bytes sent in *sent, as WSASendMsg does. fn + is WSASendMsg, which Network.Socket.Win32.Load looks up. */ +int +hsnet_ovl_sendmsg(LPFN_WSASENDMSG fn, SOCKET s, LPWSAMSG msg, DWORD flags, + LPDWORD sent, LPOVERLAPPED ov) +{ + DWORD oflags = 0; + HANDLE ev; + int r, waited; + *sent = 0; + if (ovl_begin(ov, &ev) != 0) return SOCKET_ERROR; + r = fn(s, msg, flags, sent, ov, NULL); + return ovl_end(s, ov, ev, r, sent, &oflags, &waited); +} + +/* Answers 0 and puts the bytes received in *got, as WSARecvMsg does, + and SOCKET_ERROR with WSAEMSGSIZE for a message cut short. */ +int +hsnet_ovl_recvmsg(LPFN_WSARECVMSG fn, SOCKET s, LPWSAMSG msg, LPDWORD got, + LPOVERLAPPED ov) +{ + DWORD flags = 0; + HANDLE ev; + int r, waited; + *got = 0; + if (ovl_begin(ov, &ev) != 0) return SOCKET_ERROR; + r = fn(s, msg, got, ov, NULL); + r = ovl_end(s, ov, ev, r, got, &flags, &waited); + /* Completed later: the flags come from here, not from msg. */ + if (waited) msg->dwFlags = flags; + /* The control buffer may point at garbage after a truncation. */ + if (r == SOCKET_ERROR && WSAGetLastError() == WSAEMSGSIZE) { + msg->dwFlags |= MSG_TRUNC; + msg->Control.len = 0; + msg->Control.buf = NULL; + } + return r; +} diff --git a/include/HsNet.h b/include/HsNet.h index 355ee8d6..97fda7a9 100644 --- a/include/HsNet.h +++ b/include/HsNet.h @@ -158,6 +158,21 @@ hsnet_freeaddrinfo(struct addrinfo *ai) freeaddrinfo(ai); } +#ifndef _WIN32 +#include + +/* Whether the socket is readable (or writable) now, without waiting. */ +INLINE int +hsnet_ready_now(int fd, int write) +{ + struct pollfd p; + p.fd = fd; + p.events = write ? POLLOUT : POLLIN; + p.revents = 0; + return poll(&p, 1, 0) > 0; +} +#endif + #ifndef IOV_MAX # define IOV_MAX 1024 #endif diff --git a/network.cabal b/network.cabal index 4e63d8dc..585c53c8 100644 --- a/network.cabal +++ b/network.cabal @@ -68,6 +68,7 @@ extra-source-files: include/HsNetDef.h include/afunix_compat.h cbits/asyncAccept.c + cbits/overlapped.c cbits/initWinSock.c cbits/winSockErr.c cbits/cmsg.c @@ -120,6 +121,7 @@ library Network.Socket.Syscall Network.Socket.Types Network.Socket.Unix + Network.Socket.Wait default-language: Haskell2010 include-dirs: include @@ -156,9 +158,11 @@ library cbits/initWinSock.c cbits/winSockErr.c cbits/asyncAccept.c + cbits/overlapped.c other-modules: Network.Socket.ByteString.Lazy.Windows + Network.Socket.Win32.Cancel Network.Socket.Win32.Cmsg Network.Socket.Win32.CmsgHdr Network.Socket.Win32.HostName @@ -202,6 +206,7 @@ test-suite spec directory, HUnit, network, + stm, temporary, hspec >=2.6, QuickCheck diff --git a/tests/Network/Socket/ByteString/LazySpec.hs b/tests/Network/Socket/ByteString/LazySpec.hs index a55b7be9..1c61458f 100644 --- a/tests/Network/Socket/ByteString/LazySpec.hs +++ b/tests/Network/Socket/ByteString/LazySpec.hs @@ -5,6 +5,8 @@ module Network.Socket.ByteString.LazySpec (main, spec) where import Prelude hiding (getContents) +import Control.Concurrent (threadDelay) +import Control.Concurrent.STM (STM, check, readTVar, registerDelay, retry) import Control.Monad import qualified Data.ByteString.Lazy as L import Network.Socket @@ -44,6 +46,31 @@ spec = do sendAll sock lazyTestMsg `shouldThrow` anyException tcpTest client server + describe "sendWith" $ do + it "sends as send does" $ do + let server sock = recv sock 1024 `shouldReturn` lazyTestMsg + client sock = sendWith sock lazyTestMsg never `shouldReturn` Right (L.length lazyTestMsg) + tcpTest client server + + describe "sendAllWith" $ do + it "sends as sendAll does" $ do + let server sock = recv sock 1024 `shouldReturn` lazyTestMsg + client sock = sendAllWith sock lazyTestMsg never `shouldReturn` Right () + tcpTest client server + + describe "recvWith" $ do + it "receives as recv does" $ do + let server sock = recvWith sock 1024 never `shouldReturn` Right lazyTestMsg + client sock = sendAll sock lazyTestMsg + tcpTest client server + + it "gives up when told to, with nothing to receive" $ do + let server sock = do + late <- registerDelay 100000 + recvWith sock 1024 (check =<< readTVar late) `shouldReturn` Left () + client _ = threadDelay 500000 + tcpTest client server + describe "getContents" $ do it "works well" $ do let server sock = getContents sock `shouldReturn` lazyTestMsg @@ -83,3 +110,6 @@ spec = do let client s = recv s 4096 `shouldReturn` L.empty server s = shutdown s ShutdownSend tcpTest client server + +never :: STM () +never = retry diff --git a/tests/Network/Socket/ByteStringSpec.hs b/tests/Network/Socket/ByteStringSpec.hs index ac5801df..3b3916b2 100644 --- a/tests/Network/Socket/ByteStringSpec.hs +++ b/tests/Network/Socket/ByteStringSpec.hs @@ -2,7 +2,10 @@ module Network.Socket.ByteStringSpec (main, spec) where -import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) +import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar, tryReadMVar) +import Control.Concurrent.STM (STM, check, readTVar, registerDelay, retry) +import Control.Exception (SomeException, finally, try) import Control.Monad import Data.Bits import qualified Data.ByteString as S @@ -200,6 +203,98 @@ spec = do seg2 = C.pack "test message." udpTest client server + describe "recvMsgWith" $ do + it "receives as recvMsg does" $ do + let server sock = do + r <- recvMsgWith sock 1024 0 mempty retry + case r of + Right (_, msg, _, _) -> msg `shouldBe` seg + Left () -> expectationFailure "gave up without being told" + client sock addr = sendTo sock seg addr + + seg = C.pack "This is a test message" + udpTest client server + + it "gives up when told to, with nothing to receive" $ do + sock <- socket AF_INET Datagram defaultProtocol + bind sock $ SockAddrInet 0 $ tupleToHostAddress (127, 0, 0, 1) + late <- registerDelay 100000 + r <- recvMsgWith sock 1024 0 mempty $ check =<< readTVar late + close sock + either (const True) (const False) r `shouldBe` True + + describe "sendWith" $ do + it "sends as send does" $ do + let server sock = recv sock 1024 `shouldReturn` testMsg + client sock = sendWith sock testMsg never `shouldReturn` Right (S.length testMsg) + tcpTest client server + + describe "sendAllWith" $ do + it "sends as sendAll does" $ do + let server sock = recv sock 1024 `shouldReturn` testMsg + client sock = sendAllWith sock testMsg never `shouldReturn` Right () + tcpTest client server + + it "gives up when the peer does not read" $ withStalledPeer $ \sock -> do + late <- registerDelay 200000 + gaveUp (sendAllWith sock chunk $ check =<< readTVar late) + `shouldReturn` True + + describe "giving up" $ do + -- On Windows a give-up cancels the operation; that must not + -- reach another thread's operation on the same socket. + it "leaves another thread's send alone" $ withStalledPeer $ \sock -> do + ended <- newEmptyMVar + _ <- forkIO $ do + r <- try $ forever $ sendAll sock chunk + putMVar ended (r :: Either SomeException ()) + threadDelay 300000 + late <- registerDelay 200000 + r <- recvWith sock 4096 $ check =<< readTVar late + either (const True) (const False) r `shouldBe` True + threadDelay 300000 + isNothing <$> tryReadMVar ended `shouldReturn` True + + it "leaves another thread's recv alone" $ withStalledPeer $ \sock -> do + ended <- newEmptyMVar + _ <- forkIO $ do + r <- try $ recv sock 4096 + putMVar ended (r :: Either SomeException S.ByteString) + threadDelay 100000 + late <- registerDelay 200000 + gaveUp (sendAllWith sock chunk $ check =<< readTVar late) + `shouldReturn` True + threadDelay 300000 + isNothing <$> tryReadMVar ended `shouldReturn` True + + describe "sendManyWith" $ do + it "sends as sendMany does" $ do + let server sock = recv sock 1024 `shouldReturn` S.append seg1 seg2 + client sock = sendManyWith sock [seg1, seg2] never `shouldReturn` Right () + seg1 = C.pack "This is a " + seg2 = C.pack "test message." + tcpTest client server + + it "gives up when the peer does not read" $ withStalledPeer $ \sock -> do + late <- registerDelay 200000 + gaveUp (sendManyWith sock [chunk, chunk] $ check =<< readTVar late) + `shouldReturn` True + + describe "sendToWith" $ do + it "sends as sendTo does" $ do + let server sock = recv sock 1024 `shouldReturn` testMsg + client sock addr = + sendToWith sock testMsg addr never `shouldReturn` Right (S.length testMsg) + udpTest client server + + describe "sendMsgWith" $ do + it "sends as sendMsg does" $ do + let server sock = recv sock 1024 `shouldReturn` testMsg + client sock addr = + sendMsgWith sock addr [testMsg] [] mempty never + `shouldReturn` Right (S.length testMsg) + udpTest client server + describe "recvMsg" $ do it "works well" $ do let server sock = do @@ -312,3 +407,32 @@ spec = do seg = C.pack "This is a test message" udpTest client server + +-- A connection whose other end never reads, with small buffers, so that +-- sending blocks once they are full. +withStalledPeer :: (Socket -> IO a) -> IO a +withStalledPeer body = do + let localhost = tupleToHostAddress (127, 0, 0, 1) + listener <- socket AF_INET Stream defaultProtocol + setSocketOption listener RecvBuffer 4096 + bind listener $ SockAddrInet 0 localhost + listen listener 1 + port <- socketPort listener + sock <- socket AF_INET Stream defaultProtocol + setSocketOption sock SendBuffer 4096 + connect sock $ SockAddrInet port localhost + (peer, _) <- accept listener + body sock `finally` mapM_ close [sock, peer, listener] + +-- Sending until the action gives up, and whether it did within 1 GiB. +gaveUp :: IO (Either () ()) -> IO Bool +gaveUp send1 = loop (1024 :: Int) + where + loop 0 = return False + loop n = send1 >>= either (const $ return True) (const $ loop (n - 1)) + +chunk :: S.ByteString +chunk = S.replicate (1024 * 1024) 0x41 + +never :: STM () +never = retry diff --git a/tests/Network/SocketSpec.hs b/tests/Network/SocketSpec.hs index 9c01adff..813f08aa 100644 --- a/tests/Network/SocketSpec.hs +++ b/tests/Network/SocketSpec.hs @@ -5,7 +5,14 @@ module Network.SocketSpec (main, spec) where import Control.Concurrent (threadDelay, forkIO) import Control.Concurrent.MVar (newEmptyMVar, putMVar, readMVar, takeMVar) +import Control.Exception (bracket) +import Control.Concurrent.STM import Control.Monad +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Unsafe as BU +import Foreign.Marshal.Alloc (allocaBytes) +import Foreign.Ptr (castPtr) import Data.List (nub) import Data.Maybe (fromJust) import Foreign.C.Types () @@ -24,6 +31,39 @@ import Test.QuickCheck main :: IO () main = hspec spec +-- Send through the given action to a port nobody has bound, then receive +-- on the same socket. 'Nothing' if the receive heard nothing in time. +bounce :: (Socket -> SockAddr -> IO ()) -> IO (Maybe ByteString) +bounce = bounceWith $ \sock giveUp -> fmap fst <$> recvFromWith sock 2048 giveUp + +bounceWith + :: (Socket -> STM () -> IO (Either () ByteString)) + -> (Socket -> SockAddr -> IO ()) + -> IO (Maybe ByteString) +bounceWith receive send' = do + dead <- deadPort + sock <- socket AF_INET Datagram defaultProtocol + bind sock $ SockAddrInet 0 localhost + send' sock dead + -- Long enough for the unreachable to come back before the receive. + waitFor 300000 + late <- registerDelay 300000 + r <- receive sock $ check =<< readTVar late + close sock + return $ either (const Nothing) Just r + where + localhost = tupleToHostAddress (127, 0, 0, 1) + waitFor us = do + late <- registerDelay us + atomically $ check =<< readTVar late + -- Bind a port, read it, and let it go again. + deadPort = do + s <- socket AF_INET Datagram defaultProtocol + bind s $ SockAddrInet 0 localhost + sa <- getSocketName s + close s + return sa + spec :: Spec spec = do describe "connect" $ do @@ -43,6 +83,63 @@ spec = do { clientSetup = readMVar portVar >>= connect' } + describe "connectWith" $ do + let + hints = defaultHints { addrSocketType = Stream } + connectWith' serverPort giveUp = do + addr:_ <- getAddrInfo (Just hints) (Just serverAddr) (Just $ show serverPort) + sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr) + r <- connectWith sock (addrAddress addr) giveUp + return (sock, r) + + it "connects as connect does" $ do + withPort $ \portVar -> test (tcp serverAddr return portVar) + { clientSetup = do + (sock, r) <- readMVar portVar >>= \p -> connectWith' p (retry :: STM ()) + r `shouldBe` Right () + return sock + } + + it "throws as connect does" $ do + connectWith' (8003 :: Int) (retry :: STM ()) `shouldThrow` anyIOException + + it "gives up when already told to" $ do + (sock, r) <- connectWith' (8003 :: Int) (return ()) + close sock + r `shouldBe` Left () + + it "gives up on a connection which does not come" $ do + -- Nothing answers here, unless the network says it is + -- unreachable at once, which is not giving up but is fine. + sock <- socket AF_INET Stream defaultProtocol + late <- registerDelay 300000 + r <- tryIOError $ connectWith sock (SockAddrInet 9 $ tupleToHostAddress (10, 255, 255, 1)) $ + check =<< readTVar late + close sock + r `shouldSatisfy` either (const True) (== Left ()) + + describe "sendBufMsgWith and recvBufMsgWith" $ do + it "send and receive as sendBufMsg and recvBufMsg do" $ do + let server sock = allocaBytes 1024 $ \buf -> do + r <- recvBufMsgWith sock [(buf, 1024)] 0 mempty (retry :: STM ()) + case r of + Right (_, n, _, _) -> + BS.packCStringLen (castPtr buf, n) `shouldReturn` testMsg + Left () -> expectationFailure "gave up without being told" + client sock addr = BU.unsafeUseAsCStringLen testMsg $ \(p, n) -> + sendBufMsgWith sock addr [(castPtr p, n)] [] mempty (retry :: STM ()) + `shouldReturn` Right n + udpTest client server + + it "recvBufMsgWith gives up when told to" $ do + sock <- socket AF_INET Datagram defaultProtocol + bind sock $ SockAddrInet 0 $ tupleToHostAddress (127, 0, 0, 1) + late <- registerDelay 100000 + r <- allocaBytes 1024 $ \buf -> + recvBufMsgWith sock [(buf, 1024)] 0 mempty $ check =<< readTVar late + close sock + either (const True) (const False) r `shouldBe` True + describe "bind" $ do let hints = defaultHints { addrFlags = [AI_PASSIVE] @@ -262,6 +359,25 @@ spec = do -} #endif + -- A send to a port nobody has bound brings an ICMP unreachable back. + -- POSIX hands it to the sender only once the socket is connected, and + -- Winsock hands it to every datagram socket; a reply bouncing off a + -- client which has gone would then end a server's receive loop. + describe "an ICMP unreachable from a send of one's own" $ do + let connectSend sock dead = connect sock dead >> void (send sock "bounce") + recvMsg' sock giveUp = + fmap (\(_, bs, _, _) -> bs) <$> recvMsgWith sock 2048 0 mempty giveUp + it "is not heard by a socket which was never connected" $ + bounce (\sock dead -> void $ sendTo sock "bounce" dead) + `shouldReturn` Nothing + it "is heard by a connected one" $ + bounce connectSend `shouldThrow` isPortUnreachableError + it "is heard by a connected one through recv" $ + bounceWith (\sock -> recvWith sock 2048) connectSend + `shouldThrow` isPortUnreachableError + it "is heard by a connected one through recvMsg" $ + bounceWith recvMsg' connectSend `shouldThrow` isPortUnreachableError + describe "gracefulClose" $ do it "does not send TCP RST back" $ do let server sock = do @@ -275,6 +391,75 @@ spec = do void $ recv sock 1024 tcpTest client server + describe "unsafeGracefulClose" $ do + it "does not send TCP RST back" $ do + let server sock = do + void $ recv sock 1024 -- receiving "GOAWAY" + unsafeGracefulClose sock 300 + client sock = do + sendAll sock "GOAWAY" + threadDelay 10000 + sendAll sock "PING" + threadDelay 10000 + void $ recv sock 1024 + tcpTest client server + + describe "giving up" $ do + it "ends a receive which is told to stop, leaving the socket usable" $ do + let server sock = do + stop <- newTVarIO False + done <- newEmptyMVar + void $ forkIO $ do + r <- recvWith sock 1024 (check =<< readTVar stop) + putMVar done r + threadDelay 100000 + atomically $ writeTVar stop True + takeMVar done `shouldReturn` Left () + -- Nothing was closed, so the socket is still the + -- caller's to read. + recv sock 1024 `shouldReturn` testMsg + client sock = do + threadDelay 300000 + void $ send sock testMsg + tcpTest client server + + -- A tie goes to the data: answering Left would leave bytes + -- already received unread, which for a stream loses them. + it "takes what is there even once the caller has given up" $ do + let server sock = do + threadDelay 100000 + gone <- newTVarIO True + recvWith sock 1024 (check =<< readTVar gone) + `shouldReturn` Right testMsg + client sock = void $ send sock testMsg + tcpTest client server + + it "ends an accept which is told to stop, leaving the socket usable" $ + withListener $ \sock port -> do + stop <- newTVarIO False + done <- newEmptyMVar + void $ forkIO $ do + r <- acceptWith sock (check =<< readTVar stop) + putMVar done $ + either (const True) (const False) + (r :: Either () (Socket, SockAddr)) + threadDelay 100000 + atomically $ writeTVar stop True + takeMVar done `shouldReturn` True + -- The listening socket was not closed to end that + -- accept, so it still accepts. + bracket (socket AF_INET Stream defaultProtocol) close $ \cli -> do + connect cli $ SockAddrInet port $ tupleToHostAddress (127, 0, 0, 1) + bracket (fst <$> accept sock) close $ \_ -> return () + + it "tells a socket closed under an accept from other errors" $ do + sock <- socket AF_INET Stream defaultProtocol + bind sock $ SockAddrInet 0 $ tupleToHostAddress (127, 0, 0, 1) + listen sock 1 + close sock + r <- tryIOError $ accept sock + either isClosedSocketError (const False) r `shouldBe` True + describe "socketToFd" $ do it "socketToFd can send using fd" $ do let server sock = do @@ -501,3 +686,13 @@ cmsgidPatterns = nub , CmsgIdIPv6PktInfo , CmsgIdFds ] + +-- A listening socket on the loopback, and the port it got. +withListener :: (Socket -> PortNumber -> IO a) -> IO a +withListener action = + bracket (socket AF_INET Stream defaultProtocol) close $ \sock -> do + setSocketOption sock ReuseAddr 1 + bind sock $ SockAddrInet 0 $ tupleToHostAddress (127, 0, 0, 1) + listen sock 1 + port <- socketPort sock + action sock port