packages feed

zoovisitor 0.1.3.1 → 0.1.4.0

raw patch · 5 files changed

+201/−11 lines, 5 filesdep +uuiddep ~Z-DataPVP ok

version bump matches the API change (PVP)

Dependencies added: uuid

Dependency ranges changed: Z-Data

API changes (from Hackage documentation)

+ ZooKeeper.Recipe.Election: election :: ZHandle -> CBytes -> CBytes -> IO () -> (DataCompletion -> IO ()) -> IO ()

Files

cbits/hs_zk.c view
@@ -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(); }  // ----------------------------------------------------------------------------
+ src/ZooKeeper/Recipe/Election.hs view
@@ -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
+ src/ZooKeeper/Recipe/Utils.hs view
@@ -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
test/Spec.hs view
@@ -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)  -------------------------------------------------------------------------------
zoovisitor.cabal view
@@ -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