diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -13,7 +13,6 @@ ## TODOs * Support logging text, image value. * Create project -* Register callback for the abort action from web console. * Support system channels * Get git information diff --git a/examples/Main.hs b/examples/Main.hs --- a/examples/Main.hs +++ b/examples/Main.hs @@ -16,6 +16,6 @@ main = do withNept "jiasen/sandbox" $ \_ experiment -> do - forM_ [1..10::Int] $ \i -> do + forM_ [1..200::Int] $ \i -> do nlog experiment "counter" (fromIntegral (i * i) :: Double) threadDelay 1000000 diff --git a/lib/Neptune/AbortHandler.hs b/lib/Neptune/AbortHandler.hs new file mode 100644 --- /dev/null +++ b/lib/Neptune/AbortHandler.hs @@ -0,0 +1,75 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE QuasiQuotes #-} +module Neptune.AbortHandler where + +import Control.Lens (_Right, (^?!)) +import Control.Retry (constantDelay, recoverAll) +import Data.Aeson (FromJSON (..), Value (..), + eitherDecode', (.:)) +import Data.Aeson.Types (prependFailure, typeMismatch) +import Data.Text.Encoding (encodeUtf8) +import qualified Network.WebSockets as WS +import RIO +import RIO.Text as T (pack, unpack) +import Text.URI (RText, RTextLabel (..), mkURI) +import Text.URI.Lens +import qualified Wuss + +import Neptune.Backend.Core (configHost) +import Neptune.Backend.Model (ExperimentId (..)) +import Neptune.OAuth (oas_access_token) +import Neptune.Session (ClientToken (..), Experiment, + NeptuneSession (..), exp_experiment_id) + +abortListener :: NeptuneSession -> Experiment -> ThreadId -> IO () +abortListener sess exp main_thread = do + base_url <- mkURI (sess & _neptune_client_token & _ct_api_url) + + let host = base_url ^?! uriAuthority . _Right . authHost . unRText + path = notif_path <> exp_id <> "/operations" :: String + + oauth_ref = sess & _neptune_oauth2 + + run_listener = do + oauth_current <- readMVar oauth_ref + let oauth_token = oauth_current ^. oas_access_token + Wuss.runSecureClientWith + (T.unpack host) port path + WS.defaultConnectionOptions + [("Authorization", "Bearer " <> encodeUtf8 oauth_token)] + (listener main_thread) + + recoverAll (constantDelay 500) $ \_ -> run_listener + + where + port = 443 + notif_path = "/api/notifications/v1/experiments/" + exp_id = T.unpack $ unExperimentId $ exp ^. exp_experiment_id + + +data Message = MessageAbort { _msg_experiment_id :: Text } + +instance FromJSON Message where + parseJSON (Object v) = do + typ <- v .: "messageType" + case typ of + "Abort" -> do + body <- v .: "messageBody" + MessageAbort <$> body .: "experimentId" + _ -> fail $ "Unsupported message type" <> (T.unpack typ) + parseJSON invalid = + prependFailure "parsing Message failed, " + (typeMismatch "Object" invalid) + +data AbortException = AbortException + deriving Show + +instance Exception AbortException + +listener :: ThreadId -> WS.ClientApp () +listener main_thread conn = forever $ do + msg <- WS.receiveData conn + case eitherDecode' msg of + Right (MessageAbort _) -> throwTo main_thread AbortException + Left msg -> traceM (T.pack msg) >> return () + diff --git a/lib/Neptune/Channel.hs b/lib/Neptune/Channel.hs --- a/lib/Neptune/Channel.hs +++ b/lib/Neptune/Channel.hs @@ -12,7 +12,8 @@ import Control.Concurrent.Event as E (isSet, set) import Control.Lens import Data.Typeable (Proxy (..), cast) -import RIO hiding (Lens', (^.)) +import RIO hiding (Lens', (.~), (^.), (^..), + (^?)) import qualified RIO.HashMap as M import qualified Neptune.Backend.API as NBAPI diff --git a/lib/Neptune/Client.hs b/lib/Neptune/Client.hs --- a/lib/Neptune/Client.hs +++ b/lib/Neptune/Client.hs @@ -10,6 +10,8 @@ import Control.Concurrent (forkIO, killThread) import Control.Concurrent.Event as E (new, set, waitTimeout) +import Control.Exception (AsyncException (UserInterrupt), + asyncExceptionFromException, try) import Control.Lens ((<&>), (^.)) import qualified Data.Text.Lazy as TL import qualified Data.Text.Lazy.Encoding as TL @@ -18,11 +20,14 @@ import Data.UUID.V4 as UUID (nextRandom) import qualified Network.HTTP.Client as NH import qualified Network.HTTP.Client.TLS as NH -import RIO hiding (Lens', (^.)) +import RIO hiding (Lens', try, (^.)) import qualified RIO.HashMap as M import qualified RIO.Text as T import System.Envy (decodeEnv) +import System.Posix.Signals (Handler (Catch), installHandler, + keyboardSignal) +import Neptune.AbortHandler (AbortException (..), abortListener) import qualified Neptune.Backend.API as NBAPI import Neptune.Backend.Client import Neptune.Backend.Core @@ -65,16 +70,22 @@ params "command" -- legacy (fromMaybe "Untitled" name) - tags){ experimentCreationParamsDescription = description } + tags){ experimentCreationParamsDescription = description + , experimentCreationParamsAbortable = Just True } + let exp_id = ExperimentId (exp ^. experimentIdL) chan <- newTChanIO user_channels <- newTVarIO M.empty stop_flag <- E.new transmitter_flag <- E.new - let exp = Experiment exp_id chan user_channels stop_flag transmitter_flag undefined + let exp = Experiment exp_id chan user_channels stop_flag transmitter_flag undefined undefined transmitter_thread <- forkIO $ transmitter session exp - return exp {_exp_transmitter = transmitter_thread} + parent_thread <- myThreadId + abort_handler <- forkIO $ abortListener session exp parent_thread + + return exp {_exp_transmitter = transmitter_thread, _exp_abort_handler = abort_handler} + where _mkParameter (ExperimentParamS name value) = do _id <- UUID.toText <$> UUID.nextRandom @@ -103,13 +114,25 @@ ses <- initNept project_qualified_name exp <- createExperiment ses Nothing Nothing [] [] [] + -- install an signal handler for CTRL-C, ensuring that an async- + -- exception UserInterrupt is sent to the main thread + main_thread <- myThreadId + let interrupted = throwTo main_thread UserInterrupt + installHandler keyboardSignal (Catch interrupted) Nothing + result <- try (act ses exp) case result of Left (e :: SomeException) -> do - teardownNept ses exp ExperimentState'Failed (T.pack $ displayException e) + let end_state = case fromException e of + Just AbortException -> Nothing + Nothing -> Just $ + case asyncExceptionFromException e of + Just UserInterrupt -> (ExperimentState'Failed, "User interrupted.") + Nothing -> (ExperimentState'Failed, T.pack $ displayException e) + teardownNept ses exp end_state throwM e Right a -> do - teardownNept ses exp ExperimentState'Succeeded "" + teardownNept ses exp (Just (ExperimentState'Succeeded, "")) return a -- | Initialize a neptune session @@ -120,7 +143,7 @@ ct@ClientToken{..} <- decodeEnv >>= either throwString return mgr <- NH.newManager NH.tlsManagerSettings - config0 <- withStderrLogging =<< newConfig + config0 <- pure . withNoLogging =<< newConfig let api_endpoint = TL.encodeUtf8 (TL.fromStrict _ct_api_url) config = config0 { configHost = api_endpoint } @@ -154,10 +177,9 @@ -- | Teardown a neptune session teardownNept :: NeptuneSession -- ^ session -> Experiment -- ^ experiment - -> ExperimentState -- ^ completion state - -> Text -- ^ completion message + -> Maybe (ExperimentState, Text) -- ^ completion state & message -> IO () -teardownNept NeptuneSession{..} experiment state msg = do +teardownNept NeptuneSession{..} experiment state_msg = do E.set (experiment ^. exp_stop_flag) -- wait at most 5 seconds done <- E.waitTimeout (experiment ^. exp_transmitter_flag) 5000000 @@ -166,11 +188,11 @@ killThread $ experiment ^. exp_transmitter killThread $ _neptune_oauth2_refresh - _ <- _neptune_dispatch $ NBAPI.markExperimentCompleted - (ContentType MimeJSON) - (Accept MimeNoContent) - (mkCompletedExperimentParams state msg) - (experiment ^. exp_experiment_id) :: IO NoContent - - return () - + case state_msg of + Just (state, msg) -> + void (_neptune_dispatch $ NBAPI.markExperimentCompleted + (ContentType MimeJSON) + (Accept MimeNoContent) + (mkCompletedExperimentParams state msg) + (experiment ^. exp_experiment_id) :: IO NoContent) + Nothing -> return () diff --git a/lib/Neptune/OAuth.hs b/lib/Neptune/OAuth.hs --- a/lib/Neptune/OAuth.hs +++ b/lib/Neptune/OAuth.hs @@ -17,7 +17,7 @@ ReqBodyUrlEnc (..), defaultHttpConfig, jsonResponse, req, responseBody, runReq, useHttpsURI, (=:)) -import RIO hiding (Lens', (^.)) +import RIO hiding (Lens', (.~), (^.)) import qualified RIO.Text as T import Text.URI (mkURI) import qualified Web.JWT as JWT diff --git a/lib/Neptune/Session.hs b/lib/Neptune/Session.hs --- a/lib/Neptune/Session.hs +++ b/lib/Neptune/Session.hs @@ -77,6 +77,7 @@ , _exp_stop_flag :: E.Event -- ^ Event flag to indicate the end of the session , _exp_transmitter_flag :: E.Event -- ^ Event flag to indicate completion of transmission , _exp_transmitter :: ThreadId -- ^ Background thread for transmission + , _exp_abort_handler :: ThreadId } class (Typeable a, Show a) => NeptDataType a where diff --git a/neptune-backend.cabal b/neptune-backend.cabal --- a/neptune-backend.cabal +++ b/neptune-backend.cabal @@ -1,6 +1,6 @@ cabal-version: 2.2 name: neptune-backend -version: 0.1.2 +version: 0.2.1 synopsis: Neptune Client description: . Client library for calling the Neptune API based on http-client. @@ -63,6 +63,10 @@ , envy , uuid , concurrent-extra + , websockets + , wuss + , unix + , retry other-modules: Paths_neptune_backend autogen-modules: @@ -74,6 +78,7 @@ Neptune.OAuth Neptune.Session Neptune.Utils + Neptune.AbortHandler Neptune.Backend.API Neptune.Backend.API.ApiDefault Neptune.Backend.Client