diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ MIT License -Copyright (c) 2017-2019 Michael Truog +Copyright (c) 2017-2020 Michael Truog Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), diff --git a/cloudi.cabal b/cloudi.cabal --- a/cloudi.cabal +++ b/cloudi.cabal @@ -1,5 +1,5 @@ name: cloudi -version: 1.8.0 +version: 2.0.0 synopsis: Haskell CloudI API description: Haskell CloudI API (see https://cloudi.org for more details) homepage: https://github.com/CloudI/cloudi_api_haskell @@ -7,7 +7,7 @@ license-file: LICENSE author: Michael Truog maintainer: mjtruog at protonmail dot com -copyright: 2017-2019 Michael Truog +copyright: 2017-2020 Michael Truog category: Foreign stability: provisional build-type: Simple @@ -15,9 +15,10 @@ tested-with: GHC == 7.10.3 , GHC == 8.0.2 , GHC == 8.2.2 - , GHC == 8.4.3 + , GHC == 8.4.4 , GHC == 8.6.5 - , GHC == 8.8.1 + , GHC == 8.8.3 + , GHC == 8.10.1 extra-source-files: README.markdown changelog.md flag dev diff --git a/src/Foreign/CloudI.hs b/src/Foreign/CloudI.hs --- a/src/Foreign/CloudI.hs +++ b/src/Foreign/CloudI.hs @@ -5,7 +5,7 @@ MIT License - Copyright (c) 2017-2019 Michael Truog + Copyright (c) 2017-2020 Michael Truog Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), @@ -72,9 +72,10 @@ , threadCreate , threadsWait , infoKeyValueParse + , infoKeyValueNew ) where -import Prelude hiding (init,length) +import Prelude hiding (init,length,(<>)) import Data.Bits (shiftL,(.|.)) import Data.Maybe (fromMaybe) import Data.Typeable (Typeable) @@ -97,6 +98,7 @@ import qualified Foreign.Erlang as Erlang import qualified Foreign.CloudI.Instance as Instance import qualified System.IO as SysIO +import qualified System.IO.Error as SysIOErr import qualified System.IO.Unsafe as Unsafe import qualified System.Posix.Env as POSIX (getEnv) type Array = IArray.Array @@ -158,6 +160,7 @@ | ReturnAsync (Instance.T s) | ForwardSync (Instance.T s) | ForwardAsync (Instance.T s) + | Terminate (Instance.T s) deriving (Show, Typeable) instance Typeable s => Exception.Exception (Exception s) @@ -178,10 +181,14 @@ type Result a = Either String a +infixr 4 <> +(<>) :: Monoid.Monoid m => m -> m -> m +(<>) = Monoid.mappend + -- | creates an instance of the CloudI API -api :: Typeable s => Int -> s -> +api :: Typeable s => Int -> s -> Maybe Bool -> IO (Result (Instance.T s)) -api threadIndex state = do +api threadIndex state terminateReturnValueOpt = do SysIO.hSetEncoding SysIO.stdout SysIO.utf8 SysIO.hSetBuffering SysIO.stdout SysIO.LineBuffering SysIO.hSetEncoding SysIO.stderr SysIO.utf8 @@ -190,22 +197,25 @@ bufferSizeValue <- POSIX.getEnv "CLOUDI_API_INIT_BUFFER_SIZE" case (protocolValue, bufferSizeValue) of (Just protocol, Just bufferSizeStr) -> - let bufferSize = read bufferSizeStr :: Int + let terminateReturnValue = fromMaybe True terminateReturnValueOpt + terminateException = not terminateReturnValue + bufferSize = read bufferSizeStr :: Int fd = C.CInt $ fromIntegral (threadIndex + 3) useHeader = protocol /= "udp" - timeoutTerminate' = 1000 + timeoutTerminate' = 10 -- TIMEOUT_TERMINATE_MIN initTerms = Erlang.OtpErlangAtom (Char8.pack "init") in case Erlang.termToBinary initTerms (-1) of Left err -> return $ Left $ show err Right initBinary -> do - api0 <- Instance.make state protocol fd - useHeader bufferSize timeoutTerminate' + api0 <- Instance.make state terminateException + protocol fd useHeader bufferSize timeoutTerminate' send api0 initBinary result <- pollRequest api0 (-1) False case result of Left err -> + -- Terminate exception not used here return $ Left err Right (_, api1) -> return $ Right api1 @@ -240,7 +250,9 @@ -- | returns the number of subscriptions for a single service name pattern subscribeCount :: Typeable s => Instance.T s -> ByteString -> IO (Result (Int, Instance.T s)) -subscribeCount api0 pattern = +subscribeCount api0@Instance.T{ + Instance.terminateException = terminateException} + pattern = let subscribeCountTerms = Erlang.OtpErlangTuple [ Erlang.OtpErlangAtom (Char8.pack "subscribe_count") , Erlang.OtpErlangString pattern] @@ -253,7 +265,10 @@ result <- pollRequest api0 (-1) False case result of Left err -> - return $ Left err + if err == terminateError && terminateException then + Exception.throwIO $ Terminate api0 + else + return $ Left err Right (_, api1@Instance.T{Instance.subscribeCount = count}) -> return $ Right (count, api1) @@ -277,7 +292,8 @@ Maybe Int -> Maybe ByteString -> Maybe Int -> IO (Result (ByteString, Instance.T s)) sendAsync api0@Instance.T{ - Instance.timeoutAsync = timeoutAsync' + Instance.terminateException = terminateException + , Instance.timeoutAsync = timeoutAsync' , Instance.priorityDefault = priorityDefault} name request timeoutOpt requestInfoOpt priorityOpt = let timeout = fromMaybe timeoutAsync' timeoutOpt @@ -299,7 +315,10 @@ result <- pollRequest api0 (-1) False case result of Left err -> - return $ Left err + if err == terminateError && terminateException then + Exception.throwIO $ Terminate api0 + else + return $ Left err Right (_, api1@Instance.T{Instance.transId = transId}) -> return $ Right (transId, api1) @@ -308,7 +327,8 @@ Maybe Int -> Maybe ByteString -> Maybe Int -> IO (Result (ByteString, ByteString, ByteString, Instance.T s)) sendSync api0@Instance.T{ - Instance.timeoutSync = timeoutSync' + Instance.terminateException = terminateException + , Instance.timeoutSync = timeoutSync' , Instance.priorityDefault = priorityDefault} name request timeoutOpt requestInfoOpt priorityOpt = let timeout = fromMaybe timeoutSync' timeoutOpt @@ -330,7 +350,10 @@ result <- pollRequest api0 (-1) False case result of Left err -> - return $ Left err + if err == terminateError && terminateException then + Exception.throwIO $ Terminate api0 + else + return $ Left err Right (_, api1@Instance.T{ Instance.responseInfo = responseInfo , Instance.response = response @@ -343,7 +366,8 @@ Maybe Int -> Maybe ByteString -> Maybe Int -> IO (Result (Array Int ByteString, Instance.T s)) mcastAsync api0@Instance.T{ - Instance.timeoutAsync = timeoutAsync' + Instance.terminateException = terminateException + , Instance.timeoutAsync = timeoutAsync' , Instance.priorityDefault = priorityDefault} name request timeoutOpt requestInfoOpt priorityOpt = let timeout = fromMaybe timeoutAsync' timeoutOpt @@ -365,7 +389,10 @@ result <- pollRequest api0 (-1) False case result of Left err -> - return $ Left err + if err == terminateError && terminateException then + Exception.throwIO $ Terminate api0 + else + return $ Left err Right (_, api1@Instance.T{Instance.transIds = transIds}) -> return $ Right (transIds, api1) @@ -521,7 +548,9 @@ recvAsync :: Typeable s => Instance.T s -> Maybe Int -> Maybe ByteString -> Maybe Bool -> IO (Result (ByteString, ByteString, ByteString, Instance.T s)) -recvAsync api0@Instance.T{Instance.timeoutSync = timeoutSync'} +recvAsync api0@Instance.T{ + Instance.terminateException = terminateException + , Instance.timeoutSync = timeoutSync'} timeoutOpt transIdOpt consumeOpt = let timeout = fromMaybe timeoutSync' timeoutOpt transId = fromMaybe transIdNull transIdOpt @@ -540,7 +569,10 @@ result <- pollRequest api0 (-1) False case result of Left err -> - return $ Left err + if err == terminateError && terminateException then + Exception.throwIO $ Terminate api0 + else + return $ Left err Right (_, api1@Instance.T{ Instance.responseInfo = responseInfo , Instance.response = response @@ -630,16 +662,18 @@ requestInfo request timeout priority transId pid state api1 case callbackResultAsyncValue of + Left (Terminate api2) -> + return $ ReturnI (empty, empty, state, setTerminate api2) + Left (ReturnAsync api2) -> + return $ Finished api2 Left (ReturnSync api2) -> do printException "Synchronous Call Return Invalid" - return $ Finished api2 - Left (ReturnAsync api2) -> + return $ Finished (setTerminate api2) + Left (ForwardAsync api2) -> return $ Finished api2 Left (ForwardSync api2) -> do printException "Synchronous Call Forward Invalid" - return $ Finished api2 - Left (ForwardAsync api2) -> - return $ Finished api2 + return $ Finished (setTerminate api2) Right (Instance.ResponseInfo (v0, v1, v2, v3)) -> return $ ReturnI (v0, v1, v2, v3) Right (Instance.Response (v0, v1, v2)) -> @@ -659,16 +693,18 @@ requestInfo request timeout priority transId pid state api1 case callbackResultSyncValue of + Left (Terminate api2) -> + return $ ReturnI (empty, empty, state, setTerminate api2) Left (ReturnSync api2) -> return $ Finished api2 Left (ReturnAsync api2) -> do printException "Asynchronous Call Return Invalid" - return $ Finished api2 + return $ Finished (setTerminate api2) Left (ForwardSync api2) -> return $ Finished api2 Left (ForwardAsync api2) -> do printException "Asynchronous Call Forward Invalid" - return $ Finished api2 + return $ Finished (setTerminate api2) Right (Instance.ResponseInfo (v0, v1, v2, v3)) -> return $ ReturnI (v0, v1, v2, v3) Right (Instance.Response (v0, v1, v2)) -> @@ -685,7 +721,7 @@ callbackResultType <- case callbackResultValue of Left exception -> do printException $ show (exception :: Exception.SomeException) - return $ Finished api1 + return $ ReturnI (empty, empty, state, api1) Right callbackResult -> return $ callbackResult case requestType of @@ -720,11 +756,8 @@ cmd <- if cmd0 == 0 then Get.getWord32host else return cmd0 case () of _ | cmd == messageTerm -> - let api1 = api0{ - Instance.terminate = True - , Instance.timeout = Just False} in if external then - return ([], api1) + return ([], setTerminate api0) else fail terminateError | cmd == messageReinit -> do @@ -928,16 +961,15 @@ pollRequestLoop :: Typeable s => Instance.T s -> Int -> Bool -> Clock.NominalDiffTime -> IO (Result (Bool, Instance.T s)) -pollRequestLoop api0@Instance.T{ - Instance.socketHandle = socketHandle - , Instance.bufferRecvSize = bufferRecvSize} - timeout external pollTimer = do - inputAvailable <- if bufferRecvSize > 0 then - return True - else - SysIO.hWaitForInput socketHandle timeout - if not inputAvailable then +pollRequestLoop api0 timeout external pollTimer = do + inputAvailable <- pollWait api0 timeout + if inputAvailable == Just False then return $ Right (True, api0{Instance.timeout = Nothing}) + else if inputAvailable == Nothing then + let api1 = api0{ + Instance.terminate = True + , Instance.timeout = Nothing} in + return $ Right (False, api1) else do (dataIn, _, api1) <- recv api0 dataResult <- pollRequestData api1 external dataIn @@ -973,7 +1005,10 @@ Instance.initializationComplete = initializationComplete , Instance.terminate = terminate} timeout external = if terminate then - return $ Right (False, api0) + if external then + return $ Right (False, api0) + else + return $ Left terminateError else do pollTimer <- if timeout <= 0 then return 0 @@ -992,6 +1027,23 @@ else pollRequestLoopBegin api0 timeout external pollTimer +pollWait :: Typeable s => Instance.T s -> Int -> IO (Maybe Bool) +pollWait Instance.T{ + Instance.socketHandle = socketHandle + , Instance.bufferRecvSize = bufferRecvSize} + timeout = do + if bufferRecvSize > 0 then + return $ Just True + else + SysIOErr.catchIOError + (do + inputAvailable <- SysIO.hWaitForInput socketHandle timeout + return $ Just inputAvailable) + (\e -> if SysIOErr.isEOFError e then + return Nothing + else + SysIOErr.ioError e) + -- | blocks to process incoming CloudI service requests poll :: Typeable s => Instance.T s -> Int -> IO (Result (Bool, Instance.T s)) poll api0 timeout = @@ -1111,6 +1163,11 @@ else return (t1, fromIntegral $ timeoutValue - elapsed) +setTerminate :: Instance.T s -> Instance.T s +setTerminate api0 = + api0{ Instance.terminate = True + , Instance.timeout = Just False} + threadList :: Concurrent.MVar [Concurrent.MVar ()] threadList = Unsafe.unsafePerformIO (Concurrent.newMVar []) @@ -1147,8 +1204,8 @@ Concurrent.takeMVar done threadsWait -textKeyValueParse :: ByteString -> Map ByteString [ByteString] -textKeyValueParse text = +textPairsParse :: ByteString -> Map ByteString [ByteString] +textPairsParse text = let loop m [] = m loop m [v] = if v == ByteString.empty then @@ -1160,11 +1217,31 @@ Nothing -> loop (Map.insert k [v] m) l' Just v' -> - loop (Map.insert k (v:v') m) l' + loop (Map.insert k (v' ++ [v]) m) l' in loop Map.empty (Char8.split '\0' text) --- | parses "text_pairs" in service request info +textPairsNew :: Map ByteString [ByteString] -> Maybe Bool -> ByteString +textPairsNew pairs responseOpt = + let response = fromMaybe True responseOpt in + if response && Map.size pairs == 0 then + Char8.pack "\0" + else + let pair builder _ [] = + builder + pair builder k (v:l') = + pair (builder <> + Builder.byteString k <> Builder.char8 '\0' <> + Builder.byteString v <> Builder.char8 '\0') k l' + in + LazyByteString.toStrict $ Builder.toLazyByteString $ + Map.foldlWithKey pair Monoid.mempty pairs + +-- | decode service request info key/value data infoKeyValueParse :: ByteString -> Map ByteString [ByteString] -infoKeyValueParse = textKeyValueParse +infoKeyValueParse = textPairsParse + +-- | encode service response info key/value data +infoKeyValueNew :: Map ByteString [ByteString] -> Maybe Bool -> ByteString +infoKeyValueNew = textPairsNew diff --git a/src/Foreign/CloudI/Instance.hs b/src/Foreign/CloudI/Instance.hs --- a/src/Foreign/CloudI/Instance.hs +++ b/src/Foreign/CloudI/Instance.hs @@ -5,7 +5,7 @@ MIT License - Copyright (c) 2017-2019 Michael Truog + Copyright (c) 2017-2020 Michael Truog Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), @@ -99,6 +99,7 @@ -- | an instance of the CloudI API data T s = T { state :: !s + , terminateException :: !Bool , socketHandle :: !Handle , useHeader :: !Bool , initializationComplete :: !Bool @@ -144,11 +145,13 @@ socket <- makeSocket protocol fd Socket.socketToHandle socket SysIO.ReadWriteMode -make :: s -> String -> C.CInt -> Bool -> Int -> Int -> IO (T s) -make state' protocol fd useHeader' bufferSize' timeoutTerminate' = do +make :: s -> Bool -> String -> C.CInt -> Bool -> Int -> Int -> IO (T s) +make state' terminateException' + protocol fd useHeader' bufferSize' timeoutTerminate' = do socketHandle' <- makeSocketHandle protocol fd return $ T { state = state' + , terminateException = terminateException' , socketHandle = socketHandle' , useHeader = useHeader' , initializationComplete = False