acid-state 0.8.3 → 0.10.0

raw patch · 6 files changed

+503/−175 lines, 6 filesdep ~basePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• acid-state.cabal +21/−47
• examples/RemoteClient.hs +65/−34
• examples/RemoteServer.hs +9/−12
• src/Data/Acid/Abstract.hs +6/−6
• src/Data/Acid/Common.hs +7/−3
• src/Data/Acid/Remote.hs +395/−73

@@ -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 <lemmih@gmail.com>---- 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:         -  

@@ -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 [<hostname> <port>]"

@@ -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))

@@ -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)

@@ -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

@@ -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