diff --git a/src/ZooKeeper.hs b/src/ZooKeeper.hs --- a/src/ZooKeeper.hs +++ b/src/ZooKeeper.hs @@ -3,8 +3,8 @@ , I.zooSetDebugLevel , zookeeperResInit - , Res.withResource - , Res.Resource + , U.withResource + , U.Resource , zooCreate , zooSet @@ -47,12 +47,11 @@ import Z.Data.CBytes (CBytes) import Z.Data.Vector (Bytes) import qualified Z.Foreign as Z -import qualified Z.IO.FileSystem.FilePath as ZF -import qualified Z.IO.Resource as Res import qualified ZooKeeper.Exception as E import qualified ZooKeeper.Internal.FFI as I import qualified ZooKeeper.Internal.Types as I +import qualified ZooKeeper.Internal.Utils as U import qualified ZooKeeper.Types as T ------------------------------------------------------------------------------- @@ -80,9 +79,9 @@ -- state will indicate the reason for failure (typically 'T.ZooExpiredSession'). -> CInt -- ^ flags, reserved for future use. Should be set to zero. - -> Res.Resource T.ZHandle + -> U.Resource T.ZHandle zookeeperResInit host fn timeout mclientid flags = fst <$> - Res.initResource (zookeeperInit host fn timeout mclientid flags) zookeeperClose + U.initResource (zookeeperInit host fn timeout mclientid flags) zookeeperClose -- | Create a node. -- @@ -260,11 +259,14 @@ -- | Delete a node and all its children in zookeeper. -- --- If fail will throw exceptions, check `zooDeleteAll` and `zooGetChildren` for more details +-- NOTE: the path should not have a trailing '/' +-- +-- If fails, this will throw exceptions, check `zooDeleteAll` and `zooGetChildren` +-- for more details. zooDeleteAll :: HasCallStack => T.ZHandle -> CBytes -> IO () zooDeleteAll zh path = do T.StringsCompletion (T.StringVector children) <- zooGetChildren zh path - mapM_ (zooDeleteAll zh <=< ZF.join path) children + mapM_ (zooDeleteAll zh . ((path <> "/") <>)) children zooDelete zh path Nothing -- | Checks the existence of a node in zookeeper. diff --git a/src/ZooKeeper/Internal/Utils.hs b/src/ZooKeeper/Internal/Utils.hs new file mode 100644 --- /dev/null +++ b/src/ZooKeeper/Internal/Utils.hs @@ -0,0 +1,114 @@ +module ZooKeeper.Internal.Utils + ( -- * Resource management + Resource(..) + , initResource + , initResource_ + , withResource + , withResource' + ) where + +import Control.Exception (onException) +import Control.Monad (when) +import qualified Control.Monad.Catch as MonadCatch +import Control.Monad.IO.Class (MonadIO, liftIO) +import GHC.Stack (HasCallStack) +import Z.Data.PrimRef (atomicOrCounter, newCounter) + +-------------------------------------------------------------------------------- + +-- | A 'Resource' is an 'IO' action which acquires some resource of type a and +-- also returns a finalizer of type IO () that releases the resource. +-- +-- The only safe way to use a 'Resource' is 'withResource' and 'withResource'', +-- You should not use the 'acquire' field directly, unless you want to implement +-- your own resource management. In the later case, you should 'mask_' 'acquire' +-- since some resource initializations may assume async exceptions are masked. +-- +-- 'MonadIO' instance is provided so that you can lift 'IO' computation inside +-- 'Resource', this is convenient for propagating 'Resource' around since many +-- 'IO' computations carry finalizers. +newtype Resource a = Resource { acquire :: IO (a, IO ()) } + +-- | Create 'Resource' from create and release action. +-- +-- Note, 'resource' doesn't open resource itself, resource is created when you +-- use 'with' \/ 'with''. +initResource :: IO a -> (a -> IO ()) -> Resource a +{-# INLINABLE initResource #-} +initResource create release = Resource $ do + r <- create + return (r, release r) + +-- | Create 'Resource' from create and release action. +-- +-- This function is useful when you want to add some initialization and clean +-- up action inside 'Resource' monad. +initResource_ :: IO a -> IO () -> Resource a +{-# INLINABLE initResource_ #-} +initResource_ create release = Resource $ do + r <- create + return (r, release) + +instance Functor Resource where + {-# INLINE fmap #-} + fmap f resource = Resource $ do + (a, release) <- acquire resource + return (f a, release) + +instance Applicative Resource where + {-# INLINE pure #-} + pure a = Resource (pure (a, pure ())) + {-# INLINE (<*>) #-} + resource1 <*> resource2 = Resource $ do + (f, release1) <- acquire resource1 + (x, release2) <- acquire resource2 `onException` release1 + return (f x, release2 >> release1) + +instance Monad Resource where + {-# INLINE return #-} + return = pure + {-# INLINE (>>=) #-} + m >>= f = Resource $ do + (m', release1) <- acquire m + (x , release2) <- acquire (f m') `onException` release1 + return (x, release2 >> release1) + +instance MonadIO Resource where + {-# INLINE liftIO #-} + liftIO f = Resource $ fmap (\ a -> (a, dummyRelease)) f + where dummyRelease = return () + +-- | Create a new resource and run some computation, resource is guarantee to +-- be closed. +-- +-- Be careful, don't leak the resource through the computation return value +-- because after the computation finishes, the resource is already closed. +withResource :: (MonadCatch.MonadMask m, MonadIO m, HasCallStack) + => Resource a -> (HasCallStack => a -> m b) -> m b +{-# INLINABLE withResource #-} +withResource resource k = MonadCatch.bracket + (liftIO (acquire resource)) + (\(_, release) -> liftIO release) + (\(a, _) -> k a) + +-- | Create a new resource and run some computation, resource is guarantee to +-- be closed. +-- +-- The difference from 'with' is that the computation will receive an extra +-- close action, which can be used to close the resource early before the whole +-- computation finished, the close action can be called multiple times, +-- only the first call will clean up the resource. +withResource' :: (MonadCatch.MonadMask m, MonadIO m, HasCallStack) + => Resource a -> (HasCallStack => a -> m () -> m b) -> m b +{-# INLINABLE withResource' #-} +withResource' resource k = do + c <- liftIO (newCounter 0) + MonadCatch.bracket + (liftIO $ do + (a, release) <- acquire resource + let release' = do + c' <- atomicOrCounter c 1 + when (c' == 0) release + return (a, release')) + (\(_, release) -> liftIO release) + (\(a, release) -> k a (liftIO release)) diff --git a/src/ZooKeeper/Recipe/Election.hs b/src/ZooKeeper/Recipe/Election.hs --- a/src/ZooKeeper/Recipe/Election.hs +++ b/src/ZooKeeper/Recipe/Election.hs @@ -2,15 +2,14 @@ ( election ) where -import Control.Exception (catch) +import Control.Exception (catch, throwIO) import Control.Monad -import qualified Z.Data.Builder as B import Z.Data.CBytes (CBytes) -import qualified Z.IO.Logger as Log import ZooKeeper import ZooKeeper.Exception (ZNODEEXISTS) -import ZooKeeper.Recipe.Utils (SequenceNumWithGUID (..), +import ZooKeeper.Recipe.Utils (ZkRecipeException (..), + SequenceNumWithGUID (..), createSeqEphemeralZNode, mkSequenceNumWithGUID) import ZooKeeper.Types @@ -33,7 +32,7 @@ -- remind the user that one 'step' is finished. -> IO () -- TODO: Use user-configurable logger instead -election zk electionPath guid leaderApp watchSetApp = Log.withDefaultLogger $ do +election zk electionPath guid leaderApp watchSetApp = do let electionSeqPath = electionPath <> "/" <> guid <> "_" -- Check persistent paths @@ -46,40 +45,46 @@ -- Create Ephemeral and Sequece znode, and get the seq number i (StringCompletion this) <- createSeqEphemeralZNode zk electionPath guid let thisSeqNumWithGUID = mkSequenceNumWithGUID this - Log.debug . B.stringUTF8 $ "Created SEQUENTIAL|EPHEMERAL ZNode " <> show thisSeqNumWithGUID + -- TODO: Use zookeeper log + -- Log.debug . B.stringUTF8 $ "Created SEQUENTIAL|EPHEMERAL ZNode " <> show thisSeqNumWithGUID -- Get the child that has the max seq number j < i (StringsCompletion (StringVector children)) <- zooGetChildren zk electionPath let childrenSeqNumWithGUID = mkSequenceNumWithGUID <$> children - Log.debug . B.stringUTF8 $ "Children now: " <> show childrenSeqNumWithGUID + -- TODO: Use zookeeper log + -- Log.debug . B.stringUTF8 $ "Children now: " <> show childrenSeqNumWithGUID -- find max j < i case filter (< thisSeqNumWithGUID) childrenSeqNumWithGUID of [] -> do - let smallest = minimum childrenSeqNumWithGUID - Log.debug . B.stringUTF8 $ "Leader elected: " <> show smallest + let _smallest = minimum childrenSeqNumWithGUID + -- TODO: Use zookeeper log + -- Log.debug . B.stringUTF8 $ "Leader elected: " <> show smallest leaderApp xs -> do let toWatch = electionPath <> "/" <> unSequenceNumWithGUID (maximum xs) - Log.debug . B.stringUTF8 $ "Now watching: " <> show toWatch + -- TODO: Use zookeeper log + -- Log.debug . B.stringUTF8 $ "Now watching: " <> show toWatch -- add watch zooWatchGet zk toWatch (callback electionSeqPath thisSeqNumWithGUID) watchSetApp where callback electionSeqPath thisSeqNumWithGUID HsWatcherCtx{..} = do - Log.debug . B.stringUTF8 $ "Watch triggered, some node failed." + -- TODO: Use zookeeper log + --Log.debug . B.stringUTF8 $ "Watch triggered, some node failed." (StringsCompletion (StringVector children)) <- zooGetChildren watcherCtxZHandle electionPath let childrenSeqNumWithGUID = mkSequenceNumWithGUID <$> children let smallest = minimum childrenSeqNumWithGUID - case smallest == thisSeqNumWithGUID of - True -> do - Log.debug . B.stringUTF8 $ "Leader elected: " <> show smallest - leaderApp - False -> do - -- find max j < i - case filter (< thisSeqNumWithGUID) childrenSeqNumWithGUID of - [] -> Log.fatal . B.stringUTF8 $ "The 'impossible' happened!" - xs -> do - let toWatch = electionPath <> "/" <> unSequenceNumWithGUID (maximum xs ) - Log.debug . B.stringUTF8 $ "Now watching: " <> show toWatch - -- add watch - zooWatchGet zk toWatch (callback electionSeqPath thisSeqNumWithGUID) watchSetApp + if smallest == thisSeqNumWithGUID + then do + -- TODO: Use zookeeper log + --Log.debug . B.stringUTF8 $ "Leader elected: " <> show smallest + leaderApp + else do + -- find max j < i + case filter (< thisSeqNumWithGUID) childrenSeqNumWithGUID of + [] -> throwIO $ ZkRecipeException "The 'impossible' happened!" + xs -> do + let toWatch = electionPath <> "/" <> unSequenceNumWithGUID (maximum xs ) + -- Log.debug . B.stringUTF8 $ "Now watching: " <> show toWatch + -- add watch + zooWatchGet zk toWatch (callback electionSeqPath thisSeqNumWithGUID) watchSetApp diff --git a/src/ZooKeeper/Recipe/Utils.hs b/src/ZooKeeper/Recipe/Utils.hs --- a/src/ZooKeeper/Recipe/Utils.hs +++ b/src/ZooKeeper/Recipe/Utils.hs @@ -3,9 +3,11 @@ SequenceNumWithGUID(..) , mkSequenceNumWithGUID , extractSeqNum + , ZkRecipeException (..) -- * ZNode operations , createSeqEphemeralZNode + ) where import Control.Exception @@ -65,3 +67,9 @@ case L.find (\child -> CB.unpack guid `L.isSubsequenceOf` CB.unpack child) children of Just child -> return $ StringCompletion child Nothing -> createSeqEphemeralZNode zk prefixPath guid + +-------------------------------------------------------------------------------- + +newtype ZkRecipeException = ZkRecipeException String + deriving (Show, Eq) +instance Exception ZkRecipeException diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -36,9 +36,9 @@ opSpec :: ZHandle -> Spec opSpec zh = do describe "ZooKeeper.zooVersion" $ do - it "version should be 3.4.* - 3.6.*" $ do + it "version should be 3.4.* - 3.8.*" $ do zooVersion `shouldSatisfy` (>= makeVersion [3, 4, 0]) - zooVersion `shouldSatisfy` (< makeVersion [3, 8, 0]) + zooVersion `shouldSatisfy` (<= makeVersion [3, 8, 0]) describe "zookeeper get set" $ do it "set some value to a node and get it" $ do diff --git a/zoovisitor.cabal b/zoovisitor.cabal --- a/zoovisitor.cabal +++ b/zoovisitor.cabal @@ -1,6 +1,6 @@ cabal-version: 2.2 name: zoovisitor -version: 0.2.1.2 +version: 0.2.3.0 synopsis: A haskell binding to Apache Zookeeper C library(mt) using Haskell Z project. @@ -38,14 +38,15 @@ other-modules: ZooKeeper.Internal.FFI ZooKeeper.Internal.Types + ZooKeeper.Internal.Utils ZooKeeper.Recipe.Election ZooKeeper.Recipe.Lock ZooKeeper.Recipe.Utils build-depends: - , base >=4.12 && <5 - , Z-Data >=0.7.2 && <1.2 || ^>=2.0 - , Z-IO >=0.7 && <1.1 || ^>=2.0 + , base >=4.12 && <5 + , exceptions ^>=0.10 + , Z-Data >=0.7.2 && <1.2 || ^>=2.0 includes: hs_zk.h c-sources: cbits/hs_zk.c