api_fuzzer.c 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968
  1. /*
  2. *
  3. * Copyright 2016, Google Inc.
  4. * All rights reserved.
  5. *
  6. * Redistribution and use in source and binary forms, with or without
  7. * modification, are permitted provided that the following conditions are
  8. * met:
  9. *
  10. * * Redistributions of source code must retain the above copyright
  11. * notice, this list of conditions and the following disclaimer.
  12. * * Redistributions in binary form must reproduce the above
  13. * copyright notice, this list of conditions and the following disclaimer
  14. * in the documentation and/or other materials provided with the
  15. * distribution.
  16. * * Neither the name of Google Inc. nor the names of its
  17. * contributors may be used to endorse or promote products derived from
  18. * this software without specific prior written permission.
  19. *
  20. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  21. * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  22. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  23. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  24. * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  25. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  26. * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  27. * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  28. * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  29. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  30. * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  31. *
  32. */
  33. #include <string.h>
  34. #include <grpc/grpc.h>
  35. #include <grpc/support/alloc.h>
  36. #include <grpc/support/log.h>
  37. #include <grpc/support/string_util.h>
  38. #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
  39. #include "src/core/lib/channel/channel_args.h"
  40. #include "src/core/lib/iomgr/resolve_address.h"
  41. #include "src/core/lib/iomgr/tcp_client.h"
  42. #include "src/core/lib/iomgr/timer.h"
  43. #include "src/core/lib/surface/server.h"
  44. #include "src/core/lib/transport/metadata.h"
  45. #include "test/core/util/passthru_endpoint.h"
  46. ////////////////////////////////////////////////////////////////////////////////
  47. // logging
  48. bool squelch = true;
  49. bool leak_check = true;
  50. static void dont_log(gpr_log_func_args *args) {}
  51. ////////////////////////////////////////////////////////////////////////////////
  52. // input_stream: allows easy access to input bytes, and allows reading a little
  53. // past the end (avoiding needing to check everywhere)
  54. typedef struct {
  55. const uint8_t *cur;
  56. const uint8_t *end;
  57. } input_stream;
  58. static uint8_t next_byte(input_stream *inp) {
  59. if (inp->cur == inp->end) {
  60. return 0;
  61. }
  62. return *inp->cur++;
  63. }
  64. static void end(input_stream *inp) { inp->cur = inp->end; }
  65. static char *read_string(input_stream *inp) {
  66. char *str = NULL;
  67. size_t cap = 0;
  68. size_t sz = 0;
  69. char c;
  70. do {
  71. if (cap == sz) {
  72. cap = GPR_MAX(3 * cap / 2, cap + 8);
  73. str = gpr_realloc(str, cap);
  74. }
  75. c = (char)next_byte(inp);
  76. str[sz++] = c;
  77. } while (c != 0);
  78. return str;
  79. }
  80. static void read_buffer(input_stream *inp, char **buffer, size_t *length) {
  81. *length = next_byte(inp);
  82. *buffer = gpr_malloc(*length);
  83. for (size_t i = 0; i < *length; i++) {
  84. (*buffer)[i] = (char)next_byte(inp);
  85. }
  86. }
  87. static uint32_t read_uint22(input_stream *inp) {
  88. uint8_t b = next_byte(inp);
  89. uint32_t x = b & 0x7f;
  90. if (b & 0x80) {
  91. x <<= 7;
  92. b = next_byte(inp);
  93. x |= b & 0x7f;
  94. if (b & 0x80) {
  95. x <<= 8;
  96. x |= next_byte(inp);
  97. }
  98. }
  99. return x;
  100. }
  101. static uint32_t read_uint32(input_stream *inp) {
  102. uint8_t b = next_byte(inp);
  103. uint32_t x = b & 0x7f;
  104. if (b & 0x80) {
  105. x <<= 7;
  106. b = next_byte(inp);
  107. x |= b & 0x7f;
  108. if (b & 0x80) {
  109. x <<= 7;
  110. b = next_byte(inp);
  111. x |= b & 0x7f;
  112. if (b & 0x80) {
  113. x <<= 7;
  114. b = next_byte(inp);
  115. x |= b & 0x7f;
  116. if (b & 0x80) {
  117. x = (x << 4) | (next_byte(inp) & 0x0f);
  118. }
  119. }
  120. }
  121. }
  122. return x;
  123. }
  124. static grpc_byte_buffer *read_message(input_stream *inp) {
  125. gpr_slice slice = gpr_slice_malloc(read_uint22(inp));
  126. memset(GPR_SLICE_START_PTR(slice), 0, GPR_SLICE_LENGTH(slice));
  127. grpc_byte_buffer *out = grpc_raw_byte_buffer_create(&slice, 1);
  128. gpr_slice_unref(slice);
  129. return out;
  130. }
  131. static int read_int(input_stream *inp) { return (int)read_uint32(inp); }
  132. static grpc_channel_args *read_args(input_stream *inp) {
  133. size_t n = next_byte(inp);
  134. grpc_arg *args = gpr_malloc(sizeof(*args) * n);
  135. for (size_t i = 0; i < n; i++) {
  136. bool is_string = next_byte(inp) & 1;
  137. args[i].type = is_string ? GRPC_ARG_STRING : GRPC_ARG_INTEGER;
  138. args[i].key = read_string(inp);
  139. if (is_string) {
  140. args[i].value.string = read_string(inp);
  141. } else {
  142. args[i].value.integer = read_int(inp);
  143. }
  144. }
  145. grpc_channel_args *a = gpr_malloc(sizeof(*a));
  146. a->args = args;
  147. a->num_args = n;
  148. return a;
  149. }
  150. static bool is_eof(input_stream *inp) { return inp->cur == inp->end; }
  151. ////////////////////////////////////////////////////////////////////////////////
  152. // global state
  153. static gpr_timespec g_now;
  154. static grpc_server *g_server;
  155. static grpc_channel *g_channel;
  156. static grpc_buffer_pool *g_buffer_pool;
  157. extern gpr_timespec (*gpr_now_impl)(gpr_clock_type clock_type);
  158. static gpr_timespec now_impl(gpr_clock_type clock_type) {
  159. GPR_ASSERT(clock_type != GPR_TIMESPAN);
  160. return g_now;
  161. }
  162. ////////////////////////////////////////////////////////////////////////////////
  163. // dns resolution
  164. typedef struct addr_req {
  165. grpc_timer timer;
  166. char *addr;
  167. grpc_closure *on_done;
  168. grpc_resolved_addresses **addrs;
  169. } addr_req;
  170. static void finish_resolve(grpc_exec_ctx *exec_ctx, void *arg,
  171. grpc_error *error) {
  172. addr_req *r = arg;
  173. if (error == GRPC_ERROR_NONE && 0 == strcmp(r->addr, "server")) {
  174. grpc_resolved_addresses *addrs = gpr_malloc(sizeof(*addrs));
  175. addrs->naddrs = 1;
  176. addrs->addrs = gpr_malloc(sizeof(*addrs->addrs));
  177. addrs->addrs[0].len = 0;
  178. *r->addrs = addrs;
  179. grpc_exec_ctx_sched(exec_ctx, r->on_done, GRPC_ERROR_NONE, NULL);
  180. } else {
  181. grpc_exec_ctx_sched(
  182. exec_ctx, r->on_done,
  183. GRPC_ERROR_CREATE_REFERENCING("Resolution failed", &error, 1), NULL);
  184. }
  185. gpr_free(r->addr);
  186. gpr_free(r);
  187. }
  188. void my_resolve_address(grpc_exec_ctx *exec_ctx, const char *addr,
  189. const char *default_port, grpc_closure *on_done,
  190. grpc_resolved_addresses **addresses) {
  191. addr_req *r = gpr_malloc(sizeof(*r));
  192. r->addr = gpr_strdup(addr);
  193. r->on_done = on_done;
  194. r->addrs = addresses;
  195. grpc_timer_init(exec_ctx, &r->timer,
  196. gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
  197. gpr_time_from_seconds(1, GPR_TIMESPAN)),
  198. finish_resolve, r, gpr_now(GPR_CLOCK_MONOTONIC));
  199. }
  200. ////////////////////////////////////////////////////////////////////////////////
  201. // client connection
  202. // defined in tcp_client_posix.c
  203. extern void (*grpc_tcp_client_connect_impl)(
  204. grpc_exec_ctx *exec_ctx, grpc_closure *closure, grpc_endpoint **ep,
  205. grpc_pollset_set *interested_parties, const struct sockaddr *addr,
  206. size_t addr_len, gpr_timespec deadline);
  207. static void sched_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
  208. grpc_endpoint **ep, gpr_timespec deadline);
  209. typedef struct {
  210. grpc_timer timer;
  211. grpc_closure *closure;
  212. grpc_endpoint **ep;
  213. gpr_timespec deadline;
  214. } future_connect;
  215. static void do_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
  216. future_connect *fc = arg;
  217. if (error != GRPC_ERROR_NONE) {
  218. *fc->ep = NULL;
  219. grpc_exec_ctx_sched(exec_ctx, fc->closure, GRPC_ERROR_REF(error), NULL);
  220. } else if (g_server != NULL) {
  221. grpc_endpoint *client;
  222. grpc_endpoint *server;
  223. grpc_passthru_endpoint_create(&client, &server, g_buffer_pool);
  224. *fc->ep = client;
  225. grpc_transport *transport =
  226. grpc_create_chttp2_transport(exec_ctx, NULL, server, 0);
  227. grpc_server_setup_transport(exec_ctx, g_server, transport, NULL, NULL);
  228. grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL);
  229. grpc_exec_ctx_sched(exec_ctx, fc->closure, GRPC_ERROR_NONE, NULL);
  230. } else {
  231. sched_connect(exec_ctx, fc->closure, fc->ep, fc->deadline);
  232. }
  233. gpr_free(fc);
  234. }
  235. static void sched_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
  236. grpc_endpoint **ep, gpr_timespec deadline) {
  237. if (gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) < 0) {
  238. *ep = NULL;
  239. grpc_exec_ctx_sched(exec_ctx, closure,
  240. GRPC_ERROR_CREATE("Connect deadline exceeded"), NULL);
  241. return;
  242. }
  243. future_connect *fc = gpr_malloc(sizeof(*fc));
  244. fc->closure = closure;
  245. fc->ep = ep;
  246. fc->deadline = deadline;
  247. grpc_timer_init(exec_ctx, &fc->timer,
  248. gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
  249. gpr_time_from_millis(1, GPR_TIMESPAN)),
  250. do_connect, fc, gpr_now(GPR_CLOCK_MONOTONIC));
  251. }
  252. static void my_tcp_client_connect(grpc_exec_ctx *exec_ctx,
  253. grpc_closure *closure, grpc_endpoint **ep,
  254. grpc_pollset_set *interested_parties,
  255. const struct sockaddr *addr, size_t addr_len,
  256. gpr_timespec deadline) {
  257. sched_connect(exec_ctx, closure, ep, deadline);
  258. }
  259. ////////////////////////////////////////////////////////////////////////////////
  260. // test driver
  261. typedef struct validator {
  262. void (*validate)(void *arg, bool success);
  263. void *arg;
  264. } validator;
  265. static validator *create_validator(void (*validate)(void *arg, bool success),
  266. void *arg) {
  267. validator *v = gpr_malloc(sizeof(*v));
  268. v->validate = validate;
  269. v->arg = arg;
  270. return v;
  271. }
  272. static void assert_success_and_decrement(void *counter, bool success) {
  273. GPR_ASSERT(success);
  274. --*(int *)counter;
  275. }
  276. static void decrement(void *counter, bool success) { --*(int *)counter; }
  277. typedef struct connectivity_watch {
  278. int *counter;
  279. gpr_timespec deadline;
  280. } connectivity_watch;
  281. static connectivity_watch *make_connectivity_watch(gpr_timespec s,
  282. int *counter) {
  283. connectivity_watch *o = gpr_malloc(sizeof(*o));
  284. o->deadline = s;
  285. o->counter = counter;
  286. return o;
  287. }
  288. static void validate_connectivity_watch(void *p, bool success) {
  289. connectivity_watch *w = p;
  290. if (!success) {
  291. GPR_ASSERT(gpr_time_cmp(gpr_now(w->deadline.clock_type), w->deadline) >= 0);
  292. }
  293. --*w->counter;
  294. gpr_free(w);
  295. }
  296. static void free_non_null(void *p) {
  297. GPR_ASSERT(p != NULL);
  298. gpr_free(p);
  299. }
  300. typedef enum { ROOT, CLIENT, SERVER, PENDING_SERVER } call_state_type;
  301. #define DONE_FLAG_CALL_CLOSED ((uint64_t)(1 << 0))
  302. typedef struct call_state {
  303. call_state_type type;
  304. grpc_call *call;
  305. grpc_byte_buffer *recv_message;
  306. grpc_status_code status;
  307. grpc_metadata_array recv_initial_metadata;
  308. grpc_metadata_array recv_trailing_metadata;
  309. char *recv_status_details;
  310. size_t recv_status_details_capacity;
  311. int cancelled;
  312. int pending_ops;
  313. grpc_call_details call_details;
  314. grpc_byte_buffer *send_message;
  315. // starts at 0, individual flags from DONE_FLAG_xxx are set
  316. // as different operations are completed
  317. uint64_t done_flags;
  318. // array of pointers to free later
  319. size_t num_to_free;
  320. size_t cap_to_free;
  321. void **to_free;
  322. struct call_state *next;
  323. struct call_state *prev;
  324. } call_state;
  325. static call_state *g_active_call;
  326. static call_state *new_call(call_state *sibling, call_state_type type) {
  327. call_state *c = gpr_malloc(sizeof(*c));
  328. memset(c, 0, sizeof(*c));
  329. if (sibling != NULL) {
  330. c->next = sibling;
  331. c->prev = sibling->prev;
  332. c->next->prev = c->prev->next = c;
  333. } else {
  334. c->next = c->prev = c;
  335. }
  336. c->type = type;
  337. return c;
  338. }
  339. static call_state *maybe_delete_call_state(call_state *call) {
  340. call_state *next = call->next;
  341. if (call->call != NULL) return next;
  342. if (call->pending_ops != 0) return next;
  343. if (call == g_active_call) {
  344. g_active_call = call->next;
  345. GPR_ASSERT(call != g_active_call);
  346. }
  347. call->prev->next = call->next;
  348. call->next->prev = call->prev;
  349. grpc_metadata_array_destroy(&call->recv_initial_metadata);
  350. grpc_metadata_array_destroy(&call->recv_trailing_metadata);
  351. gpr_free(call->recv_status_details);
  352. grpc_call_details_destroy(&call->call_details);
  353. for (size_t i = 0; i < call->num_to_free; i++) {
  354. gpr_free(call->to_free[i]);
  355. }
  356. gpr_free(call->to_free);
  357. gpr_free(call);
  358. return next;
  359. }
  360. static void add_to_free(call_state *call, void *p) {
  361. if (call->num_to_free == call->cap_to_free) {
  362. call->cap_to_free = GPR_MAX(8, 2 * call->cap_to_free);
  363. call->to_free =
  364. gpr_realloc(call->to_free, sizeof(*call->to_free) * call->cap_to_free);
  365. }
  366. call->to_free[call->num_to_free++] = p;
  367. }
  368. static void read_metadata(input_stream *inp, size_t *count,
  369. grpc_metadata **metadata, call_state *cs) {
  370. *count = next_byte(inp);
  371. if (*count) {
  372. *metadata = gpr_malloc(*count * sizeof(**metadata));
  373. memset(*metadata, 0, *count * sizeof(**metadata));
  374. for (size_t i = 0; i < *count; i++) {
  375. (*metadata)[i].key = read_string(inp);
  376. read_buffer(inp, (char **)&(*metadata)[i].value,
  377. &(*metadata)[i].value_length);
  378. (*metadata)[i].flags = read_uint32(inp);
  379. add_to_free(cs, (void *)(*metadata)[i].key);
  380. add_to_free(cs, (void *)(*metadata)[i].value);
  381. }
  382. } else {
  383. *metadata = gpr_malloc(1);
  384. }
  385. add_to_free(cs, *metadata);
  386. }
  387. static call_state *destroy_call(call_state *call) {
  388. grpc_call_destroy(call->call);
  389. call->call = NULL;
  390. return maybe_delete_call_state(call);
  391. }
  392. static void finished_request_call(void *csp, bool success) {
  393. call_state *cs = csp;
  394. GPR_ASSERT(cs->pending_ops > 0);
  395. --cs->pending_ops;
  396. if (success) {
  397. GPR_ASSERT(cs->call != NULL);
  398. cs->type = SERVER;
  399. } else {
  400. maybe_delete_call_state(cs);
  401. }
  402. }
  403. typedef struct {
  404. call_state *cs;
  405. uint8_t has_ops;
  406. } batch_info;
  407. static void finished_batch(void *p, bool success) {
  408. batch_info *bi = p;
  409. --bi->cs->pending_ops;
  410. if ((bi->has_ops & (1u << GRPC_OP_RECV_MESSAGE)) &&
  411. (bi->cs->done_flags & DONE_FLAG_CALL_CLOSED)) {
  412. GPR_ASSERT(bi->cs->recv_message == NULL);
  413. }
  414. if ((bi->has_ops & (1u << GRPC_OP_RECV_MESSAGE) &&
  415. bi->cs->recv_message != NULL)) {
  416. grpc_byte_buffer_destroy(bi->cs->recv_message);
  417. bi->cs->recv_message = NULL;
  418. }
  419. if ((bi->has_ops & (1u << GRPC_OP_SEND_MESSAGE))) {
  420. grpc_byte_buffer_destroy(bi->cs->send_message);
  421. bi->cs->send_message = NULL;
  422. }
  423. if ((bi->has_ops & (1u << GRPC_OP_RECV_STATUS_ON_CLIENT)) ||
  424. (bi->has_ops & (1u << GRPC_OP_RECV_CLOSE_ON_SERVER))) {
  425. bi->cs->done_flags |= DONE_FLAG_CALL_CLOSED;
  426. }
  427. maybe_delete_call_state(bi->cs);
  428. gpr_free(bi);
  429. }
  430. static validator *make_finished_batch_validator(call_state *cs,
  431. uint8_t has_ops) {
  432. batch_info *bi = gpr_malloc(sizeof(*bi));
  433. bi->cs = cs;
  434. bi->has_ops = has_ops;
  435. return create_validator(finished_batch, bi);
  436. }
  437. int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
  438. grpc_test_only_set_metadata_hash_seed(0);
  439. if (squelch) gpr_set_log_function(dont_log);
  440. input_stream inp = {data, data + size};
  441. grpc_resolve_address = my_resolve_address;
  442. grpc_tcp_client_connect_impl = my_tcp_client_connect;
  443. gpr_now_impl = now_impl;
  444. grpc_init();
  445. GPR_ASSERT(g_channel == NULL);
  446. GPR_ASSERT(g_server == NULL);
  447. bool server_shutdown = false;
  448. int pending_server_shutdowns = 0;
  449. int pending_channel_watches = 0;
  450. int pending_pings = 0;
  451. g_active_call = new_call(NULL, ROOT);
  452. g_buffer_pool = grpc_buffer_pool_create();
  453. grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
  454. while (!is_eof(&inp) || g_channel != NULL || g_server != NULL ||
  455. pending_channel_watches > 0 || pending_pings > 0 ||
  456. g_active_call->type != ROOT || g_active_call->next != g_active_call) {
  457. if (is_eof(&inp)) {
  458. if (g_channel != NULL) {
  459. grpc_channel_destroy(g_channel);
  460. g_channel = NULL;
  461. }
  462. if (g_server != NULL) {
  463. if (!server_shutdown) {
  464. grpc_server_shutdown_and_notify(
  465. g_server, cq, create_validator(assert_success_and_decrement,
  466. &pending_server_shutdowns));
  467. server_shutdown = true;
  468. pending_server_shutdowns++;
  469. } else if (pending_server_shutdowns == 0) {
  470. grpc_server_destroy(g_server);
  471. g_server = NULL;
  472. }
  473. }
  474. call_state *s = g_active_call;
  475. do {
  476. if (s->type != PENDING_SERVER && s->call != NULL) {
  477. s = destroy_call(s);
  478. } else {
  479. s = s->next;
  480. }
  481. } while (s != g_active_call);
  482. g_now = gpr_time_add(g_now, gpr_time_from_seconds(1, GPR_TIMESPAN));
  483. }
  484. switch (next_byte(&inp)) {
  485. // terminate on bad bytes
  486. default:
  487. end(&inp);
  488. break;
  489. // tickle completion queue
  490. case 0: {
  491. grpc_event ev = grpc_completion_queue_next(
  492. cq, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
  493. switch (ev.type) {
  494. case GRPC_OP_COMPLETE: {
  495. validator *v = ev.tag;
  496. v->validate(v->arg, ev.success);
  497. gpr_free(v);
  498. break;
  499. }
  500. case GRPC_QUEUE_TIMEOUT:
  501. break;
  502. case GRPC_QUEUE_SHUTDOWN:
  503. abort();
  504. break;
  505. }
  506. break;
  507. }
  508. // increment global time
  509. case 1: {
  510. g_now = gpr_time_add(
  511. g_now, gpr_time_from_micros(read_uint32(&inp), GPR_TIMESPAN));
  512. break;
  513. }
  514. // create an insecure channel
  515. case 2: {
  516. if (g_channel == NULL) {
  517. char *target = read_string(&inp);
  518. char *target_uri;
  519. gpr_asprintf(&target_uri, "dns:%s", target);
  520. grpc_channel_args *args = read_args(&inp);
  521. g_channel = grpc_insecure_channel_create(target_uri, args, NULL);
  522. GPR_ASSERT(g_channel != NULL);
  523. grpc_channel_args_destroy(args);
  524. gpr_free(target_uri);
  525. gpr_free(target);
  526. } else {
  527. end(&inp);
  528. }
  529. break;
  530. }
  531. // destroy a channel
  532. case 3: {
  533. if (g_channel != NULL) {
  534. grpc_channel_destroy(g_channel);
  535. g_channel = NULL;
  536. } else {
  537. end(&inp);
  538. }
  539. break;
  540. }
  541. // bring up a server
  542. case 4: {
  543. if (g_server == NULL) {
  544. grpc_channel_args *args = read_args(&inp);
  545. g_server = grpc_server_create(args, NULL);
  546. GPR_ASSERT(g_server != NULL);
  547. grpc_channel_args_destroy(args);
  548. grpc_server_register_completion_queue(g_server, cq, NULL);
  549. grpc_server_start(g_server);
  550. server_shutdown = false;
  551. GPR_ASSERT(pending_server_shutdowns == 0);
  552. } else {
  553. end(&inp);
  554. }
  555. break;
  556. }
  557. // begin server shutdown
  558. case 5: {
  559. if (g_server != NULL) {
  560. grpc_server_shutdown_and_notify(
  561. g_server, cq, create_validator(assert_success_and_decrement,
  562. &pending_server_shutdowns));
  563. pending_server_shutdowns++;
  564. server_shutdown = true;
  565. } else {
  566. end(&inp);
  567. }
  568. break;
  569. }
  570. // cancel all calls if shutdown
  571. case 6: {
  572. if (g_server != NULL && server_shutdown) {
  573. grpc_server_cancel_all_calls(g_server);
  574. } else {
  575. end(&inp);
  576. }
  577. break;
  578. }
  579. // destroy server
  580. case 7: {
  581. if (g_server != NULL && server_shutdown &&
  582. pending_server_shutdowns == 0) {
  583. grpc_server_destroy(g_server);
  584. g_server = NULL;
  585. } else {
  586. end(&inp);
  587. }
  588. break;
  589. }
  590. // check connectivity
  591. case 8: {
  592. if (g_channel != NULL) {
  593. uint8_t try_to_connect = next_byte(&inp);
  594. if (try_to_connect == 0 || try_to_connect == 1) {
  595. grpc_channel_check_connectivity_state(g_channel, try_to_connect);
  596. } else {
  597. end(&inp);
  598. }
  599. } else {
  600. end(&inp);
  601. }
  602. break;
  603. }
  604. // watch connectivity
  605. case 9: {
  606. if (g_channel != NULL) {
  607. grpc_connectivity_state st =
  608. grpc_channel_check_connectivity_state(g_channel, 0);
  609. if (st != GRPC_CHANNEL_SHUTDOWN) {
  610. gpr_timespec deadline = gpr_time_add(
  611. gpr_now(GPR_CLOCK_REALTIME),
  612. gpr_time_from_micros(read_uint32(&inp), GPR_TIMESPAN));
  613. grpc_channel_watch_connectivity_state(
  614. g_channel, st, deadline, cq,
  615. create_validator(validate_connectivity_watch,
  616. make_connectivity_watch(
  617. deadline, &pending_channel_watches)));
  618. pending_channel_watches++;
  619. }
  620. } else {
  621. end(&inp);
  622. }
  623. break;
  624. }
  625. // create a call
  626. case 10: {
  627. bool ok = true;
  628. if (g_channel == NULL) ok = false;
  629. grpc_call *parent_call = NULL;
  630. if (g_active_call->type != ROOT) {
  631. if (g_active_call->call == NULL || g_active_call->type == CLIENT) {
  632. end(&inp);
  633. break;
  634. }
  635. parent_call = g_active_call->call;
  636. }
  637. uint32_t propagation_mask = read_uint32(&inp);
  638. char *method = read_string(&inp);
  639. char *host = read_string(&inp);
  640. gpr_timespec deadline =
  641. gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
  642. gpr_time_from_micros(read_uint32(&inp), GPR_TIMESPAN));
  643. if (ok) {
  644. call_state *cs = new_call(g_active_call, CLIENT);
  645. cs->call =
  646. grpc_channel_create_call(g_channel, parent_call, propagation_mask,
  647. cq, method, host, deadline, NULL);
  648. } else {
  649. end(&inp);
  650. }
  651. gpr_free(method);
  652. gpr_free(host);
  653. break;
  654. }
  655. // switch the 'current' call
  656. case 11: {
  657. g_active_call = g_active_call->next;
  658. break;
  659. }
  660. // queue some ops on a call
  661. case 12: {
  662. if (g_active_call->type == PENDING_SERVER ||
  663. g_active_call->type == ROOT || g_active_call->call == NULL) {
  664. end(&inp);
  665. break;
  666. }
  667. size_t num_ops = next_byte(&inp);
  668. if (num_ops > 6) {
  669. end(&inp);
  670. break;
  671. }
  672. grpc_op *ops = gpr_malloc(sizeof(grpc_op) * num_ops);
  673. memset(ops, 0, sizeof(grpc_op) * num_ops);
  674. bool ok = true;
  675. size_t i;
  676. grpc_op *op;
  677. uint8_t has_ops = 0;
  678. for (i = 0; i < num_ops; i++) {
  679. op = &ops[i];
  680. switch (next_byte(&inp)) {
  681. default:
  682. /* invalid value */
  683. op->op = (grpc_op_type)-1;
  684. ok = false;
  685. break;
  686. case GRPC_OP_SEND_INITIAL_METADATA:
  687. op->op = GRPC_OP_SEND_INITIAL_METADATA;
  688. has_ops |= 1 << GRPC_OP_SEND_INITIAL_METADATA;
  689. read_metadata(&inp, &op->data.send_initial_metadata.count,
  690. &op->data.send_initial_metadata.metadata,
  691. g_active_call);
  692. break;
  693. case GRPC_OP_SEND_MESSAGE:
  694. op->op = GRPC_OP_SEND_MESSAGE;
  695. if (g_active_call->send_message != NULL) {
  696. ok = false;
  697. } else {
  698. has_ops |= 1 << GRPC_OP_SEND_MESSAGE;
  699. g_active_call->send_message = op->data.send_message =
  700. read_message(&inp);
  701. }
  702. break;
  703. case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
  704. op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
  705. has_ops |= 1 << GRPC_OP_SEND_CLOSE_FROM_CLIENT;
  706. break;
  707. case GRPC_OP_SEND_STATUS_FROM_SERVER:
  708. op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
  709. has_ops |= 1 << GRPC_OP_SEND_STATUS_FROM_SERVER;
  710. read_metadata(
  711. &inp,
  712. &op->data.send_status_from_server.trailing_metadata_count,
  713. &op->data.send_status_from_server.trailing_metadata,
  714. g_active_call);
  715. op->data.send_status_from_server.status = next_byte(&inp);
  716. op->data.send_status_from_server.status_details =
  717. read_string(&inp);
  718. break;
  719. case GRPC_OP_RECV_INITIAL_METADATA:
  720. op->op = GRPC_OP_RECV_INITIAL_METADATA;
  721. has_ops |= 1 << GRPC_OP_RECV_INITIAL_METADATA;
  722. op->data.recv_initial_metadata =
  723. &g_active_call->recv_initial_metadata;
  724. break;
  725. case GRPC_OP_RECV_MESSAGE:
  726. op->op = GRPC_OP_RECV_MESSAGE;
  727. has_ops |= 1 << GRPC_OP_RECV_MESSAGE;
  728. op->data.recv_message = &g_active_call->recv_message;
  729. break;
  730. case GRPC_OP_RECV_STATUS_ON_CLIENT:
  731. op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
  732. op->data.recv_status_on_client.status = &g_active_call->status;
  733. op->data.recv_status_on_client.trailing_metadata =
  734. &g_active_call->recv_trailing_metadata;
  735. op->data.recv_status_on_client.status_details =
  736. &g_active_call->recv_status_details;
  737. op->data.recv_status_on_client.status_details_capacity =
  738. &g_active_call->recv_status_details_capacity;
  739. break;
  740. case GRPC_OP_RECV_CLOSE_ON_SERVER:
  741. op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
  742. has_ops |= 1 << GRPC_OP_RECV_CLOSE_ON_SERVER;
  743. op->data.recv_close_on_server.cancelled =
  744. &g_active_call->cancelled;
  745. break;
  746. }
  747. op->reserved = NULL;
  748. op->flags = read_uint32(&inp);
  749. }
  750. if (ok) {
  751. validator *v = make_finished_batch_validator(g_active_call, has_ops);
  752. g_active_call->pending_ops++;
  753. grpc_call_error error =
  754. grpc_call_start_batch(g_active_call->call, ops, num_ops, v, NULL);
  755. if (error != GRPC_CALL_OK) {
  756. v->validate(v->arg, false);
  757. gpr_free(v);
  758. }
  759. } else {
  760. end(&inp);
  761. }
  762. if (!ok && (has_ops & (1 << GRPC_OP_SEND_MESSAGE))) {
  763. grpc_byte_buffer_destroy(g_active_call->send_message);
  764. g_active_call->send_message = NULL;
  765. }
  766. for (i = 0; i < num_ops; i++) {
  767. op = &ops[i];
  768. switch (op->op) {
  769. case GRPC_OP_SEND_STATUS_FROM_SERVER:
  770. gpr_free((void *)op->data.send_status_from_server.status_details);
  771. break;
  772. case GRPC_OP_SEND_MESSAGE:
  773. case GRPC_OP_SEND_INITIAL_METADATA:
  774. case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
  775. case GRPC_OP_RECV_INITIAL_METADATA:
  776. case GRPC_OP_RECV_MESSAGE:
  777. case GRPC_OP_RECV_STATUS_ON_CLIENT:
  778. case GRPC_OP_RECV_CLOSE_ON_SERVER:
  779. break;
  780. }
  781. }
  782. gpr_free(ops);
  783. break;
  784. }
  785. // cancel current call
  786. case 13: {
  787. if (g_active_call->type != ROOT && g_active_call->call != NULL) {
  788. grpc_call_cancel(g_active_call->call, NULL);
  789. } else {
  790. end(&inp);
  791. }
  792. break;
  793. }
  794. // get a calls peer
  795. case 14: {
  796. if (g_active_call->type != ROOT && g_active_call->call != NULL) {
  797. free_non_null(grpc_call_get_peer(g_active_call->call));
  798. } else {
  799. end(&inp);
  800. }
  801. break;
  802. }
  803. // get a channels target
  804. case 15: {
  805. if (g_channel != NULL) {
  806. free_non_null(grpc_channel_get_target(g_channel));
  807. } else {
  808. end(&inp);
  809. }
  810. break;
  811. }
  812. // send a ping on a channel
  813. case 16: {
  814. if (g_channel != NULL) {
  815. pending_pings++;
  816. grpc_channel_ping(g_channel, cq,
  817. create_validator(decrement, &pending_pings), NULL);
  818. } else {
  819. end(&inp);
  820. }
  821. break;
  822. }
  823. // enable a tracer
  824. case 17: {
  825. char *tracer = read_string(&inp);
  826. grpc_tracer_set_enabled(tracer, 1);
  827. gpr_free(tracer);
  828. break;
  829. }
  830. // disable a tracer
  831. case 18: {
  832. char *tracer = read_string(&inp);
  833. grpc_tracer_set_enabled(tracer, 0);
  834. gpr_free(tracer);
  835. break;
  836. }
  837. // request a server call
  838. case 19: {
  839. if (g_server == NULL) {
  840. end(&inp);
  841. break;
  842. }
  843. call_state *cs = new_call(g_active_call, PENDING_SERVER);
  844. cs->pending_ops++;
  845. validator *v = create_validator(finished_request_call, cs);
  846. grpc_call_error error =
  847. grpc_server_request_call(g_server, &cs->call, &cs->call_details,
  848. &cs->recv_initial_metadata, cq, cq, v);
  849. if (error != GRPC_CALL_OK) {
  850. v->validate(v->arg, false);
  851. gpr_free(v);
  852. }
  853. break;
  854. }
  855. // destroy a call
  856. case 20: {
  857. if (g_active_call->type != ROOT &&
  858. g_active_call->type != PENDING_SERVER &&
  859. g_active_call->call != NULL) {
  860. destroy_call(g_active_call);
  861. } else {
  862. end(&inp);
  863. }
  864. break;
  865. }
  866. // resize the buffer pool
  867. case 21: {
  868. grpc_buffer_pool_resize(g_buffer_pool, read_uint22(&inp));
  869. break;
  870. }
  871. }
  872. }
  873. GPR_ASSERT(g_channel == NULL);
  874. GPR_ASSERT(g_server == NULL);
  875. GPR_ASSERT(g_active_call->type == ROOT);
  876. GPR_ASSERT(g_active_call->next == g_active_call);
  877. gpr_free(g_active_call);
  878. grpc_completion_queue_shutdown(cq);
  879. GPR_ASSERT(
  880. grpc_completion_queue_next(cq, gpr_inf_past(GPR_CLOCK_REALTIME), NULL)
  881. .type == GRPC_QUEUE_SHUTDOWN);
  882. grpc_completion_queue_destroy(cq);
  883. grpc_buffer_pool_unref(g_buffer_pool);
  884. grpc_shutdown();
  885. return 0;
  886. }