{-|
Module      : Z.IO.IPC
Description : Named pipe\/Unix domain servers and clients
Copyright   : (c) Dong Han, 2018
License     : BSD
Maintainer  : winterland1989@gmail.com
Stability   : experimental
Portability : non-portable

This module provides an API for creating IPC servers and clients. IPC Support is implemented with named pipes on Windows, and UNIX domain sockets on other operating systems.

On UNIX, the local domain is also known as the UNIX domain. The path is a filesystem path name. It gets truncated to sizeof(sockaddr_un.sun_path) - 1, which varies on different operating system between 91 and 107 bytes. The typical values are 107 on Linux and 103 on macOS. The path is subject to the same naming conventions and permissions checks as would be done on file creation. It will be visible in the filesystem, and will persist until unlinked.

On Windows, the local domain is implemented using a named pipe. The path must refer to an entry in \\?\pipe\ or \\.\pipe\. Any characters are permitted, but the latter may do some processing of pipe names, such as resolving .. sequences. Despite appearances, the pipe name space is flat. Pipes will not persist, they are removed when the last reference to them is closed.

-}

module Z.IO.Network.IPC (
  -- * IPC Client
    IPCClientConfig(..)
  , UVStream
  , defaultIPCClientConfig
  , initIPCClient
  -- * IPC Server
  , IPCServerConfig(..)
  , defaultIPCServerConfig
  , startIPCServer
  -- * For test
  , helloWorld
  , echo
  -- * Internal helper
  , initIPCStream
  ) where

import           Control.Concurrent.MVar
import           Control.Monad
import           Control.Monad.IO.Class
import           Data.Primitive.PrimArray
import           Foreign.Ptr
import           GHC.Generics
import           Z.Data.CBytes
import           Z.Data.Text.Print   (Print)
import           Z.Data.JSON         (JSON)
import           Z.IO.Exception
import           Z.IO.Resource
import           Z.IO.UV.FFI
import           Z.IO.UV.Manager
import           Z.IO.UV.UVStream
import           Data.Coerce

--------------------------------------------------------------------------------

-- | A IPC client configuration
--
data IPCClientConfig = IPCClientConfig
    { IPCClientConfig -> Maybe CBytes
ipcClientName :: Maybe CBytes -- ^ bind to a local file path (Unix) or name (Windows),
                                    -- won't bind if set to 'Nothing'.
    , IPCClientConfig -> CBytes
ipcTargetName :: CBytes       -- ^ target path (Unix) or a name (Windows).
    } deriving (IPCClientConfig -> IPCClientConfig -> Bool
(IPCClientConfig -> IPCClientConfig -> Bool)
-> (IPCClientConfig -> IPCClientConfig -> Bool)
-> Eq IPCClientConfig
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
/= :: IPCClientConfig -> IPCClientConfig -> Bool
$c/= :: IPCClientConfig -> IPCClientConfig -> Bool
== :: IPCClientConfig -> IPCClientConfig -> Bool
$c== :: IPCClientConfig -> IPCClientConfig -> Bool
Eq, Eq IPCClientConfig
Eq IPCClientConfig
-> (IPCClientConfig -> IPCClientConfig -> Ordering)
-> (IPCClientConfig -> IPCClientConfig -> Bool)
-> (IPCClientConfig -> IPCClientConfig -> Bool)
-> (IPCClientConfig -> IPCClientConfig -> Bool)
-> (IPCClientConfig -> IPCClientConfig -> Bool)
-> (IPCClientConfig -> IPCClientConfig -> IPCClientConfig)
-> (IPCClientConfig -> IPCClientConfig -> IPCClientConfig)
-> Ord IPCClientConfig
IPCClientConfig -> IPCClientConfig -> Bool
IPCClientConfig -> IPCClientConfig -> Ordering
IPCClientConfig -> IPCClientConfig -> IPCClientConfig
forall a.
Eq a
-> (a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
min :: IPCClientConfig -> IPCClientConfig -> IPCClientConfig
$cmin :: IPCClientConfig -> IPCClientConfig -> IPCClientConfig
max :: IPCClientConfig -> IPCClientConfig -> IPCClientConfig
$cmax :: IPCClientConfig -> IPCClientConfig -> IPCClientConfig
>= :: IPCClientConfig -> IPCClientConfig -> Bool
$c>= :: IPCClientConfig -> IPCClientConfig -> Bool
> :: IPCClientConfig -> IPCClientConfig -> Bool
$c> :: IPCClientConfig -> IPCClientConfig -> Bool
<= :: IPCClientConfig -> IPCClientConfig -> Bool
$c<= :: IPCClientConfig -> IPCClientConfig -> Bool
< :: IPCClientConfig -> IPCClientConfig -> Bool
$c< :: IPCClientConfig -> IPCClientConfig -> Bool
compare :: IPCClientConfig -> IPCClientConfig -> Ordering
$ccompare :: IPCClientConfig -> IPCClientConfig -> Ordering
$cp1Ord :: Eq IPCClientConfig
Ord, Int -> IPCClientConfig -> ShowS
[IPCClientConfig] -> ShowS
IPCClientConfig -> String
(Int -> IPCClientConfig -> ShowS)
-> (IPCClientConfig -> String)
-> ([IPCClientConfig] -> ShowS)
-> Show IPCClientConfig
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
showList :: [IPCClientConfig] -> ShowS
$cshowList :: [IPCClientConfig] -> ShowS
show :: IPCClientConfig -> String
$cshow :: IPCClientConfig -> String
showsPrec :: Int -> IPCClientConfig -> ShowS
$cshowsPrec :: Int -> IPCClientConfig -> ShowS
Show, ReadPrec [IPCClientConfig]
ReadPrec IPCClientConfig
Int -> ReadS IPCClientConfig
ReadS [IPCClientConfig]
(Int -> ReadS IPCClientConfig)
-> ReadS [IPCClientConfig]
-> ReadPrec IPCClientConfig
-> ReadPrec [IPCClientConfig]
-> Read IPCClientConfig
forall a.
(Int -> ReadS a)
-> ReadS [a] -> ReadPrec a -> ReadPrec [a] -> Read a
readListPrec :: ReadPrec [IPCClientConfig]
$creadListPrec :: ReadPrec [IPCClientConfig]
readPrec :: ReadPrec IPCClientConfig
$creadPrec :: ReadPrec IPCClientConfig
readList :: ReadS [IPCClientConfig]
$creadList :: ReadS [IPCClientConfig]
readsPrec :: Int -> ReadS IPCClientConfig
$creadsPrec :: Int -> ReadS IPCClientConfig
Read, (forall x. IPCClientConfig -> Rep IPCClientConfig x)
-> (forall x. Rep IPCClientConfig x -> IPCClientConfig)
-> Generic IPCClientConfig
forall x. Rep IPCClientConfig x -> IPCClientConfig
forall x. IPCClientConfig -> Rep IPCClientConfig x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cto :: forall x. Rep IPCClientConfig x -> IPCClientConfig
$cfrom :: forall x. IPCClientConfig -> Rep IPCClientConfig x
Generic)
      deriving anyclass (Int -> IPCClientConfig -> Builder ()
(Int -> IPCClientConfig -> Builder ()) -> Print IPCClientConfig
forall a. (Int -> a -> Builder ()) -> Print a
toUTF8BuilderP :: Int -> IPCClientConfig -> Builder ()
$ctoUTF8BuilderP :: Int -> IPCClientConfig -> Builder ()
Print, Value -> Converter IPCClientConfig
IPCClientConfig -> Value
IPCClientConfig -> Builder ()
(Value -> Converter IPCClientConfig)
-> (IPCClientConfig -> Value)
-> (IPCClientConfig -> Builder ())
-> JSON IPCClientConfig
forall a.
(Value -> Converter a)
-> (a -> Value) -> (a -> Builder ()) -> JSON a
encodeJSON :: IPCClientConfig -> Builder ()
$cencodeJSON :: IPCClientConfig -> Builder ()
toValue :: IPCClientConfig -> Value
$ctoValue :: IPCClientConfig -> Value
fromValue :: Value -> Converter IPCClientConfig
$cfromValue :: Value -> Converter IPCClientConfig
JSON)

-- | Default config, connect to ".\/ipc".
--
defaultIPCClientConfig :: IPCClientConfig
defaultIPCClientConfig :: IPCClientConfig
defaultIPCClientConfig = Maybe CBytes -> CBytes -> IPCClientConfig
IPCClientConfig Maybe CBytes
forall a. Maybe a
Nothing CBytes
"./ipc"

-- | init a IPC client 'Resource', which open a new connect when used.
--
initIPCClient :: IPCClientConfig -> Resource UVStream
initIPCClient :: IPCClientConfig -> Resource UVStream
initIPCClient (IPCClientConfig Maybe CBytes
cname CBytes
tname) = do
    UVManager
uvm <- IO UVManager -> Resource UVManager
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO UVManager
getUVManager
    UVStream
client <- HasCallStack => UVManager -> Resource UVStream
UVManager -> Resource UVStream
initIPCStream UVManager
uvm
    let hdl :: Ptr UVHandle
hdl = UVStream -> Ptr UVHandle
uvsHandle UVStream
client
    IO () -> Resource ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> Resource ()) -> IO () -> Resource ()
forall a b. (a -> b) -> a -> b
$ do
        Maybe CBytes -> (CBytes -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe CBytes
cname ((CBytes -> IO ()) -> IO ()) -> (CBytes -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ CBytes
cname' ->
            CBytes -> (BA# Word8 -> IO ()) -> IO ()
forall a. CBytes -> (BA# Word8 -> IO a) -> IO a
withCBytesUnsafe CBytes
cname' ((BA# Word8 -> IO ()) -> IO ()) -> (BA# Word8 -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ BA# Word8
cname_p ->
                -- bind is safe without withUVManager
                IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (Ptr UVHandle -> BA# Word8 -> IO CInt
uv_pipe_bind Ptr UVHandle
hdl BA# Word8
cname_p)
        CBytes -> (BA# Word8 -> IO ()) -> IO ()
forall a. CBytes -> (BA# Word8 -> IO a) -> IO a
withCBytesUnsafe CBytes
tname ((BA# Word8 -> IO ()) -> IO ()) -> (BA# Word8 -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ BA# Word8
tname_p -> do
            IO Int -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int -> IO ())
-> ((Ptr UVLoop -> IO UVSlotUnsafe) -> IO Int)
-> (Ptr UVLoop -> IO UVSlotUnsafe)
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HasCallStack =>
UVManager -> (Ptr UVLoop -> IO UVSlotUnsafe) -> IO Int
UVManager -> (Ptr UVLoop -> IO UVSlotUnsafe) -> IO Int
withUVRequest UVManager
uvm ((Ptr UVLoop -> IO UVSlotUnsafe) -> IO ())
-> (Ptr UVLoop -> IO UVSlotUnsafe) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ Ptr UVLoop
_ -> Ptr UVHandle -> BA# Word8 -> IO UVSlotUnsafe
hs_uv_pipe_connect Ptr UVHandle
hdl BA# Word8
tname_p
    UVStream -> Resource UVStream
forall (m :: * -> *) a. Monad m => a -> m a
return UVStream
client

--------------------------------------------------------------------------------

-- | A IPC server configuration
--
data IPCServerConfig = IPCServerConfig
    { IPCServerConfig -> CBytes
ipcListenName       :: CBytes      -- ^ listening path (Unix) or a name (Windows).
    , IPCServerConfig -> Int
ipcListenBacklog    :: Int           -- ^ listening pipe's backlog size, should be large enough(>128)
    } deriving (IPCServerConfig -> IPCServerConfig -> Bool
(IPCServerConfig -> IPCServerConfig -> Bool)
-> (IPCServerConfig -> IPCServerConfig -> Bool)
-> Eq IPCServerConfig
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
/= :: IPCServerConfig -> IPCServerConfig -> Bool
$c/= :: IPCServerConfig -> IPCServerConfig -> Bool
== :: IPCServerConfig -> IPCServerConfig -> Bool
$c== :: IPCServerConfig -> IPCServerConfig -> Bool
Eq, Eq IPCServerConfig
Eq IPCServerConfig
-> (IPCServerConfig -> IPCServerConfig -> Ordering)
-> (IPCServerConfig -> IPCServerConfig -> Bool)
-> (IPCServerConfig -> IPCServerConfig -> Bool)
-> (IPCServerConfig -> IPCServerConfig -> Bool)
-> (IPCServerConfig -> IPCServerConfig -> Bool)
-> (IPCServerConfig -> IPCServerConfig -> IPCServerConfig)
-> (IPCServerConfig -> IPCServerConfig -> IPCServerConfig)
-> Ord IPCServerConfig
IPCServerConfig -> IPCServerConfig -> Bool
IPCServerConfig -> IPCServerConfig -> Ordering
IPCServerConfig -> IPCServerConfig -> IPCServerConfig
forall a.
Eq a
-> (a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
min :: IPCServerConfig -> IPCServerConfig -> IPCServerConfig
$cmin :: IPCServerConfig -> IPCServerConfig -> IPCServerConfig
max :: IPCServerConfig -> IPCServerConfig -> IPCServerConfig
$cmax :: IPCServerConfig -> IPCServerConfig -> IPCServerConfig
>= :: IPCServerConfig -> IPCServerConfig -> Bool
$c>= :: IPCServerConfig -> IPCServerConfig -> Bool
> :: IPCServerConfig -> IPCServerConfig -> Bool
$c> :: IPCServerConfig -> IPCServerConfig -> Bool
<= :: IPCServerConfig -> IPCServerConfig -> Bool
$c<= :: IPCServerConfig -> IPCServerConfig -> Bool
< :: IPCServerConfig -> IPCServerConfig -> Bool
$c< :: IPCServerConfig -> IPCServerConfig -> Bool
compare :: IPCServerConfig -> IPCServerConfig -> Ordering
$ccompare :: IPCServerConfig -> IPCServerConfig -> Ordering
$cp1Ord :: Eq IPCServerConfig
Ord, Int -> IPCServerConfig -> ShowS
[IPCServerConfig] -> ShowS
IPCServerConfig -> String
(Int -> IPCServerConfig -> ShowS)
-> (IPCServerConfig -> String)
-> ([IPCServerConfig] -> ShowS)
-> Show IPCServerConfig
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
showList :: [IPCServerConfig] -> ShowS
$cshowList :: [IPCServerConfig] -> ShowS
show :: IPCServerConfig -> String
$cshow :: IPCServerConfig -> String
showsPrec :: Int -> IPCServerConfig -> ShowS
$cshowsPrec :: Int -> IPCServerConfig -> ShowS
Show, ReadPrec [IPCServerConfig]
ReadPrec IPCServerConfig
Int -> ReadS IPCServerConfig
ReadS [IPCServerConfig]
(Int -> ReadS IPCServerConfig)
-> ReadS [IPCServerConfig]
-> ReadPrec IPCServerConfig
-> ReadPrec [IPCServerConfig]
-> Read IPCServerConfig
forall a.
(Int -> ReadS a)
-> ReadS [a] -> ReadPrec a -> ReadPrec [a] -> Read a
readListPrec :: ReadPrec [IPCServerConfig]
$creadListPrec :: ReadPrec [IPCServerConfig]
readPrec :: ReadPrec IPCServerConfig
$creadPrec :: ReadPrec IPCServerConfig
readList :: ReadS [IPCServerConfig]
$creadList :: ReadS [IPCServerConfig]
readsPrec :: Int -> ReadS IPCServerConfig
$creadsPrec :: Int -> ReadS IPCServerConfig
Read, (forall x. IPCServerConfig -> Rep IPCServerConfig x)
-> (forall x. Rep IPCServerConfig x -> IPCServerConfig)
-> Generic IPCServerConfig
forall x. Rep IPCServerConfig x -> IPCServerConfig
forall x. IPCServerConfig -> Rep IPCServerConfig x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cto :: forall x. Rep IPCServerConfig x -> IPCServerConfig
$cfrom :: forall x. IPCServerConfig -> Rep IPCServerConfig x
Generic)
      deriving anyclass (Int -> IPCServerConfig -> Builder ()
(Int -> IPCServerConfig -> Builder ()) -> Print IPCServerConfig
forall a. (Int -> a -> Builder ()) -> Print a
toUTF8BuilderP :: Int -> IPCServerConfig -> Builder ()
$ctoUTF8BuilderP :: Int -> IPCServerConfig -> Builder ()
Print, Value -> Converter IPCServerConfig
IPCServerConfig -> Value
IPCServerConfig -> Builder ()
(Value -> Converter IPCServerConfig)
-> (IPCServerConfig -> Value)
-> (IPCServerConfig -> Builder ())
-> JSON IPCServerConfig
forall a.
(Value -> Converter a)
-> (a -> Value) -> (a -> Builder ()) -> JSON a
encodeJSON :: IPCServerConfig -> Builder ()
$cencodeJSON :: IPCServerConfig -> Builder ()
toValue :: IPCServerConfig -> Value
$ctoValue :: IPCServerConfig -> Value
fromValue :: Value -> Converter IPCServerConfig
$cfromValue :: Value -> Converter IPCServerConfig
JSON)

-- | A default hello world server on @.\/ipc@
--
-- Test it with @main = startIPCServer defaultIPCServerConfig@
--
defaultIPCServerConfig :: IPCServerConfig
defaultIPCServerConfig :: IPCServerConfig
defaultIPCServerConfig = CBytes -> Int -> IPCServerConfig
IPCServerConfig
    CBytes
"./ipc"
    Int
256

-- | Start a server
--
-- Fork new worker thread upon a new connection.
--
startIPCServer :: HasCallStack
               => IPCServerConfig
               -> (UVStream -> IO ())  -- ^ worker which get an accepted IPC stream,
                                        -- run in a seperated haskell thread,
                                       --  will be closed upon exception or worker finishes.
               -> IO ()
startIPCServer :: IPCServerConfig -> (UVStream -> IO ()) -> IO ()
startIPCServer IPCServerConfig{Int
CBytes
ipcListenBacklog :: Int
ipcListenName :: CBytes
ipcListenBacklog :: IPCServerConfig -> Int
ipcListenName :: IPCServerConfig -> CBytes
..} UVStream -> IO ()
ipcServerWorker = do
    let backLog :: Int
backLog = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
ipcListenBacklog Int
128
    UVManager
serverUVManager <- IO UVManager
getUVManager
    Resource UVStream -> (UVStream -> IO ()) -> IO ()
forall (m :: * -> *) a b.
(MonadMask m, MonadIO m, HasCallStack) =>
Resource a -> (a -> m b) -> m b
withResource (HasCallStack => UVManager -> Resource UVStream
UVManager -> Resource UVStream
initIPCStream UVManager
serverUVManager) ((UVStream -> IO ()) -> IO ()) -> (UVStream -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ (UVStream Ptr UVHandle
serverHandle Int
serverSlot UVManager
_ IORef Bool
_) -> do
        CBytes -> (BA# Word8 -> IO ()) -> IO ()
forall a. CBytes -> (BA# Word8 -> IO a) -> IO a
withCBytesUnsafe CBytes
ipcListenName ((BA# Word8 -> IO ()) -> IO ()) -> (BA# Word8 -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ BA# Word8
name_p -> do
            IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (Ptr UVHandle -> BA# Word8 -> IO CInt
uv_pipe_bind Ptr UVHandle
serverHandle BA# Word8
name_p)
        IO (Ptr UVHandle)
-> (Ptr UVHandle -> IO ()) -> (Ptr UVHandle -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
bracket
            (do Ptr UVHandle
check <- IO (Ptr UVHandle) -> IO (Ptr UVHandle)
forall a. HasCallStack => IO (Ptr a) -> IO (Ptr a)
throwOOMIfNull (IO (Ptr UVHandle) -> IO (Ptr UVHandle))
-> IO (Ptr UVHandle) -> IO (Ptr UVHandle)
forall a b. (a -> b) -> a -> b
$ IO (Ptr UVHandle)
hs_uv_check_alloc
                IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (Ptr UVHandle -> Ptr UVHandle -> IO CInt
hs_uv_check_init Ptr UVHandle
check Ptr UVHandle
serverHandle)
                Ptr UVHandle -> IO (Ptr UVHandle)
forall (m :: * -> *) a. Monad m => a -> m a
return Ptr UVHandle
check)
            Ptr UVHandle -> IO ()
hs_uv_check_close ((Ptr UVHandle -> IO ()) -> IO ())
-> (Ptr UVHandle -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
            \ Ptr UVHandle
check -> do

-- The buffer passing of accept is a litte complicated here, to get maximum performance,
-- we do batch accepting. i.e. recv multiple client inside libuv's event loop:
--
-- we poke uvmanager's buffer table as a Ptr Word8, with byte size (backLog*sizeof(FD))
-- inside libuv event loop, we cast the buffer back to int32_t* pointer.
-- each accept callback push a new socket fd to the buffer, and increase a counter(buffer_size_table).
-- backLog should be large enough(>128), so under windows we can't possibly filled it up within one
-- uv_run, under unix we hacked uv internal to provide a stop and resume function, when backLog is
-- reached, we will stop receiving.
--
-- once back to haskell side, we read all accepted sockets and fork worker threads.
-- if backLog is reached, we resume receiving from haskell side.
--
-- Step 1.
-- we allocate a buffer to hold accepted FDs, pass it just like a normal reading buffer.
-- then we can start listening.
                MutablePrimArray RealWorld CInt
acceptBuf <- Int -> IO (MutablePrimArray (PrimState IO) CInt)
forall (m :: * -> *) a.
(PrimMonad m, Prim a) =>
Int -> m (MutablePrimArray (PrimState m) a)
newPinnedPrimArray Int
backLog
                let acceptBufPtr :: Ptr Word8
acceptBufPtr = Ptr CInt -> Ptr Word8
coerce (MutablePrimArray RealWorld CInt -> Ptr CInt
forall s a. MutablePrimArray s a -> Ptr a
mutablePrimArrayContents MutablePrimArray RealWorld CInt
acceptBuf :: Ptr FD)

                UVManager -> IO () -> IO ()
forall a. HasCallStack => UVManager -> IO a -> IO a
withUVManager' UVManager
serverUVManager (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                    -- We use buffersize as accepted fd count(count backwards)
                    UVManager -> Int -> Ptr Word8 -> Int -> IO ()
pokeBufferTable UVManager
serverUVManager Int
serverSlot Ptr Word8
acceptBufPtr (Int
backLogInt -> Int -> Int
forall a. Num a => a -> a -> a
-Int
1)
                    IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (Ptr UVHandle -> CInt -> IO CInt
hs_uv_listen Ptr UVHandle
serverHandle (Int -> CInt
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
backLog))
-- Step 2.
-- we start a uv_check_t for given uv_stream_t, with predefined checking callback
-- see hs_accept_check_cb in hs_uv_stream.c
                    IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (IO CInt -> IO ()) -> IO CInt -> IO ()
forall a b. (a -> b) -> a -> b
$ Ptr UVHandle -> IO CInt
hs_uv_accept_check_start Ptr UVHandle
check

                MVar Int
m <- UVManager -> Int -> IO (MVar Int)
getBlockMVar UVManager
serverUVManager Int
serverSlot
                IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                    -- wait until accept some FDs
                    Int
_ <- MVar Int -> IO Int
forall a. MVar a -> IO a
takeMVar MVar Int
m
-- Step 3.
-- After uv loop finishes, if we got some FDs, copy the FD buffer, fetch accepted FDs and fork worker threads.

                    -- we lock uv manager here in case of next uv_run overwrite current accept buffer
                    PrimArray CInt
acceptBufCopy <- UVManager -> IO (PrimArray CInt) -> IO (PrimArray CInt)
forall a. HasCallStack => UVManager -> IO a -> IO a
withUVManager' UVManager
serverUVManager (IO (PrimArray CInt) -> IO (PrimArray CInt))
-> IO (PrimArray CInt) -> IO (PrimArray CInt)
forall a b. (a -> b) -> a -> b
$ do
                        Maybe Int
_ <- MVar Int -> IO (Maybe Int)
forall a. MVar a -> IO (Maybe a)
tryTakeMVar MVar Int
m
                        Int
acceptCountDown <- UVManager -> Int -> IO Int
peekBufferSizeTable UVManager
serverUVManager Int
serverSlot
                        UVManager -> Int -> Int -> IO ()
pokeBufferSizeTable UVManager
serverUVManager Int
serverSlot (Int
backLogInt -> Int -> Int
forall a. Num a => a -> a -> a
-Int
1)

                        -- if acceptCountDown count to -1, we should resume on haskell side
                        Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
acceptCountDown Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== -Int
1) (Ptr UVHandle -> IO ()
hs_uv_listen_resume Ptr UVHandle
serverHandle)

                        -- copy accepted FDs
                        let acceptCount :: Int
acceptCount = Int
backLog Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1 Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
acceptCountDown
                        MutablePrimArray RealWorld CInt
acceptBuf' <- Int -> IO (MutablePrimArray (PrimState IO) CInt)
forall (m :: * -> *) a.
(PrimMonad m, Prim a) =>
Int -> m (MutablePrimArray (PrimState m) a)
newPrimArray Int
acceptCount
                        MutablePrimArray (PrimState IO) CInt
-> Int
-> MutablePrimArray (PrimState IO) CInt
-> Int
-> Int
-> IO ()
forall (m :: * -> *) a.
(PrimMonad m, Prim a) =>
MutablePrimArray (PrimState m) a
-> Int -> MutablePrimArray (PrimState m) a -> Int -> Int -> m ()
copyMutablePrimArray MutablePrimArray RealWorld CInt
MutablePrimArray (PrimState IO) CInt
acceptBuf' Int
0 MutablePrimArray RealWorld CInt
MutablePrimArray (PrimState IO) CInt
acceptBuf (Int
acceptCountDownInt -> Int -> Int
forall a. Num a => a -> a -> a
+Int
1) Int
acceptCount
                        MutablePrimArray (PrimState IO) CInt -> IO (PrimArray CInt)
forall (m :: * -> *) a.
PrimMonad m =>
MutablePrimArray (PrimState m) a -> m (PrimArray a)
unsafeFreezePrimArray MutablePrimArray RealWorld CInt
MutablePrimArray (PrimState IO) CInt
acceptBuf'

                    -- fork worker thread
                    [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
0..PrimArray CInt -> Int
forall a. Prim a => PrimArray a -> Int
sizeofPrimArray PrimArray CInt
acceptBufCopyInt -> Int -> Int
forall a. Num a => a -> a -> a
-Int
1] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ Int
i -> do
                        let fd :: CInt
fd = PrimArray CInt -> Int -> CInt
forall a. Prim a => PrimArray a -> Int -> a
indexPrimArray PrimArray CInt
acceptBufCopy Int
i
                        if CInt
fd CInt -> CInt -> Bool
forall a. Ord a => a -> a -> Bool
< CInt
0
                        -- minus fd indicate a server error and we should close server
                        then IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (CInt -> IO CInt
forall (m :: * -> *) a. Monad m => a -> m a
return CInt
fd)
                        -- It's important to use the worker thread's mananger instead of server's one!
                        else IO ThreadId -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO ThreadId -> IO ()) -> (IO () -> IO ThreadId) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IO () -> IO ThreadId
forkBa (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                            UVManager
uvm <- IO UVManager
getUVManager
                            Resource UVStream -> (UVStream -> IO ()) -> IO ()
forall (m :: * -> *) a b.
(MonadMask m, MonadIO m, HasCallStack) =>
Resource a -> (a -> m b) -> m b
withResource (HasCallStack =>
(Ptr UVLoop -> Ptr UVHandle -> IO ())
-> UVManager -> Resource UVStream
(Ptr UVLoop -> Ptr UVHandle -> IO ())
-> UVManager -> Resource UVStream
initUVStream (\ Ptr UVLoop
loop Ptr UVHandle
hdl -> do
                                IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (Ptr UVLoop -> Ptr UVHandle -> CInt -> IO CInt
uv_pipe_init Ptr UVLoop
loop Ptr UVHandle
hdl CInt
0)
                                IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (Ptr UVHandle -> CInt -> IO CInt
uv_pipe_open Ptr UVHandle
hdl CInt
fd)) UVManager
uvm) ((UVStream -> IO ()) -> IO ()) -> (UVStream -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ UVStream
uvs -> do
                                UVStream -> IO ()
ipcServerWorker UVStream
uvs

--------------------------------------------------------------------------------

initIPCStream :: HasCallStack => UVManager -> Resource UVStream
initIPCStream :: UVManager -> Resource UVStream
initIPCStream = HasCallStack =>
(Ptr UVLoop -> Ptr UVHandle -> IO ())
-> UVManager -> Resource UVStream
(Ptr UVLoop -> Ptr UVHandle -> IO ())
-> UVManager -> Resource UVStream
initUVStream (\ Ptr UVLoop
loop Ptr UVHandle
hdl ->
    IO CInt -> IO ()
forall a. (HasCallStack, Integral a) => IO a -> IO ()
throwUVIfMinus_ (Ptr UVLoop -> Ptr UVHandle -> CInt -> IO CInt
uv_pipe_init Ptr UVLoop
loop Ptr UVHandle
hdl CInt
0))