TestMulti.cc 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771
  1. /**
  2. * Licensed to the Apache Software Foundation (ASF) under one
  3. * or more contributor license agreements. See the NOTICE file
  4. * distributed with this work for additional information
  5. * regarding copyright ownership. The ASF licenses this file
  6. * to you under the Apache License, Version 2.0 (the
  7. * "License"); you may not use this file except in compliance
  8. * with the License. You may obtain a copy of the License at
  9. *
  10. * http://www.apache.org/licenses/LICENSE-2.0
  11. *
  12. * Unless required by applicable law or agreed to in writing, software
  13. * distributed under the License is distributed on an "AS IS" BASIS,
  14. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. * See the License for the specific language governing permissions and
  16. * limitations under the License.
  17. */
  18. #include <cppunit/extensions/HelperMacros.h>
  19. #include "CppAssertHelper.h"
  20. #include <signal.h>
  21. #include <stdlib.h>
  22. #include <unistd.h>
  23. #include <sys/select.h>
  24. #include "CollectionUtil.h"
  25. #include "ThreadingUtil.h"
  26. using namespace Util;
  27. #include "Vector.h"
  28. using namespace std;
  29. #include <cstring>
  30. #include <list>
  31. #include <zookeeper.h>
  32. #include <errno.h>
  33. #include <recordio.h>
  34. #include "Util.h"
  35. #ifdef THREADED
  36. static void yield(zhandle_t *zh, int i)
  37. {
  38. sleep(i);
  39. }
  40. #else
  41. static void yield(zhandle_t *zh, int seconds)
  42. {
  43. int fd;
  44. int interest;
  45. int events;
  46. struct timeval tv;
  47. int rc;
  48. time_t expires = time(0) + seconds;
  49. time_t timeLeft = seconds;
  50. fd_set rfds, wfds, efds;
  51. FD_ZERO(&rfds);
  52. FD_ZERO(&wfds);
  53. FD_ZERO(&efds);
  54. while(timeLeft >= 0) {
  55. zookeeper_interest(zh, &fd, &interest, &tv);
  56. if (fd != -1) {
  57. if (interest&ZOOKEEPER_READ) {
  58. FD_SET(fd, &rfds);
  59. } else {
  60. FD_CLR(fd, &rfds);
  61. }
  62. if (interest&ZOOKEEPER_WRITE) {
  63. FD_SET(fd, &wfds);
  64. } else {
  65. FD_CLR(fd, &wfds);
  66. }
  67. } else {
  68. fd = 0;
  69. }
  70. FD_SET(0, &rfds);
  71. if (tv.tv_sec > timeLeft) {
  72. tv.tv_sec = timeLeft;
  73. }
  74. rc = select(fd+1, &rfds, &wfds, &efds, &tv);
  75. timeLeft = expires - time(0);
  76. events = 0;
  77. if (FD_ISSET(fd, &rfds)) {
  78. events |= ZOOKEEPER_READ;
  79. }
  80. if (FD_ISSET(fd, &wfds)) {
  81. events |= ZOOKEEPER_WRITE;
  82. }
  83. zookeeper_process(zh, events);
  84. }
  85. }
  86. #endif
  87. typedef struct evt {
  88. string path;
  89. int type;
  90. } evt_t;
  91. typedef struct watchCtx {
  92. private:
  93. list<evt_t> events;
  94. watchCtx(const watchCtx&);
  95. watchCtx& operator=(const watchCtx&);
  96. public:
  97. bool connected;
  98. zhandle_t *zh;
  99. Mutex mutex;
  100. watchCtx() {
  101. connected = false;
  102. zh = 0;
  103. }
  104. ~watchCtx() {
  105. if (zh) {
  106. zookeeper_close(zh);
  107. zh = 0;
  108. }
  109. }
  110. evt_t getEvent() {
  111. evt_t evt;
  112. mutex.acquire();
  113. CPPUNIT_ASSERT( events.size() > 0);
  114. evt = events.front();
  115. events.pop_front();
  116. mutex.release();
  117. return evt;
  118. }
  119. int countEvents() {
  120. int count;
  121. mutex.acquire();
  122. count = events.size();
  123. mutex.release();
  124. return count;
  125. }
  126. void putEvent(evt_t evt) {
  127. mutex.acquire();
  128. events.push_back(evt);
  129. mutex.release();
  130. }
  131. bool waitForConnected(zhandle_t *zh) {
  132. time_t expires = time(0) + 10;
  133. while(!connected && time(0) < expires) {
  134. yield(zh, 1);
  135. }
  136. return connected;
  137. }
  138. bool waitForDisconnected(zhandle_t *zh) {
  139. time_t expires = time(0) + 15;
  140. while(connected && time(0) < expires) {
  141. yield(zh, 1);
  142. }
  143. return !connected;
  144. }
  145. } watchctx_t;
  146. #ifdef THREADED
  147. class Zookeeper_multi : public CPPUNIT_NS::TestFixture
  148. {
  149. CPPUNIT_TEST_SUITE(Zookeeper_multi);
  150. //FIXME: None of these tests pass in single-threaded mode. It seems to be a
  151. //flaw in the test suite setup.
  152. CPPUNIT_TEST(testCreate);
  153. CPPUNIT_TEST(testCreateDelete);
  154. CPPUNIT_TEST(testInvalidVersion);
  155. CPPUNIT_TEST(testNestedCreate);
  156. CPPUNIT_TEST(testSetData);
  157. CPPUNIT_TEST(testUpdateConflict);
  158. CPPUNIT_TEST(testDeleteUpdateConflict);
  159. CPPUNIT_TEST(testAsyncMulti);
  160. CPPUNIT_TEST(testMultiFail);
  161. CPPUNIT_TEST(testCheck);
  162. CPPUNIT_TEST(testWatch);
  163. CPPUNIT_TEST(testSequentialNodeCreateInAsyncMulti);
  164. CPPUNIT_TEST(testBigAsyncMulti);
  165. CPPUNIT_TEST_SUITE_END();
  166. static void watcher(zhandle_t *, int type, int state, const char *path,void*v){
  167. watchctx_t *ctx = (watchctx_t*)v;
  168. if (state == ZOO_CONNECTED_STATE) {
  169. ctx->connected = true;
  170. } else {
  171. ctx->connected = false;
  172. }
  173. if (type != ZOO_SESSION_EVENT) {
  174. evt_t evt;
  175. evt.path = path;
  176. evt.type = type;
  177. ctx->putEvent(evt);
  178. }
  179. }
  180. static const char hostPorts[];
  181. const char *getHostPorts() {
  182. return hostPorts;
  183. }
  184. zhandle_t *createClient(watchctx_t *ctx) {
  185. return createClient(hostPorts, ctx);
  186. }
  187. zhandle_t *createClient(const char *hp, watchctx_t *ctx) {
  188. zhandle_t *zk = zookeeper_init(hp, watcher, 10000, 0, ctx, 0);
  189. ctx->zh = zk;
  190. CPPUNIT_ASSERT_EQUAL(true, ctx->waitForConnected(zk));
  191. return zk;
  192. }
  193. FILE *logfile;
  194. public:
  195. Zookeeper_multi() {
  196. logfile = openlogfile("Zookeeper_multi");
  197. }
  198. ~Zookeeper_multi() {
  199. if (logfile) {
  200. fflush(logfile);
  201. fclose(logfile);
  202. logfile = 0;
  203. }
  204. }
  205. void setUp()
  206. {
  207. zoo_set_log_stream(logfile);
  208. }
  209. void tearDown()
  210. {
  211. }
  212. static volatile int count;
  213. static void multi_completion_fn(int rc, const void *data) {
  214. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  215. count++;
  216. }
  217. static void multi_completion_fn_no_assert(int rc, const void *data) {
  218. count++;
  219. }
  220. static void multi_completion_fn_rc(int rc, const void *data) {
  221. count++;
  222. *((int*) data) = rc;
  223. }
  224. static void create_completion_fn_rc(int rc, const char* value, const void *data) {
  225. count++;
  226. *((int*) data) = rc;
  227. }
  228. static void waitForMultiCompletion(int seconds) {
  229. time_t expires = time(0) + seconds;
  230. while(count == 0 && time(0) < expires) {
  231. sleep(1);
  232. }
  233. count--;
  234. }
  235. static void resetCounter() {
  236. count = 0;
  237. }
  238. /**
  239. * Test basic multi-op create functionality
  240. */
  241. void testCreate() {
  242. int rc;
  243. watchctx_t ctx;
  244. zhandle_t *zk = createClient(&ctx);
  245. int sz = 512;
  246. char p1[sz];
  247. char p2[sz];
  248. char p3[sz];
  249. p1[0] = p2[0] = p3[0] = '\0';
  250. int nops = 3 ;
  251. zoo_op_t ops[nops];
  252. zoo_op_result_t results[nops];
  253. zoo_create_op_init(&ops[0], "/multi1", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  254. zoo_create_op_init(&ops[1], "/multi1/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
  255. zoo_create_op_init(&ops[2], "/multi1/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
  256. rc = zoo_multi(zk, nops, ops, results);
  257. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  258. CPPUNIT_ASSERT(strcmp(p1, "/multi1") == 0);
  259. CPPUNIT_ASSERT(strcmp(p2, "/multi1/a") == 0);
  260. CPPUNIT_ASSERT(strcmp(p3, "/multi1/b") == 0);
  261. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
  262. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
  263. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2].err);
  264. }
  265. /**
  266. * Test create followed by delete
  267. */
  268. void testCreateDelete() {
  269. int rc;
  270. watchctx_t ctx;
  271. zhandle_t *zk = createClient(&ctx);
  272. int sz = 512;
  273. char p1[sz];
  274. p1[0] = '\0';
  275. int nops = 2 ;
  276. zoo_op_t ops[nops];
  277. zoo_op_result_t results[nops];
  278. zoo_create_op_init(&ops[0], "/multi2", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  279. zoo_delete_op_init(&ops[1], "/multi2", 0);
  280. rc = zoo_multi(zk, nops, ops, results);
  281. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  282. // '/multi2' should have been deleted
  283. rc = zoo_exists(zk, "/multi2", 0, NULL);
  284. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  285. }
  286. /**
  287. * Test invalid versions
  288. */
  289. void testInvalidVersion() {
  290. int rc;
  291. watchctx_t ctx;
  292. zhandle_t *zk = createClient(&ctx);
  293. int nops = 4;
  294. zoo_op_t ops[nops];
  295. zoo_op_result_t results[nops];
  296. zoo_create_op_init(&ops[0], "/multi3", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
  297. zoo_delete_op_init(&ops[1], "/multi3", 1);
  298. zoo_create_op_init(&ops[2], "/multi3", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
  299. zoo_create_op_init(&ops[3], "/multi3/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
  300. rc = zoo_multi(zk, nops, ops, results);
  301. CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
  302. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
  303. CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, results[1].err);
  304. CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[2].err);
  305. CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[3].err);
  306. }
  307. /**
  308. * Test nested creates that rely on state in earlier op in multi
  309. */
  310. void testNestedCreate() {
  311. int rc;
  312. watchctx_t ctx;
  313. zhandle_t *zk = createClient(&ctx);
  314. int sz = 512;
  315. char p1[sz];
  316. p1[0] = '\0';
  317. int nops = 6;
  318. zoo_op_t ops[nops];
  319. zoo_op_result_t results[nops];
  320. /* Create */
  321. zoo_create_op_init(&ops[0], "/multi4", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  322. zoo_create_op_init(&ops[1], "/multi4/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  323. zoo_create_op_init(&ops[2], "/multi4/a/1", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  324. /* Delete */
  325. zoo_delete_op_init(&ops[3], "/multi4/a/1", 0);
  326. zoo_delete_op_init(&ops[4], "/multi4/a", 0);
  327. zoo_delete_op_init(&ops[5], "/multi4", 0);
  328. rc = zoo_multi(zk, nops, ops, results);
  329. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  330. // Verify tree deleted
  331. rc = zoo_exists(zk, "/multi4/a/1", 0, NULL);
  332. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  333. rc = zoo_exists(zk, "/multi4/a", 0, NULL);
  334. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  335. rc = zoo_exists(zk, "/multi4", 0, NULL);
  336. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  337. }
  338. /**
  339. * Test setdata functionality
  340. */
  341. void testSetData() {
  342. int rc;
  343. watchctx_t ctx;
  344. zhandle_t *zk = createClient(&ctx);
  345. int sz = 512;
  346. struct Stat s1;
  347. char buf[sz];
  348. int blen = sz ;
  349. char p1[sz], p2[sz];
  350. int nops = 2;
  351. zoo_op_t ops[nops];
  352. zoo_op_result_t results[nops];
  353. zoo_create_op_init(&ops[0], "/multi5", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  354. zoo_create_op_init(&ops[1], "/multi5/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
  355. rc = zoo_multi(zk, nops, ops, results);
  356. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  357. yield(zk, 5);
  358. zoo_op_t setdata_ops[nops];
  359. zoo_op_result_t setdata_results[nops];
  360. zoo_set_op_init(&setdata_ops[0], "/multi5", "1", 1, 0, &s1);
  361. zoo_set_op_init(&setdata_ops[1], "/multi5/a", "2", 1, 0, &s1);
  362. rc = zoo_multi(zk, nops, setdata_ops, setdata_results);
  363. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  364. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
  365. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
  366. memset(buf, '\0', blen);
  367. rc = zoo_get(zk, "/multi5", 0, buf, &blen, &s1);
  368. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  369. CPPUNIT_ASSERT_EQUAL(1, blen);
  370. CPPUNIT_ASSERT(strcmp("1", buf) == 0);
  371. memset(buf, '\0', blen);
  372. rc = zoo_get(zk, "/multi5/a", 0, buf, &blen, &s1);
  373. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  374. CPPUNIT_ASSERT_EQUAL(1, blen);
  375. CPPUNIT_ASSERT(strcmp("2", buf) == 0);
  376. }
  377. /**
  378. * Test update conflicts
  379. */
  380. void testUpdateConflict() {
  381. int rc;
  382. watchctx_t ctx;
  383. zhandle_t *zk = createClient(&ctx);
  384. int sz = 512;
  385. char buf[sz];
  386. int blen = sz;
  387. char p1[sz];
  388. p1[0] = '\0';
  389. struct Stat s1;
  390. int nops = 3;
  391. zoo_op_t ops[nops];
  392. zoo_op_result_t results[nops];
  393. zoo_create_op_init(&ops[0], "/multi6", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  394. zoo_set_op_init(&ops[1], "/multi6", "X", 1, 0, &s1);
  395. zoo_set_op_init(&ops[2], "/multi6", "Y", 1, 0, &s1);
  396. rc = zoo_multi(zk, nops, ops, results);
  397. CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
  398. //Updating version solves conflict -- order matters
  399. ops[2].set_op.version = 1;
  400. rc = zoo_multi(zk, nops, ops, results);
  401. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  402. memset(buf, 0, sz);
  403. rc = zoo_get(zk, "/multi6", 0, buf, &blen, &s1);
  404. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  405. CPPUNIT_ASSERT_EQUAL(blen, 1);
  406. CPPUNIT_ASSERT(strncmp(buf, "Y", 1) == 0);
  407. }
  408. /**
  409. * Test delete-update conflicts
  410. */
  411. void testDeleteUpdateConflict() {
  412. int rc;
  413. watchctx_t ctx;
  414. zhandle_t *zk = createClient(&ctx);
  415. int sz = 512;
  416. char buf[sz];
  417. int blen;
  418. char p1[sz];
  419. p1[0] = '\0';
  420. struct Stat stat;
  421. int nops = 3;
  422. zoo_op_t ops[nops];
  423. zoo_op_result_t results[nops];
  424. zoo_create_op_init(&ops[0], "/multi7", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  425. zoo_delete_op_init(&ops[1], "/multi7", 0);
  426. zoo_set_op_init(&ops[2], "/multi7", "Y", 1, 0, &stat);
  427. rc = zoo_multi(zk, nops, ops, results);
  428. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  429. // '/multi' should never have been created as entire op should fail
  430. rc = zoo_exists(zk, "/multi7", 0, NULL);
  431. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  432. }
  433. void testAsyncMulti() {
  434. int rc;
  435. watchctx_t ctx;
  436. zhandle_t *zk = createClient(&ctx);
  437. int sz = 512;
  438. char p1[sz], p2[sz], p3[sz];
  439. p1[0] = '\0';
  440. p2[0] = '\0';
  441. p3[0] = '\0';
  442. int nops = 3;
  443. zoo_op_t ops[nops];
  444. zoo_op_result_t results[nops];
  445. zoo_create_op_init(&ops[0], "/multi8", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  446. zoo_create_op_init(&ops[1], "/multi8/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
  447. zoo_create_op_init(&ops[2], "/multi8/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
  448. rc = zoo_amulti(zk, nops, ops, results, multi_completion_fn, 0);
  449. waitForMultiCompletion(10);
  450. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  451. CPPUNIT_ASSERT(strcmp(p1, "/multi8") == 0);
  452. CPPUNIT_ASSERT(strcmp(p2, "/multi8/a") == 0);
  453. CPPUNIT_ASSERT(strcmp(p3, "/multi8/b") == 0);
  454. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
  455. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
  456. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2].err);
  457. }
  458. void testMultiFail() {
  459. int rc;
  460. watchctx_t ctx;
  461. zhandle_t *zk = createClient(&ctx);
  462. int sz = 512;
  463. char p1[sz], p2[sz], p3[sz];
  464. p1[0] = '\0';
  465. p2[0] = '\0';
  466. p3[0] = '\0';
  467. int nops = 3;
  468. zoo_op_t ops[nops];
  469. zoo_op_result_t results[nops];
  470. zoo_create_op_init(&ops[0], "/multi9", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  471. zoo_create_op_init(&ops[1], "/multi9", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
  472. zoo_create_op_init(&ops[2], "/multi9/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
  473. rc = zoo_multi(zk, nops, ops, results);
  474. CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, rc);
  475. }
  476. /**
  477. * Test basic multi-op check functionality
  478. */
  479. void testCheck() {
  480. int rc;
  481. watchctx_t ctx;
  482. zhandle_t *zk = createClient(&ctx);
  483. int sz = 512;
  484. char p1[sz];
  485. p1[0] = '\0';
  486. struct Stat s1;
  487. rc = zoo_create(zk, "/multi0", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  488. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  489. // Conditionally create /multi0/a' only if '/multi0' at version 0
  490. int nops = 2;
  491. zoo_op_t ops[nops];
  492. zoo_op_result_t results[nops];
  493. zoo_check_op_init(&ops[0], "/multi0", 0);
  494. zoo_create_op_init(&ops[1], "/multi0/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  495. rc = zoo_multi(zk, nops, ops, results);
  496. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  497. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
  498. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
  499. // '/multi0/a' should have been created as it passed version check
  500. rc = zoo_exists(zk, "/multi0/a", 0, NULL);
  501. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  502. // Only create '/multi0/b' if '/multi0' at version 10 (which it's not)
  503. zoo_op_t ops2[nops];
  504. zoo_check_op_init(&ops2[0], "/multi0", 10);
  505. zoo_create_op_init(&ops2[1], "/multi0/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
  506. rc = zoo_multi(zk, nops, ops2, results);
  507. CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
  508. CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, results[0].err);
  509. CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[1].err);
  510. // '/multi0/b' should NOT have been created
  511. rc = zoo_exists(zk, "/multi0/b", 0, NULL);
  512. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  513. }
  514. /**
  515. * Do a multi op inside a watch callback context.
  516. */
  517. static void doMultiInWatch(zhandle_t *zk, int type, int state, const char *path, void *ctx) {
  518. int rc;
  519. int sz = 512;
  520. char p1[sz];
  521. p1[0] = '\0';
  522. struct Stat s1;
  523. int nops = 1;
  524. zoo_op_t ops[nops];
  525. zoo_op_result_t results[nops];
  526. zoo_set_op_init(&ops[0], "/multiwatch", "1", 1, -1, NULL);
  527. rc = zoo_multi(zk, nops, ops, results);
  528. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  529. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
  530. memset(p1, '\0', sz);
  531. rc = zoo_get(zk, "/multiwatch", 0, p1, &sz, &s1);
  532. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  533. CPPUNIT_ASSERT_EQUAL(1, sz);
  534. CPPUNIT_ASSERT(strcmp("1", p1) == 0);
  535. count++;
  536. }
  537. /**
  538. * Test multi-op called from a watch
  539. */
  540. void testWatch() {
  541. int rc;
  542. watchctx_t ctx;
  543. zhandle_t *zk = createClient(&ctx);
  544. int sz = 512;
  545. char p1[sz];
  546. p1[0] = '\0';
  547. rc = zoo_create(zk, "/multiwatch", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
  548. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  549. // create a watch on node '/multiwatch'
  550. rc = zoo_wget(zk, "/multiwatch", doMultiInWatch, &ctx, p1, &sz, NULL);
  551. // setdata on node '/multiwatch' this should trip the watch
  552. rc = zoo_set(zk, "/multiwatch", NULL, -1, -1);
  553. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  554. // wait for multi completion in doMultiInWatch
  555. waitForMultiCompletion(5);
  556. }
  557. /**
  558. * ZOOKEEPER-1636: If request is too large, the server will cut the
  559. * connection without sending response packet. The client will try to
  560. * process completion on multi request and eventually cause SIGSEGV
  561. */
  562. void testBigAsyncMulti() {
  563. int rc;
  564. int callback_rc = (int) ZOK;
  565. watchctx_t ctx;
  566. zhandle_t *zk = createClient(&ctx);
  567. // The request should be more than 1MB which exceeds the default
  568. // jute.maxbuffer and causes the server to drop client connection
  569. const int iteration = 500;
  570. const int type_count = 3;
  571. const int nops = iteration * type_count;
  572. char buff[1024];
  573. zoo_op_result_t results[nops];
  574. zoo_op_t ops[nops];
  575. struct Stat* s[nops];
  576. int index = 0;
  577. // Test that we deliver error to 3 types of sub-request
  578. for (int i = 0; i < iteration; ++i) {
  579. zoo_set_op_init(&ops[index++], "/x", buff, sizeof(buff), -1, s[i]);
  580. zoo_create_op_init(&ops[index++], "/x", buff, sizeof(buff),
  581. &ZOO_OPEN_ACL_UNSAFE, ZOO_SEQUENCE, NULL, 0);
  582. zoo_delete_op_init(&ops[index++], "/x", -1);
  583. }
  584. rc = zoo_amulti(zk, nops, ops, results, multi_completion_fn_rc,
  585. + &callback_rc);
  586. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  587. waitForMultiCompletion(10);
  588. // With the bug, we will get SIGSEGV before reaching this point
  589. CPPUNIT_ASSERT_EQUAL((int) ZCONNECTIONLOSS, callback_rc);
  590. // Make sure that all sub-request completions get processed
  591. for (int i = 0; i < nops; ++i) {
  592. CPPUNIT_ASSERT_EQUAL((int) ZCONNECTIONLOSS, results[i].err);
  593. }
  594. // The handle should be able to recover itself.
  595. ctx.waitForConnected(zk);
  596. // Try to submit another async request to see if it get processed
  597. // correctly
  598. rc = zoo_acreate(zk, "/target", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0,
  599. create_completion_fn_rc, &callback_rc);
  600. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  601. waitForMultiCompletion(10);
  602. CPPUNIT_ASSERT_EQUAL((int) ZOK, callback_rc);
  603. }
  604. /**
  605. * ZOOKEEPER-1624: PendingChanges of create sequential node request didn't
  606. * get rollbacked correctly when multi-op failed. This caused
  607. * create sequential node request in subsequent multi-op to failed because
  608. * sequential node name generation is incorrect.
  609. *
  610. * The check is to make sure that each request in multi-op failed with
  611. * the correct reason.
  612. */
  613. void testSequentialNodeCreateInAsyncMulti() {
  614. int rc;
  615. watchctx_t ctx;
  616. zhandle_t *zk = createClient(&ctx);
  617. int iteration = 4;
  618. int nops = 2;
  619. zoo_op_result_t results[iteration][nops];
  620. zoo_op_t ops[nops];
  621. zoo_create_op_init(&ops[0], "/node-", "", 0, &ZOO_OPEN_ACL_UNSAFE, ZOO_SEQUENCE, NULL, 0);
  622. zoo_create_op_init(&ops[1], "/dup", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
  623. for (int i = 0; i < iteration ; ++i) {
  624. rc = zoo_amulti(zk, nops, ops, results[i], multi_completion_fn_no_assert, 0);
  625. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  626. }
  627. waitForMultiCompletion(10);
  628. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0][0].err);
  629. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1][0].err);
  630. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2][0].err);
  631. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[3][0].err);
  632. CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0][1].err);
  633. CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[1][1].err);
  634. CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[2][1].err);
  635. CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[3][1].err);
  636. resetCounter();
  637. }
  638. };
  639. volatile int Zookeeper_multi::count;
  640. const char Zookeeper_multi::hostPorts[] = "127.0.0.1:22181";
  641. CPPUNIT_TEST_SUITE_REGISTRATION(Zookeeper_multi);
  642. #endif