123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767 |
- /**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
- #include <cppunit/extensions/HelperMacros.h>
- #include "CppAssertHelper.h"
- #include <signal.h>
- #include <stdlib.h>
- #include <unistd.h>
- #include <sys/select.h>
- #include "CollectionUtil.h"
- #include "ThreadingUtil.h"
- using namespace Util;
- #include "Vector.h"
- using namespace std;
- #include <cstring>
- #include <list>
- #include <zookeeper.h>
- #include <errno.h>
- #include <recordio.h>
- #include "Util.h"
- #ifdef THREADED
- static void yield(zhandle_t *zh, int i)
- {
- sleep(i);
- }
- #else
- static void yield(zhandle_t *zh, int seconds)
- {
- int fd;
- int interest;
- int events;
- struct timeval tv;
- time_t expires = time(0) + seconds;
- time_t timeLeft = seconds;
- fd_set rfds, wfds, efds;
- FD_ZERO(&rfds);
- FD_ZERO(&wfds);
- FD_ZERO(&efds);
- while(timeLeft >= 0) {
- zookeeper_interest(zh, &fd, &interest, &tv);
- if (fd != -1) {
- if (interest&ZOOKEEPER_READ) {
- FD_SET(fd, &rfds);
- } else {
- FD_CLR(fd, &rfds);
- }
- if (interest&ZOOKEEPER_WRITE) {
- FD_SET(fd, &wfds);
- } else {
- FD_CLR(fd, &wfds);
- }
- } else {
- fd = 0;
- }
- FD_SET(0, &rfds);
- if (tv.tv_sec > timeLeft) {
- tv.tv_sec = timeLeft;
- }
- select(fd+1, &rfds, &wfds, &efds, &tv);
- timeLeft = expires - time(0);
- events = 0;
- if (FD_ISSET(fd, &rfds)) {
- events |= ZOOKEEPER_READ;
- }
- if (FD_ISSET(fd, &wfds)) {
- events |= ZOOKEEPER_WRITE;
- }
- zookeeper_process(zh, events);
- }
- }
- #endif
- typedef struct evt {
- string path;
- int type;
- } evt_t;
- typedef struct watchCtx {
- private:
- list<evt_t> events;
- watchCtx(const watchCtx&);
- watchCtx& operator=(const watchCtx&);
- public:
- bool connected;
- zhandle_t *zh;
- Mutex mutex;
- watchCtx() {
- connected = false;
- zh = 0;
- }
- ~watchCtx() {
- if (zh) {
- zookeeper_close(zh);
- zh = 0;
- }
- }
- evt_t getEvent() {
- evt_t evt;
- mutex.acquire();
- CPPUNIT_ASSERT( events.size() > 0);
- evt = events.front();
- events.pop_front();
- mutex.release();
- return evt;
- }
- int countEvents() {
- int count;
- mutex.acquire();
- count = events.size();
- mutex.release();
- return count;
- }
- void putEvent(evt_t evt) {
- mutex.acquire();
- events.push_back(evt);
- mutex.release();
- }
- bool waitForConnected(zhandle_t *zh) {
- time_t expires = time(0) + 10;
- while(!connected && time(0) < expires) {
- yield(zh, 1);
- }
- return connected;
- }
- bool waitForDisconnected(zhandle_t *zh) {
- time_t expires = time(0) + 15;
- while(connected && time(0) < expires) {
- yield(zh, 1);
- }
- return !connected;
- }
- } watchctx_t;
- #ifdef THREADED
- class Zookeeper_multi : public CPPUNIT_NS::TestFixture
- {
- CPPUNIT_TEST_SUITE(Zookeeper_multi);
- //FIXME: None of these tests pass in single-threaded mode. It seems to be a
- //flaw in the test suite setup.
- CPPUNIT_TEST(testCreate);
- CPPUNIT_TEST(testCreateDelete);
- CPPUNIT_TEST(testInvalidVersion);
- CPPUNIT_TEST(testNestedCreate);
- CPPUNIT_TEST(testSetData);
- CPPUNIT_TEST(testUpdateConflict);
- CPPUNIT_TEST(testDeleteUpdateConflict);
- CPPUNIT_TEST(testAsyncMulti);
- CPPUNIT_TEST(testMultiFail);
- CPPUNIT_TEST(testCheck);
- CPPUNIT_TEST(testWatch);
- CPPUNIT_TEST(testSequentialNodeCreateInAsyncMulti);
- CPPUNIT_TEST(testBigAsyncMulti);
- CPPUNIT_TEST_SUITE_END();
- static void watcher(zhandle_t *, int type, int state, const char *path,void*v){
- watchctx_t *ctx = (watchctx_t*)v;
- if (state == ZOO_CONNECTED_STATE) {
- ctx->connected = true;
- } else {
- ctx->connected = false;
- }
- if (type != ZOO_SESSION_EVENT) {
- evt_t evt;
- evt.path = path;
- evt.type = type;
- ctx->putEvent(evt);
- }
- }
- static const char hostPorts[];
- const char *getHostPorts() {
- return hostPorts;
- }
-
- zhandle_t *createClient(watchctx_t *ctx) {
- return createClient(hostPorts, ctx);
- }
- zhandle_t *createClient(const char *hp, watchctx_t *ctx) {
- zhandle_t *zk = zookeeper_init(hp, watcher, 10000, 0, ctx, 0);
- ctx->zh = zk;
- CPPUNIT_ASSERT_EQUAL(true, ctx->waitForConnected(zk));
- return zk;
- }
-
- FILE *logfile;
- public:
- Zookeeper_multi() {
- logfile = openlogfile("Zookeeper_multi");
- }
- ~Zookeeper_multi() {
- if (logfile) {
- fflush(logfile);
- fclose(logfile);
- logfile = 0;
- }
- }
- void setUp()
- {
- zoo_set_log_stream(logfile);
- }
- void tearDown()
- {
- }
-
- static volatile int count;
- static void multi_completion_fn(int rc, const void *data) {
- CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
- count++;
- }
- static void multi_completion_fn_no_assert(int rc, const void *data) {
- count++;
- }
- static void multi_completion_fn_rc(int rc, const void *data) {
- count++;
- *((int*) data) = rc;
- }
-
- static void create_completion_fn_rc(int rc, const char* value, const void *data) {
- count++;
- *((int*) data) = rc;
- }
-
- static void waitForMultiCompletion(int seconds) {
- time_t expires = time(0) + seconds;
- while(count == 0 && time(0) < expires) {
- sleep(1);
- }
- count--;
- }
- static void resetCounter() {
- count = 0;
- }
- /**
- * Test basic multi-op create functionality
- */
- void testCreate() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
-
- int sz = 512;
- char p1[sz];
- char p2[sz];
- char p3[sz];
- p1[0] = p2[0] = p3[0] = '\0';
- int nops = 3 ;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_create_op_init(&ops[0], "/multi1", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_create_op_init(&ops[1], "/multi1/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
- zoo_create_op_init(&ops[2], "/multi1/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT(strcmp(p1, "/multi1") == 0);
- CPPUNIT_ASSERT(strcmp(p2, "/multi1/a") == 0);
- CPPUNIT_ASSERT(strcmp(p3, "/multi1/b") == 0);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2].err);
- }
- /**
- * Test create followed by delete
- */
- void testCreateDelete() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int sz = 512;
- char p1[sz];
- p1[0] = '\0';
- int nops = 2 ;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_create_op_init(&ops[0], "/multi2", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_delete_op_init(&ops[1], "/multi2", 0);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- // '/multi2' should have been deleted
- rc = zoo_exists(zk, "/multi2", 0, NULL);
- CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
- }
- /**
- * Test invalid versions
- */
- void testInvalidVersion() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int nops = 4;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_create_op_init(&ops[0], "/multi3", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
- zoo_delete_op_init(&ops[1], "/multi3", 1);
- zoo_create_op_init(&ops[2], "/multi3", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
- zoo_create_op_init(&ops[3], "/multi3/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, results[1].err);
- CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[2].err);
- CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[3].err);
- }
- /**
- * Test nested creates that rely on state in earlier op in multi
- */
- void testNestedCreate() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int sz = 512;
- char p1[sz];
- p1[0] = '\0';
- int nops = 6;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- /* Create */
- zoo_create_op_init(&ops[0], "/multi4", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_create_op_init(&ops[1], "/multi4/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_create_op_init(&ops[2], "/multi4/a/1", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- /* Delete */
- zoo_delete_op_init(&ops[3], "/multi4/a/1", 0);
- zoo_delete_op_init(&ops[4], "/multi4/a", 0);
- zoo_delete_op_init(&ops[5], "/multi4", 0);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- // Verify tree deleted
- rc = zoo_exists(zk, "/multi4/a/1", 0, NULL);
- CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
-
- rc = zoo_exists(zk, "/multi4/a", 0, NULL);
- CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
-
- rc = zoo_exists(zk, "/multi4", 0, NULL);
- CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
- }
- /**
- * Test setdata functionality
- */
- void testSetData() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int sz = 512;
- struct Stat s1;
- char buf[sz];
- int blen = sz ;
- char p1[sz], p2[sz];
-
- int nops = 2;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_create_op_init(&ops[0], "/multi5", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_create_op_init(&ops[1], "/multi5/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
-
- yield(zk, 5);
- zoo_op_t setdata_ops[nops];
- zoo_op_result_t setdata_results[nops];
- zoo_set_op_init(&setdata_ops[0], "/multi5", "1", 1, 0, &s1);
- zoo_set_op_init(&setdata_ops[1], "/multi5/a", "2", 1, 0, &s1);
- rc = zoo_multi(zk, nops, setdata_ops, setdata_results);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
-
- memset(buf, '\0', blen);
- rc = zoo_get(zk, "/multi5", 0, buf, &blen, &s1);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT_EQUAL(1, blen);
- CPPUNIT_ASSERT(strcmp("1", buf) == 0);
-
- memset(buf, '\0', blen);
- rc = zoo_get(zk, "/multi5/a", 0, buf, &blen, &s1);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT_EQUAL(1, blen);
- CPPUNIT_ASSERT(strcmp("2", buf) == 0);
- }
- /**
- * Test update conflicts
- */
- void testUpdateConflict() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int sz = 512;
- char buf[sz];
- int blen = sz;
- char p1[sz];
- p1[0] = '\0';
- struct Stat s1;
- int nops = 3;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_create_op_init(&ops[0], "/multi6", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_set_op_init(&ops[1], "/multi6", "X", 1, 0, &s1);
- zoo_set_op_init(&ops[2], "/multi6", "Y", 1, 0, &s1);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
- //Updating version solves conflict -- order matters
- ops[2].set_op.version = 1;
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
-
- memset(buf, 0, sz);
- rc = zoo_get(zk, "/multi6", 0, buf, &blen, &s1);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT_EQUAL(blen, 1);
- CPPUNIT_ASSERT(strncmp(buf, "Y", 1) == 0);
- }
- /**
- * Test delete-update conflicts
- */
- void testDeleteUpdateConflict() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int sz = 512;
- char p1[sz];
- p1[0] = '\0';
- struct Stat stat;
- int nops = 3;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_create_op_init(&ops[0], "/multi7", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_delete_op_init(&ops[1], "/multi7", 0);
- zoo_set_op_init(&ops[2], "/multi7", "Y", 1, 0, &stat);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
- // '/multi' should never have been created as entire op should fail
- rc = zoo_exists(zk, "/multi7", 0, NULL);
- CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
- }
- void testAsyncMulti() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
-
- int sz = 512;
- char p1[sz], p2[sz], p3[sz];
- p1[0] = '\0';
- p2[0] = '\0';
- p3[0] = '\0';
- int nops = 3;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_create_op_init(&ops[0], "/multi8", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_create_op_init(&ops[1], "/multi8/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
- zoo_create_op_init(&ops[2], "/multi8/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
-
- rc = zoo_amulti(zk, nops, ops, results, multi_completion_fn, 0);
- waitForMultiCompletion(10);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT(strcmp(p1, "/multi8") == 0);
- CPPUNIT_ASSERT(strcmp(p2, "/multi8/a") == 0);
- CPPUNIT_ASSERT(strcmp(p3, "/multi8/b") == 0);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2].err);
- }
- void testMultiFail() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
-
- int sz = 512;
- char p1[sz], p2[sz], p3[sz];
- p1[0] = '\0';
- p2[0] = '\0';
- p3[0] = '\0';
- int nops = 3;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_create_op_init(&ops[0], "/multi9", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- zoo_create_op_init(&ops[1], "/multi9", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
- zoo_create_op_init(&ops[2], "/multi9/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, rc);
- }
-
- /**
- * Test basic multi-op check functionality
- */
- void testCheck() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int sz = 512;
- char p1[sz];
- p1[0] = '\0';
- rc = zoo_create(zk, "/multi0", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- // Conditionally create /multi0/a' only if '/multi0' at version 0
- int nops = 2;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_check_op_init(&ops[0], "/multi0", 0);
- zoo_create_op_init(&ops[1], "/multi0/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
-
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
- // '/multi0/a' should have been created as it passed version check
- rc = zoo_exists(zk, "/multi0/a", 0, NULL);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
-
- // Only create '/multi0/b' if '/multi0' at version 10 (which it's not)
- zoo_op_t ops2[nops];
- zoo_check_op_init(&ops2[0], "/multi0", 10);
- zoo_create_op_init(&ops2[1], "/multi0/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
-
- rc = zoo_multi(zk, nops, ops2, results);
- CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
- CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, results[0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[1].err);
- // '/multi0/b' should NOT have been created
- rc = zoo_exists(zk, "/multi0/b", 0, NULL);
- CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
- }
- /**
- * Do a multi op inside a watch callback context.
- */
- static void doMultiInWatch(zhandle_t *zk, int type, int state, const char *path, void *ctx) {
- int rc;
- int sz = 512;
- char p1[sz];
- p1[0] = '\0';
- struct Stat s1;
- int nops = 1;
- zoo_op_t ops[nops];
- zoo_op_result_t results[nops];
- zoo_set_op_init(&ops[0], "/multiwatch", "1", 1, -1, NULL);
- rc = zoo_multi(zk, nops, ops, results);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
- memset(p1, '\0', sz);
- rc = zoo_get(zk, "/multiwatch", 0, p1, &sz, &s1);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- CPPUNIT_ASSERT_EQUAL(1, sz);
- CPPUNIT_ASSERT(strcmp("1", p1) == 0);
- count++;
- }
- /**
- * Test multi-op called from a watch
- */
- void testWatch() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int sz = 512;
- char p1[sz];
- p1[0] = '\0';
- rc = zoo_create(zk, "/multiwatch", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- // create a watch on node '/multiwatch'
- rc = zoo_wget(zk, "/multiwatch", doMultiInWatch, &ctx, p1, &sz, NULL);
- // setdata on node '/multiwatch' this should trip the watch
- rc = zoo_set(zk, "/multiwatch", NULL, -1, -1);
- CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
- // wait for multi completion in doMultiInWatch
- waitForMultiCompletion(5);
- }
-
- /**
- * ZOOKEEPER-1636: If request is too large, the server will cut the
- * connection without sending response packet. The client will try to
- * process completion on multi request and eventually cause SIGSEGV
- */
- void testBigAsyncMulti() {
- int rc;
- int callback_rc = (int) ZOK;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
-
- // The request should be more than 1MB which exceeds the default
- // jute.maxbuffer and causes the server to drop client connection
- const int iteration = 500;
- const int type_count = 3;
- const int nops = iteration * type_count;
- char buff[1024];
- zoo_op_result_t results[nops];
- zoo_op_t ops[nops];
- struct Stat* s[nops];
- int index = 0;
- // Test that we deliver error to 3 types of sub-request
- for (int i = 0; i < iteration; ++i) {
- zoo_set_op_init(&ops[index++], "/x", buff, sizeof(buff), -1, s[i]);
- zoo_create_op_init(&ops[index++], "/x", buff, sizeof(buff),
- &ZOO_OPEN_ACL_UNSAFE, ZOO_SEQUENCE, NULL, 0);
- zoo_delete_op_init(&ops[index++], "/x", -1);
- }
-
- rc = zoo_amulti(zk, nops, ops, results, multi_completion_fn_rc,
- + &callback_rc);
- CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
-
- waitForMultiCompletion(10);
- // With the bug, we will get SIGSEGV before reaching this point
- CPPUNIT_ASSERT_EQUAL((int) ZCONNECTIONLOSS, callback_rc);
-
- // Make sure that all sub-request completions get processed
- for (int i = 0; i < nops; ++i) {
- CPPUNIT_ASSERT_EQUAL((int) ZCONNECTIONLOSS, results[i].err);
- }
- // The handle should be able to recover itself.
- ctx.waitForConnected(zk);
- // Try to submit another async request to see if it get processed
- // correctly
- rc = zoo_acreate(zk, "/target", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0,
- create_completion_fn_rc, &callback_rc);
- CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
- waitForMultiCompletion(10);
- CPPUNIT_ASSERT_EQUAL((int) ZOK, callback_rc);
- }
- /**
- * ZOOKEEPER-1624: PendingChanges of create sequential node request didn't
- * get rollbacked correctly when multi-op failed. This caused
- * create sequential node request in subsequent multi-op to failed because
- * sequential node name generation is incorrect.
- *
- * The check is to make sure that each request in multi-op failed with
- * the correct reason.
- */
- void testSequentialNodeCreateInAsyncMulti() {
- int rc;
- watchctx_t ctx;
- zhandle_t *zk = createClient(&ctx);
- int iteration = 4;
- int nops = 2;
- zoo_op_result_t results[iteration][nops];
- zoo_op_t ops[nops];
- zoo_create_op_init(&ops[0], "/node-", "", 0, &ZOO_OPEN_ACL_UNSAFE, ZOO_SEQUENCE, NULL, 0);
- zoo_create_op_init(&ops[1], "/dup", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
- for (int i = 0; i < iteration ; ++i) {
- rc = zoo_amulti(zk, nops, ops, results[i], multi_completion_fn_no_assert, 0);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
- }
- waitForMultiCompletion(10);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0][0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1][0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2][0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[3][0].err);
- CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0][1].err);
- CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[1][1].err);
- CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[2][1].err);
- CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[3][1].err);
- resetCounter();
- }
- };
- volatile int Zookeeper_multi::count;
- const char Zookeeper_multi::hostPorts[] = "127.0.0.1:22181";
- CPPUNIT_TEST_SUITE_REGISTRATION(Zookeeper_multi);
- #endif
|