api_fuzzer.c 29 KB

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