diff --git a/acid-state.cabal b/acid-state.cabal --- a/acid-state.cabal +++ b/acid-state.cabal @@ -1,71 +1,50 @@ --- acid-state.cabal auto-generated by cabal init. For additional --- options, see --- http://www.haskell.org/cabal/release/cabal-latest/doc/users-guide/authors.html#pkg-descr. --- The name of the package. Name: acid-state - --- The package version. See the Haskell package versioning policy --- (http://www.haskell.org/haskellwiki/Package_versioning_policy) for --- standards guiding when and how versions should be incremented. -Version: 0.8.3 - --- A short (one-line) description of the package. +Version: 0.10.0 Synopsis: Add ACID guarantees to any serializable Haskell data structure. - --- A longer description of the package. Description: Use regular Haskell data structures as your database and get stronger ACID guarantees than most RDBMS offer. - --- URL for the project homepage or repository. Homepage: http://acid-state.seize.it/ - --- The license under which the package is released. License: PublicDomain - --- The package author(s). Author: David Himmelstrup - --- An email address to which users can send suggestions, bug reports, --- and patches. Maintainer: Lemmih - --- A copyright notice. --- Copyright: - +-- Copyright: Category: Database - Build-type: Simple - --- Extra files to be distributed with the package, such as examples or --- a README. -Extra-source-files: examples/*.hs, examples/errors/*.hs, src-win32/*.hs, src-unix/*.hs - --- Constraint on the version of Cabal needed to build this package. Cabal-version: >=1.6 +Extra-source-files: + examples/*.hs + examples/errors/*.hs + src-win32/*.hs + src-unix/*.hs Source-repository head type: darcs location: http://hub.darcs.net/Lemmih/acid-state - Library - -- Modules exported by the library. Exposed-Modules: Data.Acid, Data.Acid.Local, Data.Acid.Memory, Data.Acid.Memory.Pure, Data.Acid.Remote, Data.Acid.Advanced - -- Modules not exported by this package. Other-modules: Data.Acid.Log, Data.Acid.Archive, Data.Acid.CRC, Paths_acid_state, Data.Acid.TemplateHaskell, Data.Acid.Common, FileIO, Data.Acid.Abstract, Data.Acid.Core - - -- Packages needed in order to build this package. - -- We need hGetSome from bytestring, added in 0.9.1.8 - Build-depends: base >= 4 && < 5, cereal >= 0.3.2.0, safecopy >= 0.6, - bytestring >= 0.9.1.8, stm, extensible-exceptions, - filepath, directory, mtl, array, containers, template-haskell, network + Build-depends: array, + base >= 4 && < 5, + bytestring >= 0.9.1.8, + cereal >= 0.3.2.0, + containers, + extensible-exceptions, + safecopy >= 0.6, + stm, + directory, + filepath, + mtl, + network, + template-haskell + if os(windows) Build-depends: Win32 else @@ -78,9 +57,4 @@ else Hs-Source-Dirs: src-unix/ - GHC-Options: -fwarn-unused-imports -fwarn-unused-binds - - -- Extra tools (e.g. alex, hsc2hs, ...) needed to build the source. - -- Build-tools: - diff --git a/examples/RemoteClient.hs b/examples/RemoteClient.hs --- a/examples/RemoteClient.hs +++ b/examples/RemoteClient.hs @@ -1,48 +1,79 @@ {-# LANGUAGE DeriveDataTypeable, TypeFamilies, TemplateHaskell #-} module Main (main) where -import Data.Acid -import Data.Acid.Advanced -import Data.Acid.Remote +import Control.Monad ( replicateM_ ) +import Data.Acid ( AcidState, closeAcidState, createCheckpoint, query, update ) +import Data.Acid.Advanced ( scheduleUpdate ) +import Data.Acid.Remote ( openRemoteState, sharedSecretPerform ) +import Data.ByteString.Char8 ( pack ) +import Network ( PortID(..) ) +import RemoteCommon ( StressState(..), ClearState(..), PokeState(..), QueryState(..) ) +import System.Environment ( getArgs ) +import System.IO ( hFlush, stdout ) -import Control.Monad.State -import Control.Monad.Reader -import System.Environment -import System.IO -import Data.SafeCopy -import Data.Typeable -import Network +------------------------------------------------------ +-- printHelp -import RemoteCommon +printHelp :: IO () +printHelp + = do putStrLn $ "Commands:" + putStrLn $ " query Prints out the current state." + putStrLn $ " poke Spawn 100k transactions." + putStrLn $ " checkpoint Create a new checkpoint." + putStrLn $ " clear Clear the state and create a new checkpoint." + putStrLn $ " quit Exit with out creating a checkpoint." ------------------------------------------------------ --- This is how AcidState is used: - -open :: IO (AcidState StressState) -open = openRemoteState "localhost" (PortNumber 8080) +-- interactive command loop -main :: IO () -main = do args <- getArgs - case args of - ["checkpoint"] - -> do acid <- open - createCheckpoint acid - ["query"] - -> do acid <- open - n <- query acid QueryState +commandLoop :: AcidState StressState -> IO () +commandLoop acid + = do printHelp + go + where + go = do + putStr "> " + hFlush stdout + cmd <- getLine + case cmd of + "checkpoint" + -> do createCheckpoint acid + go + "query" + -> do n <- query acid QueryState putStrLn $ "State value: " ++ show n - ["poke"] - -> do acid <- open - putStr "Issuing 100k transactions... " + go + "poke" + -> do putStr "Issuing 100k transactions... " hFlush stdout replicateM_ (100000-1) (scheduleUpdate acid PokeState) update acid PokeState putStrLn "Done" - ["clear"] - -> do acid <- open - update acid ClearState + go + "clear" + -> do update acid ClearState createCheckpoint acid - _ -> do putStrLn $ "Commands:" - putStrLn $ " query Prints out the current state." - putStrLn $ " poke Spawn 100k transactions." - putStrLn $ " checkpoint Create a new checkpoint." + go + "quit" + -> do closeAcidState acid + return () + + _ + -> do printHelp + go + +------------------------------------------------------ +-- connect to remote server and start command-loop + +main :: IO () +main + = do args <- getArgs + case args of + [] -> + do acid <- openRemoteState (sharedSecretPerform $ (pack "12345")) "localhost" (PortNumber $ fromIntegral 8080) + commandLoop acid + + [hostname, port] -> + do acid <- openRemoteState (sharedSecretPerform $ (pack "12345")) hostname (PortNumber $ fromIntegral $ read port) + commandLoop acid + _ -> putStrLn "Usage: RemoteClientTLS [ ]" diff --git a/examples/RemoteServer.hs b/examples/RemoteServer.hs --- a/examples/RemoteServer.hs +++ b/examples/RemoteServer.hs @@ -1,21 +1,18 @@ {-# LANGUAGE DeriveDataTypeable, TypeFamilies, TemplateHaskell #-} module Main (main) where -import Data.Acid -import Data.Acid.Remote (acidServer) - -import Control.Exception (bracket) -import Data.Typeable - -import Network - -import RemoteCommon - --- open a server on port 8080 +import Control.Exception ( bracket ) +import Data.Acid ( closeAcidState, openLocalState ) +import Data.Acid.Remote ( acidServer, sharedSecretCheck ) +import Data.ByteString.Char8 ( pack ) +import Data.Set ( singleton ) +import Network ( PortID(PortNumber) ) +import RemoteCommon ( StressState(..) ) +-- | open a server on port 8080 main :: IO () main = bracket (openLocalState $ StressState 0) closeAcidState - (\s -> acidServer s (PortNumber 8080)) + (acidServer (sharedSecretCheck (singleton $ pack "12345")) (PortNumber 8080)) diff --git a/src/Data/Acid/Abstract.hs b/src/Data/Acid/Abstract.hs --- a/src/Data/Acid/Abstract.hs +++ b/src/Data/Acid/Abstract.hs @@ -38,12 +38,12 @@ (both those caused by hardware and software). -} data AcidState st - = AcidState { + = AcidState { _scheduleUpdate :: forall event. (UpdateEvent event, EventState event ~ st) => event -> IO (MVar (EventResult event)) , scheduleColdUpdate :: Tagged ByteString -> IO (MVar ByteString) , _query :: (QueryEvent event, EventState event ~ st) => event -> IO (EventResult event) , queryCold :: Tagged ByteString -> IO ByteString - , + , -- | Take a snapshot of the state and save it to disk. Creating checkpoints -- makes it faster to resume AcidStates and you're free to create them as -- often or seldom as fits your needs. Transactions can run concurrently @@ -51,7 +51,7 @@ -- -- This call will not return until the operation has succeeded. createCheckpoint :: IO () - , + , -- | Close an AcidState and associated resources. -- Any subsequent usage of the AcidState will throw an exception. closeAcidState :: IO () @@ -107,10 +107,10 @@ downcast :: Typeable1 sub => AcidState st -> sub st downcast AcidState{acidSubState = AnyState sub} - = r + = r where r = case gcast1 (Just sub) of Just (Just x) -> x - Nothing -> + _ -> error $ - "Data.Acid: Invalid subtype cast: " ++ show (typeOf1 sub) ++ " -> " ++ show (typeOf1 r) + "Data.Acid: Invalid subtype cast: " ++ show (typeOf1 sub) ++ " -> " ++ show (typeOf1 r) diff --git a/src/Data/Acid/Common.hs b/src/Data/Acid/Common.hs --- a/src/Data/Acid/Common.hs +++ b/src/Data/Acid/Common.hs @@ -15,15 +15,19 @@ import Control.Monad.State import Control.Monad.Reader +import Data.ByteString.Lazy ( ByteString ) import Data.SafeCopy -import Data.Serialize (runGet, runGetLazy) +import Data.Serialize ( Get, runGet, runGetLazy ) import Control.Applicative import qualified Data.ByteString as Strict -- Silly fix for bug in cereal-0.3.3.0's version of runGetLazy. -runGetLazyFix getter inp +runGetLazyFix :: Get a + -> ByteString + -> Either String a +runGetLazyFix getter inp = case runGet getter Strict.empty of - Left msg -> runGetLazy getter inp + Left _msg -> runGetLazy getter inp Right val -> Right val class (SafeCopy st) => IsAcidic st where diff --git a/src/Data/Acid/Remote.hs b/src/Data/Acid/Remote.hs --- a/src/Data/Acid/Remote.hs +++ b/src/Data/Acid/Remote.hs @@ -1,5 +1,4 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE CPP #-} +{-# LANGUAGE CPP, DeriveDataTypeable, RankNTypes, RecordWildCards, ScopedTypeVariables #-} ----------------------------------------------------------------------------- {- | Module : Data.Acid.Remote @@ -8,47 +7,256 @@ Maintainer : lemmih@gmail.com Portability : non-portable (uses GHC extensions) - Network backend. + This module provides the ability perform 'update' and 'query' calls +from a remote process. +On the server-side you: + + 1. open your 'AcidState' normally + + 2. then use 'acidServer' to share the state + +On the client-side you: + + 1. use 'openRemoteState' to connect to the remote state + + 2. use the returned 'AcidState' like any other 'AcidState' handle + +'openRemoteState' and 'acidServer' communicate over an unencrypted +socket. If you need an encrypted connection, see @acid-state-tls@. + +On Unix®-like systems you can use 'UnixSocket' to create a socket file for +local communication between the client and server. Access can be +controlled by setting the permissions of the parent directory +containing the socket file. + +It is also possible to perform some simple authentication using +'sharedSecretCheck' and 'sharedSecretPerform'. Keep in mind that +secrets will be sent in plain-text if you do not use +@acid-state-tls@. If you are using a 'UnixSocket' additional +authentication may not be required, so you can use +'skipAuthenticationCheck' and 'skipAuthenticationPerform'. + +Working with a remote 'AcidState' is nearly identical to working with +a local 'AcidState' with a few important differences. + +The connection to the remote 'AcidState' can be lost. The client will +automatically attempt to reconnect every second. Because 'query' +events do not affect the state, an aborted 'query' will be retried +automatically after the server is reconnected. + +If the connection was lost during an 'update' event, the event will +not be retried. Instead 'RemoteConnectionError' will be raised. This +is because it is impossible for the client to know if the aborted +update completed on the server-side or not. + +When using a local 'AcidState', an update event in one thread does not +block query events taking place in other threads. With a remote +connection, all queries and requests are channeled over a single +connection. As a result, updates and queries are performed in the +order they are executed and do block each other. In the rare case +where this is an issue, you could create one remote connection per +thread. + +When working with local state, a query or update which returns the +whole state is not usually a problem due to memory sharing. The +update/query event basically just needs to return a pointer to the +data already in memory. But, when working remotely, the entire result +will be serialized and sent to the remote client. Hence, it is good +practice to create queries and updates that will only return the +required data. + +This module is designed to be extenible. You can easily add your own +authentication methods by creating a suitable pair of functions and +passing them to 'acidServer' and 'openRemoteState'. + +It is also possible to create alternative communication layers using +'CommChannel', 'process', and 'processRemoteState'. + -} module Data.Acid.Remote - ( acidServer + ( + -- * Server/Client + acidServer , openRemoteState + -- * Authentication + , skipAuthenticationCheck + , skipAuthenticationPerform + , sharedSecretCheck + , sharedSecretPerform + -- * Exception type + , AcidRemoteException(..) + -- * Low-Level functions needed to implement additional communication channels + , CommChannel(..) + , process + , processRemoteState ) where - +import Control.Concurrent.STM ( atomically ) +import Control.Concurrent.STM.TMVar ( newEmptyTMVar, readTMVar, takeTMVar, tryTakeTMVar, putTMVar ) +import Control.Concurrent.STM.TQueue +import Control.Exception ( AsyncException(ThreadKilled) + , Exception(fromException), IOException, Handler(..) + , SomeException, catch, catches, throw ) +import Control.Exception ( throwIO, finally ) +import Control.Monad ( forever, liftM, join, when ) +import Control.Concurrent ( ThreadId, forkIO, threadDelay, killThread, myThreadId ) +import Control.Concurrent.MVar ( MVar, newEmptyMVar, putMVar, takeMVar ) +import Control.Concurrent.Chan ( newChan, readChan, writeChan ) import Data.Acid.Abstract import Data.Acid.Core import Data.Acid.Common - -import Network -import qualified Data.ByteString as Strict -import qualified Data.ByteString.Lazy as Lazy -import Control.Exception ( throwIO, ErrorCall(..), finally ) -import Control.Monad ( forever, liftM, join ) -import Control.Concurrent +import qualified Data.ByteString as Strict +import Data.ByteString.Char8 ( pack ) +import qualified Data.ByteString.Lazy as Lazy import Data.IORef ( newIORef, readIORef, writeIORef ) import Data.Serialize import Data.SafeCopy ( SafeCopy, safeGet, safePut ) -import System.Directory ( removeFile ) -import System.IO ( Handle, hFlush, hClose ) -import qualified Data.Sequence as Seq +import Data.Set ( Set, member ) import Data.Typeable ( Typeable ) +import GHC.IO.Exception ( IOErrorType(..) ) +import Network ( HostName, PortID(..), connectTo, listenOn, withSocketsDo ) +import Network.Socket ( Socket, accept, sClose ) +import Network.Socket.ByteString ( recv, sendAll ) +import System.Directory ( removeFile ) +import System.IO ( Handle, hPrint, hFlush, hClose, stderr ) +import System.IO.Error ( ioeGetErrorType, isFullError, isDoesNotExistError ) -{- | Accept connections on @port@ and serve requests using the given 'AcidState'. +debugStrLn :: String -> IO () +debugStrLn s = + do -- putStrLn s -- uncomment to enable debugging + return () + +-- | 'CommChannel' is a record containing the IO functions we need for communication between the server and client. +-- +-- We abstract this out of the core processing function so that we can easily add support for SSL/TLS and Unit testing. +data CommChannel = CommChannel + { ccPut :: Strict.ByteString -> IO () + , ccGetSome :: Int -> IO (Strict.ByteString) + , ccClose :: IO () + } + +data AcidRemoteException + = RemoteConnectionError + | AcidStateClosed + | SerializeError String + | AuthenticationError String + deriving (Eq, Show, Typeable) +instance Exception AcidRemoteException + + +-- | create a 'CommChannel' from a 'Handle'. The 'Handle' should be +-- some two-way communication channel, such as a socket +-- connection. Passing in a 'Handle' to a normal is file is unlikely +-- to do anything useful. +handleToCommChannel :: Handle -> CommChannel +handleToCommChannel handle = + CommChannel { ccPut = \bs -> Strict.hPut handle bs >> hFlush handle + , ccGetSome = Strict.hGetSome handle + , ccClose = hClose handle + } + +{- | create a 'CommChannel' from a 'Socket'. The 'Socket' should be + an accepted socket, not a listen socket. +-} +socketToCommChannel :: Socket -> CommChannel +socketToCommChannel socket = + CommChannel { ccPut = sendAll socket + , ccGetSome = recv socket + , ccClose = sClose socket + } + +{- | skip server-side authentication checking entirely. -} +skipAuthenticationCheck :: CommChannel -> IO Bool +skipAuthenticationCheck _ = return True + +{- | skip client-side authentication entirely. -} +skipAuthenticationPerform :: CommChannel -> IO Bool +skipAuthenticationPerform _ = return True + +{- | check that the client knows a shared secret. + +The function takes a 'Set' of shared secrets. If a client knows any +of them, it is considered to be trusted. + +The shared secret is any 'ByteString' of your choice. + +If you give each client a different shared secret then you can +revoke access individually. + +see also: 'sharedSecretPerform' +-} +sharedSecretCheck :: Set Strict.ByteString -- ^ set of shared secrets + -> (CommChannel -> IO Bool) +sharedSecretCheck secrets cc = + do bs <- ccGetSome cc 1024 + if member bs secrets + then do ccPut cc (pack "OK") + return True + else do ccPut cc (pack "FAIL") + return False + +-- | attempt to authenticate with the server using a shared secret. +sharedSecretPerform :: Strict.ByteString -- ^ shared secret + -> (CommChannel -> IO ()) +sharedSecretPerform pw cc = + do ccPut cc pw + r <- ccGetSome cc 1024 + if r == (pack "OK") + then return () + else throwIO (AuthenticationError "shared secret authentication failed.") + +{- | Accept connections on @port@ and handle requests using the given 'AcidState'. This call doesn't return. + + On Unix®-like systems you can use 'UnixSocket' to communicate + using a socket file. To control access, you can set the permissions of + the parent directory which contains the socket file. + + see also: 'openRemoteState' and 'sharedSecretCheck'. -} -acidServer :: SafeCopy st => AcidState st -> PortID -> IO () -acidServer acidState port - = do socket <- listenOn port - forever (do (handle, _host, _port) <- accept socket - forkIO (process acidState handle)) - `finally` do sClose socket - case port of +acidServer :: SafeCopy st => + (CommChannel -> IO Bool) -- ^ check authentication, see 'sharedSecretPerform' + -> PortID -- ^ Port to listen on + -> AcidState st -- ^ state to serve + -> IO () +acidServer checkAuth port acidState + = withSocketsDo $ + do listenSocket <- listenOn port + let loop = forever $ + do (socket, _sockAddr) <- accept listenSocket + let commChannel = socketToCommChannel socket + forkIO $ do authorized <- checkAuth commChannel + when authorized $ + process commChannel acidState + ccClose commChannel -- FIXME: `finally` ? + infi = loop `catchSome` logError >> infi + infi `finally` (cleanup listenSocket) + where + logError :: (Show e) => e -> IO () + logError e = hPrint stderr e + + isResourceVanishedError :: IOException -> Bool + isResourceVanishedError = isResourceVanishedType . ioeGetErrorType + + isResourceVanishedType :: IOErrorType -> Bool + isResourceVanishedType ResourceVanished = True + isResourceVanishedType _ = False + + catchSome :: IO () -> (Show e => e -> IO ()) -> IO () + catchSome op _h = + op `catches` [ Handler $ \(e :: IOException) -> + if isFullError e || isDoesNotExistError e || isResourceVanishedError e + then return () -- h (toException e) -- we could log the exception, but there could be thousands of them + else throw e + ] + cleanup socket = + do sClose socket + case port of #if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32) - UnixSocket path -> removeFile path + UnixSocket path -> removeFile path #endif - _ -> return () + _ -> return () data Command = RunQuery (Tagged Lazy.ByteString) | RunUpdate (Tagged Lazy.ByteString) @@ -64,30 +272,40 @@ 0 -> liftM RunQuery get 1 -> liftM RunUpdate get 2 -> return CreateCheckpoint + _ -> error $ "Serialize.get for Command, invalid tag: " ++ show tag -data Response = Result Lazy.ByteString | Acknowledgement +data Response = Result Lazy.ByteString | Acknowledgement | ConnectionError instance Serialize Response where put resp = case resp of Result result -> do putWord8 0; put result Acknowledgement -> putWord8 1 + ConnectionError -> putWord8 2 get = do tag <- getWord8 case tag of 0 -> liftM Result get 1 -> return Acknowledgement + 2 -> return ConnectionError + _ -> error $ "Serialize.get for Response, invalid tag: " ++ show tag -process :: SafeCopy st => AcidState st -> Handle -> IO () -process acidState handle +{- | Server inner-loop + + This function is generally only needed if you are adding a new communication channel. +-} +process :: SafeCopy st => + CommChannel -- ^ a connected, authenticated communication channel + -> AcidState st -- ^ state to share + -> IO () +process CommChannel{..} acidState = do chan <- newChan forkIO $ forever $ do response <- join (readChan chan) - Strict.hPut handle (encode response) - hFlush handle + ccPut (encode response) worker chan (runGetPartial get Strict.empty) where worker chan inp = case inp of - Fail msg -> return () -- error msg - Partial cont -> do inp <- Strict.hGetSome handle 1024 - worker chan (cont inp) + Fail msg -> throwIO (SerializeError msg) + Partial cont -> do bs <- ccGetSome 1024 + worker chan (cont bs) Done cmd rest -> do processCommand chan cmd; worker chan (runGetPartial get rest) processCommand chan cmd = case cmd of @@ -102,46 +320,145 @@ data RemoteState st = RemoteState (Command -> IO (MVar Response)) (IO ()) deriving (Typeable) -{- | Connect to a remotely running 'AcidState'. -} -openRemoteState :: IsAcidic st => HostName -> PortID -> IO (AcidState st) -openRemoteState host port - = do handle <- connectTo host port - writeLock <- newMVar () - -- callbacks are added to the right and read from the left - callbacks <- newMVar (Seq.empty :: Seq.Seq (Response -> IO ())) - isClosed <- newIORef False - let getCallback = - modifyMVar callbacks $ \s -> return $ - case Seq.viewl s of - Seq.EmptyL -> noCallback - (cb Seq.:< s') -> (s', cb) - noCallback = error "openRemote: Internal error: Missing callback." - newCallback cb = modifyMVar_ callbacks (\s -> return (s Seq.|> cb)) - - listener inp - = case inp of - Fail msg -> error msg - Partial cont -> do inp <- Strict.hGetSome handle 1024 - listener (cont inp) - Done resp rest -> do callback <- getCallback - callback (resp :: Response) - listener (runGetPartial get rest) - actor cmd = do readIORef isClosed >>= closedError - ref <- newEmptyMVar - withMVar writeLock $ \() -> do - newCallback (putMVar ref) - Strict.hPut handle (encode cmd) >> hFlush handle - return ref +{- | Connect to an acid-state server which is sharing an 'AcidState'. -} +openRemoteState :: IsAcidic st => + (CommChannel -> IO ()) -- ^ authentication function, see 'sharedSecretPerform' + -> HostName -- ^ remote host to connect to (ignored when 'PortID' is 'UnixSocket') + -> PortID -- ^ remote port to connect to + -> IO (AcidState st) +openRemoteState performAuthorization host port + = withSocketsDo $ + do processRemoteState reconnect + where + -- | reconnect + reconnect :: IO CommChannel + reconnect + = (do debugStrLn "Reconnecting." + handle <- connectTo host port + let cc = handleToCommChannel handle + performAuthorization cc + debugStrLn "Reconnected." + return cc + ) + `catch` + ((\_ -> threadDelay 1000000 >> reconnect) :: IOError -> IO CommChannel) - closedError False = return () - closedError True = throwIO $ ErrorCall "The AcidState has been closed" - tid <- forkIO (listener (runGetPartial get Strict.empty)) - let shutdown = do writeIORef isClosed True - killThread tid - hClose handle - return (toAcidState $ RemoteState actor shutdown) +{- | Client inner-loop + This function is generally only needed if you are adding a new communication channel. +-} +processRemoteState :: IsAcidic st => + IO CommChannel -- ^ (re-)connect function + -> IO (AcidState st) +processRemoteState reconnect + = do cmdQueue <- atomically newTQueue + ccTMV <- atomically newEmptyTMVar + isClosed <- newIORef False + + let actor :: Command -> IO (MVar Response) + actor command = + do debugStrLn "actor: begin." + readIORef isClosed >>= flip when (throwIO AcidStateClosed) + ref <- newEmptyMVar + atomically $ writeTQueue cmdQueue (command, ref) + debugStrLn "actor: end." + return ref + + expireQueue listenQueue = + do mCallback <- atomically $ tryReadTQueue listenQueue + case mCallback of + Nothing -> return () + (Just callback) -> + do callback ConnectionError + expireQueue listenQueue + + handleReconnect :: SomeException -> IO () + handleReconnect e + = case fromException e of + (Just ThreadKilled) -> + do debugStrLn "handleReconnect: ThreadKilled. Not attempting to reconnect." + return () + _ -> + do debugStrLn $ "handleReconnect begin." + tmv <- atomically $ tryTakeTMVar ccTMV + case tmv of + Nothing -> + do debugStrLn $ "handleReconnect: error handling already in progress." + debugStrLn $ "handleReconnect end." + return () + (Just (oldCC, oldListenQueue, oldListenerTID)) -> + do thisTID <- myThreadId + when (thisTID /= oldListenerTID) (killThread oldListenerTID) + ccClose oldCC + expireQueue oldListenQueue + cc <- reconnect + listenQueue <- atomically $ newTQueue + listenerTID <- forkIO $ listener cc listenQueue + atomically $ putTMVar ccTMV (cc, listenQueue, listenerTID) + debugStrLn $ "handleReconnect end." + return () + + listener :: CommChannel -> TQueue (Response -> IO ()) -> IO () + listener cc listenQueue + = getResponse Strict.empty `catch` handleReconnect + where + getResponse leftover = + do debugStrLn $ "listener: listening for Response." + let go inp = case inp of + Fail msg -> error msg + Partial cont -> do debugStrLn $ "listener: ccGetSome" + bs <- ccGetSome cc 1024 + go (cont bs) + Done resp rest -> do debugStrLn $ "listener: getting callback" + callback <- atomically $ readTQueue listenQueue + debugStrLn $ "listener: passing Response to callback" + callback (resp :: Response) + return rest + rest <- go (runGetPartial get leftover) -- `catch` (\e -> do handleReconnect e + -- throwIO e + -- ) + getResponse rest + + actorThread :: IO () + actorThread = forever $ + do debugStrLn "actorThread: waiting for something to do." + (cc, cmd) <- atomically $ + do (cmd, ref) <- readTQueue cmdQueue + (cc, listenQueue, _) <- readTMVar ccTMV + writeTQueue listenQueue (putMVar ref) + return (cc, cmd) + debugStrLn "actorThread: sending command." + ccPut cc (encode cmd) `catch` handleReconnect + debugStrLn "actorThread: sent." + return () + + shutdown :: ThreadId -> IO () + shutdown actorTID = + do debugStrLn "shutdown: update isClosed IORef to True." + writeIORef isClosed True + debugStrLn "shutdown: killing actor thread." + killThread actorTID + debugStrLn "shutdown: taking ccTMV." + (cc, listenQueue, listenerTID) <- atomically $ takeTMVar ccTMV -- FIXME: or should this by tryTakeTMVar + debugStrLn "shutdown: killing listener thread." + killThread listenerTID + debugStrLn "shutdown: expiring listen queue." + expireQueue listenQueue + debugStrLn "shutdown: closing connection." + ccClose cc + return () + + cc <- reconnect + listenQueue <- atomically $ newTQueue + + actorTID <- forkIO $ actorThread + listenerTID <- forkIO $ listener cc listenQueue + + atomically $ putTMVar ccTMV (cc, listenQueue, listenerTID) + + return (toAcidState $ RemoteState actor (shutdown actorTID)) + remoteQuery :: QueryEvent event => RemoteState (EventState event) -> event -> IO (EventResult event) remoteQuery acidState event = do let encoded = runPutLazy (safePut event) @@ -151,9 +468,14 @@ Right result -> result) remoteQueryCold :: RemoteState st -> Tagged Lazy.ByteString -> IO Lazy.ByteString -remoteQueryCold (RemoteState fn _shutdown) event - = do Result resp <- takeMVar =<< fn (RunQuery event) - return resp +remoteQueryCold rs@(RemoteState fn _shutdown) event + = do resp <- takeMVar =<< fn (RunQuery event) + case resp of + (Result result) -> return result + ConnectionError -> do debugStrLn "retrying query event." + remoteQueryCold rs event + Acknowledgement -> error "remoteQueryCold got Acknowledgement. That should never happen." + scheduleRemoteUpdate :: UpdateEvent event => RemoteState (EventState event) -> event -> IO (MVar (EventResult event)) scheduleRemoteUpdate (RemoteState fn _shutdown) event