diff --git a/cbits/HaskellPosix.c b/cbits/HaskellPosix.c --- a/cbits/HaskellPosix.c +++ b/cbits/HaskellPosix.c @@ -1,6 +1,19 @@ +#define _GNU_SOURCE #include +#include +#include +#include #include "HaskellPosix.h" +#include "Rts.h" +#ifdef __GNUC__ +#define likely(x) __builtin_expect(!!(x), 1) +#define unlikely(x) __builtin_expect(!!(x), 0) +#else +#define likely(x) (x) +#define unlikely(x) (x) +#endif + // Generally, this library tries to avoid wrapping POSIX functions // in an additional function. However, for some functions whose wrappers // unsafe FFI wrappers use unpinned ByteArray instead of Addr, the only @@ -16,9 +29,163 @@ ssize_t sendto_offset(int socket, const char *message, int offset, size_t length, int flags, const struct sockaddr *dest_addr, socklen_t dest_len){ return sendto(socket, (const void*)(message + offset), length, flags, dest_addr, dest_len); } +ssize_t sendto_inet_offset(int socket, const char *message, int offset, size_t length, int flags, uint16_t port, uint32_t inet_addr){ + struct sockaddr_in dest; + memset(&dest, 0, sizeof(dest)); + dest.sin_family = AF_INET; + dest.sin_addr.s_addr = inet_addr; + dest.sin_port = port; + return sendto(socket, (const void*)(message + offset), length, flags, (struct sockaddr*)&dest, sizeof(dest)); +} ssize_t recvfrom_offset(int socket, char *restrict buffer, int offset, size_t length, int flags, struct sockaddr *restrict address, socklen_t *restrict address_len) { return recvfrom(socket, (void*)(buffer + offset), length, flags, address, address_len); } +ssize_t recvfrom_offset_inet + ( int socket + , char *restrict buffer_base + , HsInt offset + , size_t length + , int flags + , struct sockaddr_in *restrict addresses + , HsInt address_offset + ) { + void* buffer = (void*)(buffer_base + offset); + struct sockaddr_in* address = addresses + address_offset; + socklen_t address_len[1] = {sizeof(struct sockaddr_in)}; + ssize_t r = recvfrom(socket, buffer, length, flags, address, address_len); + if (likely(address_len[0] == sizeof(struct sockaddr_in))) { + return r; + } else { + fprintf(stderr, "posix-api: recvfrom_offset_bufs"); + exit(EXIT_FAILURE); + } +} int setsockopt_int(int socket, int level, int option_name, int option_value) { return setsockopt(socket,level,option_name,&option_value,sizeof(int)); } + +// The second buffer is char* instead of void* because we need +// to apply an offset to it. +int sendmsg_a + ( int sockfd + , void *bufA + , size_t lenA + , char *bufB + , HsInt offB + , size_t lenB + , int flags + ) { + struct iovec bufs[2] = + { { .iov_base = bufA, .iov_len = lenA } + , { .iov_base = (void*)(bufB + offB), .iov_len = lenB } + }; + struct msghdr msg = + { .msg_name = NULL + , .msg_namelen = 0 + , .msg_iov = bufs + , .msg_iovlen = 2 + , .msg_control = NULL + , .msg_controllen = 0 + }; + return sendmsg(sockfd,&msg,flags); +} + +// The first buffer is char* instead of void* because we need +// to apply an offset to it. +int sendmsg_b + ( int sockfd + , char *bufA + , HsInt offA + , size_t lenA + , void *bufB + , size_t lenB + , int flags + ) { + struct iovec bufs[2] = + { { .iov_base = (void*)(bufA + offA), .iov_len = lenA } + , { .iov_base = bufB, .iov_len = lenB } + }; + struct msghdr msg = + { .msg_name = NULL + , .msg_namelen = 0 + , .msg_iov = bufs + , .msg_iovlen = 2 + , .msg_control = NULL + , .msg_controllen = 0 + }; + return sendmsg(sockfd,&msg,flags); +} + +int recvmmsg_sockaddr_in + ( int sockfd + , int *lens // used for output + , struct sockaddr_in *addrs // used for output + , StgMutArrPtrs *arr // used for output + , unsigned int vlen + , int flags + ) { + StgClosure **bufsX = arr->payload; + StgArrBytes **bufs = (StgArrBytes**)bufsX; + struct mmsghdr msgs[vlen]; + struct iovec vecs[vlen]; + for(unsigned int i = 0; i < vlen; i++) { + vecs[i].iov_base = (void*)(bufs[i]->payload); + vecs[i].iov_len = (size_t)(bufs[i]->bytes); + // We deliberately leave msg_len unassigned. + msgs[i].msg_hdr.msg_name = addrs + i; + msgs[i].msg_hdr.msg_namelen = sizeof(struct sockaddr_in); + msgs[i].msg_hdr.msg_iov = vecs + i; + msgs[i].msg_hdr.msg_iovlen = 1; + msgs[i].msg_hdr.msg_control = NULL; + msgs[i].msg_hdr.msg_controllen = 0; + msgs[i].msg_hdr.msg_flags = flags; + } + int r = recvmmsg(sockfd,msgs,vlen,flags,NULL); + // If no errors occurred, copy all of the lengths into the + // length buffer. This copy makes me feel a little sad. + // It is the only copy in a wrapper that otherwise is + // able to share buffers perfectly between Haskell and C. + if(r > (-1)) { + for(int i = 0; i < r; i++) { + lens[i] = msgs[i].msg_len; + } + } + return r; +} + +int recvmmsg_sockaddr_discard + ( int sockfd + , int *lens // used for output + , StgMutArrPtrs *arr // used for output + , unsigned int vlen + , int flags + ) { + StgClosure **bufsX = arr->payload; + StgArrBytes **bufs = (StgArrBytes**)bufsX; + struct mmsghdr msgs[vlen]; + struct iovec vecs[vlen]; + for(unsigned int i = 0; i < vlen; i++) { + vecs[i].iov_base = (void*)(bufs[i]->payload); + vecs[i].iov_len = (size_t)(bufs[i]->bytes); + // We deliberately leave msg_len unassigned. + msgs[i].msg_hdr.msg_name = NULL; + msgs[i].msg_hdr.msg_namelen = 0; + msgs[i].msg_hdr.msg_iov = vecs + i; + msgs[i].msg_hdr.msg_iovlen = 1; + msgs[i].msg_hdr.msg_control = NULL; + msgs[i].msg_hdr.msg_controllen = 0; + msgs[i].msg_hdr.msg_flags = flags; + } + int r = recvmmsg(sockfd,msgs,vlen,flags,NULL); + // If no errors occurred, copy all of the lengths into the + // length buffer. This copy makes me feel a little sad. + // It is the only copy in a wrapper that otherwise is + // able to share buffers perfectly between Haskell and C. + if(r > (-1)) { + for(int i = 0; i < r; i++) { + lens[i] = msgs[i].msg_len; + } + } + return r; +} + diff --git a/include/HaskellPosix.h b/include/HaskellPosix.h --- a/include/HaskellPosix.h +++ b/include/HaskellPosix.h @@ -1,12 +1,17 @@ #include #include +#include +#include "Rts.h" ssize_t recv_offset(int socket, char *buffer, int offset, size_t length, int flags); ssize_t send_offset(int socket, const char *buffer, int offset, size_t length, int flags); ssize_t sendto_offset(int socket, const char *message, int offset, size_t length, int flags, const struct sockaddr *dest_addr, socklen_t dest_len); +ssize_t sendto_inet_offset(int socket, const char *message, int offset, size_t length, int flags, uint16_t port, uint32_t inet_addr); ssize_t recvfrom_offset(int socket, char *restrict buffer, int offset, size_t length, int flags, struct sockaddr *restrict address, socklen_t *restrict address_len); int setsockopt_int(int socket, int level, int option_name, int option_value); +int recvmmsg_sockaddr_in (int sockfd , int *lens , struct sockaddr_in *addrs , StgMutArrPtrs *arr , unsigned int vlen , int flags); +int recvmmsg_sockaddr_discard (int sockfd , int *lens , StgMutArrPtrs *arr , unsigned int vlen , int flags); diff --git a/include/custom.h b/include/custom.h --- a/include/custom.h +++ b/include/custom.h @@ -5,83 +5,71 @@ // The macro FIELD_SIZEOF is defined in the linux kernel. // It is written out here for portability. -#define INTERNAL_FIELD_SIZEOF(t, f) (sizeof(((t*)0)->f)) +#ifndef FIELD_SIZEOF +#define FIELD_SIZEOF(t, f) (sizeof(((t*)0)->f)) +#endif -// This is a lot like peek except that it is designed to -// work with MutableByteArray instead of Ptr. GHC requires that all -// indexing into ByteArrays is aligned. So, we do some trickery -// based on the size of what the user is requesting. -#define hsc_read(t, f) \ - switch (INTERNAL_FIELD_SIZEOF(t,f)) { \ - case 1: \ - hsc_printf ("(\\hsc_arr -> readByteArray hsc_arr %ld)", (long) offsetof (t, f)); \ - break; \ - case 2: \ - if (offsetof (t, f) % 2 == 0) \ - hsc_printf ("(\\hsc_arr -> readByteArray hsc_arr %ld)", ((long) offsetof (t, f)) / 2); \ - else \ - hsc_printf ("BAD_READ_ALIGNMENT"); \ - break; \ - default: \ - hsc_printf ("(BAD_READ_SIZE_%ld)", (long) INTERNAL_FIELD_SIZEOF(t,f)); \ - break; \ + +#define hsc_readByteArray(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> readByteArray hsc_arr (%ld + (hsc_ix * %ld)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ + } else { \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ } -// This is like peek except that it is designed to work with ByteArray -// instead of Ptr. See hsc_read. This function becomes indexByteArray. -#define hsc_index(t, f) \ - switch (INTERNAL_FIELD_SIZEOF(t,f)) { \ - case 1: \ - hsc_printf ("(\\hsc_arr -> indexByteArray hsc_arr %ld)", (long) offsetof (t, f)); \ - break; \ - case 2: \ - if (offsetof (t, f) % 2 == 0) \ - hsc_printf ("(\\hsc_arr -> indexByteArray hsc_arr %ld)", ((long) offsetof (t, f)) / 2); \ - else \ - hsc_printf ("BAD_INDEX_ALIGNMENT"); \ - break; \ - case 4: \ - if (offsetof (t, f) % 4 == 0) \ - hsc_printf ("(\\hsc_arr -> indexByteArray hsc_arr %ld)", ((long) offsetof (t, f)) / 4); \ - else \ - hsc_printf ("BAD_INDEX_ALIGNMENT"); \ - break; \ - default: \ - hsc_printf ("(BAD_INDEX_SIZE_%ld)", (long) INTERNAL_FIELD_SIZEOF(t,f)); \ - break; \ +#define hsc_writeByteArray(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> writeByteArray hsc_arr (%ld + (hsc_ix * %ld)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ + } else { \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ } -// This is a lot like poke except that it is designed to -// work with MutableByteArray instead of Ptr. See hsc_read. -#define hsc_write(t, f) \ - switch (INTERNAL_FIELD_SIZEOF(t,f)) { \ - case 1: \ - hsc_printf ("(\\hsc_arr -> writeByteArray hsc_arr %ld)", (long) offsetof (t, f)); \ - break; \ - case 2: \ - if (offsetof (t, f) % 2 == 0) \ - hsc_printf ("(\\hsc_arr -> writeByteArray hsc_arr %ld)", ((long) offsetof (t, f)) / 2); \ - else \ - hsc_printf ("BAD_WRITE_ALIGNMENT"); \ - break; \ - case 4: \ - if (offsetof (t, f) % 4 == 0) \ - hsc_printf ("(\\hsc_arr -> writeByteArray hsc_arr %ld)", ((long) offsetof (t, f)) / 4); \ - else \ - hsc_printf ("BAD_WRITE_ALIGNMENT"); \ - break; \ - default: \ - hsc_printf ("(BAD_WRITE_SIZE_%ld)", (long) INTERNAL_FIELD_SIZEOF(t,f)); \ - break; \ +#define hsc_indexByteArray(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> indexByteArray hsc_arr (%ld + (hsc_ix * %ld)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ + } else { \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ } -// Compute an element offset from a byte offset and the element size. -// This causes a compile-time failure if the element size does not -// divide evenly into the byte offset. This is commonly used as: -#define hsc_elementize(boff, sz) \ - if ((boff) % (sz) == 0) { \ - hsc_printf ("%ld", ((boff) / (sz))); \ +#define hsc_readByteArrayHash(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> readByteArray# hsc_arr (%ld# +# (hsc_ix *# %ld#)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ } else { \ - hsc_printf ("BAD_ELEMENT_OFFSET"); \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ + } + +#define hsc_writeByteArrayHash(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> writeByteArray# hsc_arr (%ld# +# (hsc_ix *# %ld#)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ + } else { \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ + } + +#define hsc_indexByteArrayHash(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> indexByteArray# hsc_arr (%ld# +# (hsc_ix *# %ld#)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ + } else { \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ + } + +#define hsc_readOffAddrHash(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> readOffAddr# hsc_arr (%ld# +# (hsc_ix *# %ld#)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ + } else { \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ + } + +#define hsc_writeOffAddrHash(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> writeOffAddr# hsc_arr (%ld# +# (hsc_ix *# %ld#)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ + } else { \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ + } + +#define hsc_indexOffAddrHash(t, f) \ + if ((offsetof(t,f)) % (FIELD_SIZEOF(t,f)) == 0 && (sizeof(t) % (FIELD_SIZEOF(t,f))) == 0) { \ + hsc_printf ("(\\hsc_arr hsc_ix -> indexOffAddr# hsc_arr (%ld# +# (hsc_ix *# %ld#)))", ((offsetof(t,f)) / (FIELD_SIZEOF(t,f))), (sizeof(t) / (FIELD_SIZEOF(t,f)))); \ + } else { \ + hsc_printf ("BAD_BYTEARRAY_ALIGNMENT"); \ } diff --git a/posix-api.cabal b/posix-api.cabal --- a/posix-api.cabal +++ b/posix-api.cabal @@ -1,6 +1,6 @@ cabal-version: 2.2 name: posix-api -version: 0.2.1.0 +version: 0.3.0.0 synopsis: posix bindings description: This library provides a very thin wrapper around POSIX APIs. It can be @@ -71,19 +71,38 @@ include/HaskellPosix.h include/custom.h +flag assertions + manual: True + description: Extra run-time invariant checking + default: False + library exposed-modules: + Linux.Epoll Linux.Socket Posix.Directory + Posix.Poll + Posix.Select Posix.Socket + Posix.Types other-modules: + Linux.Epoll.Types Linux.Socket.Types - Posix.Socket.Types + Posix.Poll.Types Posix.Socket.Platform + Posix.Socket.Types + Assertion build-depends: , base >=4.11.1 && <5 - , primitive >= 0.6.4 + , primitive >= 0.7 && <0.8 + , primitive-addr >= 0.1 && <0.2 + , primitive-offset >= 0.2 && <0.3 + , primitive-unlifted >= 0.1 && <0.2 hs-source-dirs: src + if flag(assertions) + hs-source-dirs: src-assertions + else + hs-source-dirs: src-noassertions if os(linux) hs-source-dirs: src-linux default-language: Haskell2010 @@ -91,7 +110,7 @@ c-sources: cbits/HaskellPosix.c include-dirs: include includes: HaskellPosix.h - build-tools: hsc2hs + build-tool-depends: hsc2hs:hsc2hs test-suite test type: exitcode-stdio-1.0 diff --git a/src-assertions/Assertion.hs b/src-assertions/Assertion.hs new file mode 100644 --- /dev/null +++ b/src-assertions/Assertion.hs @@ -0,0 +1,19 @@ +{-# language MagicHash #-} +{-# language UnboxedTuples #-} + +module Assertion + ( assertMutablePrimArrayPinned + ) where + +import GHC.Exts (isTrue#) + +import qualified Data.Primitive as PM +import qualified GHC.Exts as Exts + +assertMutablePrimArrayPinned :: PM.MutablePrimArray s a -> PM.MutablePrimArray s a +assertMutablePrimArrayPinned x = if isMutablePrimArrayPinned x + then x + else error "assertMutablePrimArrayPinned" + +isMutablePrimArrayPinned :: PM.MutablePrimArray s a -> Bool +isMutablePrimArrayPinned (PM.MutablePrimArray marr#) = isTrue# (Exts.isMutableByteArrayPinned# marr#) diff --git a/src-linux/Posix/Socket/Platform.hsc b/src-linux/Posix/Socket/Platform.hsc --- a/src-linux/Posix/Socket/Platform.hsc +++ b/src-linux/Posix/Socket/Platform.hsc @@ -14,8 +14,6 @@ #include #include "custom.h" --- | All of the data constructors provided by this module are unsafe. --- Only use them if you really know what you are doing. module Posix.Socket.Platform ( -- * Encoding Socket Addresses encodeSocketAddressInternet @@ -28,7 +26,8 @@ ) where import Control.Monad (when) -import Data.Primitive (Addr,MutableByteArray,ByteArray(..),writeByteArray,indexByteArray) +import Data.Primitive (MutableByteArray,ByteArray(..),writeByteArray,indexByteArray) +import Data.Primitive.Addr (Addr(..)) import Data.Word (Word8) import Foreign.C.Types (CUShort,CInt) import GHC.Exts (ByteArray##,State##,RealWorld,runRW##,Ptr(..)) @@ -38,6 +37,7 @@ import Foreign.Storable (peekByteOff) import qualified Data.Primitive as PM +import qualified Data.Primitive.Addr as PMA import qualified Foreign.Storable as FS -- | The size of a serialized internet socket address. @@ -58,11 +58,11 @@ -- I am not sure of how to do this. At any rate, I do not expect -- that linux will ever change the bit size of sa_family_t, so I -- am not too concerned. - #{write struct sockaddr_in, sin_family} bs (#{const AF_INET} :: CUShort) + #{writeByteArray struct sockaddr_in, sin_family} bs 0 (#{const AF_INET} :: CUShort) -- The port and the address are already supposed to be in network -- byte order in the SocketAddressInternet data type. - #{write struct sockaddr_in, sin_port} bs port - #{write struct sockaddr_in, sin_addr.s_addr} bs address + #{writeByteArray struct sockaddr_in, sin_port} bs 0 port + #{writeByteArray struct sockaddr_in, sin_addr.s_addr} bs 0 address -- | Serialize a IPv4 socket address so that it may be passed to @bind@. -- This serialization is operating-system dependent. @@ -82,10 +82,10 @@ if PM.sizeofByteArray arr == (#{size struct sockaddr_in}) -- We assume that AF_INET takes up 16 bits. See the comment in -- encodeSocketAddressInternet for more detail. - then if (#{index struct sockaddr_in, sin_family} arr) == (#{const AF_INET} :: CUShort) + then if (#{indexByteArray struct sockaddr_in, sin_family} arr 0) == (#{const AF_INET} :: CUShort) then Just $ SocketAddressInternet - { port = #{index struct sockaddr_in, sin_port} arr - , address = #{index struct sockaddr_in, sin_addr.s_addr} arr + { port = #{indexByteArray struct sockaddr_in, sin_port} arr 0 + , address = #{indexByteArray struct sockaddr_in, sin_addr.s_addr} arr 0 } else Nothing else Nothing @@ -104,7 +104,7 @@ pure (Right (SocketAddressInternet { port, address })) else pure (Left (cushortToCInt fam)) where - !(PM.Addr offAddr) = PM.plusAddr addr (ix * (#{size struct sockaddr_in})) + !(Addr offAddr) = PMA.plusAddr addr (ix * (#{size struct sockaddr_in})) ptr = Ptr offAddr -- | Serialize a unix domain socket address so that it may be passed to @bind@. diff --git a/src-noassertions/Assertion.hs b/src-noassertions/Assertion.hs new file mode 100644 --- /dev/null +++ b/src-noassertions/Assertion.hs @@ -0,0 +1,8 @@ +module Assertion + ( assertMutablePrimArrayPinned + ) where + +import qualified Data.Primitive as PM + +assertMutablePrimArrayPinned :: PM.MutablePrimArray s a -> PM.MutablePrimArray s a +assertMutablePrimArrayPinned = id diff --git a/src/Linux/Epoll.hs b/src/Linux/Epoll.hs new file mode 100644 --- /dev/null +++ b/src/Linux/Epoll.hs @@ -0,0 +1,173 @@ +{-# language BangPatterns #-} +{-# language DataKinds #-} +{-# language GADTSyntax #-} +{-# language KindSignatures #-} +{-# language MagicHash #-} +{-# language UnliftedFFITypes #-} +{-# language NamedFieldPuns #-} +{-# language UnboxedTuples #-} + +module Linux.Epoll + ( -- * Functions + -- ** Create + uninterruptibleCreate + , uninterruptibleCreate1 + -- ** Wait + , waitMutablePrimArray + , uninterruptibleWaitMutablePrimArray + -- ** Control + , uninterruptibleControlMutablePrimArray + -- * Types + , EpollFlags(..) + , ControlOperation(..) + , Events(..) + , Event(..) + , Exchange(..) + -- * Classes + , PrimEpollData + -- * Constants + , T.closeOnExec + , T.add + , T.modify + , T.delete + , T.input + , T.output + , T.priority + , T.hangup + , T.readHangup + , T.error + , T.edgeTriggered + -- * Events Combinators + , T.containsAnyEvents + , T.containsAllEvents + -- * Marshalling + , T.sizeofEvent + , T.peekEventEvents + , T.peekEventDataFd + , T.peekEventDataPtr + , T.peekEventDataU32 + , T.peekEventDataU64 + , T.pokeEventDataU64 + -- , T.readEventDataU64 + -- , T.writeEventDataU64 + -- , T.writeEventEvents + ) where + +import Prelude hiding (error) + +import Assertion (assertMutablePrimArrayPinned) +import Data.Primitive (MutablePrimArray(..)) +import Foreign.C.Error (Errno,getErrno) +import Foreign.C.Types (CInt(..)) +import GHC.Exts (RealWorld,MutableByteArray#) +import Linux.Epoll.Types (EpollFlags(..),ControlOperation(..),Events(..),Exchange(..)) +import Linux.Epoll.Types (Event(..),PrimEpollData(..)) +import System.Posix.Types (Fd(..)) + +import qualified Linux.Epoll.Types as T + +foreign import ccall unsafe "sys/epoll.h epoll_create" + c_epoll_create :: CInt -> IO Fd + +foreign import ccall unsafe "sys/epoll.h epoll_create1" + c_epoll_create1 :: EpollFlags -> IO Fd + +foreign import ccall unsafe "sys/epoll.h epoll_wait" + c_epoll_wait_unsafe :: Fd -> MutableByteArray# RealWorld -> CInt -> CInt -> IO CInt + +foreign import ccall safe "sys/epoll.h epoll_wait" + c_epoll_wait_safe :: Fd -> MutableByteArray# RealWorld -> CInt -> CInt -> IO CInt + +foreign import ccall unsafe "sys/epoll.h epoll_ctl" + c_epoll_ctl_unsafe :: Fd -> ControlOperation -> Fd -> MutableByteArray# RealWorld -> IO CInt + +-- -- | Write @data.u64@ from @struct epoll_event@. +-- writeEventEvents :: +-- MutableByteArray RealWorld +-- -> Int -- ^ Index. Element are @struct epoll_event@. +-- -> Events e +-- -> IO () +-- writeEventEvents !arr !ix !payload = do +-- -- See the comments on readEventDataU64 +-- PM.writeByteArray arr (ix * 3 + 1) (word64ToWord32 (unsafeShiftR payload 32)) +-- PM.writeByteArray arr (ix * 3 + 2) (word64ToWord32 payload) + +uninterruptibleCreate :: + CInt -- ^ Size, ignored since Linux 2.6.8 + -> IO (Either Errno Fd) +{-# inline uninterruptibleCreate #-} +uninterruptibleCreate !sz = c_epoll_create sz >>= errorsFromFd + +uninterruptibleCreate1 :: + EpollFlags -- ^ Flags + -> IO (Either Errno Fd) +{-# inline uninterruptibleCreate1 #-} +uninterruptibleCreate1 !flags = + c_epoll_create1 flags >>= errorsFromFd + +-- | Wait for an I/O event on an epoll file descriptor. The +-- +-- includes more details. The @timeout@ argument is omitted +-- since it is nonsense to choose anything other than 0 when +-- using the unsafe FFI. +uninterruptibleWaitMutablePrimArray :: + Fd -- ^ EPoll file descriptor + -> MutablePrimArray RealWorld (Event 'Response a) -- ^ Event buffer + -> CInt -- ^ Maximum events + -> IO (Either Errno CInt) -- ^ Number of events received +{-# inline uninterruptibleWaitMutablePrimArray #-} +uninterruptibleWaitMutablePrimArray !epfd (MutablePrimArray evs) !maxEvents = + c_epoll_wait_unsafe epfd evs maxEvents 0 >>= errorsFromInt + +-- | Wait for an I/O event on an epoll file descriptor. The +-- +-- includes more details. The event buffer must be a pinned +-- byte array. +waitMutablePrimArray :: + Fd -- ^ EPoll file descriptor + -> MutablePrimArray RealWorld (Event 'Response a) -- ^ Event buffer, must be pinned + -> CInt -- ^ Maximum events + -> CInt -- ^ Timeout in milliseconds, use @-1@ to block forever. + -> IO (Either Errno CInt) -- ^ Number of events received +{-# inline waitMutablePrimArray #-} +waitMutablePrimArray !epfd !evs !maxEvents !timeout = + let !(MutablePrimArray evs#) = assertMutablePrimArrayPinned evs + in c_epoll_wait_safe epfd evs# maxEvents timeout >>= errorsFromInt + +-- | Add, modify, or remove entries in the interest list of the +-- epoll instance referred to by the file descriptor @epfd@. +-- +-- includes more details. +uninterruptibleControlMutablePrimArray :: + Fd -- ^ EPoll file descriptor (@epfd@) + -> ControlOperation + -- ^ Operation: @EPOLL_CTL_ADD@, @EPOLL_CTL_MOD@, or @EPOLL_CTL_DEL@ + -> Fd -- ^ File descriptor whose registration will be affected + -> MutablePrimArray RealWorld (Event 'Request a) + -- ^ A single event. This is read from, not written to. + -> IO (Either Errno ()) +{-# inline uninterruptibleControlMutablePrimArray #-} +uninterruptibleControlMutablePrimArray !epfd !op !fd (MutablePrimArray ev) = + c_epoll_ctl_unsafe epfd op fd ev >>= errorsFromInt_ + +errorsFromFd :: Fd -> IO (Either Errno Fd) +{-# inline errorsFromFd #-} +errorsFromFd r = if r > (-1) + then pure (Right r) + else fmap Left getErrno + +errorsFromInt :: CInt -> IO (Either Errno CInt) +{-# inline errorsFromInt #-} +errorsFromInt r = if r > (-1) + then pure (Right r) + else fmap Left getErrno + +-- Sometimes, functions that return an int use zero to indicate +-- success and negative one to indicate failure without including +-- additional information in the value. +errorsFromInt_ :: CInt -> IO (Either Errno ()) +{-# inline errorsFromInt_ #-} +errorsFromInt_ r = if r == 0 + then pure (Right ()) + else fmap Left getErrno + diff --git a/src/Linux/Epoll/Types.hsc b/src/Linux/Epoll/Types.hsc new file mode 100644 --- /dev/null +++ b/src/Linux/Epoll/Types.hsc @@ -0,0 +1,363 @@ +{-# language BangPatterns #-} +{-# language BinaryLiterals #-} +{-# language DataKinds #-} +{-# language DerivingStrategies #-} +{-# language DuplicateRecordFields #-} +{-# language GADTSyntax #-} +{-# language GeneralizedNewtypeDeriving #-} +{-# language KindSignatures #-} +{-# language MagicHash #-} +{-# language NamedFieldPuns #-} +{-# language PolyKinds #-} +{-# language ScopedTypeVariables #-} +{-# language TypeApplications #-} +{-# language TypeInType #-} +{-# language UnboxedTuples #-} + +-- This is needed because hsc2hs does not currently handle ticked +-- promoted data constructors correctly. +{-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-} + +#define _GNU_SOURCE +#include +#include "custom.h" + +-- | All of the data constructors provided by this module are unsafe. +-- Only use them if you really know what you are doing. +module Linux.Epoll.Types + ( EpollFlags(..) + , ControlOperation(..) + , Exchange(..) + , Events(..) + , Event(..) + , PrimEpollData(..) + -- * Flags + , closeOnExec + , add + , modify + , delete + -- * Events + , input + , output + , priority + , hangup + , readHangup + , error + , edgeTriggered + -- * Events Combinators + , containsAnyEvents + , containsAllEvents + -- * Marshalling + , sizeofEvent + , peekEventEvents + , peekEventDataFd + , peekEventDataPtr + , peekEventDataU32 + , peekEventDataU64 + , pokeEventDataU64 + -- , readEventDataU64 + -- , writeEventDataU64 + ) where + +import Prelude hiding (truncate,error) + +import Data.Bits (Bits,(.&.),(.|.),unsafeShiftL,unsafeShiftR) +import Data.Kind (Type) +import Data.Primitive.Addr (Addr(..)) +import Data.Primitive (Prim) +import Data.Primitive (indexByteArray##,writeByteArray##,readByteArray##) +import Data.Primitive (indexOffAddr##,readOffAddr##,writeOffAddr##) +import Data.Word (Word32,Word64) +import Foreign.C.Types (CInt(..)) +import Foreign.Storable (Storable,peekByteOff,pokeByteOff) +import GHC.Exts (Int(I##),(+##),(*##)) +import GHC.Exts (State##,Int##,Addr##,MutableByteArray##,ByteArray##) +import GHC.Ptr (Ptr(..)) +import Posix.Poll (Exchange(..)) +import System.Posix.Types (Fd(..)) + +import qualified Data.Primitive as PM + +newtype ControlOperation = ControlOperation CInt + deriving stock (Eq) + +newtype EpollFlags = EpollFlags CInt + deriving stock (Eq) + deriving newtype (Bits) + +instance Semigroup EpollFlags where (<>) = (.|.) +instance Monoid EpollFlags where mempty = EpollFlags 0 + +newtype Events :: Exchange -> Type where + Events :: Word32 -> Events e + deriving stock (Eq) + deriving newtype (Bits,Storable,Prim) + +instance Semigroup (Events e) where (<>) = (.|.) +instance Monoid (Events e) where mempty = Events 0 + +-- | A data type corresponding to @struct epoll_event@. Linux +-- defines this as: +-- +-- > typedef union epoll_data { +-- > void *ptr; +-- > int fd; +-- > uint32_t u32; +-- > uint64_t u64; +-- > } epoll_data_t; +-- > +-- > struct epoll_event { +-- > uint32_t events; /* Epoll events */ +-- > epoll_data_t data; /* User data variable */ +-- > }; +-- +-- It is a little difficult to capture what this type conveys, but +-- we make an attempt. The second argument to the @Event@ type +-- constructor is either @Addr@, @Fd@, @Word32@, or @Word64@. This +-- corresponds to the four possibilities in the @epoll_data@ union +-- type. As long as the user monomorphizes this type when using +-- it, there should not be any performance penalty for the +-- flexibility afforded by this approach. +data Event :: Exchange -> Type -> Type where + Event :: + { events :: !(Events e) + -- ^ Epoll events + , payload :: !a + -- ^ User data variable, named @data@ in @struct epoll_event@. + } -> Event e a + +class PrimEpollData a where + indexByteArrayEpoll :: ByteArray## -> Int## -> Event e a + readByteArrayEpoll :: MutableByteArray## s -> Int## -> State## s -> (## State## s, Event e a ##) + writeByteArrayEpoll :: MutableByteArray## s -> Int## -> Event e a -> State## s -> State## s + indexOffAddrEpoll :: Addr## -> Int## -> Event e a + readOffAddrEpoll :: Addr## -> Int## -> State## s -> (## State## s, Event e a ##) + writeOffAddrEpoll :: Addr## -> Int## -> Event e a -> State## s -> State## s + +instance PrimEpollData a => Prim (Event e a) where + {-# inline sizeOf# #-} + {-# inline alignment# #-} + {-# inline indexByteArray# #-} + {-# inline readByteArray# #-} + {-# inline writeByteArray# #-} + {-# inline setByteArray# #-} + {-# inline indexOffAddr# #-} + {-# inline readOffAddr# #-} + {-# inline writeOffAddr# #-} + {-# inline setOffAddr# #-} + sizeOf## _ = unI #{size struct epoll_event} + alignment## _ = PM.alignment## (undefined :: Word32) + indexByteArray## = indexByteArrayEpoll + readByteArray## = readByteArrayEpoll + writeByteArray## = writeByteArrayEpoll + setByteArray## = PM.defaultSetByteArray## + indexOffAddr## = indexOffAddrEpoll + readOffAddr## = readOffAddrEpoll + writeOffAddr## = writeOffAddrEpoll + setOffAddr## = PM.defaultSetOffAddr## + +instance PrimEpollData Fd where + {-# inline indexByteArrayEpoll #-} + {-# inline readByteArrayEpoll #-} + {-# inline writeByteArrayEpoll #-} + {-# inline indexOffAddrEpoll #-} + {-# inline readOffAddrEpoll #-} + {-# inline writeOffAddrEpoll #-} + indexByteArrayEpoll arr i = Event + { events = #{indexByteArrayHash struct epoll_event, events} arr i + , payload = #{indexByteArrayHash struct epoll_event, data.fd} arr i + } + writeByteArrayEpoll arr i Event{events,payload} s0 = + case #{writeByteArrayHash struct epoll_event, events} arr i events s0 of + s1 -> #{writeByteArrayHash struct epoll_event, data.fd} arr i payload s1 + readByteArrayEpoll arr i s0 = + case #{readByteArrayHash struct epoll_event, events} arr i s0 of + (## s1, events ##) -> case #{readByteArrayHash struct epoll_event, data.fd} arr i s1 of + (## s2, payload ##) -> (## s2, Event{events,payload} ##) + indexOffAddrEpoll arr i = Event + { events = #{indexOffAddrHash struct epoll_event, events} arr i + , payload = #{indexOffAddrHash struct epoll_event, data.fd} arr i + } + writeOffAddrEpoll arr i Event{events,payload} s0 = + case #{writeOffAddrHash struct epoll_event, events} arr i events s0 of + s1 -> #{writeOffAddrHash struct epoll_event, data.fd} arr i payload s1 + readOffAddrEpoll arr i s0 = + case #{readOffAddrHash struct epoll_event, events} arr i s0 of + (## s1, events ##) -> case #{readOffAddrHash struct epoll_event, data.fd} arr i s1 of + (## s2, payload ##) -> (## s2, Event{events,payload} ##) + +-- | Since @epoll_event@ includes an unaligned 64-bit word, it is +-- difficult to use @hsc2hs@ to generate the marshalling code. Consequently, +-- the offsets of @events@ and @data@ are currently hardcoded. Open an +-- issue in this causes a problem on your platform. +instance PrimEpollData Word64 where + {-# inline indexByteArrayEpoll #-} + {-# inline readByteArrayEpoll #-} + {-# inline writeByteArrayEpoll #-} + {-# inline indexOffAddrEpoll #-} + {-# inline readOffAddrEpoll #-} + {-# inline writeOffAddrEpoll #-} + indexByteArrayEpoll arr i = Event + { events = PM.indexByteArray## arr (i *## 3##) + , payload = composePayload + (PM.indexByteArray## arr ((i *## 3##) +# 1##)) + (PM.indexByteArray## arr ((i *## 3##) +# 2##)) + } + writeByteArrayEpoll arr i Event{events,payload} s0 = case PM.writeByteArray## arr (i *## 3##) events s0 of + s1 -> case PM.writeByteArray## arr ((i *## 3##) +## 1##) pa s1 of + s2 -> PM.writeByteArray## arr ((i *## 3##) +## 2##) pb s2 + where + !(pa,pb) = decomposePayload payload + readByteArrayEpoll arr i s0 = case PM.readByteArray## arr (i *## 3##) s0 of + (## s1, events ##) -> case PM.readByteArray## arr ((i *## 3##) +## 1##) s1 of + (## s2, pa ##) -> case PM.readByteArray## arr ((i *## 3##) +## 2##) s2 of + (## s3, pb ##) -> let payload = composePayload pa pb in + (## s3, Event{events,payload} ##) + indexOffAddrEpoll arr i = Event + { events = PM.indexOffAddr## arr (i *## 3##) + , payload = composePayload + (PM.indexOffAddr## arr ((i *## 3##) +## 1##)) + (PM.indexOffAddr## arr ((i *## 3##) +## 2##)) + } + writeOffAddrEpoll arr i Event{events,payload} s0 = case PM.writeOffAddr## arr (i *## 3##) events s0 of + s1 -> case PM.writeOffAddr## arr ((i *## 3##) +## 1##) pa s1 of + s2 -> PM.writeOffAddr## arr ((i *## 3##) +## 2##) pb s2 + where + !(pa,pb) = decomposePayload payload + readOffAddrEpoll arr i s0 = case PM.readOffAddr## arr (i *## 3##) s0 of + (## s1, events ##) -> case PM.readOffAddr## arr ((i *## 3##) +## 1##) s1 of + (## s2, pa ##) -> case PM.readOffAddr## arr ((i *## 3##) +## 2##) s2 of + (## s3, pb ##) -> let payload = composePayload pa pb in + (## s3, Event{events,payload} ##) + +-- | The @EPOLL_CTL_ADD@ control operation. +add :: ControlOperation +add = ControlOperation #{const EPOLL_CTL_ADD} + +-- | The @EPOLL_CTL_MOD@ control operation. +modify :: ControlOperation +modify = ControlOperation #{const EPOLL_CTL_MOD} + +-- | The @EPOLL_CTL_DEL@ control operation. +delete :: ControlOperation +delete = ControlOperation #{const EPOLL_CTL_DEL} + +-- | The @EPOLL_CLOEXEC@ flag. +closeOnExec :: EpollFlags +closeOnExec = EpollFlags #{const EPOLL_CLOEXEC} + +-- | The @EPOLLIN@ event. Can appear in a request or a response. +input :: Events e +input = Events #{const EPOLLIN} + +-- | The @EPOLLOUT@ event. Can appear in a request or a response. +output :: Events e +output = Events #{const EPOLLOUT} + +-- | The @EPOLLPRI@ event. Can appear in a request or a response. +priority :: Events e +priority = Events #{const EPOLLPRI} + +-- | The @EPOLLERR@ event. The +-- says +-- "@epoll_wait@ will always wait for this event; it is not necessary to set it in @events@". +-- Consequently, in this library, it has been marked as only appearing in @Response@ positions. +error :: Events Response +error = Events #{const EPOLLERR} + +-- | The @EPOLLHUP@ event. The +-- says +-- "@epoll_wait@ will always wait for this event; it is not necessary to set it in @events@". +-- Consequently, in this library, it has been marked as only appearing in @Response@ positions. +hangup :: Events Response +hangup = Events #{const EPOLLHUP} + +-- | The @EPOLLRDHUP@ event. Can appear in a request or a response. +readHangup :: Events e +readHangup = Events #{const EPOLLRDHUP} + +-- | The @EPOLLET@ event. Only appears in requests. +edgeTriggered :: Events Request +edgeTriggered = Events #{const EPOLLET} + +-- | Does the first event set entirely contain the second one? That is, +-- is the second argument a subset of the first? +containsAllEvents :: Events e -> Events e -> Bool +containsAllEvents (Events a) (Events b) = a .&. b == b + +-- | Does the first event set contain any of the events from the second one? +containsAnyEvents :: Events e -> Events e -> Bool +containsAnyEvents (Events a) (Events b) = (a .&. b) /= 0 + +sizeofEvent :: Int +sizeofEvent = #{size struct epoll_event} + +-- | Read @events@ from @struct epoll_event@. +peekEventEvents :: Addr -> IO (Events e) +peekEventEvents (Addr p) = #{peek struct epoll_event, events} (Ptr p) + +-- | Read @data.fd@ from @struct epoll_event@. +peekEventDataFd :: Addr -> IO Fd +peekEventDataFd (Addr p) = #{peek struct epoll_event, data.fd} (Ptr p) + +-- | Read @data.ptr@ from @struct epoll_event@. +peekEventDataPtr :: Addr -> IO Addr +peekEventDataPtr (Addr p) = do + Ptr q <- #{peek struct epoll_event, data.ptr} (Ptr p) + pure (Addr q) + +-- | Read @data.u32@ from @struct epoll_event@. +peekEventDataU32 :: Addr -> IO Word32 +peekEventDataU32 (Addr p) = #{peek struct epoll_event, data.u32} (Ptr p) + +-- | Read @data.u64@ from @struct epoll_event@. +peekEventDataU64 :: Addr -> IO Word64 +peekEventDataU64 (Addr p) = #{peek struct epoll_event, data.u64} (Ptr p) + +-- | Write @data.u64@ from @struct epoll_event@. +pokeEventDataU64 :: Addr -> Word64 -> IO () +pokeEventDataU64 (Addr p) w = #{poke struct epoll_event, data.u64} (Ptr p) w + +composePayload :: Word32 -> Word32 -> Word64 +{-# inline composePayload #-} +composePayload a b = unsafeShiftL (word32ToWord64 a) 32 .|. word32ToWord64 b + +decomposePayload :: Word64 -> (Word32,Word32) +{-# inline decomposePayload #-} +decomposePayload w = (word64ToWord32 (unsafeShiftR w 32), word64ToWord32 w) + +word32ToWord64 :: Word32 -> Word64 +word32ToWord64 = fromIntegral + +word64ToWord32 :: Word64 -> Word32 +word64ToWord32 = fromIntegral + +unI :: Int -> Int## +unI (I## i) = i + +-- -- | Read @data.u64@ from @struct epoll_event@. +-- readEventDataU64 :: +-- MutableByteArray RealWorld +-- -> Int -- ^ Index. Elements are @struct epoll_event@. +-- -> IO Word64 +-- readEventDataU64 !arr !ix = do +-- -- On 64-bit platforms, Linux bitpacks this structure, causing the +-- -- data (a 64-bit word) to be misaligned. Consequently, we must +-- -- hardcode the assumed offsets to perform only aligned accesses. +-- -- The behavior is deterministic across platforms of different +-- -- endianness only if the only use of this function is paired with +-- -- writeEventDataU64. +-- (a :: Word32) <- PM.readByteArray arr (ix * 3 + 1) +-- (b :: Word32) <- PM.readByteArray arr (ix * 3 + 2) +-- pure (unsafeShiftL (word32ToWord64 a) 32 .|. word32ToWord64 b) +-- +-- -- | Write @data.u64@ from @struct epoll_event@. +-- writeEventDataU64 :: +-- MutableByteArray RealWorld +-- -> Int -- ^ Index. Element are @struct epoll_event@. +-- -> Word64 -- ^ Data +-- -> IO () +-- writeEventDataU64 !arr !ix !payload = do +-- -- See the comments on readEventDataU64 +-- PM.writeByteArray arr (ix * 3 + 1) (word64ToWord32 (unsafeShiftR payload 32)) +-- PM.writeByteArray arr (ix * 3 + 2) (word64ToWord32 payload) diff --git a/src/Linux/Socket.hs b/src/Linux/Socket.hs --- a/src/Linux/Socket.hs +++ b/src/Linux/Socket.hs @@ -9,6 +9,9 @@ ( -- * Functions uninterruptibleReceiveMultipleMessageA , uninterruptibleReceiveMultipleMessageB + , uninterruptibleReceiveMultipleMessageC + , uninterruptibleReceiveMultipleMessageD + , uninterruptibleAccept4 -- * Types , SocketFlags(..) -- * Option Names @@ -46,18 +49,21 @@ import Control.Monad (when) import Data.Bits ((.|.)) -import Data.Primitive (MutableByteArray(..),Addr(..),ByteArray(..)) -import Data.Primitive (MutableUnliftedArray(..),UnliftedArray) +import Data.Primitive.Addr (Addr(..),plusAddr,nullAddr) +import Data.Primitive (MutableByteArray(..),ByteArray(..),MutablePrimArray(..)) +import Data.Primitive.Unlifted.Array (MutableUnliftedArray(..),UnliftedArray) +import Data.Word (Word8) import Foreign.C.Error (Errno,getErrno) import Foreign.C.Types (CInt(..),CSize(..),CUInt(..)) -import GHC.Exts (Ptr,RealWorld,ByteArray#,MutableByteArray#,Addr#,MutableArrayArray#,Int(I#)) +import GHC.Exts (Ptr(..),RealWorld,MutableByteArray#,Addr#,MutableArrayArray#,Int(I#)) import GHC.Exts (shrinkMutableByteArray#,touch#,nullAddr#) import GHC.IO (IO(..)) import Linux.Socket.Types (SocketFlags(..)) -import Posix.Socket (Type(..),MessageFlags(..),Message(Receive)) +import Posix.Socket (Type(..),MessageFlags(..),Message(Receive),SocketAddress(..)) import System.Posix.Types (Fd(..),CSsize(..)) import qualified Data.Primitive as PM +import qualified Data.Primitive.Unlifted.Array as PM import qualified Control.Monad.Primitive as PM import qualified Posix.Socket as S import qualified Linux.Socket.Types as LST @@ -70,6 +76,32 @@ -> Addr# -- Timeout -> IO CSsize +foreign import ccall unsafe "sys/socket.h accept4" + c_unsafe_accept4 :: Fd + -> MutableByteArray# RealWorld -- SocketAddress + -> MutableByteArray# RealWorld -- Ptr CInt + -> SocketFlags + -> IO Fd + +foreign import ccall unsafe "HaskellPosix.h recvmmsg_sockaddr_in" + c_unsafe_recvmmsg_sockaddr_in :: + Fd + -> MutableByteArray# RealWorld -- lengths + -> MutableByteArray# RealWorld -- sockaddrs + -> MutableArrayArray# RealWorld -- buffers + -> CUInt -- Length of msghdr array + -> MessageFlags 'Receive + -> IO CInt + +foreign import ccall unsafe "HaskellPosix.h recvmmsg_sockaddr_discard" + c_unsafe_recvmmsg_sockaddr_discard :: + Fd + -> MutableByteArray# RealWorld -- lengths + -> MutableArrayArray# RealWorld -- buffers + -> CUInt -- Length of msghdr array + -> MessageFlags 'Receive + -> IO CInt + -- | Linux extends the @type@ argument of -- to allow -- setting two socket flags on socket creation: @SOCK_CLOEXEC@ and @@ -108,8 +140,8 @@ bufs <- PM.unsafeNewUnliftedArray (cuintToInt msgCount) mmsghdrsBuf <- PM.newPinnedByteArray (cuintToInt msgCount * cintToInt LST.sizeofMultipleMessageHeader) iovecsBuf <- PM.newPinnedByteArray (cuintToInt msgCount * cintToInt S.sizeofIOVector) - let !mmsghdrsAddr@(Addr mmsghdrsAddr#) = PM.mutableByteArrayContents mmsghdrsBuf - let iovecsAddr = PM.mutableByteArrayContents iovecsBuf + let !mmsghdrsAddr@(Addr mmsghdrsAddr#) = ptrToAddr (PM.mutableByteArrayContents mmsghdrsBuf) + let iovecsAddr = ptrToAddr (PM.mutableByteArrayContents iovecsBuf) initializeMultipleMessageHeadersWithoutSockAddr bufs iovecsAddr mmsghdrsAddr msgSize msgCount r <- c_unsafe_addr_recvmmsg s mmsghdrsAddr# msgCount flags nullAddr# if r > (-1) @@ -164,9 +196,9 @@ sockaddrsBuf <- PM.newPinnedByteArray (cuintToInt msgCount * cintToInt expSockAddrSize) -- Linux does not require zeroing out sockaddr_in before using it, -- so we leave sockaddrsBuf alone after initialization. - let sockaddrsAddr = PM.mutableByteArrayContents sockaddrsBuf - let !mmsghdrsAddr@(Addr mmsghdrsAddr#) = PM.mutableByteArrayContents mmsghdrsBuf - let iovecsAddr = PM.mutableByteArrayContents iovecsBuf + let sockaddrsAddr = ptrToAddr (PM.mutableByteArrayContents sockaddrsBuf) + let !mmsghdrsAddr@(Addr mmsghdrsAddr#) = ptrToAddr (PM.mutableByteArrayContents mmsghdrsBuf) + let iovecsAddr = ptrToAddr (PM.mutableByteArrayContents iovecsBuf) initializeMultipleMessageHeadersWithSockAddr bufs iovecsAddr mmsghdrsAddr sockaddrsAddr expSockAddrSize msgSize msgCount r <- c_unsafe_addr_recvmmsg s mmsghdrsAddr# msgCount flags nullAddr# if r > (-1) @@ -185,7 +217,30 @@ touchMutableByteArray sockaddrsBuf fmap Left getErrno +-- | All three buffer arguments need to have the same length (in elements, not bytes). +uninterruptibleReceiveMultipleMessageC :: + Fd -- ^ Socket + -> MutablePrimArray RealWorld CInt -- ^ Buffer for payload lengths + -> MutablePrimArray RealWorld S.SocketAddressInternet -- ^ Buffer for @sockaddr_in@s + -> MutableUnliftedArray RealWorld (MutableByteArray RealWorld) -- ^ Buffers for payloads + -> CUInt -- ^ Maximum number of datagrams to receive, length of buffers + -> MessageFlags 'Receive -- ^ Flags + -> IO (Either Errno CInt) +uninterruptibleReceiveMultipleMessageC !s (MutablePrimArray lens) (MutablePrimArray addrs) (PM.MutableUnliftedArray payloads) !msgCount !flags = + c_unsafe_recvmmsg_sockaddr_in s lens addrs payloads msgCount flags >>= errorsFromInt +-- | All three buffer arguments need to have the same length (in elements, not bytes). +-- This discards the source addresses. +uninterruptibleReceiveMultipleMessageD :: + Fd -- ^ Socket + -> MutablePrimArray RealWorld CInt -- ^ Buffer for payload lengths + -> MutableUnliftedArray RealWorld (MutableByteArray RealWorld) -- ^ Buffers for payloads + -> CUInt -- ^ Maximum number of datagrams to receive, length of buffers + -> MessageFlags 'Receive -- ^ Flags + -> IO (Either Errno CInt) +uninterruptibleReceiveMultipleMessageD !s (MutablePrimArray lens) (PM.MutableUnliftedArray payloads) !msgCount !flags = + c_unsafe_recvmmsg_sockaddr_discard s lens payloads msgCount flags >>= errorsFromInt + -- This sets up an array of mmsghdr. Each msghdr has msg_iov set to -- be an array of iovec with a single element. initializeMultipleMessageHeadersWithoutSockAddr :: @@ -198,9 +253,9 @@ initializeMultipleMessageHeadersWithoutSockAddr bufs iovecsAddr mmsgHdrsAddr msgSize msgCount = let go !ix !iovecAddr !mmsgHdrAddr = if ix < cuintToInt msgCount then do - pokeMultipleMessageHeader mmsgHdrAddr PM.nullAddr 0 iovecAddr 1 PM.nullAddr 0 mempty 0 + pokeMultipleMessageHeader mmsgHdrAddr nullAddr 0 iovecAddr 1 nullAddr 0 mempty 0 initializeIOVector bufs iovecAddr msgSize ix - go (ix + 1) (PM.plusAddr iovecAddr (cintToInt S.sizeofIOVector)) (PM.plusAddr mmsgHdrAddr (cintToInt LST.sizeofMultipleMessageHeader)) + go (ix + 1) (plusAddr iovecAddr (cintToInt S.sizeofIOVector)) (plusAddr mmsgHdrAddr (cintToInt LST.sizeofMultipleMessageHeader)) else pure () in go 0 iovecsAddr mmsgHdrsAddr @@ -219,15 +274,18 @@ initializeMultipleMessageHeadersWithSockAddr bufs iovecsAddr0 mmsgHdrsAddr0 sockaddrsAddr0 sockaddrSize msgSize msgCount = let go !ix !iovecAddr !mmsgHdrAddr !sockaddrAddr = if ix < cuintToInt msgCount then do - pokeMultipleMessageHeader mmsgHdrAddr sockaddrAddr sockaddrSize iovecAddr 1 PM.nullAddr 0 mempty 0 + pokeMultipleMessageHeader mmsgHdrAddr sockaddrAddr sockaddrSize iovecAddr 1 nullAddr 0 mempty 0 initializeIOVector bufs iovecAddr msgSize ix go (ix + 1) - (PM.plusAddr iovecAddr (cintToInt S.sizeofIOVector)) - (PM.plusAddr mmsgHdrAddr (cintToInt LST.sizeofMultipleMessageHeader)) - (PM.plusAddr sockaddrAddr (cintToInt sockaddrSize)) + (plusAddr iovecAddr (cintToInt S.sizeofIOVector)) + (plusAddr mmsgHdrAddr (cintToInt LST.sizeofMultipleMessageHeader)) + (plusAddr sockaddrAddr (cintToInt sockaddrSize)) else pure () in go 0 iovecsAddr0 mmsgHdrsAddr0 sockaddrsAddr0 +ptrToAddr :: Ptr Word8 -> Addr +ptrToAddr (Ptr x) = Addr x + -- Initialize a single iovec. We write the pinned byte array into -- both the iov_base field and into an unlifted array. initializeIOVector :: @@ -239,7 +297,7 @@ initializeIOVector bufs iovecAddr msgSize ix = do buf <- PM.newPinnedByteArray (csizeToInt msgSize) PM.writeUnliftedArray bufs ix buf - S.pokeIOVectorBase iovecAddr (PM.mutableByteArrayContents buf) + S.pokeIOVectorBase iovecAddr (ptrToAddr (PM.mutableByteArrayContents buf)) S.pokeIOVectorLength iovecAddr msgSize -- Freeze a slice of the mutable byte arrays inside the unlifted array, @@ -263,7 +321,7 @@ when (cuintToInt sz < csizeToInt bufSize) (shrinkMutableByteArray buf (cuintToInt sz)) PM.writeUnliftedArray r ix =<< PM.unsafeFreezeByteArray buf go r (validation .|. (sockaddrSz - expSockAddrSize)) (ix + 1) (max maxMsgSz sz) - (PM.plusAddr mmsghdr (cintToInt LST.sizeofMultipleMessageHeader)) + (plusAddr mmsghdr (cintToInt LST.sizeofMultipleMessageHeader)) else do a <- PM.unsafeFreezeUnliftedArray r pure (validation,maxMsgSz,a) @@ -283,6 +341,29 @@ shrinkMutableByteArray (MutableByteArray arr) (I# sz) = PM.primitive_ (shrinkMutableByteArray# arr sz) +-- | Variant of 'Posix.Socket.uninterruptibleAccept' that allows setting +-- flags on the newly-accepted connection. +uninterruptibleAccept4 :: + Fd -- ^ Listening socket + -> CInt -- ^ Maximum socket address size + -> SocketFlags -- ^ Set non-blocking and close-on-exec without extra syscall + -> IO (Either Errno (CInt,SocketAddress,Fd)) -- ^ Peer information and connected socket +{-# inline uninterruptibleAccept4 #-} +uninterruptibleAccept4 !sock !maxSz !flags = do + sockAddrBuf@(MutableByteArray sockAddrBuf#) <- PM.newByteArray (cintToInt maxSz) + lenBuf@(MutableByteArray lenBuf#) <- PM.newByteArray (PM.sizeOf (undefined :: CInt)) + PM.writeByteArray lenBuf 0 maxSz + r <- c_unsafe_accept4 sock sockAddrBuf# lenBuf# flags + if r > (-1) + then do + (sz :: CInt) <- PM.readByteArray lenBuf 0 + if sz < maxSz + then shrinkMutableByteArray sockAddrBuf (cintToInt sz) + else pure () + sockAddr <- PM.unsafeFreezeByteArray sockAddrBuf + pure (Right (sz,SocketAddress sockAddr,r)) + else fmap Left getErrno + cintToInt :: CInt -> Int cintToInt = fromIntegral @@ -294,6 +375,12 @@ cssizeToInt :: CSsize -> Int cssizeToInt = fromIntegral + +errorsFromInt :: CInt -> IO (Either Errno CInt) +{-# inline errorsFromInt #-} +errorsFromInt r = if r > (-1) + then pure (Right r) + else fmap Left getErrno touchMutableUnliftedArray :: MutableUnliftedArray RealWorld a -> IO () touchMutableUnliftedArray (MutableUnliftedArray x) = touchMutableUnliftedArray# x diff --git a/src/Linux/Socket/Types.hsc b/src/Linux/Socket/Types.hsc --- a/src/Linux/Socket/Types.hsc +++ b/src/Linux/Socket/Types.hsc @@ -61,17 +61,14 @@ import Prelude hiding (truncate) -import Data.Bits (Bits((.|.)),unsafeShiftL,unsafeShiftR) +import Data.Bits (Bits((.|.))) import Data.Word (Word8,Word16,Word32) -import Data.Primitive (Addr(..),MutableByteArray,writeByteArray) +import Data.Primitive.Addr (Addr(..),writeOffAddr) import Foreign.C.Types (CInt(..),CSize,CUInt) import Posix.Socket (MessageFlags(..),Message(Receive),OptionName(..)) import Foreign.Storable (peekByteOff,pokeByteOff) import GHC.Ptr (Ptr(..)) -import GHC.Exts (RealWorld) -import qualified Data.Primitive as PM - newtype SocketFlags = SocketFlags CInt deriving stock (Eq) deriving newtype (Bits) @@ -113,7 +110,7 @@ -- | The @MSG_TRUNC@ receive flag. truncate :: MessageFlags Receive -truncate = MessageFlags #{const MSG_DONTWAIT} +truncate = MessageFlags #{const MSG_TRUNC} -- | The @MSG_CTRUNC@ receive flag. controlTruncate :: MessageFlags Receive @@ -192,12 +189,10 @@ -- will marshal the value appropriately depending on the platform's -- bit-endianness. pokeIpHeaderVersionIhl :: Addr -> Word8 -> IO () --- TODO: Verify if this is correct. +-- TODO: Verify if this is correct. Also, something bad is going +-- on here. Fix this. #if defined(__LITTLE_ENDIAN_BITFIELD) -pokeIpHeaderVersionIhl p _ = PM.writeOffAddr p 0 (0b01000101 :: Word8) - -- PM.writeOffAddr p 0 (unsafeShiftL w 4 .|. unsafeShiftR w 4) - -- where - -- rev = +pokeIpHeaderVersionIhl p _ = writeOffAddr p 0 (0b01000101 :: Word8) #elif defined (__BIG_ENDIAN_BITFIELD) pokeIpHeaderVersionIhl p w = PM.writeOffAddr p 0 w #else diff --git a/src/Posix/Directory.hs b/src/Posix/Directory.hs --- a/src/Posix/Directory.hs +++ b/src/Posix/Directory.hs @@ -7,7 +7,7 @@ ( getCurrentWorkingDirectory ) where -import Data.Primitive (Addr(..),ByteArray) +import Data.Primitive (ByteArray) import GHC.Exts (Ptr(..)) import Foreign.Ptr (nullPtr) import Foreign.C.Error (Errno,eRANGE,getErrno) @@ -34,7 +34,7 @@ -- functions where these repeated 4KB allocations might trigger -- GC very quickly. marr <- PM.newPinnedByteArray sz - let !(Addr addr) = PM.mutableByteArrayContents marr + let !(Ptr addr) = PM.mutableByteArrayContents marr ptr <- c_getcwd (Ptr addr) (intToCSize sz) -- We probably want to use touch# or with# here. if ptr /= nullPtr diff --git a/src/Posix/Poll.hs b/src/Posix/Poll.hs new file mode 100644 --- /dev/null +++ b/src/Posix/Poll.hs @@ -0,0 +1,55 @@ +{-# language BangPatterns #-} +{-# language MagicHash #-} +{-# language UnliftedFFITypes #-} +{-# language ScopedTypeVariables #-} + +module Posix.Poll + ( uninterruptiblePoll + , uninterruptiblePollMutablePrimArray + , PollFd(..) + , Exchange(..) + , PT.input + , PT.output + , PT.error + , PT.hangup + , PT.invalid + , PT.isSubeventOf + ) where + +import Posix.Types (CNfds(..)) +import Foreign.C.Error (Errno,getErrno) +import Foreign.C.Types (CInt(..)) +import GHC.Ptr (Ptr) +import GHC.Exts (RealWorld,MutableByteArray#) +import Posix.Poll.Types (PollFd(..),Exchange(..)) +import Data.Primitive (MutablePrimArray(..)) +import Data.Word (Word64) + +import qualified Posix.Poll.Types as PT + +foreign import ccall unsafe "poll.h poll" + c_poll_ptr :: Ptr PollFd -> CNfds -> CInt -> IO CInt + +foreign import ccall unsafe "poll.h poll" + c_poll_prim_array :: MutableByteArray# RealWorld -> CNfds -> CInt -> IO CInt + +-- | The @timeout@ argument is omitted since it is nonsense to choose +-- anything other than 0 when using the unsafe FFI. +uninterruptiblePoll :: + Ptr PollFd + -> CNfds + -> IO (Either Errno CInt) +uninterruptiblePoll pfds n = + c_poll_ptr pfds n 0 >>= errorsFromInt + +uninterruptiblePollMutablePrimArray :: + MutablePrimArray RealWorld PollFd + -> CNfds + -> IO (Either Errno CInt) +uninterruptiblePollMutablePrimArray (MutablePrimArray pfds) n = + c_poll_prim_array pfds n 0 >>= errorsFromInt + +errorsFromInt :: CInt -> IO (Either Errno CInt) +errorsFromInt r = if r >= 0 + then pure (Right r) + else fmap Left getErrno diff --git a/src/Posix/Poll/Types.hsc b/src/Posix/Poll/Types.hsc new file mode 100644 --- /dev/null +++ b/src/Posix/Poll/Types.hsc @@ -0,0 +1,141 @@ +{-# language BangPatterns #-} +{-# language BinaryLiterals #-} +{-# language DataKinds #-} +{-# language DerivingStrategies #-} +{-# language DuplicateRecordFields #-} +{-# language GADTSyntax #-} +{-# language GeneralizedNewtypeDeriving #-} +{-# language KindSignatures #-} +{-# language NamedFieldPuns #-} +{-# language TypeApplications #-} +{-# language MagicHash #-} +{-# language UnboxedTuples #-} +{-# language PolyKinds #-} +{-# language TypeInType #-} + +-- This is needed because hsc2hs does not currently handle ticked +-- promoted data constructors correctly. +{-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-} + +#define _GNU_SOURCE +#include +#include "custom.h" + +-- | All of the data constructors provided by this module are unsafe. +-- Only use them if you really know what you are doing. +module Posix.Poll.Types + ( PollFd(..) + , Exchange(..) + , input + , output + , error + , hangup + , invalid + , isSubeventOf + ) where + +import Prelude hiding (truncate,error) + +import Data.Bits ((.|.),(.&.)) +import Data.Word (Word8,Word16,Word32,Word64) +import Data.Primitive (Prim(..)) +import Foreign.C.Types (CInt(..),CShort) +import Foreign.Storable (Storable(..)) +import GHC.Ptr (Ptr(..)) +import GHC.Exts (RealWorld,Int(I##),Int##,(+##),(*##)) +import System.Posix.Types (Fd(..)) + +import qualified Data.Kind +import qualified Data.Primitive as PM + +data PollFd = PollFd + { descriptor :: !Fd + -- ^ The @fd@ field of @struct pollfd@ + , request :: !(Event Request) + -- ^ The @events@ field of @struct pollfd@ + , response :: !(Event Response) + -- ^ The @revents@ field of @struct pollfd@ + } + +newtype Event :: Exchange -> Data.Kind.Type where + Event :: CShort -> Event e + deriving newtype (Eq,Storable,Prim) + +instance Semigroup (Event e) where + Event a <> Event b = Event (a .|. b) + +instance Monoid (Event e) where + mempty = Event 0 + +data Exchange = Request | Response + +instance Storable PollFd where + sizeOf _ = #{size struct pollfd} + alignment _ = alignment (undefined :: CInt) + peek ptr = do + descriptor <- #{peek struct pollfd, fd} ptr + request <- #{peek struct pollfd, events} ptr + response <- #{peek struct pollfd, revents} ptr + let !pollfd = PollFd{descriptor,request,response} + pure pollfd + poke ptr PollFd{descriptor,request,response} = do + #{poke struct pollfd, fd} ptr descriptor + #{poke struct pollfd, events} ptr request + #{poke struct pollfd, revents} ptr response + +unI :: Int -> Int## +unI (I## i) = i + +instance Prim PollFd where + sizeOf## _ = unI #{size struct pollfd} + alignment## _ = alignment## (undefined :: CInt) + indexByteArray## arr i = PollFd + { descriptor = #{indexByteArrayHash struct pollfd, fd} arr i + , request = #{indexByteArrayHash struct pollfd, events} arr i + , response = #{indexByteArrayHash struct pollfd, revents} arr i + } + writeByteArray## arr i PollFd{descriptor,request,response} s0 = case #{writeByteArrayHash struct pollfd, fd} arr i descriptor s0 of + s1 -> case #{writeByteArrayHash struct pollfd, events} arr i request s1 of + s2 -> #{writeByteArrayHash struct pollfd, revents} arr i response s2 + readByteArray## arr i s0 = case #{readByteArrayHash struct pollfd, fd} arr i s0 of + (## s1, descriptor ##) -> case #{readByteArrayHash struct pollfd, events} arr i s1 of + (## s2, request ##) -> case #{readByteArrayHash struct pollfd, revents} arr i s2 of + (## s3, response ##) -> (## s3, PollFd{descriptor,request,response} ##) + setByteArray## = PM.defaultSetByteArray## + indexOffAddr## arr i = PollFd + { descriptor = #{indexOffAddrHash struct pollfd, fd} arr i + , request = #{indexOffAddrHash struct pollfd, events} arr i + , response = #{indexOffAddrHash struct pollfd, revents} arr i + } + writeOffAddr## arr i PollFd{descriptor,request,response} s0 = case #{writeOffAddrHash struct pollfd, fd} arr i descriptor s0 of + s1 -> case #{writeOffAddrHash struct pollfd, events} arr i request s1 of + s2 -> #{writeOffAddrHash struct pollfd, revents} arr i response s2 + readOffAddr## arr i s0 = case #{readOffAddrHash struct pollfd, fd} arr i s0 of + (## s1, fdVal ##) -> case #{readOffAddrHash struct pollfd, events} arr i s1 of + (## s2, eventsVal ##) -> case #{readOffAddrHash struct pollfd, revents} arr i s2 of + (## s3, reventsVal ##) -> (## s3, PollFd fdVal eventsVal reventsVal ##) + setOffAddr## = PM.defaultSetOffAddr## + +-- | The @POLLIN@ event. +input :: Event e +input = Event #{const POLLIN} + +-- | The @POLLOUT@ event. +output :: Event e +output = Event #{const POLLOUT} + +-- | The @POLLERR@ event. +error :: Event Response +error = Event #{const POLLERR} + +-- | The @POLLHUP@ event. +hangup :: Event Response +hangup = Event #{const POLLHUP} + +-- | The @POLLNVAL@ event. +invalid :: Event Response +invalid = Event #{const POLLNVAL} + +-- | Is the first argument a subset of the second argument? +isSubeventOf :: Event e -> Event e -> Bool +isSubeventOf (Event a) (Event b) = a .&. b == a diff --git a/src/Posix/Select.hs b/src/Posix/Select.hs new file mode 100644 --- /dev/null +++ b/src/Posix/Select.hs @@ -0,0 +1,5 @@ +module Posix.Select + ( + ) where + + diff --git a/src/Posix/Socket.hs b/src/Posix/Socket.hs --- a/src/Posix/Socket.hs +++ b/src/Posix/Socket.hs @@ -1,6 +1,11 @@ {-# language BangPatterns #-} {-# language DataKinds #-} +{-# language DuplicateRecordFields #-} +{-# language GADTSyntax #-} +{-# language KindSignatures #-} +{-# language LambdaCase #-} {-# language MagicHash #-} +{-# language NamedFieldPuns #-} {-# language ScopedTypeVariables #-} {-# language UnboxedTuples #-} {-# language UnliftedFFITypes #-} @@ -60,6 +65,8 @@ -- ** Send To , uninterruptibleSendToByteArray , uninterruptibleSendToMutableByteArray + , uninterruptibleSendToInternetByteArray + , uninterruptibleSendToInternetMutableByteArray -- ** Write Vector , writeVector -- ** Receive @@ -70,10 +77,14 @@ -- ** Receive From , uninterruptibleReceiveFromMutableByteArray , uninterruptibleReceiveFromMutableByteArray_ + , uninterruptibleReceiveFromInternetMutableByteArray -- ** Receive Message -- $receiveMessage , uninterruptibleReceiveMessageA , uninterruptibleReceiveMessageB + -- ** Send Message + , uninterruptibleSendMessageA + , uninterruptibleSendMessageB -- ** Byte-Order Conversion -- $conversion , hostToNetworkLong @@ -166,24 +177,31 @@ import GHC.ByteOrder (ByteOrder(BigEndian,LittleEndian),targetByteOrder) import GHC.IO (IO(..)) -import Data.Primitive (MutablePrimArray(..),MutableByteArray(..),Addr(..),ByteArray(..)) -import Data.Primitive (MutableUnliftedArray(..),UnliftedArray(..)) -import Data.Word (Word16,Word32,byteSwap16,byteSwap32) +import Data.Primitive.Addr (Addr(..),plusAddr,nullAddr) +import Data.Primitive (MutablePrimArray(..),MutableByteArray(..),ByteArray(..)) +import Data.Primitive.Unlifted.Array (MutableUnliftedArray(..),UnliftedArray(..)) +import Data.Primitive.ByteArray.Offset (MutableByteArrayOffset(..)) +import Data.Primitive.PrimArray.Offset (MutablePrimArrayOffset(..)) +import Data.Word (Word8,Word16,Word32,byteSwap16,byteSwap32) import Data.Void (Void) import Foreign.C.Error (Errno,getErrno) import Foreign.C.Types (CInt(..),CSize(..)) import Foreign.Ptr (nullPtr) -import GHC.Exts (Ptr,RealWorld,ByteArray#,MutableByteArray#,Addr#) +import GHC.Exts (Ptr,RealWorld,ByteArray#,MutableByteArray#) +import GHC.Exts (Addr#,TYPE,RuntimeRep(UnliftedRep)) import GHC.Exts (ArrayArray#,MutableArrayArray#,Int(I#)) import GHC.Exts (shrinkMutableByteArray#,touch#) import Posix.Socket.Types (Domain(..),Protocol(..),Type(..),SocketAddress(..)) +import Posix.Socket.Types (SocketAddressInternet(..)) import Posix.Socket.Types (MessageFlags(..),Message(..),ShutdownType(..)) import Posix.Socket.Types (Level(..),OptionName(..),OptionValue(..)) import System.Posix.Types (Fd(..),CSsize(..)) import qualified Posix.Socket.Types as PST import qualified Data.Primitive as PM +import qualified Data.Primitive.Unlifted.Array as PM import qualified Control.Monad.Primitive as PM +import qualified GHC.Exts as Exts -- This module include operating-system specific code used -- to serialize some of various kind of socket address types. @@ -298,7 +316,17 @@ -- The ByteArray# (second to last argument) is a SocketAddress. foreign import ccall unsafe "sys/socket.h sendto_offset" c_unsafe_mutable_bytearray_sendto :: Fd -> MutableByteArray# RealWorld -> CInt -> CSize -> MessageFlags 'Send -> ByteArray# -> CInt -> IO CSsize +foreign import ccall unsafe "sys/socket.h sendto_inet_offset" + c_unsafe_mutable_bytearray_sendto_inet :: Fd -> MutableByteArray# RealWorld -> CInt -> CSize -> MessageFlags 'Send -> Word16 -> Word32 -> IO CSsize +foreign import ccall unsafe "HaskellPosix.h sendto_inet_offset" + c_unsafe_bytearray_sendto_inet :: Fd -> ByteArray# -> CInt -> CSize -> MessageFlags 'Send -> Word16 -> Word32 -> IO CSsize +foreign import ccall unsafe "HaskellPosix.h sendmsg_a" + c_unsafe_sendmsg_a :: Fd -> Addr# -> CSize -> MutableByteArray# RealWorld -> Int -> CSize -> MessageFlags 'Send -> IO CSsize + +foreign import ccall unsafe "HaskellPosix.h sendmsg_b" + c_unsafe_sendmsg_b :: Fd -> MutableByteArray# RealWorld -> Int -> CSize -> Addr# -> CSize -> MessageFlags 'Send -> IO CSsize + foreign import ccall safe "sys/uio.h writev" c_safe_writev :: Fd -> MutableByteArray# RealWorld -> CInt -> IO CSsize @@ -314,7 +342,19 @@ foreign import ccall unsafe "sys/socket.h recvfrom_offset" c_unsafe_mutable_byte_array_recvfrom :: Fd -> MutableByteArray# RealWorld -> CInt -> CSize -> MessageFlags 'Receive -> MutableByteArray# RealWorld -> MutableByteArray# RealWorld -> IO CSsize foreign import ccall unsafe "sys/socket.h recvfrom_offset" - c_unsafe_mutable_byte_array_ptr_recvfrom :: Fd -> MutableByteArray# RealWorld -> CInt -> CSize -> MessageFlags 'Receive -> Ptr Void -> Ptr CInt -> IO CSsize + c_unsafe_mutable_byte_array_ptr_recvfrom :: + Fd -> MutableByteArray# RealWorld -> CInt -> CSize + -> MessageFlags 'Receive -> Ptr Void -> Ptr CInt -> IO CSsize +foreign import ccall unsafe "sys/socket.h recvfrom_offset_inet" + c_unsafe_recvfrom_inet :: + Fd + -> MutableByteArray# RealWorld + -> Int + -> CSize + -> MessageFlags 'Receive + -> MutableByteArray# RealWorld + -> Int + -> IO CSsize foreign import ccall unsafe "sys/socket.h recvmsg" c_unsafe_addr_recvmsg :: Fd @@ -371,7 +411,7 @@ -> SocketAddress -- ^ Socket address, extensible tagged union -> IO (Either Errno ()) uninterruptibleBind fd (SocketAddress b@(ByteArray b#)) = - c_bind fd b# (intToCInt (PM.sizeofByteArray b)) >>= errorsFromInt + c_bind fd b# (intToCInt (PM.sizeofByteArray b)) >>= errorsFromInt_ -- | Mark the socket as a passive socket, that is, as a socket that -- will be used to accept incoming connection requests using @accept@. @@ -383,7 +423,7 @@ Fd -- ^ Socket -> CInt -- ^ Backlog -> IO (Either Errno ()) -uninterruptibleListen fd backlog = c_listen fd backlog >>= errorsFromInt +uninterruptibleListen fd backlog = c_listen fd backlog >>= errorsFromInt_ -- | Connect the socket to the specified socket address. -- The @@ -393,13 +433,13 @@ -> SocketAddress -- ^ Socket address, extensible tagged union -> IO (Either Errno ()) connect fd (SocketAddress sockAddr@(ByteArray sockAddr#)) = - case PM.isByteArrayPinned sockAddr of - True -> c_safe_connect fd sockAddr# (intToCInt (PM.sizeofByteArray sockAddr)) >>= errorsFromInt + case isByteArrayPinned sockAddr of + True -> c_safe_connect fd sockAddr# (intToCInt (PM.sizeofByteArray sockAddr)) >>= errorsFromInt_ False -> do let len = PM.sizeofByteArray sockAddr x@(MutableByteArray x#) <- PM.newPinnedByteArray len PM.copyByteArray x 0 sockAddr 0 len - c_safe_mutablebytearray_connect fd x# (intToCInt len) >>= errorsFromInt + c_safe_mutablebytearray_connect fd x# (intToCInt len) >>= errorsFromInt_ -- | Connect the socket to the specified socket address. -- The @@ -410,7 +450,7 @@ -> SocketAddress -- ^ Socket address, extensible tagged union -> IO (Either Errno ()) uninterruptibleConnect fd (SocketAddress sockAddr@(ByteArray sockAddr#)) = - c_unsafe_connect fd sockAddr# (intToCInt (PM.sizeofByteArray sockAddr)) >>= errorsFromInt + c_unsafe_connect fd sockAddr# (intToCInt (PM.sizeofByteArray sockAddr)) >>= errorsFromInt_ -- | Extract the first connection on the queue of pending connections. The -- @@ -455,7 +495,6 @@ x <- PM.newByteArray (cintToInt minSz) PM.copyMutableByteArray x 0 sockAddrBuf 0 (cintToInt minSz) sockAddr <- PM.unsafeFreezeByteArray x - -- sockAddr <- PM.unsafeFreezeByteArray =<< PM.resizeMutableByteArray sockAddrBuf (cintToInt sz) pure (Right (sz,SocketAddress sockAddr,r)) else fmap Left getErrno @@ -551,7 +590,7 @@ -> CInt -- ^ Option value -> IO (Either Errno ()) uninterruptibleSetSocketOptionInt sock level optName optValue = - c_unsafe_setsockopt_int sock level optName optValue >>= errorsFromInt + c_unsafe_setsockopt_int sock level optName optValue >>= errorsFromInt_ -- | Send data from a byte array over a network socket. Users -- may specify an offset and a length to send fewer bytes than are @@ -565,7 +604,7 @@ -> CSize -- ^ Length in bytes -> MessageFlags 'Send -- ^ Flags -> IO (Either Errno CSize) -- ^ Number of bytes pushed to send buffer -sendByteArray fd b@(ByteArray b#) off len flags = if PM.isByteArrayPinned b +sendByteArray fd b@(ByteArray b#) off len flags = if isByteArrayPinned b then errorsFromSize =<< c_safe_bytearray_send fd b# off len flags else do x@(MutableByteArray x#) <- PM.newPinnedByteArray (csizeToInt len) @@ -573,7 +612,9 @@ errorsFromSize =<< c_safe_mutablebytearray_no_offset_send fd x# len flags -- | Write data from multiple byte arrays to the file/socket associated --- with the file descriptor. This does not support slicing. +-- with the file descriptor. This does not support slicing. The +-- +-- of @writev@ includes more details. writeVector :: Fd -- ^ Socket -> UnliftedArray ByteArray -- ^ Source byte arrays @@ -583,47 +624,97 @@ PM.newPinnedByteArray (cintToInt PST.sizeofIOVector * PM.sizeofUnliftedArray buffers) - downward (PM.sizeofUnliftedArray buffers) $ \i -> do - buffer <- pinByteArray (PM.indexUnliftedArray buffers i) - - let - targetAddr :: Addr - targetAddr = - PM.mutableByteArrayContents iovecs `PM.plusAddr` - (i * cintToInt PST.sizeofIOVector) - - PST.pokeIOVectorBase targetAddr (PM.byteArrayContents buffer) - PST.pokeIOVectorLength targetAddr (intToCSize (PM.sizeofByteArray buffer)) - + -- We construct a list of the new buffers for the sole purpose + -- of ensuring that we can touch the list later to keep all + -- the new buffers live. + newBufs <- foldDownward (PM.sizeofUnliftedArray buffers) UNil $ \newBufs i -> do + let !buf = PM.indexUnliftedArray buffers i + pinByteArray buf >>= \case + Nothing -> do + let buffer = buf + let targetAddr :: Addr + targetAddr = ptrToAddr (PM.mutableByteArrayContents iovecs) `plusAddr` + (i * cintToInt PST.sizeofIOVector) + PST.pokeIOVectorBase targetAddr (ptrToAddr (PM.byteArrayContents buffer)) + PST.pokeIOVectorLength targetAddr (intToCSize (PM.sizeofByteArray buffer)) + pure newBufs + Just buffer -> do + let targetAddr :: Addr + targetAddr = ptrToAddr (PM.mutableByteArrayContents iovecs) `plusAddr` + (i * cintToInt PST.sizeofIOVector) + PST.pokeIOVectorBase targetAddr (ptrToAddr (PM.byteArrayContents buffer)) + PST.pokeIOVectorLength targetAddr (intToCSize (PM.sizeofByteArray buffer)) + pure (UCons (unByteArray buffer) newBufs) r <- errorsFromSize =<< c_safe_writev fd iovecs# (intToCInt (PM.sizeofUnliftedArray buffers)) - -- Touching the unlifted array here is crucial to ensuring that + -- Touching both the unlifted array and the list of new buffers + -- here is crucial to ensuring that -- the buffers do not get GCed before c_safe_writev. Just touching - -- the unlifted array should keep all of its children alive too. + -- them should keep all of their children alive too. touchUnliftedArray buffers + touchLifted newBufs pure r --- Internal function. Upper bound is exclusive. Hits every --- int in the range [0,hi) from highest to lowest. -downward :: Int -> (Int -> IO a) -> IO () -downward !hi f = go (hi - 1) where - go !ix = if ix >= 0 - then f ix *> go (ix - 1) - else pure () +data UList (a :: TYPE 'UnliftedRep) where + UNil :: UList a + UCons :: a -> UList a -> UList a +-- Internal function. Fold with strict accumulator. Upper bound is exclusive. +-- Hits every int in the range [0,hi) from highest to lowest. +foldDownward :: forall a. Int -> a -> (a -> Int -> IO a) -> IO a +{-# INLINE foldDownward #-} +foldDownward !hi !a0 f = go (hi - 1) a0 where + go :: Int -> a -> IO a + go !ix !a = if ix >= 0 + then f a ix >>= go (ix - 1) + else pure a + -- | Copy and pin a byte array if, it's not already pinned. -pinByteArray :: ByteArray -> IO ByteArray +pinByteArray :: ByteArray -> IO (Maybe ByteArray) +{-# INLINE pinByteArray #-} pinByteArray byteArray = - if PM.isByteArrayPinned byteArray + if isByteArrayPinned byteArray then - pure byteArray + pure Nothing else do pinnedByteArray <- PM.newPinnedByteArray len PM.copyByteArray pinnedByteArray 0 byteArray 0 len - PM.unsafeFreezeByteArray pinnedByteArray + r <- PM.unsafeFreezeByteArray pinnedByteArray + pure (Just r) where len = PM.sizeofByteArray byteArray +-- | Send two payloads (one from unmanaged memory and one from +-- managed memory) over a network socket. +uninterruptibleSendMessageA :: + Fd -- ^ Socket + -> Addr -- ^ Source address (payload A) + -> CSize -- ^ Length in bytes (payload A) + -> MutableByteArrayOffset RealWorld -- ^ Source and offset (payload B) + -> CSize -- ^ Length in bytes (payload B) + -> MessageFlags 'Send -- ^ Flags + -> IO (Either Errno CSize) -- ^ Number of bytes pushed to send buffer +uninterruptibleSendMessageA fd (Addr addr) lenA + (MutableByteArrayOffset{array,offset}) lenB flags = + c_unsafe_sendmsg_a fd addr lenA (unMba array) offset lenB flags + >>= errorsFromSize + +-- | Send two payloads (one from managed memory and one from +-- unmanaged memory) over a network socket. +uninterruptibleSendMessageB :: + Fd -- ^ Socket + -> MutableByteArrayOffset RealWorld -- ^ Source and offset (payload B) + -> CSize -- ^ Length in bytes (payload B) + -> Addr -- ^ Source address (payload A) + -> CSize -- ^ Length in bytes (payload A) + -> MessageFlags 'Send -- ^ Flags + -> IO (Either Errno CSize) -- ^ Number of bytes pushed to send buffer +uninterruptibleSendMessageB fd + (MutableByteArrayOffset{array,offset}) lenB + (Addr addr) lenA flags = + c_unsafe_sendmsg_b fd (unMba array) offset lenB addr lenA flags + >>= errorsFromSize + -- | Send data from a mutable byte array over a network socket. Users -- may specify an offset and a length to send fewer bytes than are -- actually present in the array. Since this uses the safe @@ -636,7 +727,7 @@ -> CSize -- ^ Length in bytes -> MessageFlags 'Send -- ^ Flags -> IO (Either Errno CSize) -- ^ Number of bytes pushed to send buffer -sendMutableByteArray fd b@(MutableByteArray b#) off len flags = if PM.isMutableByteArrayPinned b +sendMutableByteArray fd b@(MutableByteArray b#) off len flags = if isMutableByteArrayPinned b then errorsFromSize =<< c_safe_mutablebytearray_send fd b# off len flags else do x@(MutableByteArray x#) <- PM.newPinnedByteArray (csizeToInt len) @@ -713,8 +804,23 @@ uninterruptibleSendToByteArray fd (ByteArray b) off len flags (SocketAddress a@(ByteArray a#)) = c_unsafe_bytearray_sendto fd b off len flags a# (intToCInt (PM.sizeofByteArray a)) >>= errorsFromSize +-- | Variant of 'uninterruptibleSendToByteArray' that requires +-- that @sockaddr_in@ be used as the socket address. This is used to +-- avoid allocating a buffer for the socket address when the caller +-- knows in advance that they are sending to an IPv4 address. +uninterruptibleSendToInternetByteArray :: + Fd -- ^ Socket + -> ByteArray -- ^ Source byte array + -> CInt -- ^ Offset into source array + -> CSize -- ^ Length in bytes + -> MessageFlags 'Send -- ^ Flags + -> SocketAddressInternet -- ^ Socket Address + -> IO (Either Errno CSize) -- ^ Number of bytes pushed to send buffer +uninterruptibleSendToInternetByteArray fd (ByteArray b) off len flags (SocketAddressInternet{port,address}) = + c_unsafe_bytearray_sendto_inet fd b off len flags port address >>= errorsFromSize + -- | Send data from a mutable byte array over an unconnected network socket. --- This uses the unsafe FFI; considerations pertaining to 'sendToUnsafe' +-- This uses the unsafe FFI; concerns pertaining to 'uninterruptibleSend' -- apply to this function as well. The offset and length arguments -- cause a slice of the mutable byte array to be sent rather than the entire -- byte array. @@ -729,6 +835,21 @@ uninterruptibleSendToMutableByteArray fd (MutableByteArray b) off len flags (SocketAddress a@(ByteArray a#)) = c_unsafe_mutable_bytearray_sendto fd b off len flags a# (intToCInt (PM.sizeofByteArray a)) >>= errorsFromSize +-- | Variant of 'uninterruptibleSendToMutableByteArray' that requires +-- that @sockaddr_in@ be used as the socket address. This is used to +-- avoid allocating a buffer for the socket address when the caller +-- knows in advance that they are sending to an IPv4 address. +uninterruptibleSendToInternetMutableByteArray :: + Fd -- ^ Socket + -> MutableByteArray RealWorld -- ^ Source byte array + -> CInt -- ^ Offset into source array + -> CSize -- ^ Length in bytes + -> MessageFlags 'Send -- ^ Flags + -> SocketAddressInternet -- ^ Socket Address + -> IO (Either Errno CSize) -- ^ Number of bytes pushed to send buffer +uninterruptibleSendToInternetMutableByteArray fd (MutableByteArray b) off len flags (SocketAddressInternet{port,address}) = + c_unsafe_mutable_bytearray_sendto_inet fd b off len flags port address >>= errorsFromSize + -- | Receive data into an address from a network socket. This wraps @recv@ using -- the safe FFI. When the returned size is zero, there are no -- additional bytes to receive and the peer has performed an orderly shutdown. @@ -751,7 +872,7 @@ -> IO (Either Errno ByteArray) receiveByteArray !fd !len !flags = do m <- PM.newPinnedByteArray (csizeToInt len) - let !(Addr addr) = PM.mutableByteArrayContents m + let !(Addr addr) = ptrToAddr (PM.mutableByteArrayContents m) r <- c_safe_addr_recv fd addr len flags if r /= (-1) then do @@ -776,6 +897,7 @@ -> CSize -- ^ Length in bytes -> MessageFlags 'Receive -- ^ Flags -> IO (Either Errno CSize) +{-# inline uninterruptibleReceive #-} uninterruptibleReceive !fd (Addr !addr) !len !flags = c_unsafe_addr_recv fd addr len flags >>= errorsFromSize @@ -790,6 +912,7 @@ -> CSize -- ^ Maximum bytes to receive -> MessageFlags 'Receive -- ^ Flags -> IO (Either Errno CSize) -- ^ Bytes received into array +{-# inline uninterruptibleReceiveMutableByteArray #-} uninterruptibleReceiveMutableByteArray !fd (MutableByteArray !b) !off !len !flags = c_unsafe_mutable_byte_array_recv fd b off len flags >>= errorsFromSize @@ -805,11 +928,15 @@ -> CInt -- ^ Maximum socket address size -> IO (Either Errno (CInt,SocketAddress,CSize)) -- ^ Remote host, bytes received into array, bytes needed for @addrlen@. -{-# INLINE uninterruptibleReceiveFromMutableByteArray #-} +{-# inline uninterruptibleReceiveFromMutableByteArray #-} -- GHC does not inline this unless we give it the pragma. We really -- want this to inline since inlining typically avoids the Left/Right -- data constructor allocation. uninterruptibleReceiveFromMutableByteArray !fd (MutableByteArray !b) !off !len !flags !maxSz = do + -- TODO: We currently allocate one buffer for the size and + -- one for the peer. We could improve this by allocating + -- a single buffer instead. We would need to add some + -- cleverness in the cbits directory. sockAddrBuf@(MutableByteArray sockAddrBuf#) <- PM.newByteArray (cintToInt maxSz) lenBuf@(MutableByteArray lenBuf#) <- PM.newByteArray (PM.sizeOf (undefined :: CInt)) PM.writeByteArray lenBuf 0 maxSz @@ -824,6 +951,20 @@ pure (Right (sz,SocketAddress sockAddr,cssizeToCSize r)) else fmap Left getErrno +uninterruptibleReceiveFromInternetMutableByteArray :: + Fd -- ^ Socket + -> MutableByteArrayOffset RealWorld -- ^ Destination byte array + -> CSize -- ^ Maximum bytes to receive + -> MessageFlags 'Receive -- ^ Flags + -> MutablePrimArrayOffset RealWorld SocketAddressInternet -- ^ Address + -> IO (Either Errno CSize) -- ^ Number of bytes received into array +{-# inline uninterruptibleReceiveFromInternetMutableByteArray #-} +uninterruptibleReceiveFromInternetMutableByteArray !fd + (MutableByteArrayOffset (MutableByteArray b) off) !len !flags + (MutablePrimArrayOffset (MutablePrimArray sockAddrBuf) addrOff) = + c_unsafe_recvfrom_inet fd b off len flags sockAddrBuf addrOff + >>= errorsFromSize + -- | Receive data into an address from a network socket. This uses the unsafe -- FFI. This does not return the socket address of the remote host that -- sent the packet received. @@ -834,11 +975,10 @@ -> CSize -- ^ Maximum bytes to receive -> MessageFlags 'Receive -- ^ Flags -> IO (Either Errno CSize) -- ^ Number of bytes received into array +{-# inline uninterruptibleReceiveFromMutableByteArray_ #-} uninterruptibleReceiveFromMutableByteArray_ !fd (MutableByteArray !b) !off !len !flags = c_unsafe_mutable_byte_array_ptr_recvfrom fd b off len flags nullPtr nullPtr >>= errorsFromSize - - -- | Receive a message, scattering the input. This does not provide -- the socket address or the control messages. All of the chunks -- must have the same maximum size. All resulting byte arrays have @@ -852,11 +992,11 @@ uninterruptibleReceiveMessageA !s !chunkSize !chunkCount !flags = do bufs <- PM.unsafeNewUnliftedArray (csizeToInt chunkCount) iovecsBuf <- PM.newPinnedByteArray (csizeToInt chunkCount * cintToInt PST.sizeofIOVector) - let iovecsAddr = PM.mutableByteArrayContents iovecsBuf + let iovecsAddr = ptrToAddr (PM.mutableByteArrayContents iovecsBuf) initializeIOVectors bufs iovecsAddr chunkSize chunkCount msgHdrBuf <- PM.newPinnedByteArray (cintToInt PST.sizeofMessageHeader) - let !msgHdrAddr@(Addr msgHdrAddr#) = PM.mutableByteArrayContents msgHdrBuf - pokeMessageHeader msgHdrAddr PM.nullAddr 0 iovecsAddr chunkCount PM.nullAddr 0 flags + let !msgHdrAddr@(Addr msgHdrAddr#) = ptrToAddr (PM.mutableByteArrayContents msgHdrBuf) + pokeMessageHeader msgHdrAddr nullAddr 0 iovecsAddr chunkCount nullAddr 0 flags r <- c_unsafe_addr_recvmsg s msgHdrAddr# flags if r > (-1) then do @@ -872,6 +1012,9 @@ touchMutableByteArray msgHdrBuf fmap Left getErrno +ptrToAddr :: Ptr Word8 -> Addr +ptrToAddr (Exts.Ptr a) = Addr a + -- | Receive a message, scattering the input. This provides the socket -- address but does not include control messages. All of the chunks -- must have the same maximum size. All resulting byte arrays have @@ -887,11 +1030,13 @@ sockAddrBuf <- PM.newPinnedByteArray (cintToInt maxSockAddrSz) bufs <- PM.unsafeNewUnliftedArray (csizeToInt chunkCount) iovecsBuf <- PM.newPinnedByteArray (csizeToInt chunkCount * cintToInt PST.sizeofIOVector) - let iovecsAddr = PM.mutableByteArrayContents iovecsBuf + let iovecsAddr = ptrToAddr (PM.mutableByteArrayContents iovecsBuf) initializeIOVectors bufs iovecsAddr chunkSize chunkCount msgHdrBuf <- PM.newPinnedByteArray (cintToInt PST.sizeofMessageHeader) - let !msgHdrAddr@(Addr msgHdrAddr#) = PM.mutableByteArrayContents msgHdrBuf - pokeMessageHeader msgHdrAddr (PM.mutableByteArrayContents sockAddrBuf) maxSockAddrSz iovecsAddr chunkCount PM.nullAddr 0 flags + let !msgHdrAddr@(Addr msgHdrAddr#) = ptrToAddr (PM.mutableByteArrayContents msgHdrBuf) + pokeMessageHeader msgHdrAddr + (ptrToAddr (PM.mutableByteArrayContents sockAddrBuf)) maxSockAddrSz iovecsAddr + chunkCount nullAddr 0 flags r <- c_unsafe_addr_recvmsg s msgHdrAddr# flags if r > (-1) then do @@ -919,7 +1064,7 @@ close :: Fd -- ^ Socket -> IO (Either Errno ()) -close fd = c_safe_close fd >>= errorsFromInt +close fd = c_safe_close fd >>= errorsFromInt_ -- | Close a socket. This uses the unsafe FFI. According to the -- , @@ -931,7 +1076,7 @@ uninterruptibleClose :: Fd -- ^ Socket -> IO (Either Errno ()) -uninterruptibleClose fd = c_unsafe_close fd >>= errorsFromInt +uninterruptibleClose fd = c_unsafe_close fd >>= errorsFromInt_ -- | Close a socket with the unsafe FFI. Do not check for errors. It is only -- appropriate to use this when a socket is being closed to handle an @@ -952,7 +1097,7 @@ -> ShutdownType -> IO (Either Errno ()) uninterruptibleShutdown fd typ = - c_unsafe_shutdown fd typ >>= errorsFromInt + c_unsafe_shutdown fd typ >>= errorsFromInt_ errorsFromSize :: CSsize -> IO (Either Errno CSize) errorsFromSize r = if r > (-1) @@ -967,8 +1112,8 @@ -- Sometimes, functions that return an int use zero to indicate -- success and negative one to indicate failure without including -- additional information in the value. -errorsFromInt :: CInt -> IO (Either Errno ()) -errorsFromInt r = if r == 0 +errorsFromInt_ :: CInt -> IO (Either Errno ()) +errorsFromInt_ r = if r == 0 then pure (Right ()) else fmap Left getErrno @@ -1043,7 +1188,7 @@ let go !ix !iovecAddr = if ix < csizeToInt chunkCount then do initializeIOVector bufs iovecAddr chunkSize ix - go (ix + 1) (PM.plusAddr iovecAddr (cintToInt PST.sizeofIOVector)) + go (ix + 1) (plusAddr iovecAddr (cintToInt PST.sizeofIOVector)) else pure () in go 0 iovecsAddr @@ -1059,7 +1204,9 @@ initializeIOVector bufs iovecAddr chunkSize ix = do buf <- PM.newPinnedByteArray (csizeToInt chunkSize) PM.writeUnliftedArray bufs ix buf - PST.pokeIOVectorBase iovecAddr (PM.mutableByteArrayContents buf) + let !(Exts.Ptr bufAddr#) = PM.mutableByteArrayContents buf + bufAddr = Addr bufAddr# + PST.pokeIOVectorBase iovecAddr bufAddr PST.pokeIOVectorLength iovecAddr chunkSize -- This is intended to be called on an array of iovec after recvmsg @@ -1104,6 +1251,9 @@ else PM.unsafeFreezeUnliftedArray x go 0 +unByteArray :: ByteArray -> ByteArray# +unByteArray (ByteArray x) = x + touchMutableUnliftedArray :: MutableUnliftedArray RealWorld a -> IO () touchMutableUnliftedArray (MutableUnliftedArray x) = touchMutableUnliftedArray# x @@ -1122,6 +1272,9 @@ touchMutableByteArray# :: MutableByteArray# RealWorld -> IO () touchMutableByteArray# x = IO $ \s -> case touch# x s of s' -> (# s', () #) +touchLifted :: a -> IO () +touchLifted x = IO $ \s -> case touch# x s of s' -> (# s', () #) + {- $conversion These functions are used to convert IPv4 addresses and ports between network byte order and host byte order. They are essential when working with @@ -1146,3 +1299,16 @@ ending in @A@, @B@, etc. -} +isByteArrayPinned :: ByteArray -> Bool +{-# inline isByteArrayPinned #-} +isByteArrayPinned (ByteArray arr#) = + Exts.isTrue# (Exts.isByteArrayPinned# arr#) + +isMutableByteArrayPinned :: MutableByteArray s -> Bool +{-# inline isMutableByteArrayPinned #-} +isMutableByteArrayPinned (MutableByteArray marr#) = + Exts.isTrue# (Exts.isMutableByteArrayPinned# marr#) + +unMba :: MutableByteArray s -> MutableByteArray# s +{-# inline unMba #-} +unMba (MutableByteArray x) = x diff --git a/src/Posix/Socket/Types.hsc b/src/Posix/Socket/Types.hsc --- a/src/Posix/Socket/Types.hsc +++ b/src/Posix/Socket/Types.hsc @@ -4,6 +4,9 @@ {-# language GADTSyntax #-} {-# language GeneralizedNewtypeDeriving #-} {-# language KindSignatures #-} +{-# language MagicHash #-} +{-# language UnboxedTuples #-} +{-# language NamedFieldPuns #-} -- This is needed because hsc2hs does not currently handle ticked -- promoted data constructors correctly. @@ -11,6 +14,7 @@ #include #include +#include "custom.h" -- | All of the data constructors provided by this module are unsafe. -- Only use them if you really know what you are doing. @@ -97,13 +101,16 @@ import Prelude hiding (read) import Data.Bits (Bits,(.|.)) -import Data.Primitive (ByteArray,Addr(..)) +import Data.Primitive (ByteArray,Prim(..)) +import Data.Primitive.Addr (Addr(..)) import Data.Word (Word16,Word32,Word64) import Foreign.C.Types (CInt(..),CSize) import Foreign.Storable (peekByteOff,pokeByteOff) import GHC.Ptr (Ptr(..)) +import GHC.Exts (Int(I##),Int##,(*##),(+##)) import qualified Data.Kind +import qualified Data.Primitive as PM -- | A socket communications domain, sometimes referred to as a family. The spec -- mandates @AF_UNIX@, @AF_UNSPEC@, and @AF_INET@. @@ -167,9 +174,42 @@ data SocketAddressInternet = SocketAddressInternet { port :: !Word16 , address :: !Word32 - } + } deriving (Eq,Show) +-- | The index and read functions ignore @sin_family@. The write functions +-- will set @sin_family@ to @AF_INET@. +instance Prim SocketAddressInternet where + sizeOf## _ = unI #{size struct sockaddr_in} + alignment## _ = PM.alignment## (undefined :: Word) + indexByteArray## arr i = SocketAddressInternet + { port = #{indexByteArrayHash struct sockaddr_in, sin_port} arr i + , address = #{indexByteArrayHash struct sockaddr_in, sin_addr} arr i + } + indexOffAddr## arr i = SocketAddressInternet + { port = #{indexOffAddrHash struct sockaddr_in, sin_port} arr i + , address = #{indexOffAddrHash struct sockaddr_in, sin_addr} arr i + } + readByteArray## arr i s0 = + case #{readByteArrayHash struct sockaddr_in, sin_port} arr i s0 of + (## s1, port ##) -> case #{readByteArrayHash struct sockaddr_in, sin_addr} arr i s1 of + (## s2, address ##) -> (## s2, SocketAddressInternet{port,address} ##) + readOffAddr## arr i s0 = + case #{readOffAddrHash struct sockaddr_in, sin_port} arr i s0 of + (## s1, port ##) -> case #{readOffAddrHash struct sockaddr_in, sin_addr} arr i s1 of + (## s2, address ##) -> (## s2, SocketAddressInternet{port,address} ##) + writeByteArray## arr i SocketAddressInternet{port,address} s0 = + case #{writeByteArrayHash struct sockaddr_in, sin_family} arr i (#{const AF_INET} :: #{type sa_family_t}) s0 of + s1 -> case #{writeByteArrayHash struct sockaddr_in, sin_port} arr i port s1 of + s2 -> #{writeByteArrayHash struct sockaddr_in, sin_addr} arr i address s2 + writeOffAddr## arr i SocketAddressInternet{port,address} s0 = + case #{writeOffAddrHash struct sockaddr_in, sin_family} arr i (#{const AF_INET} :: #{type sa_family_t}) s0 of + s1 -> case #{writeOffAddrHash struct sockaddr_in, sin_port} arr i port s1 of + s2 -> #{writeOffAddrHash struct sockaddr_in, sin_addr} arr i address s2 + setByteArray## = PM.defaultSetByteArray## + setOffAddr## = PM.defaultSetOffAddr## + -- Revisit this. We really need a standard Word128 type somewhere. +-- Solution: use the wideword package. data SocketAddressInternet6 = SocketAddressInternet6 { port :: !Word16 , flowInfo :: !Word32 @@ -394,4 +434,7 @@ peekMessageHeaderFlags (Addr p) = do i <- #{peek struct msghdr, msg_flags} (Ptr p) pure (MessageFlags i) + +unI :: Int -> Int## +unI (I## i) = i diff --git a/src/Posix/Types.hsc b/src/Posix/Types.hsc new file mode 100644 --- /dev/null +++ b/src/Posix/Types.hsc @@ -0,0 +1,18 @@ +{-# language DataKinds #-} +{-# language DerivingStrategies #-} +{-# language DuplicateRecordFields #-} +{-# language GeneralizedNewtypeDeriving #-} + +module Posix.Types + ( CNfds(..) + ) where + +import Data.Word + +import Foreign.Storable (Storable) +import Data.Bits (FiniteBits,Bits) + +#include + +newtype CNfds = CNfds #{type nfds_t} + deriving newtype (Eq,Real,Integral,Enum,Num,Ord,Storable,FiniteBits,Bits) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -1,14 +1,19 @@ {-# language BangPatterns #-} +{-# language BinaryLiterals #-} +{-# language LambdaCase #-} {-# language NamedFieldPuns #-} {-# language ScopedTypeVariables #-} +{-# language TypeApplications #-} import Control.Concurrent (forkIO) import Control.Concurrent (threadWaitRead,threadWaitWrite) import Control.Monad (when) -import Data.Primitive (ByteArray) +import Data.Primitive (ByteArray,MutablePrimArray(..),MutableByteArray(..)) import Data.Word (Word32,Word8) import Foreign.C.Error (Errno,errnoToIOError) import Foreign.C.Types (CInt,CSize) +import GHC.Exts (RealWorld) +import Numeric (showIntAtBase) import Test.Tasty import Test.Tasty.HUnit @@ -17,6 +22,7 @@ import qualified Data.Primitive.MVar as PM import qualified Posix.Socket as S import qualified Linux.Socket as L +import qualified Linux.Epoll as Epoll main :: IO () main = defaultMain tests @@ -38,7 +44,11 @@ [ testGroup "sockets" [ testCase "A" testLinuxSocketsA , testCase "B" testLinuxSocketsB + , testCase "C" testLinuxSocketsC ] + , testGroup "epoll" + [ testCase "A" testLinuxEpollA + ] ] ] @@ -163,25 +173,151 @@ testLinuxSocketsB = do a <- demand =<< S.uninterruptibleSocket S.internet S.datagram S.defaultProtocol demand =<< S.uninterruptibleBind a (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = 0, S.address = localhost})) - (expectedSzA,expectedSockAddrA) <- demand =<< S.uninterruptibleGetSocketName a 128 - when (expectedSzA /= S.sizeofSocketAddressInternet) (fail "testLinixSocketsB: bad socket address size for socket A") + (expectedSzA,expectedSockAddrA) <- demand + =<< S.uninterruptibleGetSocketName a 128 + when (expectedSzA /= S.sizeofSocketAddressInternet) + (fail "testLinixSocketsB: bad socket address size for socket A") portA <- case S.decodeSocketAddressInternet expectedSockAddrA of Nothing -> fail "testLinixSocketsB: not a sockaddr_in" Just (S.SocketAddressInternet {S.port}) -> pure port b <- demand =<< S.uninterruptibleSocket S.internet S.datagram S.defaultProtocol - demand =<< S.uninterruptibleBind b (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = 0, S.address = localhost})) + demand =<< S.uninterruptibleBind b + (S.encodeSocketAddressInternet $ S.SocketAddressInternet + { S.port = 0 + , S.address = localhost + } + ) threadWaitWrite b - bytesSent1 <- demand =<< S.uninterruptibleSendToByteArray b sample 0 5 mempty (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = portA, S.address = localhost})) - when (bytesSent1 /= 5) (fail "testLinixSocketsB: bytesSent1 was wrong") + bytesSent1 <- demand =<< S.uninterruptibleSendToByteArray b sample 0 5 mempty + (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = portA, S.address = localhost})) + when (bytesSent1 /= 5) + (fail "testLinixSocketsB: bytesSent1 was wrong") threadWaitWrite b - bytesSent2 <- demand =<< S.uninterruptibleSendToByteArray b sample2 0 4 mempty (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = portA, S.address = localhost})) - when (bytesSent2 /= 4) (fail "testLinixSocketsB: bytesSent2 was wrong") + bytesSent2 <- demand =<< S.uninterruptibleSendToByteArray b sample2 0 4 mempty + (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = portA, S.address = localhost})) + when (bytesSent2 /= 4) + (fail "testLinixSocketsB: bytesSent2 was wrong") threadWaitRead a - actual <- demand =<< L.uninterruptibleReceiveMultipleMessageB a S.sizeofSocketAddressInternet 6 3 L.dontWait + actual <- demand + =<< L.uninterruptibleReceiveMultipleMessageB a S.sizeofSocketAddressInternet 6 3 L.dontWait (expectedSzB,S.SocketAddress sabytesB) <- demand =<< S.uninterruptibleGetSocketName b 128 - when (expectedSzB /= S.sizeofSocketAddressInternet) (fail "testLinixSocketsB: bad socket address size for socket B") + when (expectedSzB /= S.sizeofSocketAddressInternet) + (fail "testLinixSocketsB: bad socket address size for socket B") (0,sabytesB <> sabytesB,5,E.fromList [sample,sample2]) @=? actual +testLinuxSocketsC :: Assertion +testLinuxSocketsC = do + a <- demand =<< S.uninterruptibleSocket S.internet S.datagram S.defaultProtocol + demand =<< S.uninterruptibleBind a (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = 0, S.address = localhost})) + (expectedSzA,expectedSockAddrA) <- demand + =<< S.uninterruptibleGetSocketName a 128 + when (expectedSzA /= S.sizeofSocketAddressInternet) + (fail "testLinuxSocketsC: bad socket address size for socket A") + portA <- case S.decodeSocketAddressInternet expectedSockAddrA of + Nothing -> fail "testLinuxSocketsC: not a sockaddr_in" + Just (S.SocketAddressInternet {S.port}) -> pure port + b <- demand =<< S.uninterruptibleSocket S.internet S.datagram S.defaultProtocol + demand =<< S.uninterruptibleBind b + (S.encodeSocketAddressInternet $ S.SocketAddressInternet + { S.port = 0 + , S.address = localhost + } + ) + threadWaitWrite b + bytesSent1 <- demand =<< S.uninterruptibleSendToByteArray b sample 0 5 mempty + (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = portA, S.address = localhost})) + when (bytesSent1 /= 5) + (fail "testLinuxSocketsC: bytesSent1 was wrong") + threadWaitWrite b + bytesSent2 <- demand =<< S.uninterruptibleSendToByteArray b sample2 0 4 mempty + (S.encodeSocketAddressInternet (S.SocketAddressInternet {S.port = portA, S.address = localhost})) + when (bytesSent2 /= 4) + (fail "testLinuxSocketsC: bytesSent2 was wrong") + threadWaitRead a + lens <- PM.newPrimArray 2 + addrs <- PM.newPrimArray 2 + payloadsMut <- PM.unsafeNewUnliftedArray 2 + PM.newByteArray 6 >>= PM.writeUnliftedArray payloadsMut 0 + PM.newByteArray 6 >>= PM.writeUnliftedArray payloadsMut 1 + msgCount <- demand =<< L.uninterruptibleReceiveMultipleMessageC a lens addrs payloadsMut 2 L.dontWait + when (msgCount /= 2) (fail "wrong number of messages") + addrsFrozen <- PM.unsafeFreezePrimArray addrs + payloads <- PM.unsafeFreezeUnliftedArray payloadsMut + len0 <- PM.readPrimArray lens 0 + len1 <- PM.readPrimArray lens 1 + buf0 <- PM.unsafeFreezeByteArray + =<< PM.resizeMutableByteArray (PM.indexUnliftedArray payloads 0) (fromIntegral @CInt @Int len0) + buf1 <- PM.unsafeFreezeByteArray + =<< PM.resizeMutableByteArray (PM.indexUnliftedArray payloads 1) (fromIntegral @CInt @Int len1) + (expectedSzB,S.SocketAddress sabytesB) <- demand =<< S.uninterruptibleGetSocketName b 128 + when (expectedSzB /= S.sizeofSocketAddressInternet) + (fail "testLinuxSocketsC: bad socket address size for socket B") + let primSockAddr = case sabytesB of PM.ByteArray x -> PM.PrimArray x + (primSockAddr <> primSockAddr,E.fromList [sample,sample2]) @=? (addrsFrozen,[buf0,buf1]) + +-- This test opens two datagram sockets and send a message from each +-- one to the other. Then it checks that epoll's event-triggered +-- interface correctly notifies the user about the read-readiness +-- that has happened. +testLinuxEpollA :: Assertion +testLinuxEpollA = do + (a,b) <- demand =<< S.uninterruptibleSocketPair S.unix S.datagram S.defaultProtocol + epfd <- demand =<< Epoll.uninterruptibleCreate 1 + reg <- PM.newPrimArray 1 + PM.writePrimArray reg 0 $ Epoll.Event + { Epoll.events = Epoll.input <> Epoll.edgeTriggered + , Epoll.payload = a + } + demand =<< Epoll.uninterruptibleControlMutablePrimArray epfd Epoll.add a reg + PM.writePrimArray reg 0 $ Epoll.Event + { Epoll.events = Epoll.input <> Epoll.edgeTriggered + , Epoll.payload = b + } + demand =<< Epoll.uninterruptibleControlMutablePrimArray epfd Epoll.add b reg + threadWaitWrite b + bytesSentB <- demand =<< S.uninterruptibleSendByteArray b sample 0 5 mempty + when (bytesSentB /= 5) (fail "testLinuxEpollA: bytesSentB was wrong") + threadWaitWrite a + bytesSentA <- demand =<< S.uninterruptibleSendByteArray a sample 0 5 mempty + when (bytesSentA /= 5) (fail "testLinuxEpollA: bytesSentA was wrong") + evs <- PM.newPrimArray 3 + loadGarbage evs + evCount <- demand =<< Epoll.waitMutablePrimArray epfd evs 3 (-1) + when (evCount /= 2) (fail ("testLinuxEpollA: evCount was " ++ show evCount)) + r <- case () of + _ -> do + Epoll.Event{Epoll.events,Epoll.payload} <- PM.readPrimArray evs 0 + when (payload /= a && payload /= b) (fail ("testLinuxEpollA: payload x was " ++ show payload)) + let Epoll.Events e = events + when (not (Epoll.containsAnyEvents events Epoll.input)) $ do + fail ("testLinuxEpollA: events x bitmask " ++ showIntAtBase 2 binChar e " missing EPOLLIN") + pure payload + Epoll.Event{Epoll.events,Epoll.payload} <- PM.readPrimArray evs 1 + when (payload == r) (fail ("testLinuxEpollA: same payload " ++ show payload ++ " for both events")) + when (payload /= a && payload /= b) (fail ("testLinuxEpollA: payload y was " ++ show payload)) + let Epoll.Events e = events + when (not (Epoll.containsAnyEvents events Epoll.input)) $ do + fail ("testLinuxEpollA: events y bitmask " ++ showIntAtBase 2 binChar e " missing EPOLLIN") + pure () + +binChar :: Int -> Char +binChar = \case + 0 -> '0' + 1 -> '1' + _ -> 'x' + +loadGarbage :: MutablePrimArray RealWorld a -> IO () +loadGarbage (MutablePrimArray x) = do + let arr = MutableByteArray x + go :: Int -> IO () + go !ix = if ix > (-1) + then do + PM.writeByteArray arr ix ((0b01010101 :: Word8) + fromIntegral ix) + go (ix - 1) + else pure () + n <- PM.getSizeofMutableByteArray arr + go (n - 1) + sample :: ByteArray sample = E.fromList [1,2,3,4,5] @@ -196,4 +332,3 @@ localhost :: Word32 localhost = S.hostToNetworkLong 2130706433 -