diff --git a/cbits/hs_zk.c b/cbits/hs_zk.c --- a/cbits/hs_zk.c +++ b/cbits/hs_zk.c @@ -12,7 +12,6 @@ watcher_ctx->state = state; watcher_ctx->path = strdup(path); hs_try_putmvar(watcher_ctx->cap, watcher_ctx->mvar); - hs_thread_done(); } /** @@ -40,7 +39,6 @@ string_completion->value = strdup(value); } hs_try_putmvar(string_completion->cap, string_completion->mvar); - hs_thread_done(); } /** @@ -79,7 +77,6 @@ data_completion->stat = dup_stat(stat); } hs_try_putmvar(data_completion->cap, data_completion->mvar); - hs_thread_done(); } /** @@ -109,7 +106,6 @@ stat_completion->stat = dup_stat(stat); } hs_try_putmvar(stat_completion->cap, stat_completion->mvar); - hs_thread_done(); } /** @@ -133,7 +129,6 @@ hs_void_completion_t* void_completion = (hs_void_completion_t*)data; void_completion->rc = rc; hs_try_putmvar(void_completion->cap, void_completion->mvar); - hs_thread_done(); } /** @@ -165,7 +160,6 @@ strings_completion->strings = dup_string_vector(strings); } hs_try_putmvar(strings_completion->cap, strings_completion->mvar); - hs_thread_done(); } /** @@ -204,7 +198,6 @@ strings_stat->stat = dup_stat(stat); } hs_try_putmvar(strings_stat->cap, strings_stat->mvar); - hs_thread_done(); } /** @@ -239,7 +232,6 @@ acl_completion->stat = dup_stat(stat); } hs_try_putmvar(acl_completion->cap, acl_completion->mvar); - hs_thread_done(); } // ---------------------------------------------------------------------------- diff --git a/src/ZooKeeper/Recipe/Election.hs b/src/ZooKeeper/Recipe/Election.hs new file mode 100644 --- /dev/null +++ b/src/ZooKeeper/Recipe/Election.hs @@ -0,0 +1,80 @@ +module ZooKeeper.Recipe.Election where + +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.Recipe.Utils (SequenceNumWithGUID (..), + createSeqEphemeralZNode, + mkSequenceNumWithGUID) +import ZooKeeper.Types + +-- | Run a leader election process. +-- __IMPORTANT__: This function may run endlessly until it is selected +-- as the leader. +election :: ZHandle + -- ^ The zookeeper handle obtained by a call to 'zookeeperResInit' + -> CBytes + -- ^ The path to start the election from. Ephemeral znodes will be + -- put on it + -> CBytes + -- ^ The GUID for this zookeeper session. To handle recoverable execptions + -- correctly, it should be distinct from different sessions. + -> IO () + -- ^ The action to be executed when an leader is elected. + -> (DataCompletion -> IO ()) + -- ^ The action to be executed when a watcher is set. It can be used to + -- remind the user that one 'step' is finished. + -> IO () +-- TODO: Use user-configurable logger instead +election zk electionPath guid leaderApp watchSetApp = Log.withDefaultLogger $ do + let electionSeqPath = electionPath <> "/" <> guid <> "_" + + -- Check persistent paths + electionExists <- zooExists zk electionPath + case electionExists of + Just _ -> return () + Nothing -> void $ zooCreate zk electionPath Nothing zooOpenAclUnsafe ZooPersistent + + -- 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 + + -- 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 + + -- find max j < i + case filter (< thisSeqNumWithGUID) childrenSeqNumWithGUID of + [] -> do + let smallest = minimum childrenSeqNumWithGUID + 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 + -- add watch + zooWatchGet zk toWatch (callback electionSeqPath thisSeqNumWithGUID) watchSetApp + where + callback electionSeqPath thisSeqNumWithGUID HsWatcherCtx{..} = do + 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 diff --git a/src/ZooKeeper/Recipe/Utils.hs b/src/ZooKeeper/Recipe/Utils.hs new file mode 100644 --- /dev/null +++ b/src/ZooKeeper/Recipe/Utils.hs @@ -0,0 +1,67 @@ +module ZooKeeper.Recipe.Utils + ( -- * Types + SequenceNumWithGUID(..) + , mkSequenceNumWithGUID + , extractSeqNum + + -- * ZNode operations + , createSeqEphemeralZNode + ) where + +import Control.Exception +import qualified Data.List as L +import Z.Data.CBytes (CBytes) +import qualified Z.Data.CBytes as CB +import ZooKeeper +import ZooKeeper.Exception +import ZooKeeper.Types + +-------------------------------------------------------------------------------- + +-- | Represenets a name of a SEQUENCE|EPHEMERAL znode. It contains two parts, +-- a GUID, and a sequence number. The GUID is used for handleing recoverable +-- exceptions so we only care about the sequence number part when we comparing +-- two of them. +newtype SequenceNumWithGUID = SequenceNumWithGUID + { unSequenceNumWithGUID :: CBytes + } + +mkSequenceNumWithGUID :: CBytes -> SequenceNumWithGUID +mkSequenceNumWithGUID = SequenceNumWithGUID + +instance Eq SequenceNumWithGUID where + (SequenceNumWithGUID s1) == (SequenceNumWithGUID s2) = + extractSeqNum s1 == extractSeqNum s2 + +instance Ord SequenceNumWithGUID where + (SequenceNumWithGUID s1) <= (SequenceNumWithGUID s2) = + extractSeqNum s1 <= extractSeqNum s2 + +instance Show SequenceNumWithGUID where + show (SequenceNumWithGUID s) = CB.unpack s + +-- | Exrtact only the sequence number part from an `SequenceNumWithGUID`. +extractSeqNum :: CBytes -> CBytes +extractSeqNum = CB.pack . reverse . takeWhile (/= '_') . reverse . CB.unpack + +-------------------------------------------------------------------------------- + +-- | Creates a sequential and ephemeral znode with specified prefix +-- and GUID. The created znode is as `prefixPath/GUID-n_0000000001`. +-- Note that it uses a GUID to handle recoverable exceptions, see +-- [this](https://zookeeper.apache.org/doc/r3.7.0/recipes.html#sc_recipes_GuidNote) +-- for more details. +createSeqEphemeralZNode :: ZHandle -> CBytes -> CBytes -> IO StringCompletion +createSeqEphemeralZNode zk prefixPath guid = do + let seqPath = prefixPath <> "/" <> guid <> "_" + catches (zooCreate zk seqPath Nothing zooOpenAclUnsafe ZooEphemeralSequential) + [ Handler (\(_ :: ZCONNECTIONLOSS ) -> retry) + , Handler (\(_ :: ZOPERATIONTIMEOUT) -> retry) + ] + where + retry :: IO StringCompletion + retry = do + (StringsCompletion (StringVector children)) <- zooGetChildren zk prefixPath + case L.find (\child -> CB.unpack guid `L.isSubsequenceOf` CB.unpack child) children of + Just child -> return $ StringCompletion child + Nothing -> createSeqEphemeralZNode zk prefixPath guid diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -3,12 +3,17 @@ module Main where import Control.Concurrent -import Control.Monad (void) -import Data.Version (makeVersion) +import Control.Monad (replicateM, void) +import qualified Data.UUID as UUID +import Data.UUID.V4 +import Data.Version (makeVersion) import Foreign.C import Test.Hspec +import qualified Z.Data.CBytes as CB + import ZooKeeper import ZooKeeper.Exception +import ZooKeeper.Recipe.Election (election) import ZooKeeper.Types recvTimeout :: CInt @@ -22,6 +27,8 @@ hspec $ opSpec zh hspec $ multiOpSpec zh hspec $ propSpec zh + hspec $ electionSpec1 zh + hspec $ electionSpec2 client opSpec :: ZHandle -> Spec opSpec zh = do @@ -127,6 +134,46 @@ head results' `shouldBe` ZooDeleteOpResult CZOK results' !! 1 `shouldBe` ZooDeleteOpResult CZOK zooExists zh "/multi" `shouldReturn` Nothing + +electionSpec1 :: ZHandle -> Spec +electionSpec1 zh = describe "ZooKeeper.zooElection (single session)" $ do + it "Test single node situation" $ do + leader <- newEmptyMVar + uuid <- nextRandom + _ <- forkIO $ election zh "/election1" (CB.pack . UUID.toString $ uuid) + (putMVar leader 1) (\_ -> return ()) + readMVar leader `shouldReturn` (1 :: Int) + +electionSpec2 :: Resource ZHandle -> Spec +electionSpec2 client_ = describe "ZooKeeper.zooElection (multi sessions)" $ do + it "Test multi node situation" $ do + leader <- newMVar 0 + [finished1, finished2, finished3, finished4] <- replicateM 4 newEmptyMVar + [uuid1, uuid2, uuid3] <- replicateM 3 nextRandom + + t1 <- forkIO $ withResource client_ $ \zh1 -> do + t <- election zh1 "/election2" (CB.pack . UUID.toString $ uuid1) + (do {void $ swapMVar leader 1; putMVar finished1 ()}) + (\_ -> putMVar finished1 ()) + threadDelay (maxBound :: Int) + return t + void $ takeMVar finished1 + + _ <- forkIO $ withResource client_ $ \zh2 -> do + election zh2 "/election2" (CB.pack . UUID.toString $ uuid2) + (do {void $ swapMVar leader 2; putMVar finished4 ()}) + (\_ -> putMVar finished2 ()) + void $ takeMVar finished2 + + _ <- forkIO $ withResource client_ $ \zh3 -> do + election zh3 "/election2" (CB.pack . UUID.toString $ uuid3) + (void $ swapMVar leader 3) + (\_ -> putMVar finished3 ()) + void $ takeMVar finished3 + + killThread t1 + takeMVar finished4 + readMVar leader `shouldReturn` (2 :: Int) ------------------------------------------------------------------------------- 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.1.3.1 +version: 0.1.4.0 synopsis: A haskell binding to Apache Zookeeper C library(mt) using Haskell Z project. @@ -32,11 +32,13 @@ exposed-modules: ZooKeeper ZooKeeper.Exception + ZooKeeper.Recipe.Election ZooKeeper.Types other-modules: ZooKeeper.Internal.FFI ZooKeeper.Internal.Types + ZooKeeper.Recipe.Utils build-depends: , base >=4.12 && <5 @@ -80,6 +82,8 @@ build-depends: , base >=4.12 && <5 , hspec + , uuid + , Z-Data , zoovisitor default-language: Haskell2010