stress_worker.cc 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533
  1. /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
  2. *
  3. * Gearmand client and server library.
  4. *
  5. * Copyright (C) 2012 Data Differential, http://datadifferential.com/
  6. * All rights reserved.
  7. *
  8. * Redistribution and use in source and binary forms, with or without
  9. * modification, are permitted provided that the following conditions are
  10. * met:
  11. *
  12. * * Redistributions of source code must retain the above copyright
  13. * notice, this list of conditions and the following disclaimer.
  14. *
  15. * * Redistributions in binary form must reproduce the above
  16. * copyright notice, this list of conditions and the following disclaimer
  17. * in the documentation and/or other materials provided with the
  18. * distribution.
  19. *
  20. * * The names of its contributors may not be used to endorse or
  21. * promote products derived from this software without specific prior
  22. * written permission.
  23. *
  24. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  25. * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  26. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  27. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  28. * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  29. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  30. * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  31. * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  32. * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  33. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  34. * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  35. *
  36. */
  37. #include "gear_config.h"
  38. #include <libtest/test.hpp>
  39. using namespace libtest;
  40. #include <libgearman/gearman.h>
  41. #include <libhostile/hostile.h>
  42. #include <tests/start_worker.h>
  43. #include "tests/burnin.h"
  44. #include "tests/workers/v2/echo_or_react.h"
  45. #include <cerrno>
  46. #include <cstdio>
  47. #include <cstdlib>
  48. #include <cstring>
  49. #include <unistd.h>
  50. #include <sys/time.h>
  51. #define WORKER_FUNCTION_NAME "foo"
  52. static bool has_hostile()
  53. {
  54. #if defined(HAVE_LIBHOSTILE) && HAVE_LIBHOSTILE
  55. if (1)
  56. {
  57. return true;
  58. }
  59. #endif
  60. return false;
  61. }
  62. static in_port_t hostile_server= 0;
  63. static in_port_t stress_server= 0;
  64. static in_port_t& current_server_= stress_server;
  65. static void reset_server()
  66. {
  67. current_server_= stress_server;
  68. }
  69. static in_port_t current_server()
  70. {
  71. return current_server_;
  72. }
  73. static void set_server(in_port_t& arg)
  74. {
  75. current_server_= arg;
  76. }
  77. struct client_thread_context_st
  78. {
  79. size_t count;
  80. size_t payload_size;
  81. client_thread_context_st() :
  82. count(0),
  83. payload_size(0)
  84. { }
  85. void increment()
  86. {
  87. count++;
  88. }
  89. };
  90. extern "C" {
  91. static void client_cleanup(void *client)
  92. {
  93. int oldstate;
  94. pthread_setcanceltype(PTHREAD_CANCEL_DISABLE, &oldstate);
  95. gearman_client_free((gearman_client_st *)client);
  96. pthread_setcanceltype(oldstate, NULL);
  97. }
  98. static __attribute__((noreturn)) void *client_thread(void *object)
  99. {
  100. client_thread_context_st *success= (client_thread_context_st *)object;
  101. fatal_assert(success);
  102. fatal_assert(success->count == 0);
  103. gearman_return_t rc;
  104. gearman_client_st *client;
  105. {
  106. int oldstate;
  107. pthread_setcanceltype(PTHREAD_CANCEL_DISABLE, &oldstate);
  108. client= gearman_client_create(NULL);
  109. if (client == NULL)
  110. {
  111. pthread_exit(0);
  112. }
  113. rc= gearman_client_add_server(client, NULL, current_server());
  114. pthread_setcanceltype(oldstate, NULL);
  115. }
  116. pthread_cleanup_push(client_cleanup, client);
  117. if (gearman_success(rc))
  118. {
  119. gearman_client_set_timeout(client, 400);
  120. for (size_t x= 0; x < 100; x++)
  121. {
  122. int oldstate;
  123. pthread_setcanceltype(PTHREAD_CANCEL_DISABLE, &oldstate);
  124. libtest::vchar_t payload;
  125. payload.resize(success->payload_size);
  126. void *value= gearman_client_do(client, WORKER_FUNCTION_NAME,
  127. NULL,
  128. &payload[0],
  129. payload.size() ? random() % payload.size() : 0,
  130. NULL, &rc);
  131. pthread_setcanceltype(oldstate, NULL);
  132. if (gearman_success(rc))
  133. {
  134. success->increment();
  135. }
  136. if (value)
  137. {
  138. free(value);
  139. }
  140. }
  141. }
  142. pthread_cleanup_pop(1);
  143. pthread_exit(0);
  144. }
  145. }
  146. static bool join_thread(pthread_t& thread_arg, struct timespec& ts)
  147. {
  148. int error;
  149. ts.tv_sec+= 10;
  150. if (HAVE_PTHREAD_TIMEDJOIN_NP)
  151. {
  152. #if defined(HAVE_PTHREAD_TIMEDJOIN_NP) && HAVE_PTHREAD_TIMEDJOIN_NP
  153. int limit= 2;
  154. while (--limit)
  155. {
  156. switch ((error= pthread_timedjoin_np(thread_arg, NULL, &ts)))
  157. {
  158. case ETIMEDOUT:
  159. libtest::dream(1, 0);
  160. continue;
  161. case 0:
  162. return true;
  163. case ESRCH:
  164. return false;
  165. default:
  166. Error << "pthread_timedjoin_np() " << strerror(error);
  167. return false;
  168. }
  169. }
  170. Out << "pthread_timedjoin_np() " << strerror(error);
  171. if ((error= pthread_cancel(thread_arg)) != 0)
  172. {
  173. Error << "pthread_cancel() " << strerror(error);
  174. return false;
  175. }
  176. #endif
  177. return true;
  178. }
  179. if ((error= pthread_join(thread_arg, NULL)) != 0)
  180. {
  181. Error << "pthread_join() " << strerror(error);
  182. return false;
  183. }
  184. return true;
  185. }
  186. static test_return_t worker_ramp_exec(const size_t payload_size)
  187. {
  188. set_alarm(1200, 0);
  189. std::vector<pthread_t> children;
  190. children.resize(number_of_cpus());
  191. std::vector<client_thread_context_st> success;
  192. success.resize(children.size());
  193. for (size_t x= 0; x < children.size(); x++)
  194. {
  195. success[x].payload_size= payload_size;
  196. pthread_create(&children[x], NULL, client_thread, &success[x]);
  197. }
  198. for (size_t x= 0; x < children.size(); x++)
  199. {
  200. struct timespec ts;
  201. bool join_success= false;
  202. int limit= 2;
  203. while (join_success == false and --limit)
  204. {
  205. #if defined(HAVE_LIBRT) && HAVE_LIBRT
  206. if (HAVE_LIBRT) // This won't be called on OSX, etc,...
  207. {
  208. if (clock_gettime(CLOCK_REALTIME, &ts) == -1)
  209. {
  210. Error << "clock_gettime(CLOCK_REALTIME) " << strerror(errno);
  211. continue;
  212. }
  213. join_success= join_thread(children[x], ts);
  214. }
  215. else
  216. #endif
  217. {
  218. struct timeval tv;
  219. if (gettimeofday(&tv, NULL) == -1)
  220. {
  221. Error << "gettimeofday() " << strerror(errno);
  222. continue;
  223. }
  224. TIMEVAL_TO_TIMESPEC(&tv, &ts);
  225. join_success= join_thread(children[x], ts);
  226. }
  227. }
  228. if (join_success == false)
  229. {
  230. pthread_cancel(children[x]);
  231. Error << "Something went very wrong, it is likely threads were not cleaned up";
  232. }
  233. }
  234. return TEST_SUCCESS;
  235. }
  236. static test_return_t worker_ramp_TEST(void *)
  237. {
  238. return worker_ramp_exec(0);
  239. }
  240. static test_return_t worker_ramp_1K_TEST(void *)
  241. {
  242. return worker_ramp_exec(1024);
  243. }
  244. static test_return_t worker_ramp_10K_TEST(void *)
  245. {
  246. return worker_ramp_exec(1024*10);
  247. }
  248. static test_return_t worker_ramp_SETUP(void *object)
  249. {
  250. test_skip_valgrind();
  251. worker_handles_st *handles= (worker_handles_st*)object;
  252. gearman_function_t echo_react_fn= gearman_function_create(echo_or_react_worker_v2);
  253. for (uint32_t x= 0; x < 10; x++)
  254. {
  255. worker_handle_st *worker;
  256. if ((worker= test_worker_start(current_server(), NULL, WORKER_FUNCTION_NAME, echo_react_fn, NULL, gearman_worker_options_t())) == NULL)
  257. {
  258. return TEST_FAILURE;
  259. }
  260. handles->push(worker);
  261. }
  262. return TEST_SUCCESS;
  263. }
  264. static test_return_t worker_ramp_TEARDOWN(void* object)
  265. {
  266. worker_handles_st *handles= (worker_handles_st*)object;
  267. handles->reset();
  268. reset_server();
  269. return TEST_SUCCESS;
  270. }
  271. static test_return_t hostile_gearmand_SETUP(void* object)
  272. {
  273. test_skip(true, has_hostile());
  274. test_skip_valgrind();
  275. test_skip(true, libtest::is_massive());
  276. // Programmer error
  277. assert(hostile_server);
  278. set_server(hostile_server);
  279. worker_ramp_SETUP(object);
  280. return TEST_SUCCESS;
  281. }
  282. static test_return_t recv_SETUP(void* object)
  283. {
  284. test_skip_valgrind();
  285. test_skip(true, libtest::is_massive());
  286. worker_ramp_SETUP(object);
  287. set_recv_close(true, 20, 20);
  288. return TEST_SUCCESS;
  289. }
  290. static test_return_t recv_corrupt_SETUP(void* object)
  291. {
  292. test_skip_valgrind();
  293. test_skip(true, libtest::is_massive());
  294. worker_ramp_SETUP(object);
  295. set_recv_corrupt(true, 20, 20);
  296. return TEST_SUCCESS;
  297. }
  298. static test_return_t resv_TEARDOWN(void* object)
  299. {
  300. set_recv_close(true, 0, 0);
  301. worker_handles_st *handles= (worker_handles_st*)object;
  302. handles->kill_all();
  303. reset_server();
  304. return TEST_SUCCESS;
  305. }
  306. static test_return_t send_SETUP(void* object)
  307. {
  308. test_skip_valgrind();
  309. test_skip(true, libtest::is_massive());
  310. worker_ramp_SETUP(object);
  311. set_send_close(true, 20, 20);
  312. return TEST_SUCCESS;
  313. }
  314. static test_return_t accept_SETUP(void* object)
  315. {
  316. test_skip_valgrind();
  317. test_skip(true, libtest::is_massive());
  318. worker_ramp_SETUP(object);
  319. set_accept_close(true, 20, 20);
  320. return TEST_SUCCESS;
  321. }
  322. static test_return_t poll_HOSTILE_POLL_CLOSED_SETUP(void* object)
  323. {
  324. test_skip_valgrind();
  325. test_skip(true, libtest::is_massive());
  326. worker_ramp_SETUP(object);
  327. set_poll_close(true, 4, 0, HOSTILE_POLL_CLOSED);
  328. return TEST_SUCCESS;
  329. }
  330. static test_return_t poll_HOSTILE_POLL_SHUT_WR_SETUP(void* object)
  331. {
  332. test_skip_valgrind();
  333. test_skip(true, libtest::is_massive());
  334. worker_ramp_SETUP(object);
  335. set_poll_close(true, 4, 0, HOSTILE_POLL_SHUT_WR);
  336. return TEST_SUCCESS;
  337. }
  338. static test_return_t poll_HOSTILE_POLL_SHUT_RD_SETUP(void* object)
  339. {
  340. test_skip_valgrind();
  341. test_skip(true, libtest::is_massive());
  342. worker_ramp_SETUP(object);
  343. set_poll_close(true, 4, 0, HOSTILE_POLL_SHUT_RD);
  344. return TEST_SUCCESS;
  345. }
  346. static test_return_t poll_TEARDOWN(void* object)
  347. {
  348. set_poll_close(false, 0, 0, HOSTILE_POLL_CLOSED);
  349. worker_handles_st *handles= (worker_handles_st*)object;
  350. handles->kill_all();
  351. reset_server();
  352. return TEST_SUCCESS;
  353. }
  354. static test_return_t send_TEARDOWN(void* object)
  355. {
  356. set_send_close(true, 0, 0);
  357. worker_handles_st *handles= (worker_handles_st*)object;
  358. handles->kill_all();
  359. reset_server();
  360. return TEST_SUCCESS;
  361. }
  362. static test_return_t accept_TEARDOWN(void* object)
  363. {
  364. set_accept_close(true, 0, 0);
  365. worker_handles_st *handles= (worker_handles_st*)object;
  366. handles->kill_all();
  367. reset_server();
  368. return TEST_SUCCESS;
  369. }
  370. /*********************** World functions **************************************/
  371. static void *world_create(server_startup_st& servers, test_return_t& error)
  372. {
  373. stress_server= libtest::default_port();
  374. if (server_startup(servers, "gearmand", stress_server, 0, NULL) == false)
  375. {
  376. error= TEST_SKIPPED;
  377. return NULL;
  378. }
  379. if (has_hostile())
  380. {
  381. hostile_server= libtest::get_free_port();
  382. if (server_startup(servers, "gearmand", hostile_server, 0, NULL) == false)
  383. {
  384. hostile_server= 0;
  385. }
  386. }
  387. return new worker_handles_st;
  388. }
  389. static bool world_destroy(void *object)
  390. {
  391. worker_handles_st *handles= (worker_handles_st *)object;
  392. delete handles;
  393. return TEST_SUCCESS;
  394. }
  395. test_st burnin_TESTS[] ={
  396. {"burnin", 0, burnin_TEST },
  397. {0, 0, 0}
  398. };
  399. test_st worker_TESTS[] ={
  400. {"first pass", 0, worker_ramp_TEST },
  401. {"second pass", 0, worker_ramp_TEST },
  402. {"first pass 1K jobs", 0, worker_ramp_1K_TEST },
  403. {"first pass 10K jobs", 0, worker_ramp_10K_TEST },
  404. {0, 0, 0}
  405. };
  406. collection_st collection[] ={
  407. {"burnin", burnin_setup, burnin_cleanup, burnin_TESTS },
  408. {"plain", worker_ramp_SETUP, worker_ramp_TEARDOWN, worker_TESTS },
  409. {"plain against hostile server", hostile_gearmand_SETUP, worker_ramp_TEARDOWN, worker_TESTS },
  410. {"hostile recv()", recv_SETUP, resv_TEARDOWN, worker_TESTS },
  411. {"hostile recv() corrupt", recv_corrupt_SETUP, resv_TEARDOWN, worker_TESTS },
  412. {"hostile send()", send_SETUP, send_TEARDOWN, worker_TESTS },
  413. {"hostile accept()", accept_SETUP, accept_TEARDOWN, worker_TESTS },
  414. {"hostile poll(CLOSED)", poll_HOSTILE_POLL_CLOSED_SETUP, poll_TEARDOWN, worker_TESTS },
  415. {"hostile poll(SHUT_RD)", poll_HOSTILE_POLL_SHUT_RD_SETUP, poll_TEARDOWN, worker_TESTS },
  416. {"hostile poll(SHUT_WR)", poll_HOSTILE_POLL_SHUT_WR_SETUP, poll_TEARDOWN, worker_TESTS },
  417. {0, 0, 0, 0}
  418. };
  419. void get_world(libtest::Framework *world)
  420. {
  421. world->collections(collection);
  422. world->create(world_create);
  423. world->destroy(world_destroy);
  424. }