happstack-state 0.3.4 → 0.4.1

raw patch · 12 files changed

+558/−317 lines, 12 filesdep +SHAdep +bencodedep +hS3dep −hspreaddep ~happstack-datadep ~happstack-utilnew-uploaderPVP ok

version bump matches the API change (PVP)

Dependencies added: SHA, bencode, hS3, process

Dependencies removed: hspread

Dependency ranges changed: happstack-data, happstack-util

API changes (from Hackage documentation)

Files

• happstack-state.cabal +20/−8
• src/Happstack/State.hs +4/−4
• src/Happstack/State/CentralLogServer.hs +251/−0
• src/Happstack/State/Checkpoint.hs +1/−28
• src/Happstack/State/ComponentTH.hs +96/−26
• src/Happstack/State/Control.hs +22/−31
• src/Happstack/State/NetworkChan.hs +64/−0
• src/Happstack/State/Spread.hs +0/−173
• src/Happstack/State/Transaction.hs +20/−29
• src/Happstack/State/TxControl.hs +79/−17
• src/Happstack/State/Types.hs +1/−1
• src/Happstack/State/Util.hs +0/−0

@@ -1,5 +1,5 @@ Name:                happstack-state-Version:             0.3.4+Version:             0.4.1 Synopsis:            Event-based distributed state. Description:         Unplug your machine and restart and have your app recover to exactly where it left off. Happstack-State spares you the need to deal with all the marshalling, consistency, and configuration headache that you would have if you used an external DBMS for this purpose. Its component model makes it easy to compose big applications from smaller reliable parts. Use event subscription to trigger IO actions and support comet-style or irc-bot applications.  License:             BSD3@@ -14,7 +14,7 @@ source-repository head     type:     darcs     subdir:   happstack-state-    location: http://patch-tag.com/publicrepos/happstack+    location: http://patch-tag.com/r/mae/happstack/pullrepo  flag base4 @@ -22,6 +22,10 @@     Description: Build the testsuite, and include the tests in the library     Default: False +flag replication+    Description: Enable state replication through a central serialization service.+    Default: False+ Library    Exposed-modules:     @@ -40,13 +44,11 @@                        Happstack.State.Tests.SaverProperties   Other-modules:                               Happstack.State.Checkpoint-                       Happstack.State.Spread                        Happstack.State.Monad                        Happstack.State.Saver.Impl.File                        Happstack.State.Saver.Impl.Memory                        Happstack.State.Saver.Impl.Queue                        Happstack.State.Saver.Types---                     Happstack.State.OperationModes                        Happstack.State.Types                        Happstack.State.Util                        Happstack.State.TxControl@@ -58,15 +60,23 @@                        extensible-exceptions,                        filepath,                        hslogger >= 1.0.2,-                       hspread >= 0.3,-                       happstack-util >= 0.3.2 && < 0.4,-                       happstack-data >= 0.3.2 && < 0.4,+                       happstack-util >= 0.4.1 && < 0.5,+                       happstack-data >= 0.4.1 && < 0.5,                        mtl,                        old-time,                        random,                        stm,                        template-haskell +  if flag(replication)+    Other-modules:     Happstack.State.CentralLogServer+                       Happstack.State.NetworkChan+    Build-Depends:     hS3 >= 0.5.1 && < 0.6,+                       bencode >= 0.5,+                       process,+                       SHA >= 1.4.0 && < 2+    cpp-options:       -DREPLICATION+   if flag(base4)     Build-Depends:     base >= 4 && < 5, syb   else@@ -87,8 +97,10 @@                        OverlappingInstances, UndecidableInstances, CPP,                        TypeFamilies, FunctionalDependencies, TypeOperators,                        GADTs, EmptyDataDecls, ExistentialQuantification, RankNTypes,-                       ScopedTypeVariables, PatternSignatures, PatternGuards,+                       ScopedTypeVariables, PatternGuards,                        MagicHash, TypeSynonymInstances+  if impl(ghc < 6.10)+     Extensions:       PatternSignatures   ghc-options:         -Wall   GHC-Prof-Options:    -auto-all 

@@ -3,7 +3,7 @@      Ev, AnyEv,      TxControl, query, update, Update, Query,      -- * Types-     TxId, EpochMilli, TxConfig(..),nullTxConfig,Saver(..),+     TxId, EpochMilli, Saver(..),      -- * Misc utilities      module Happstack.State.Monad,      getEventId, getTime, getEventClockTime, getEventStream, EventItem(..),@@ -13,11 +13,12 @@      module Happstack.Data.SerializeTH,       module Happstack.State.Control,-     module Happstack.State.TxControl,      module Happstack.State.ComponentTH,      module Happstack.State.ComponentSystem,-     closeTxControl,+     --closeTxControl,+     runTxSystem,      createCheckpoint,+     shutdownSystem,      -- * Unsafe things      unsafeIOToEv     ) where@@ -28,7 +29,6 @@ import Happstack.Data.Serialize import Happstack.Data.SerializeTH import Happstack.State.Transaction-import Happstack.State.Checkpoint import Happstack.State.ComponentSystem import Happstack.State.Types import Happstack.State.Util

@@ -0,0 +1,251 @@+module Happstack.State.CentralLogServer+    ( ApplicationName+    , connectToCluster+    , changeEventMapping+    , createCheckpoint+    ) where++import Data.Binary+import qualified Data.ByteString.Lazy.Char8 as L+import Control.Concurrent+import System.Process+import System.Exit+import System.IO+import Control.Applicative ((<$>), (<*>))+import Data.Int+import Data.List+import Control.Exception+import qualified Data.Map as Map+import Control.Monad+import System.Random (mkStdGen)+import Network.AWS.AWSConnection+import Network.AWS.S3Bucket hiding (key)+import Network.AWS.S3Object++import Happstack.Data.Serialize+import Happstack.State.Transaction+import Happstack.State.Types+import Happstack.State.NetworkChan+import Data.BEncode+import Data.BEncode.Parser++import Data.Digest.Pure.SHA++import System.Log.Logger++logLS :: Priority -> String -> IO ()+logLS = logM "Happstack.State.CentralLogServer"++type ApplicationName = String++data Entry = Entry EntryId EpochMilli EntryData deriving (Show)+type EntryId = Int64+type EntryData = L.ByteString+type URL = String++instance Binary Entry where+    get = Entry <$> get <*> get <*> get+    put (Entry eid epoch edata)+        = put eid >> put epoch >> put edata++data ForeignEvent = ForeignEvent Int L.ByteString+instance Binary ForeignEvent where+    get = ForeignEvent <$> get <*> get+    put (ForeignEvent hash event)+        = put hash >> put event++data CheckpointMsg = NoCheckpoint | Checkpoint EntryId URL deriving (Show)+instance BEncodeable CheckpointMsg where+    bencode NoCheckpoint = BString (L.pack "NoCheckpoint")+    bencode (Checkpoint eid url) = BDict (Map.singleton "Checkpoint" $ BDict $ Map.fromList [ ("entry_id", BInt $ fromIntegral eid)+                                                                                            , ("url", BString (encode url))])+    bdecode = do str <- bstring token+                 guard (str == "NoCheckpoint")+                 return NoCheckpoint+              <|>+              do setInput =<< dict "Checkpoint"+                 eid <- bint (dict "entry_id")+                 url <- bbytestring (dict "url")+                 return (Checkpoint (fromIntegral eid) $ decode url)++instance BEncodeable Entry where+    bencode entry = BString (encode entry)+    bdecode = fmap decode (bbytestring token)++data Response = Response Entry Bool+instance BEncodeable Response where+    bencode (Response entry isOwner)+        = BDict (Map.fromList [("entry",bencode entry)+                              ,("owner",BString $ L.pack $ show isOwner)])+    bdecode = do owner <- bstring (dict "owner")+                 setInput =<< dict "entry"+                 entry <- bdecode+                 case owner of+                   "True" -> return $ Response entry True+                   _      -> return $ Response entry False++instance BEncodeable L.ByteString where+    bencode bs = BString bs+    bdecode = bbytestring token++data UserMsg = NewEvent L.ByteString | NewCheckpoint EntryId URL deriving (Show)+instance BEncodeable UserMsg where+    bencode (NewEvent event)+        = BDict (Map.singleton "NewEvent" (BString event))+    bencode (NewCheckpoint eid url)+        = BDict $ Map.singleton "NewCheckpoint" $ BDict $ Map.fromList [ ("entry_id", BInt $ fromIntegral eid)+                                                                       , ("url", BString $ L.pack url) ]+    bdecode = do setInput =<< dict "NewEvent"+                 fmap NewEvent (bbytestring token)+              <|>+              do setInput =<< dict "NewCheckpoint"+                 eid <- bint (dict "entry_id")+                 url <- bbytestring (dict "url")+                 return (NewCheckpoint (fromIntegral eid) (L.unpack url))++data Cluster+    = Cluster { clusterChan :: NetworkChan+              , clusterS3   :: AWSConnection+              , clusterBucket :: String+              , clusterLastId :: MVar EntryId }++-- FIXME: The logging service should have a domain name.+logServerAddress :: URL+logServerAddress = "174.129.13.114"++-- Connect to the central log server via SSH.+-- The appropriate identities have to be added to the autentication agent. See ssh-add(1).+connectToCluster :: ApplicationName -> EventMap -> IO Cluster+connectToCluster appName localEventMap+    = do logLS NOTICE "Retrieving S3 credentials from environment."+         Just conn <- amazonS3ConnectionFromEnv+         bucket <- initiateBucket conn -- FIXME: Run this function in 'createCheckpoint' instead?+         logLS NOTICE "Running LogRelay through ssh."+         (inh, outh) <- sshConnect ("app_" ++ appName) logServerAddress "LogRelay"+         logLS NOTICE "Creating network channel."+         chan <- newNetworkChan inh outh+         checkpoint <- peek chan :: IO CheckpointMsg+         lastId <- newEmptyMVar+         case checkpoint of+           NoCheckpoint       -> do putMVar lastId 0+                                    logLS NOTICE "No checkpoint available."+           Checkpoint eid url -> do putMVar lastId eid+                                    let (checkpoint_bucket,key) = read url+                                    logLS NOTICE $ "Checkpoint available from this location: " ++ show (checkpoint_bucket,key)+                                    ret <- getObject conn (S3Object { obj_bucket = checkpoint_bucket+                                                                    , obj_name   = key+                                                                    , content_type = ""+                                                                    , obj_headers = []+                                                                    , obj_data    = L.empty })+                                    case ret of+                                      Left err  -> error (show err)+                                      Right obj -> do (stateMap, _rest) <- evaluate $ deserialize (obj_data obj)+                                                      forM_ (Map.toList stateMap) $ \(stateTy, state) ->+                                                        setNewState' localEventMap stateTy state+         return Cluster { clusterChan = chan+                        , clusterS3   = conn+                        , clusterBucket = bucket+                        , clusterLastId = lastId }++-- Update a snapshot of the component states to S3 and notify the log-server about it.+createCheckpoint :: MVar TxControl -> Cluster -> IO ()+createCheckpoint ctlVar cluster+    = withMVar ctlVar $ \ctl ->+      do -- This ID number is used as the cutoff for the checkpoint. After this checkpoint has been+         -- saved, events younger than this ID will not be sent when restoring the state.+         lastId <- readMVar (clusterLastId cluster)+         allStates <- mapM getState (ctlAllComponents ctl)+         let stateMap = Map.fromList (zip (ctlAllComponents ctl) allStates)+             checkpoint = serialize stateMap+         let bucket = clusterBucket cluster+             key    = "checkpoint_" ++ show (sha1 checkpoint)+         sendObject (clusterS3 cluster) (S3Object { obj_bucket   = bucket+                                                  , obj_name     = key+                                                  , content_type = "application/octet-stream"+                                                  , obj_headers  = []+                                                  , obj_data     = checkpoint })+         let url = show (bucket, key)+         poke (clusterChan cluster) (NewCheckpoint lastId url)++{- Find a bucket to place new checkpoints in. If no suitable bucket exists, create one.+   Race conditions aren't important since we're just trying to keep the happs buckets to a +   minimum (preferable just one).+-}+initiateBucket :: AWSConnection -> IO String+initiateBucket conn+    = do response <- listBuckets conn+         case response of+           Left err      -> error (show err)+           Right buckets -> case [ bucket_name bucket | bucket <- buckets, "happs_checkpoints" `isPrefixOf` bucket_name bucket ] of+                              (bucket:_) -> return bucket+                              [] -> do ret <- createBucketWithPrefix conn "happs_checkpoints"+                                       case ret of+                                         Left err -> error (show err)+                                         Right bucket -> return bucket++{-+  The new event mapping pipes Update events through the logging server. Query events+  do not need to be sequenced and are therefore executed normally.+  Incoming Update events may have been sent from another client and hence will not+  have a local listener attached. Those alien events are still executed (to update the+  state) but their computed result is discarded. For events that originated locally,+  the computed response is sent to the appropriate listener.+-}+changeEventMapping :: EventMap -> Cluster -> IO EventMap+changeEventMapping localEventMap cluster+    = do logLS NOTICE "Create new event mapper"+         responseIndex <- newMVar Map.empty+         let chan = clusterChan cluster+         let insertEID uniq = do mv <- newEmptyMVar+                                 modifyMVar_ responseIndex $ \idx -> return $ Map.insert uniq mv idx+                                 return $ takeMVar mv+             returnResponse uniq object = modifyMVar_ responseIndex $ \idx ->+                                          case Map.lookup uniq idx of+                                            Nothing -> return idx -- This shouldn't happen.+                                            Just mv -> do putMVar mv object -- Notify the caller about the new response.+                                                          return $ Map.delete uniq idx+             listener = forever $+                        do Response (Entry eid time foreign) owner <- peek chan+                           let ForeignEvent uniq object = decode foreign+                           logLS NOTICE $ "Received entry with id: " ++ show uniq ++ " " ++ show owner+                           let txContext = TxContext { txId = eid+                                                     , txRand = 0+                                                     , txTime = time+                                                     , txStdGen = mkStdGen 0 }+                           response <- runColdEventFunc txContext (fst $ deserialize object) localEventMap+                           swapMVar (clusterLastId cluster) eid+                           -- If we sent this event, pipe the result to the listener.+                           when owner $ returnResponse uniq response+         forkIO listener+         uniqVar <- newMVar 0+         let newEventMap = flip Map.map localEventMap $ \handler ->+                           case handler of+                             UpdateHandler runCold _ parse+                                 -> let runHot ev = do uniq <- modifyMVar uniqVar (\a -> return (a+1, a))+                                                       let obj = serialize $ mkObject ev+                                                           foreign = ForeignEvent uniq obj+                                                       logLS NOTICE $ "New locally unique id: " ++ show uniq+                                                       wait <- insertEID uniq+                                                       poke chan (NewEvent (encode foreign))+                                                       response <- wait+                                                       logLS NOTICE $ "Received response for: " ++ show uniq+                                                       return $ parseObject response+                                    in UpdateHandler runCold runHot parse+                             QueryHandler{} -> handler+         return newEventMap++++type UserName = String+type ServerAddress = String+type Command = String++-- Run a command through an SSH tunnel.+sshConnect :: UserName -> ServerAddress -> Command -> IO (Handle, Handle)+sshConnect userName serverAddress command+    = do (inh, outh, errh, _pid) <- runInteractiveProcess "ssh" ["-o", "BatchMode=yes", userName ++ "@" ++ serverAddress, command] Nothing Nothing+         tid <- myThreadId+         forkIO $ do errMsg <- hGetLine errh+                     hPutStrLn stderr $ "SSH tunneling failed: " ++ errMsg+                     throwTo tid (ExitFailure 255)+         return (inh, outh)

@@ -1,8 +1,6 @@ {-# LANGUAGE CPP, TemplateHaskell, DeriveDataTypeable #-} module Happstack.State.Checkpoint-    ( createTxControl-    , closeTxControl-    , restoreState+    ( restoreState     , createCheckpoint     ) where @@ -44,33 +42,8 @@ instance Version State $(deriveSerialize ''State) --- | Given a Saver and a Proxy, createTxControl will --- initialize a TxControl.  This does not actually start the--- state system.-createTxControl :: (Methods state, Component state) =>-                   Saver -> Proxy state -> IO (MVar TxControl)-createTxControl saver prox-    = do --         -- The state hasn't been loaded yet. Ignore events.-         eventSaverVar   <- newMVar =<< createWriter NullSaver "events" 0-         -- obtain a prefix lock-         lock <- obtainLock saver-         newMVar $ TxControl-                       { ctlSaver             = saver-                       , ctlEventSaver        = eventSaverVar-                       , ctlAllComponents     = allStateTypes prox-                       , ctlComponentVersions = componentVersions prox-                       , ctlChildren          = []-                       , ctlPrefixLock = lock }                         --- | Saves the state and closes the serialization-closeTxControl :: MVar TxControl -> IO ()-closeTxControl ctlVar-    = do ctl <- takeMVar ctlVar-         writerClose =<< takeMVar (ctlEventSaver ctl)-         releaseLock (ctlPrefixLock ctl)  -- FIXME: It may be nice to print out what components were saved on disk --        compared to the components actually used in the application.

@@ -19,7 +19,15 @@  import Data.Generics.Basics -nubCxt :: [TypeQ] -> Q [Type]+#if MIN_VERSION_template_haskell(2,4,0)+type CtxElemQ = PredQ+type CtxElem  = Pred+#else+type CtxElemQ = TypeQ+type CtxElem  = Type+#endif++nubCxt :: [CtxElemQ] -> Q [CtxElem] nubCxt tsQ     = do ts <- cxt tsQ          return $ nub ts@@ -55,23 +63,38 @@          ds5 <- genSerializeInstances methodInfos          return (ds1 ++ [ds2] ++ ds4 ++  ds5 ) -mkKeyConstraints :: [Name] -> [TypeQ]+mkKeyConstraints :: [Name] -> [CtxElemQ] mkKeyConstraints keys-    = [ appT (conT ''Typeable) (varT key) | key <- keys ] ++-      [ appT (conT ''Serialize) (varT key) | key <- keys ]+    = [ mkCtxt ''Typeable (varT key) | key <- keys ] +++      [ mkCtxt ''Serialize (varT key) | key <- keys ] -mkMethodConstraints :: [Name] -> MethodInfo -> [TypeQ]+mkMethodConstraints :: [Name] -> MethodInfo -> [CtxElemQ] mkMethodConstraints keys method     = map return (substMethodContext method keys) -substMethodContext :: MethodInfo -> [Name] -> [Type]+substMethodContext :: MethodInfo -> [Name] -> [CtxElem] substMethodContext method keys     = let relation = zip (methodKeys method) keys-          worker (VarT old) | Just new <- lookup old relation++          workerT (VarT old) | Just new <- lookup old relation                    = VarT new-          worker (AppT l r) = AppT (worker l) (worker r)-          worker (ForallT c names t) = ForallT c names (worker t)-          worker t = t+          workerT (AppT l r) = AppT (workerT l) (workerT r)+          workerT (ForallT c names t) = ForallT c names (workerT t)+          workerT t = t++#if MIN_VERSION_template_haskell(2,4,0)+          workerP (ClassP old typs)+              = let new = case lookup old relation of+                            Nothing -> old+                            Just nm -> nm+                in ClassP new $ map workerT typs+          workerP p = p++          worker = workerP+#else+          worker = workerT+#endif+       in map worker (methodContext method)  mkType :: Name -> [Name] -> TypeQ@@ -120,14 +143,21 @@ genEventInstance method     = do let keys = methodKeys method              eventType = foldl appT (conT (upperName (methodName method))) (map varT keys)-         instanceD (nubCxt $ [appT (conT ''Serialize) eventType-                             ,appT (conT ''Serialize) (return (methodResult method))]+         instanceD (nubCxt $ [mkCtxt ''Serialize eventType+                             ,mkCtxt ''Serialize (return (methodResult method))+                             ]                           ++ mkKeyConstraints keys                           ++ mkMethodConstraints keys method                    )                    (appT (appT (conT (methodClass method)) eventType) (return (methodResult method)))                    [] +mkCtxt :: Name -> TypeQ -> CtxElemQ+#if MIN_VERSION_template_haskell(2,4,0)+mkCtxt name typ = classP name [typ]+#else+mkCtxt name typ = appT (conT name) typ+#endif  genMethodStructs :: [Name] -> [MethodInfo] -> Q [Dec] genMethodStructs derv meths@@ -137,8 +167,16 @@ genMethodStruct :: [Name] -> MethodInfo -> Q [Dec] genMethodStruct derv method     = do let c = NormalC (upperName (methodName method)) (zip (repeat NotStrict ) (methodArgs method))-         return [ DataD [] (upperName (methodName method)) (methodKeys method) [c] (derv) ]+         return [ DataD [] (upperName (methodName method)) (map mkTyVarBndr $ methodKeys method) [c] (derv) ] + where++#if MIN_VERSION_template_haskell(2,4,0)+   mkTyVarBndr = PlainTV+#else+   mkTyVarBndr = id+#endif+ upperName :: Name -> Name upperName = mkName . upperFirst . nameBase @@ -148,7 +186,7 @@  data MethodInfo = Method { methodName   :: Name                          , methodKeys   :: [Name]-                         , methodContext:: [Type]+                         , methodContext:: [CtxElem]                          , methodArgs   :: [Type]                          , methodClass  :: Name                          , methodEv     :: Name@@ -231,24 +269,56 @@ getStateKeys (ConT _st) = [] getStateKeys v = error $ "Bad state type: " ++ pprint v ++ " (expected a constant, an application or a type variable)" -isMonadState :: [Type] -> Type -> Maybe Type-isMonadState contxt name = listToMaybe [ state | AppT (AppT (ConT m) state) mName <- contxt, -                                         mName == name, m == ''MonadState ]+-- |Ignoring EqualP predicates, returns all names associated with a+-- context predicate+getPredKeys :: CtxElem -> [Name]+#if MIN_VERSION_template_haskell(2,4,0)+getPredKeys (ClassP nm typs) = nm : concatMap getStateKeys typs+getPredKeys _ = []+#else+getPredKeys = getStateKeys+#endif -isMonadReader :: [Type] -> Type -> Maybe Type-isMonadReader contxt name = listToMaybe [ state | AppT (AppT (ConT m) state) mName <- contxt, -                                          mName == name, m == ''MonadReader ] -isRelevant :: [Name] -> Type -> Bool-isRelevant keys t = isAcceptableContext t && any (`elem` keys) (getStateKeys t)+isMonadState :: [CtxElem] -> Type -> Maybe Type+isMonadState = isMonadX ''MonadState -isAcceptableContext :: Type -> Bool+isMonadReader :: [CtxElem] -> Type -> Maybe Type+isMonadReader = isMonadX ''MonadReader++isMonadX :: Name -> [CtxElem] -> Type -> Maybe Type+#if MIN_VERSION_template_haskell(2,4,0)+isMonadX monadType contxt name = listToMaybe [ state | ClassP m [state,mName] <- contxt, +                                         mName == name, m == monadType ]+#else+isMonadX monadType contxt name = listToMaybe [ state | AppT (AppT (ConT m) state) mName <- contxt,+                                         mName == name, m == monadType ]+#endif++++isRelevant :: [Name] -> CtxElem -> Bool+isRelevant keys p = isAcceptableContext p && any (`elem` keys) (getPredKeys p)++isAcceptableContext :: CtxElem -> Bool+#if MIN_VERSION_template_haskell(2,4,0)+isAcceptableContext (ClassP con _) = con `notElem` [''MonadState, ''MonadReader] +#else isAcceptableContext (AppT r r') = isAcceptableContext r && isAcceptableContext r' isAcceptableContext (ConT con) = con `notElem` [''MonadState, ''MonadReader]+#endif isAcceptableContext _ = True  requireSimpleCon :: Name -> Info -> [Name]-requireSimpleCon _ (TyConI (DataD _ _ names _ _derv)) = names-requireSimpleCon _ (TyConI (NewtypeD _ _ names _ _derv)) = names-requireSimpleCon _ (TyConI (TySynD _ names _)) = names+requireSimpleCon _ (TyConI (DataD _ _ names _ _derv)) = map conv names+requireSimpleCon _ (TyConI (NewtypeD _ _ names _ _derv)) = map conv names+requireSimpleCon _ (TyConI (TySynD _ names _)) = map conv names requireSimpleCon name _ = error $ "Cannot create component from '"++pprint name++"'. Expected a data structure."++#if MIN_VERSION_template_haskell(2,4,0)+conv :: TyVarBndr -> Name+conv (PlainTV nm) = nm+conv (KindedTV nm _) = nm+#else+conv = id+#endif

@@ -2,7 +2,8 @@  module Happstack.State.Control     ( startSystemState-    , startSystemStateMultimaster+    , startSystemStateAmazon+    , processLoggingFlags     , stdSaver     , waitForTermination     ) where@@ -31,27 +32,31 @@  import Happstack.State.Transaction import Happstack.State.Saver+#ifdef REPLICATION+import Happstack.State.CentralLogServer (ApplicationName)+#endif import Happstack.State.TxControl import Happstack.State.ComponentSystem import Happstack.Data.Proxy hiding (proxy) --- | Starts the MACID system without multimaster support.  Uses the default behavior+-- | Starts the MACID system without replication support.  Uses the default behavior -- of saving the state into the _local directory. startSystemState :: (Methods a, Component a) =>                     Proxy a -> IO (MVar TxControl) startSystemState proxy-    = do _txConfig <- parseArgs-         saver <- stdSaver+    = do saver <- stdSaver          runTxSystem saver proxy --- | Starts the MACID system with multimaster support.  Uses the default behavior--- of saving the state into the _local directory.-startSystemStateMultimaster :: (Methods a, Component a) =>-                               Proxy a -> IO (MVar TxControl)-startSystemStateMultimaster proxy-    = do _txConfig <- parseArgs-         saver <- stdSaver-         runTxSystem' True saver proxy+#ifdef REPLICATION+startSystemStateAmazon :: (Methods a, Component a) => ApplicationName -> Proxy a -> IO (MVar TxControl)+startSystemStateAmazon appName proxy+    = runTxSystemAmazon appName proxy+#else+type ApplicationName = String+startSystemStateAmazon :: (Methods a, Component a) => ApplicationName -> Proxy a -> IO (MVar TxControl)+startSystemStateAmazon appName proxy+    = runTxSystemAmazon appName proxy+#endif  -- | Returns the default Saver.  It will save the application state into -- the _local directory.@@ -80,12 +85,6 @@          loop 'c' #endif -mkTxConfig :: [Flag] -> TxConfig-mkTxConfig = foldr worker nullTxConfig-    where worker (Cluster serv) c = c{txcOperationMode = ClusterMode (fromMaybe "" serv)}-          worker (ClusterPort port) c = c{txcClusterPort = port}-          worker _ c = c- data NullLogger instance SLH.LogHandler NullLogger where     setLevel = error "Don't do this! This logger should not be used!"@@ -107,7 +106,6 @@           worker (LogLevel pri) = do updateGlobalLogger "Happstack" (setLevel pri)                                      rlogger <- getLogger "Happstack"                                      logM "" WARNING ("Set logging priority to " ++ show (getLevel rlogger))-          worker _ = return ()  -- order should not matter, though it does now. -- we should ALSO allow multiple loggers at the same time! --log-target=stdout --log-target=syslog@@ -116,17 +114,11 @@                       "Log level: DEBUG, INFO, NOTICE, WARNING, ERROR, CRITICAL, ALERT, EMERGENCY. Default: WARNING"           ,Option "" ["log-target"] (ReqArg (LogTarget . readTarget) "target")                       "Log target: stdout, syslog, or a FilePath such as /home/foo/bar.log . Default: stdout"-          -- FIXME: use a better flag name-          ,Option "" ["cluster-mode"] (OptArg (Cluster) "servers")-                      "Start in multimaster mode. Will join cluster if optional arg is given."-          ,Option "" ["cluster-port"] (ReqArg (ClusterPort . read) "port")-                      "Multimaster server will listen on this port."           ]  data Target = File FilePath | StdOut | SysLog deriving (Read,Show,Eq,Ord) -data Flag = LogLevel Priority | LogTarget Target-          | Cluster (Maybe String) | ClusterPort Int deriving Show+data Flag = LogLevel Priority | LogTarget Target deriving Show  readTarget :: String -> Target readTarget arg = case map toLower arg of@@ -140,8 +132,8 @@                     worker (ReqArg _ f) = ReqArg (const ()) f                     worker (OptArg _ f) = OptArg (const ()) f -parseArgs :: IO TxConfig-parseArgs = do+processLoggingFlags :: IO a -> IO a+processLoggingFlags action = do     args <- liftIO getArgs     pn   <- liftIO getProgName     let err n ls = -- XXX these next lines should be written to stderr!@@ -150,8 +142,7 @@                       putStrLn ("Usage "++usageInfo pn castOptions)                       exitWith (ExitFailure 1)     case getOpt' Permute options args of-      (flags,_fs,_args',[]) -> do setLoggingSettings flags-                                  -- FIXME: replace system args with fs++args'-                                  return (mkTxConfig flags)+      (flags,fs,args',[]) -> do setLoggingSettings flags+                                withArgs (fs ++ args') action       (_,_,_,es)          -> err "errors" es 

@@ -0,0 +1,64 @@+module Happstack.State.NetworkChan+    ( BEncodeable(..)+    , NetworkChan+    , newNetworkChan+    , peek+    , poke+    ) where++import Data.BEncode+import Data.BEncode.Parser++import Control.Monad (when, forever, forM_ )+import System.IO+import qualified Data.ByteString.Lazy as L+import Data.Binary+import Control.Concurrent.STM+import Control.Concurrent (forkIO, ThreadId)++class BEncodeable a where+    bencode :: a -> BEncode+    bdecode :: BParser a+++data NetworkChan = NetworkChan { outgoingChan :: TVar [L.ByteString]+                               , incomingHandle :: Handle+                               , writerThread :: ThreadId }++newNetworkChan :: Handle -> Handle -> IO NetworkChan+newNetworkChan outgoing incoming+    = do cache <- newTVarIO []+         tid <- forkIO $ forever $ do ls <- atomically $ do ls <- readTVar cache+                                                            when (null ls) retry+                                                            writeTVar cache []+                                                            return (reverse ls)+                                      forM_ ls $ \entry -> do L.hPut outgoing (encode (L.length entry))+                                                              L.hPut outgoing entry+                                      hFlush outgoing+         return NetworkChan { outgoingChan = cache+                            , incomingHandle = incoming+                            , writerThread = tid }++peek :: BEncodeable a => NetworkChan -> IO a+peek chan+    = do size <- L.hGet (incomingHandle chan) 8+         when (L.length size < 8) $ error "NetworkChan.peek: Connection closed."+         body <- L.hGet (incomingHandle chan) (decode size)+         case bRead body of+           Nothing      -> error "NetworkChan.peek: Malformed bencoded data."+           Just encoded -> case runParser bdecode encoded of+                             Left err  -> error "NetworkChan.peek: Parsing bencoded data failed."+                             Right msg -> return msg++poke :: BEncodeable a => NetworkChan -> a -> IO ()+poke chan msg+    = atomically $+      do ls <- readTVar (outgoingChan chan)+         writeTVar (outgoingChan chan) (bPack (bencode msg):ls)+{-+do L.hPut (inputHandle chan) (encode (L.length body))+          L.hPut (inputHandle chan) body+          hFlush (inputHandle chan)+    where body = bPack (bencode msg)+-}+

@@ -1,173 +0,0 @@-{-# LANGUAGE TemplateHaskell, DeriveDataTypeable #-}-module Happstack.State.Spread-    ( connectToCluster-    , changeEventMapping-    ) where--import Happstack.State.Transaction-import Happstack.Data.Serialize-import Happstack.Data.SerializeTH--import qualified Data.ByteString.Char8 as B-import qualified Data.ByteString.Lazy.Char8 as L-import qualified Data.Map as Map--import Data.Typeable-import Spread.Client-import qualified Control.Concurrent.Chan.Closeable as Closeable--import System.Log.Logger-import Control.Monad hiding (join)-import Control.Concurrent-import System.Random--logSP :: Priority -> String -> IO ()-logSP = logM "Happstack.State.Spread"---data Cluster = Cluster-    { clusterChan       :: Closeable.Chan Closeable.R Message-    , clusterConnection :: Connection-    }--type EventId = Int--data ClusterMsg-    = RequestState-    | NewState B.ByteString -- A piece of state (map of all components)-    | StateTransferred      -- The entire state has been transmitted.-    | ForeignEvent EventId Object-    | EventResponse EventId Object-      deriving (Typeable)-instance Version ClusterMsg -- version 0-$(deriveSerialize ''ClusterMsg)--{- | Connects the Happstack process to the local running spread daemon.-   The spread daemon must already be running.--}--connectToCluster :: IO Cluster-connectToCluster-    = do logSP NOTICE "Connecting to spread daemon on localhost"-         uniq <- randomRIO (0,30000) -- FIXME: this is a hack to get unique names.-         let name = mkPrivateName (B.pack $ "node" ++ show (uniq::Int))-         (chan, conn) <- connect defaultConf{desiredName = name}-         startReceive conn-         join receiverGroup conn-         return $ Cluster chan conn--receiverGroup :: Group-receiverGroup = let Just g = makeGroup "receiver"-                in g--{- | This function modifies any event handlers in the given EventMap-   to make them route the update information to the rest of the cluster---}--changeEventMapping :: MVar TxControl -> EventMap -> Cluster -> IO EventMap-changeEventMapping ctlVar localEventMap cluster-    = do logSP NOTICE "Create new event mapper"-         mems <- getClusterMembers cluster-         responseIndex <- newMVar Map.empty-         ready <- newEmptyMVar-         eventQueue <- newChan-         stateVar <- newMVar L.empty-         eidStore <- newMVar 0 -- eid's are locally unique.-         let newEID = modifyMVar eidStore (\store -> return (store+1,store+1))-             pushStateBlock st = modifyMVar_ stateVar (\acc -> return $ acc `L.append` L.fromChunks [st])-             insertEID eid = do mv <- newEmptyMVar-                                modifyMVar_ responseIndex $ \idx -> return $ Map.insert eid mv idx-                                return $ takeMVar mv-             returnResponse eid object = modifyMVar_ responseIndex $ \idx ->-                                         case Map.lookup eid idx of-                                           Nothing -> return idx -- We're already received the result from another node.-                                           Just mv -> do putMVar mv object -- Notify the caller about the new response.-                                                         return $ Map.delete eid idx-             listener = forever $-                        do mbMsg <- Closeable.readChan (clusterChan cluster)-                           case mbMsg of-                             Nothing -> error "Disconnected from the spread daemon."-                             Just (Regular msg) ->-                                 case fst (deserialize (L.fromChunks [inData msg])) of-                                   RequestState            -> sendState ctlVar (inSender msg) cluster-                                   ForeignEvent eid object -> writeChan eventQueue (inSender msg, eid, object)-                                   NewState st             -> pushStateBlock st-                                   -- FIXME: check that stateVar is non-empty?-                                   StateTransferred        -> do restoreState =<< takeMVar stateVar-                                                                 putMVar ready ()-                                   EventResponse eid object-> returnResponse eid object-                             Just _  -> return ()-             responder = forever $-                         do (sender, eid, object) <- readChan eventQueue-                            response <- runObjectEventFunc object localEventMap-                            sendClusterMsg [sender] (EventResponse eid response) cluster-         forkIO listener-         forkIO responder-         case mems of-           [] -> putMVar ready () -- The cluster is empty, use state from disk.-           (x:_) -> do logSP NOTICE $ "Requesting state from: " ++ show x-                       sendClusterMsg [x] RequestState cluster-         logSP NOTICE "Waiting for ready signal"-         takeMVar ready-         logSP NOTICE "Ready signal received"-         let newEventMap = flip Map.map localEventMap $ \handler ->-                           case handler of-                             UpdateHandler runCold _ parse-                                 -> let runHot ev = do eid <- newEID-                                                       logSP NOTICE $ "New eid: " ++ show eid-                                                       wait <- insertEID eid-                                                       sendClusterMsg [receiverGroup] (ForeignEvent eid (mkObject ev)) cluster-                                                       response <- wait-                                                       logSP NOTICE $ "Received response for: " ++ show eid-                                                       return $ parseObject response-                                    in UpdateHandler runCold runHot parse-                             QueryHandler{} -> handler-         return newEventMap--sendState :: MVar TxControl -> Group -> Cluster -> IO ()-sendState ctlVar sender cluster-    = withMVar ctlVar $ \ctl ->-      do logSP NOTICE $ "Sending state to: " ++ show sender-         allStates <- mapM getState (ctlAllComponents ctl)-         let chunks = L.toChunks $ serialize $ zip (ctlAllComponents ctl) allStates-         forM_ chunks $ \chunk ->-             do sendClusterMsg [sender] (NewState chunk) cluster-         sendClusterMsg [sender] StateTransferred cluster-         logSP NOTICE $ "The state has been transmitted."--restoreState :: L.ByteString -> IO ()-restoreState stateObject-    = do logSP NOTICE $ "Loading components from network."-         forM_ state $ \(stateType, stateData) ->-             do setNewState stateType stateData-         logSP NOTICE $ "All components successfully loaded"-    where state = fst (deserialize stateObject)--sendClusterMsg :: [Group] -> ClusterMsg -> Cluster -> IO ()-sendClusterMsg to msg cluster-    = send outMsg (clusterConnection cluster)-    where outMsg = Outgoing { outOrdering = Safe-                            , outDiscard = False-                            , outData = B.concat (L.toChunks (serialize msg))-                            , outGroups = to-                            , outMsgType = 0 }---- Cluster members not including self.-getClusterMembers :: Cluster -> IO [PrivateGroup]-getClusterMembers cluster-    = do mbMsg <- Closeable.readChan (clusterChan cluster)-         case mbMsg of-           Just (Membership membership)-               -> case membership of-                    Reg{members=m} -> return $ filter (/= privateGroup (clusterConnection cluster)) m-                    Transient{} -> error $ "Received a Transient message on connect."-                    SelfLeave{} -> error $ "Received a SelfLeave message on connect."-           Just msg -> error $ "Expected membership message from cluster. Received a " ++ showMsgType msg ++ " message."-           Nothing  -> error "Cluster channel unexpectedly closed."--showMsgType :: Message -> String-showMsgType (Regular _) = "regular"-showMsgType (Membership _) = "membership"-showMsgType (Rejected _) = "rejected"-

@@ -91,7 +91,7 @@  data EventHandler where     UpdateHandler :: UpdateEvent ev res =>-                     (TxContext -> ev -> IO ()) ->+                     (TxContext -> ev -> IO res) ->                      (ev -> IO res) ->                      (Object -> ev) ->                      EventHandler@@ -171,9 +171,10 @@          return $ M.unions maps     where (componentTree, _versions, _ioActions) = collectHandlers componentProxy           eventHandler tx (Update fn)-              = let updateCold' cxt ev+              = let updateCold' ev getCxt                         = do mv <- newEmptyMVar-                             let handler = do lastCxt <- readTVar (txLastTxContext tx)+                             let handler = do cxt <- getCxt+                                              lastCxt <- readTVar (txLastTxContext tx)                                               if txId lastCxt < txId cxt                                                  then do writeTVar (txLastTxContext tx) cxt                                                          writeTChan (txProcessQueue tx) $ IHR cxt ev $@@ -185,10 +186,10 @@                              case me of                                  Left e -> throwIO e                                  Right e -> return e-                    updateCold cxt ev = do updateCold' cxt ev; return ()+                    updateCold cxt ev = do updateCold' ev $ return cxt                     updateHot ev-                        = do cxt <- atomically . addTxId tx =<< newTxContext-                             updateCold' cxt ev+                        = do cxt <- newTxContext+                             updateCold' ev $ addTxId tx cxt                 in UpdateHandler updateCold updateHot parseObject           eventHandler tx (Query fn)               = let queryEmitter ev@@ -261,6 +262,10 @@ setNewState stateType state     = emitEvent' (setNewStateType stateType) (SetCheckpointState state) +setNewState' :: EventMap -> TypeString -> L.ByteString -> IO ()+setNewState' eventMap stateType state+    = emitFunc eventMap (setNewStateType stateType) (SetCheckpointState state)+ getState :: TypeString -> IO L.ByteString getState stateType     = emitEvent' (getStateType stateType) GetCheckpointState@@ -346,19 +351,19 @@                -> do res <- runHot (parse obj)                      return $ mkObject res -runColdEvent :: TxContext -> Object -> IO ()+runColdEvent :: TxContext -> Object -> IO Object runColdEvent cxt obj     = do EmitInternal eventMap <- readIORef emitRef          runColdEventFunc cxt obj eventMap -runColdEventFunc :: TxContext -> Object -> EventMap -> IO ()+runColdEventFunc :: TxContext -> Object -> EventMap -> IO Object runColdEventFunc cxt obj eventMap     = do handler <- lookupEventHandler (objectType obj) eventMap          case handler of            QueryHandler{} -> error $ "Cold event was a query: " ++ objectType obj            UpdateHandler runCold _runHotObj parse-               -> do runCold cxt (parse obj)-                     return ()+               -> do res <- runCold cxt (parse obj)+                     return $ mkObject res  lookupEventHandler :: TypeString -> EventMap -> IO EventHandler lookupEventHandler eventType eventMap@@ -427,14 +432,6 @@ globalRandomGen = unsafePerformIO (newMVar =<< getStdGen)  -data TxConfig = TxConfig-    { txcCheckpointSeconds   :: Seconds,   -- ^ Perform checkpoint at least every N seconds.-      txcOperationMode       :: OperationMode,-      txcClusterSize         :: Int,       -- ^ Number of active nodes in the cluster (not counting this node).-      txcClusterPort         :: Int,       ---      txcCommitFrequency     :: Int        -- ^ Commits per second. Only applies to cluster mode.-    }- data TxControl = TxControl     { ctlSaver      :: Saver           -- ^ Saver given by the user.     , ctlEventSaver :: MVar (WriterStream EventLogEntry)@@ -442,6 +439,7 @@     , ctlComponentVersions :: M.Map String [L.ByteString] -- ^ Map listing all versions of a component     , ctlChildren   :: [(ThreadId, MVar ())] --      , ctlPrefixLock :: Maybe PrefixLock -- ^ Stores exclusive prefix lock (implemented in filesystem)+    , ctlCreateCheckpoint :: IO ()     }  data EventLogEntry = EventLogEntry TxContext Object deriving (Typeable, Show)@@ -452,18 +450,7 @@               do (context, obj) <- safeGet                  return $ EventLogEntry context obj -data OperationMode-    = SingleMode-    | ClusterMode String -nullTxConfig :: TxConfig-nullTxConfig = TxConfig { txcCheckpointSeconds   = 60*60*24,-                          txcOperationMode       = SingleMode,-                          txcClusterSize         = 0,-                          txcClusterPort         = 8500,-                          txcCommitFrequency     = 50-                        }- runTxLoop :: MVar (WriterStream EventLogEntry) -> ProcessQueue st -> st -> IO () runTxLoop eventSaverVar queue st0 =   let loop st = do@@ -497,6 +484,10 @@                            writerAdd eventSaver (EventLogEntry context (mkObject ev)) (logMT NOTICE "> disk " >> ra)                            loop st'   in do forkIO $ (loop st0) `catch` (\ThreadKilled -> return ())+#if MIN_VERSION_base(4,2,0)+                            `catch` (\BlockedIndefinitelyOnSTM -> return ())+#else                             `catch` (\BlockedIndefinitely -> return ())+#endif         return () 

@@ -1,7 +1,9 @@+{-# LANGUAGE CPP #-} module Happstack.State.TxControl     ( runTxSystem-    , runTxSystem'+    , runTxSystemAmazon     , shutdownSystem+    , createCheckpoint     ) where  import System.Log.Logger@@ -10,24 +12,55 @@ import Control.Exception import Control.Concurrent -import Happstack.State.Checkpoint+import qualified Happstack.State.Checkpoint as Checkpoint import Happstack.State.Saver import Happstack.State.Transaction-import Happstack.State.Spread++#ifdef REPLICATION+import qualified Happstack.State.CentralLogServer as LogServer+#endif+ import Happstack.State.ComponentSystem import Happstack.Data.Proxy  logMM :: Priority -> String -> IO () logMM = logM "Happstack.State.TxControl" ++-- | Given a Saver and a Proxy, createTxControl will +-- initialize a TxControl.  This does not actually start the+-- state system.+createTxControl :: (Methods state, Component state) =>+                   Saver -> Proxy state -> IO (MVar TxControl)+createTxControl saver prox+    = do ++         -- The state hasn't been loaded yet. Ignore events.+         eventSaverVar   <- newMVar =<< createWriter NullSaver "events" 0+         -- obtain a prefix lock+         lock <- obtainLock saver+         newMVar $ TxControl+                       { ctlSaver             = saver+                       , ctlEventSaver        = eventSaverVar+                       , ctlAllComponents     = allStateTypes prox+                       , ctlComponentVersions = componentVersions prox+                       , ctlChildren          = []+                       , ctlPrefixLock        = lock+                       , ctlCreateCheckpoint  = return () }+++-- | Saves the state and closes the serialization+closeTxControl :: MVar TxControl -> IO ()+closeTxControl ctlVar+    = do ctl <- takeMVar ctlVar+         writerClose =<< takeMVar (ctlEventSaver ctl)+         releaseLock (ctlPrefixLock ctl)+++ -- | Run the MACID system without multimaster support and with the given Saver. runTxSystem :: (Methods st, Component st) => Saver -> Proxy st -> IO (MVar TxControl)-runTxSystem = runTxSystem' False---- | Run the MACID system with multimaster support turned on if the first--- argument is True.-runTxSystem' :: (Methods st, Component st) => Bool -> Saver -> Proxy st -> IO (MVar TxControl)-runTxSystem' withMultimaster saver stateProxy =+runTxSystem saver stateProxy =     do logMM NOTICE "Initializing system control."        ctl <- createTxControl saver stateProxy        -- insert code to lock based on the saver@@ -35,26 +68,55 @@        localEventMap <- createEventMap ctl stateProxy        setNewEventMap localEventMap        logMM NOTICE "Restoring state."-       enableLogging <- restoreState ctl-       when withMultimaster-                $ do logMM NOTICE "Multimaster mode"-                     cluster <- connectToCluster-                     eventMap <- changeEventMapping ctl localEventMap cluster-                     setNewEventMap eventMap+       enableLogging <- Checkpoint.restoreState ctl+       -- Multimaster support used to be here. --        enableLogging        let ioActions = componentIO stateProxy        logMM NOTICE "Forking children."        children <- forM ioActions $ \action -> do mv <- newEmptyMVar                                                   tid <- forkIO (action `finally` putMVar mv ())                                                   return (tid,mv)-       modifyMVar_ ctl $ \c -> return c{ctlChildren = children}+       modifyMVar_ ctl $ \c -> return c{ ctlChildren = children+                                       , ctlCreateCheckpoint = Checkpoint.createCheckpoint ctl }        return ctl +#ifdef REPLICATION+runTxSystemAmazon :: (Methods st, Component st) => LogServer.ApplicationName -> Proxy st -> IO (MVar TxControl)+runTxSystemAmazon appName stateProxy+    = do logMM NOTICE "Initializing system control"+         ctl <- createTxControl NullSaver stateProxy+         logMM NOTICE "Creating local event mapper."+         localEventMap <- createEventMap ctl stateProxy+         logMM NOTICE "Connecting to central log server."+         cluster <- LogServer.connectToCluster appName localEventMap+         logMM NOTICE "Modifying local event map."+         eventMap <- LogServer.changeEventMapping localEventMap cluster+         setNewEventMap eventMap+         let ioActions = componentIO stateProxy+         logMM NOTICE "Forking children."+         children <- forM ioActions $ \action -> do mv <- newEmptyMVar+                                                    tid <- forkIO (action `finally` putMVar mv ())+                                                    return (tid,mv)+         modifyMVar_ ctl $ \c -> return c{ ctlChildren = children+                                         , ctlCreateCheckpoint = LogServer.createCheckpoint ctl cluster }+         return ctl+#else+type ApplicationName = String -- Hm, this should actually be defined in CentralLogServer.hs++runTxSystemAmazon :: (Methods st, Component st) => ApplicationName -> Proxy st -> IO (MVar TxControl)+runTxSystemAmazon appName stateProxy+    = error "Happstack-state has been built without replication support."+#endif+++createCheckpoint :: MVar TxControl -> IO ()+createCheckpoint+    = join . fmap ctlCreateCheckpoint . readMVar+ -- | Shuts down a transaction system shutdownSystem :: MVar TxControl -> IO () shutdownSystem ctl     = do logMM NOTICE "Shutting down."-         saver <- liftM ctlSaver $ readMVar ctl          children <- liftM ctlChildren $ readMVar ctl          logMM NOTICE "Killing children."          mapM_ (killThread . fst) children

@@ -1,7 +1,7 @@ {-# LANGUAGE UndecidableInstances, TemplateHaskell,     ExistentialQuantification, RankNTypes,     FlexibleInstances, DeriveDataTypeable,-    ScopedTypeVariables, PatternSignatures+    ScopedTypeVariables  #-}