libevent/evrpc.c

   1 /*
   2  * Copyright (c) 2000-2007 Niels Provos <provos@citi.umich.edu>
   3  * Copyright (c) 2007-2012 Niels Provos and Nick Mathewson
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  * 1. Redistributions of source code must retain the above copyright
   9  *    notice, this list of conditions and the following disclaimer.
  10  * 2. Redistributions in binary form must reproduce the above copyright
  11  *    notice, this list of conditions and the following disclaimer in the
  12  *    documentation and/or other materials provided with the distribution.
  13  * 3. The name of the author may not be used to endorse or promote products
  14  *    derived from this software without specific prior written permission.
  15  *
  16  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
  17  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
  18  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
  19  * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
  20  * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
  21  * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  22  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  23  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  24  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
  25  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26  */
  27 #include "event2/event-config.h"
  28
  29 #ifdef WIN32
  30 #define WIN32_LEAN_AND_MEAN
  31 #include <winsock2.h>
  32 #include <windows.h>
  33 #undef WIN32_LEAN_AND_MEAN
  34 #endif
  35
  36 #include <sys/types.h>
  37 #ifndef WIN32
  38 #include <sys/socket.h>
  39 #endif
  40 #ifdef _EVENT_HAVE_SYS_TIME_H
  41 #include <sys/time.h>
  42 #endif
  43 #include <sys/queue.h>
  44 #include <stdio.h>
  45 #include <stdlib.h>
  46 #ifndef WIN32
  47 #include <unistd.h>
  48 #endif
  49 #include <errno.h>
  50 #include <signal.h>
  51 #include <string.h>
  52
  53 #include <sys/queue.h>
  54
  55 #include "event2/event.h"
  56 #include "event2/event_struct.h"
  57 #include "event2/rpc.h"
  58 #include "event2/rpc_struct.h"
  59 #include "evrpc-internal.h"
  60 #include "event2/http.h"
  61 #include "event2/buffer.h"
  62 #include "event2/tag.h"
  63 #include "event2/http_struct.h"
  64 #include "event2/http_compat.h"
  65 #include "event2/util.h"
  66 #include "util-internal.h"
  67 #include "log-internal.h"
  68 #include "mm-internal.h"
  69
  70 struct evrpc_base *
  71 evrpc_init(struct evhttp *http_server)
  72 {
  73         struct evrpc_base* base = mm_calloc(1, sizeof(struct evrpc_base));
  74         if (base == NULL)
  75                 return (NULL);
  76
  77         /* we rely on the tagging sub system */
  78         evtag_init();
  79
  80         TAILQ_INIT(&base->registered_rpcs);
  81         TAILQ_INIT(&base->input_hooks);
  82         TAILQ_INIT(&base->output_hooks);
  83
  84         TAILQ_INIT(&base->paused_requests);
  85
  86         base->http_server = http_server;
  87
  88         return (base);
  89 }
  90
  91 void
  92 evrpc_free(struct evrpc_base *base)
  93 {
  94         struct evrpc *rpc;
  95         struct evrpc_hook *hook;
  96         struct evrpc_hook_ctx *pause;
  97         int r;
  98
  99         while ((rpc = TAILQ_FIRST(&base->registered_rpcs)) != NULL) {
 100                 r = evrpc_unregister_rpc(base, rpc->uri);
 101                 EVUTIL_ASSERT(r == 0);
 102         }
 103         while ((pause = TAILQ_FIRST(&base->paused_requests)) != NULL) {
 104                 TAILQ_REMOVE(&base->paused_requests, pause, next);
 105                 mm_free(pause);
 106         }
 107         while ((hook = TAILQ_FIRST(&base->input_hooks)) != NULL) {
 108                 r = evrpc_remove_hook(base, EVRPC_INPUT, hook);
 109                 EVUTIL_ASSERT(r);
 110         }
 111         while ((hook = TAILQ_FIRST(&base->output_hooks)) != NULL) {
 112                 r = evrpc_remove_hook(base, EVRPC_OUTPUT, hook);
 113                 EVUTIL_ASSERT(r);
 114         }
 115         mm_free(base);
 116 }
 117
 118 void *
 119 evrpc_add_hook(void *vbase,
 120     enum EVRPC_HOOK_TYPE hook_type,
 121     int (*cb)(void *, struct evhttp_request *, struct evbuffer *, void *),
 122     void *cb_arg)
 123 {
 124         struct _evrpc_hooks *base = vbase;
 125         struct evrpc_hook_list *head = NULL;
 126         struct evrpc_hook *hook = NULL;
 127         switch (hook_type) {
 128         case EVRPC_INPUT:
 129                 head = &base->in_hooks;
 130                 break;
 131         case EVRPC_OUTPUT:
 132                 head = &base->out_hooks;
 133                 break;
 134         default:
 135                 EVUTIL_ASSERT(hook_type == EVRPC_INPUT || hook_type == EVRPC_OUTPUT);
 136         }
 137
 138         hook = mm_calloc(1, sizeof(struct evrpc_hook));
 139         EVUTIL_ASSERT(hook != NULL);
 140
 141         hook->process = cb;
 142         hook->process_arg = cb_arg;
 143         TAILQ_INSERT_TAIL(head, hook, next);
 144
 145         return (hook);
 146 }
 147
 148 static int
 149 evrpc_remove_hook_internal(struct evrpc_hook_list *head, void *handle)
 150 {
 151         struct evrpc_hook *hook = NULL;
 152         TAILQ_FOREACH(hook, head, next) {
 153                 if (hook == handle) {
 154                         TAILQ_REMOVE(head, hook, next);
 155                         mm_free(hook);
 156                         return (1);
 157                 }
 158         }
 159
 160         return (0);
 161 }
 162
 163 /*
 164  * remove the hook specified by the handle
 165  */
 166
 167 int
 168 evrpc_remove_hook(void *vbase, enum EVRPC_HOOK_TYPE hook_type, void *handle)
 169 {
 170         struct _evrpc_hooks *base = vbase;
 171         struct evrpc_hook_list *head = NULL;
 172         switch (hook_type) {
 173         case EVRPC_INPUT:
 174                 head = &base->in_hooks;
 175                 break;
 176         case EVRPC_OUTPUT:
 177                 head = &base->out_hooks;
 178                 break;
 179         default:
 180                 EVUTIL_ASSERT(hook_type == EVRPC_INPUT || hook_type == EVRPC_OUTPUT);
 181         }
 182
 183         return (evrpc_remove_hook_internal(head, handle));
 184 }
 185
 186 static int
 187 evrpc_process_hooks(struct evrpc_hook_list *head, void *ctx,
 188     struct evhttp_request *req, struct evbuffer *evbuf)
 189 {
 190         struct evrpc_hook *hook;
 191         TAILQ_FOREACH(hook, head, next) {
 192                 int res = hook->process(ctx, req, evbuf, hook->process_arg);
 193                 if (res != EVRPC_CONTINUE)
 194                         return (res);
 195         }
 196
 197         return (EVRPC_CONTINUE);
 198 }
 199
 200 static void evrpc_pool_schedule(struct evrpc_pool *pool);
 201 static void evrpc_request_cb(struct evhttp_request *, void *);
 202
 203 /*
 204  * Registers a new RPC with the HTTP server.   The evrpc object is expected
 205  * to have been filled in via the EVRPC_REGISTER_OBJECT macro which in turn
 206  * calls this function.
 207  */
 208
 209 static char *
 210 evrpc_construct_uri(const char *uri)
 211 {
 212         char *constructed_uri;
 213         size_t constructed_uri_len;
 214
 215         constructed_uri_len = strlen(EVRPC_URI_PREFIX) + strlen(uri) + 1;
 216         if ((constructed_uri = mm_malloc(constructed_uri_len)) == NULL)
 217                 event_err(1, "%s: failed to register rpc at %s",
 218                     __func__, uri);
 219         memcpy(constructed_uri, EVRPC_URI_PREFIX, strlen(EVRPC_URI_PREFIX));
 220         memcpy(constructed_uri + strlen(EVRPC_URI_PREFIX), uri, strlen(uri));
 221         constructed_uri[constructed_uri_len - 1] = '\0';
 222
 223         return (constructed_uri);
 224 }
 225
 226 int
 227 evrpc_register_rpc(struct evrpc_base *base, struct evrpc *rpc,
 228     void (*cb)(struct evrpc_req_generic *, void *), void *cb_arg)
 229 {
 230         char *constructed_uri = evrpc_construct_uri(rpc->uri);
 231
 232         rpc->base = base;
 233         rpc->cb = cb;
 234         rpc->cb_arg = cb_arg;
 235
 236         TAILQ_INSERT_TAIL(&base->registered_rpcs, rpc, next);
 237
 238         evhttp_set_cb(base->http_server,
 239             constructed_uri,
 240             evrpc_request_cb,
 241             rpc);
 242
 243         mm_free(constructed_uri);
 244
 245         return (0);
 246 }
 247
 248 int
 249 evrpc_unregister_rpc(struct evrpc_base *base, const char *name)
 250 {
 251         char *registered_uri = NULL;
 252         struct evrpc *rpc;
 253         int r;
 254
 255         /* find the right rpc; linear search might be slow */
 256         TAILQ_FOREACH(rpc, &base->registered_rpcs, next) {
 257                 if (strcmp(rpc->uri, name) == 0)
 258                         break;
 259         }
 260         if (rpc == NULL) {
 261                 /* We did not find an RPC with this name */
 262                 return (-1);
 263         }
 264         TAILQ_REMOVE(&base->registered_rpcs, rpc, next);
 265
 266         registered_uri = evrpc_construct_uri(name);
 267
 268         /* remove the http server callback */
 269         r = evhttp_del_cb(base->http_server, registered_uri);
 270         EVUTIL_ASSERT(r == 0);
 271
 272         mm_free(registered_uri);
 273
 274         mm_free((char *)rpc->uri);
 275         mm_free(rpc);
 276         return (0);
 277 }
 278
 279 static int evrpc_pause_request(void *vbase, void *ctx,
 280     void (*cb)(void *, enum EVRPC_HOOK_RESULT));
 281 static void evrpc_request_cb_closure(void *, enum EVRPC_HOOK_RESULT);
 282
 283 static void
 284 evrpc_request_cb(struct evhttp_request *req, void *arg)
 285 {
 286         struct evrpc *rpc = arg;
 287         struct evrpc_req_generic *rpc_state = NULL;
 288
 289         /* let's verify the outside parameters */
 290         if (req->type != EVHTTP_REQ_POST ||
 291             evbuffer_get_length(req->input_buffer) <= 0)
 292                 goto error;
 293
 294         rpc_state = mm_calloc(1, sizeof(struct evrpc_req_generic));
 295         if (rpc_state == NULL)
 296                 goto error;
 297         rpc_state->rpc = rpc;
 298         rpc_state->http_req = req;
 299         rpc_state->rpc_data = NULL;
 300
 301         if (TAILQ_FIRST(&rpc->base->input_hooks) != NULL) {
 302                 int hook_res;
 303
 304                 evrpc_hook_associate_meta(&rpc_state->hook_meta, req->evcon);
 305
 306                 /*
 307                  * allow hooks to modify the outgoing request
 308                  */
 309                 hook_res = evrpc_process_hooks(&rpc->base->input_hooks,
 310                     rpc_state, req, req->input_buffer);
 311                 switch (hook_res) {
 312                 case EVRPC_TERMINATE:
 313                         goto error;
 314                 case EVRPC_PAUSE:
 315                         evrpc_pause_request(rpc->base, rpc_state,
 316                             evrpc_request_cb_closure);
 317                         return;
 318                 case EVRPC_CONTINUE:
 319                         break;
 320                 default:
 321                         EVUTIL_ASSERT(hook_res == EVRPC_TERMINATE ||
 322                             hook_res == EVRPC_CONTINUE ||
 323                             hook_res == EVRPC_PAUSE);
 324                 }
 325         }
 326
 327         evrpc_request_cb_closure(rpc_state, EVRPC_CONTINUE);
 328         return;
 329
 330 error:
 331         if (rpc_state != NULL)
 332                 evrpc_reqstate_free(rpc_state);
 333         evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
 334         return;
 335 }
 336
 337 static void
 338 evrpc_request_cb_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
 339 {
 340         struct evrpc_req_generic *rpc_state = arg;
 341         struct evrpc *rpc;
 342         struct evhttp_request *req;
 343
 344         EVUTIL_ASSERT(rpc_state);
 345         rpc = rpc_state->rpc;
 346         req = rpc_state->http_req;
 347
 348         if (hook_res == EVRPC_TERMINATE)
 349                 goto error;
 350
 351         /* let's check that we can parse the request */
 352         rpc_state->request = rpc->request_new(rpc->request_new_arg);
 353         if (rpc_state->request == NULL)
 354                 goto error;
 355
 356         if (rpc->request_unmarshal(
 357                     rpc_state->request, req->input_buffer) == -1) {
 358                 /* we failed to parse the request; that's a bummer */
 359                 goto error;
 360         }
 361
 362         /* at this point, we have a well formed request, prepare the reply */
 363
 364         rpc_state->reply = rpc->reply_new(rpc->reply_new_arg);
 365         if (rpc_state->reply == NULL)
 366                 goto error;
 367
 368         /* give the rpc to the user; they can deal with it */
 369         rpc->cb(rpc_state, rpc->cb_arg);
 370
 371         return;
 372
 373 error:
 374         if (rpc_state != NULL)
 375                 evrpc_reqstate_free(rpc_state);
 376         evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
 377         return;
 378 }
 379
 380
 381 void
 382 evrpc_reqstate_free(struct evrpc_req_generic* rpc_state)
 383 {
 384         struct evrpc *rpc;
 385         EVUTIL_ASSERT(rpc_state != NULL);
 386         rpc = rpc_state->rpc;
 387
 388         /* clean up all memory */
 389         if (rpc_state->hook_meta != NULL)
 390                 evrpc_hook_context_free(rpc_state->hook_meta);
 391         if (rpc_state->request != NULL)
 392                 rpc->request_free(rpc_state->request);
 393         if (rpc_state->reply != NULL)
 394                 rpc->reply_free(rpc_state->reply);
 395         if (rpc_state->rpc_data != NULL)
 396                 evbuffer_free(rpc_state->rpc_data);
 397         mm_free(rpc_state);
 398 }
 399
 400 static void
 401 evrpc_request_done_closure(void *, enum EVRPC_HOOK_RESULT);
 402
 403 void
 404 evrpc_request_done(struct evrpc_req_generic *rpc_state)
 405 {
 406         struct evhttp_request *req;
 407         struct evrpc *rpc;
 408
 409         EVUTIL_ASSERT(rpc_state);
 410
 411         req = rpc_state->http_req;
 412         rpc = rpc_state->rpc;
 413
 414         if (rpc->reply_complete(rpc_state->reply) == -1) {
 415                 /* the reply was not completely filled in.  error out */
 416                 goto error;
 417         }
 418
 419         if ((rpc_state->rpc_data = evbuffer_new()) == NULL) {
 420                 /* out of memory */
 421                 goto error;
 422         }
 423
 424         /* serialize the reply */
 425         rpc->reply_marshal(rpc_state->rpc_data, rpc_state->reply);
 426
 427         if (TAILQ_FIRST(&rpc->base->output_hooks) != NULL) {
 428                 int hook_res;
 429
 430                 evrpc_hook_associate_meta(&rpc_state->hook_meta, req->evcon);
 431
 432                 /* do hook based tweaks to the request */
 433                 hook_res = evrpc_process_hooks(&rpc->base->output_hooks,
 434                     rpc_state, req, rpc_state->rpc_data);
 435                 switch (hook_res) {
 436                 case EVRPC_TERMINATE:
 437                         goto error;
 438                 case EVRPC_PAUSE:
 439                         if (evrpc_pause_request(rpc->base, rpc_state,
 440                                 evrpc_request_done_closure) == -1)
 441                                 goto error;
 442                         return;
 443                 case EVRPC_CONTINUE:
 444                         break;
 445                 default:
 446                         EVUTIL_ASSERT(hook_res == EVRPC_TERMINATE ||
 447                             hook_res == EVRPC_CONTINUE ||
 448                             hook_res == EVRPC_PAUSE);
 449                 }
 450         }
 451
 452         evrpc_request_done_closure(rpc_state, EVRPC_CONTINUE);
 453         return;
 454
 455 error:
 456         if (rpc_state != NULL)
 457                 evrpc_reqstate_free(rpc_state);
 458         evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
 459         return;
 460 }
 461
 462 void *
 463 evrpc_get_request(struct evrpc_req_generic *req)
 464 {
 465         return req->request;
 466 }
 467
 468 void *
 469 evrpc_get_reply(struct evrpc_req_generic *req)
 470 {
 471         return req->reply;
 472 }
 473
 474 static void
 475 evrpc_request_done_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
 476 {
 477         struct evrpc_req_generic *rpc_state = arg;
 478         struct evhttp_request *req;
 479         EVUTIL_ASSERT(rpc_state);
 480         req = rpc_state->http_req;
 481
 482         if (hook_res == EVRPC_TERMINATE)
 483                 goto error;
 484
 485         /* on success, we are going to transmit marshaled binary data */
 486         if (evhttp_find_header(req->output_headers, "Content-Type") == NULL) {
 487                 evhttp_add_header(req->output_headers,
 488                     "Content-Type", "application/octet-stream");
 489         }
 490         evhttp_send_reply(req, HTTP_OK, "OK", rpc_state->rpc_data);
 491
 492         evrpc_reqstate_free(rpc_state);
 493
 494         return;
 495
 496 error:
 497         if (rpc_state != NULL)
 498                 evrpc_reqstate_free(rpc_state);
 499         evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
 500         return;
 501 }
 502
 503
 504 /* Client implementation of RPC site */
 505
 506 static int evrpc_schedule_request(struct evhttp_connection *connection,
 507     struct evrpc_request_wrapper *ctx);
 508
 509 struct evrpc_pool *
 510 evrpc_pool_new(struct event_base *base)
 511 {
 512         struct evrpc_pool *pool = mm_calloc(1, sizeof(struct evrpc_pool));
 513         if (pool == NULL)
 514                 return (NULL);
 515
 516         TAILQ_INIT(&pool->connections);
 517         TAILQ_INIT(&pool->requests);
 518
 519         TAILQ_INIT(&pool->paused_requests);
 520
 521         TAILQ_INIT(&pool->input_hooks);
 522         TAILQ_INIT(&pool->output_hooks);
 523
 524         pool->base = base;
 525         pool->timeout = -1;
 526
 527         return (pool);
 528 }
 529
 530 static void
 531 evrpc_request_wrapper_free(struct evrpc_request_wrapper *request)
 532 {
 533         if (request->hook_meta != NULL)
 534                 evrpc_hook_context_free(request->hook_meta);
 535         mm_free(request->name);
 536         mm_free(request);
 537 }
 538
 539 void
 540 evrpc_pool_free(struct evrpc_pool *pool)
 541 {
 542         struct evhttp_connection *connection;
 543         struct evrpc_request_wrapper *request;
 544         struct evrpc_hook_ctx *pause;
 545         struct evrpc_hook *hook;
 546         int r;
 547
 548         while ((request = TAILQ_FIRST(&pool->requests)) != NULL) {
 549                 TAILQ_REMOVE(&pool->requests, request, next);
 550                 evrpc_request_wrapper_free(request);
 551         }
 552
 553         while ((pause = TAILQ_FIRST(&pool->paused_requests)) != NULL) {
 554                 TAILQ_REMOVE(&pool->paused_requests, pause, next);
 555                 mm_free(pause);
 556         }
 557
 558         while ((connection = TAILQ_FIRST(&pool->connections)) != NULL) {
 559                 TAILQ_REMOVE(&pool->connections, connection, next);
 560                 evhttp_connection_free(connection);
 561         }
 562
 563         while ((hook = TAILQ_FIRST(&pool->input_hooks)) != NULL) {
 564                 r = evrpc_remove_hook(pool, EVRPC_INPUT, hook);
 565                 EVUTIL_ASSERT(r);
 566         }
 567
 568         while ((hook = TAILQ_FIRST(&pool->output_hooks)) != NULL) {
 569                 r = evrpc_remove_hook(pool, EVRPC_OUTPUT, hook);
 570                 EVUTIL_ASSERT(r);
 571         }
 572
 573         mm_free(pool);
 574 }
 575
 576 /*
 577  * Add a connection to the RPC pool.   A request scheduled on the pool
 578  * may use any available connection.
 579  */
 580
 581 void
 582 evrpc_pool_add_connection(struct evrpc_pool *pool,
 583     struct evhttp_connection *connection)
 584 {
 585         EVUTIL_ASSERT(connection->http_server == NULL);
 586         TAILQ_INSERT_TAIL(&pool->connections, connection, next);
 587
 588         /*
 589          * associate an event base with this connection
 590          */
 591         if (pool->base != NULL)
 592                 evhttp_connection_set_base(connection, pool->base);
 593
 594         /*
 595          * unless a timeout was specifically set for a connection,
 596          * the connection inherits the timeout from the pool.
 597          */
 598         if (connection->timeout == -1)
 599                 connection->timeout = pool->timeout;
 600
 601         /*
 602          * if we have any requests pending, schedule them with the new
 603          * connections.
 604          */
 605
 606         if (TAILQ_FIRST(&pool->requests) != NULL) {
 607                 struct evrpc_request_wrapper *request =
 608                     TAILQ_FIRST(&pool->requests);
 609                 TAILQ_REMOVE(&pool->requests, request, next);
 610                 evrpc_schedule_request(connection, request);
 611         }
 612 }
 613
 614 void
 615 evrpc_pool_remove_connection(struct evrpc_pool *pool,
 616     struct evhttp_connection *connection)
 617 {
 618         TAILQ_REMOVE(&pool->connections, connection, next);
 619 }
 620
 621 void
 622 evrpc_pool_set_timeout(struct evrpc_pool *pool, int timeout_in_secs)
 623 {
 624         struct evhttp_connection *evcon;
 625         TAILQ_FOREACH(evcon, &pool->connections, next) {
 626                 evcon->timeout = timeout_in_secs;
 627         }
 628         pool->timeout = timeout_in_secs;
 629 }
 630
 631
 632 static void evrpc_reply_done(struct evhttp_request *, void *);
 633 static void evrpc_request_timeout(evutil_socket_t, short, void *);
 634
 635 /*
 636  * Finds a connection object associated with the pool that is currently
 637  * idle and can be used to make a request.
 638  */
 639 static struct evhttp_connection *
 640 evrpc_pool_find_connection(struct evrpc_pool *pool)
 641 {
 642         struct evhttp_connection *connection;
 643         TAILQ_FOREACH(connection, &pool->connections, next) {
 644                 if (TAILQ_FIRST(&connection->requests) == NULL)
 645                         return (connection);
 646         }
 647
 648         return (NULL);
 649 }
 650
 651 /*
 652  * Prototypes responsible for evrpc scheduling and hooking
 653  */
 654
 655 static void evrpc_schedule_request_closure(void *ctx, enum EVRPC_HOOK_RESULT);
 656
 657 /*
 658  * We assume that the ctx is no longer queued on the pool.
 659  */
 660 static int
 661 evrpc_schedule_request(struct evhttp_connection *connection,
 662     struct evrpc_request_wrapper *ctx)
 663 {
 664         struct evhttp_request *req = NULL;
 665         struct evrpc_pool *pool = ctx->pool;
 666         struct evrpc_status status;
 667
 668         if ((req = evhttp_request_new(evrpc_reply_done, ctx)) == NULL)
 669                 goto error;
 670
 671         /* serialize the request data into the output buffer */
 672         ctx->request_marshal(req->output_buffer, ctx->request);
 673
 674         /* we need to know the connection that we might have to abort */
 675         ctx->evcon = connection;
 676
 677         /* if we get paused we also need to know the request */
 678         ctx->req = req;
 679
 680         if (TAILQ_FIRST(&pool->output_hooks) != NULL) {
 681                 int hook_res;
 682
 683                 evrpc_hook_associate_meta(&ctx->hook_meta, connection);
 684
 685                 /* apply hooks to the outgoing request */
 686                 hook_res = evrpc_process_hooks(&pool->output_hooks,
 687                     ctx, req, req->output_buffer);
 688
 689                 switch (hook_res) {
 690                 case EVRPC_TERMINATE:
 691                         goto error;
 692                 case EVRPC_PAUSE:
 693                         /* we need to be explicitly resumed */
 694                         if (evrpc_pause_request(pool, ctx,
 695                                 evrpc_schedule_request_closure) == -1)
 696                                 goto error;
 697                         return (0);
 698                 case EVRPC_CONTINUE:
 699                         /* we can just continue */
 700                         break;
 701                 default:
 702                         EVUTIL_ASSERT(hook_res == EVRPC_TERMINATE ||
 703                             hook_res == EVRPC_CONTINUE ||
 704                             hook_res == EVRPC_PAUSE);
 705                 }
 706         }
 707
 708         evrpc_schedule_request_closure(ctx, EVRPC_CONTINUE);
 709         return (0);
 710
 711 error:
 712         memset(&status, 0, sizeof(status));
 713         status.error = EVRPC_STATUS_ERR_UNSTARTED;
 714         (*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
 715         evrpc_request_wrapper_free(ctx);
 716         return (-1);
 717 }
 718
 719 static void
 720 evrpc_schedule_request_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
 721 {
 722         struct evrpc_request_wrapper *ctx = arg;
 723         struct evhttp_connection *connection = ctx->evcon;
 724         struct evhttp_request *req = ctx->req;
 725         struct evrpc_pool *pool = ctx->pool;
 726         struct evrpc_status status;
 727         char *uri = NULL;
 728         int res = 0;
 729
 730         if (hook_res == EVRPC_TERMINATE)
 731                 goto error;
 732
 733         uri = evrpc_construct_uri(ctx->name);
 734         if (uri == NULL)
 735                 goto error;
 736
 737         if (pool->timeout > 0) {
 738                 /*
 739                  * a timeout after which the whole rpc is going to be aborted.
 740                  */
 741                 struct timeval tv;
 742                 evutil_timerclear(&tv);
 743                 tv.tv_sec = pool->timeout;
 744                 evtimer_add(&ctx->ev_timeout, &tv);
 745         }
 746
 747         /* start the request over the connection */
 748         res = evhttp_make_request(connection, req, EVHTTP_REQ_POST, uri);
 749         mm_free(uri);
 750
 751         if (res == -1)
 752                 goto error;
 753
 754         return;
 755
 756 error:
 757         memset(&status, 0, sizeof(status));
 758         status.error = EVRPC_STATUS_ERR_UNSTARTED;
 759         (*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
 760         evrpc_request_wrapper_free(ctx);
 761 }
 762
 763 /* we just queue the paused request on the pool under the req object */
 764 static int
 765 evrpc_pause_request(void *vbase, void *ctx,
 766     void (*cb)(void *, enum EVRPC_HOOK_RESULT))
 767 {
 768         struct _evrpc_hooks *base = vbase;
 769         struct evrpc_hook_ctx *pause = mm_malloc(sizeof(*pause));
 770         if (pause == NULL)
 771                 return (-1);
 772
 773         pause->ctx = ctx;
 774         pause->cb = cb;
 775
 776         TAILQ_INSERT_TAIL(&base->pause_requests, pause, next);
 777         return (0);
 778 }
 779
 780 int
 781 evrpc_resume_request(void *vbase, void *ctx, enum EVRPC_HOOK_RESULT res)
 782 {
 783         struct _evrpc_hooks *base = vbase;
 784         struct evrpc_pause_list *head = &base->pause_requests;
 785         struct evrpc_hook_ctx *pause;
 786
 787         TAILQ_FOREACH(pause, head, next) {
 788                 if (pause->ctx == ctx)
 789                         break;
 790         }
 791
 792         if (pause == NULL)
 793                 return (-1);
 794
 795         (*pause->cb)(pause->ctx, res);
 796         TAILQ_REMOVE(head, pause, next);
 797         mm_free(pause);
 798         return (0);
 799 }
 800
 801 int
 802 evrpc_make_request(struct evrpc_request_wrapper *ctx)
 803 {
 804         struct evrpc_pool *pool = ctx->pool;
 805
 806         /* initialize the event structure for this rpc */
 807         evtimer_assign(&ctx->ev_timeout, pool->base, evrpc_request_timeout, ctx);
 808
 809         /* we better have some available connections on the pool */
 810         EVUTIL_ASSERT(TAILQ_FIRST(&pool->connections) != NULL);
 811
 812         /*
 813          * if no connection is available, we queue the request on the pool,
 814          * the next time a connection is empty, the rpc will be send on that.
 815          */
 816         TAILQ_INSERT_TAIL(&pool->requests, ctx, next);
 817
 818         evrpc_pool_schedule(pool);
 819
 820         return (0);
 821 }
 822
 823
 824 struct evrpc_request_wrapper *
 825 evrpc_make_request_ctx(
 826         struct evrpc_pool *pool, void *request, void *reply,
 827         const char *rpcname,
 828         void (*req_marshal)(struct evbuffer*, void *),
 829         void (*rpl_clear)(void *),
 830         int (*rpl_unmarshal)(void *, struct evbuffer *),
 831         void (*cb)(struct evrpc_status *, void *, void *, void *),
 832         void *cbarg)
 833 {
 834         struct evrpc_request_wrapper *ctx = (struct evrpc_request_wrapper *)
 835             mm_malloc(sizeof(struct evrpc_request_wrapper));
 836         if (ctx == NULL)
 837                 return (NULL);
 838
 839         ctx->pool = pool;
 840         ctx->hook_meta = NULL;
 841         ctx->evcon = NULL;
 842         ctx->name = mm_strdup(rpcname);
 843         if (ctx->name == NULL) {
 844                 mm_free(ctx);
 845                 return (NULL);
 846         }
 847         ctx->cb = cb;
 848         ctx->cb_arg = cbarg;
 849         ctx->request = request;
 850         ctx->reply = reply;
 851         ctx->request_marshal = req_marshal;
 852         ctx->reply_clear = rpl_clear;
 853         ctx->reply_unmarshal = rpl_unmarshal;
 854
 855         return (ctx);
 856 }
 857
 858 static void
 859 evrpc_reply_done_closure(void *, enum EVRPC_HOOK_RESULT);
 860
 861 static void
 862 evrpc_reply_done(struct evhttp_request *req, void *arg)
 863 {
 864         struct evrpc_request_wrapper *ctx = arg;
 865         struct evrpc_pool *pool = ctx->pool;
 866         int hook_res = EVRPC_CONTINUE;
 867
 868         /* cancel any timeout we might have scheduled */
 869         event_del(&ctx->ev_timeout);
 870
 871         ctx->req = req;
 872
 873         /* we need to get the reply now */
 874         if (req == NULL) {
 875                 evrpc_reply_done_closure(ctx, EVRPC_CONTINUE);
 876                 return;
 877         }
 878
 879         if (TAILQ_FIRST(&pool->input_hooks) != NULL) {
 880                 evrpc_hook_associate_meta(&ctx->hook_meta, ctx->evcon);
 881
 882                 /* apply hooks to the incoming request */
 883                 hook_res = evrpc_process_hooks(&pool->input_hooks,
 884                     ctx, req, req->input_buffer);
 885
 886                 switch (hook_res) {
 887                 case EVRPC_TERMINATE:
 888                 case EVRPC_CONTINUE:
 889                         break;
 890                 case EVRPC_PAUSE:
 891                         /*
 892                          * if we get paused we also need to know the
 893                          * request.  unfortunately, the underlying
 894                          * layer is going to free it.  we need to
 895                          * request ownership explicitly
 896                          */
 897                         if (req != NULL)
 898                                 evhttp_request_own(req);
 899
 900                         evrpc_pause_request(pool, ctx,
 901                             evrpc_reply_done_closure);
 902                         return;
 903                 default:
 904                         EVUTIL_ASSERT(hook_res == EVRPC_TERMINATE ||
 905                             hook_res == EVRPC_CONTINUE ||
 906                             hook_res == EVRPC_PAUSE);
 907                 }
 908         }
 909
 910         evrpc_reply_done_closure(ctx, hook_res);
 911
 912         /* http request is being freed by underlying layer */
 913 }
 914
 915 static void
 916 evrpc_reply_done_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
 917 {
 918         struct evrpc_request_wrapper *ctx = arg;
 919         struct evhttp_request *req = ctx->req;
 920         struct evrpc_pool *pool = ctx->pool;
 921         struct evrpc_status status;
 922         int res = -1;
 923
 924         memset(&status, 0, sizeof(status));
 925         status.http_req = req;
 926
 927         /* we need to get the reply now */
 928         if (req == NULL) {
 929                 status.error = EVRPC_STATUS_ERR_TIMEOUT;
 930         } else if (hook_res == EVRPC_TERMINATE) {
 931                 status.error = EVRPC_STATUS_ERR_HOOKABORTED;
 932         } else {
 933                 res = ctx->reply_unmarshal(ctx->reply, req->input_buffer);
 934                 if (res == -1)
 935                         status.error = EVRPC_STATUS_ERR_BADPAYLOAD;
 936         }
 937
 938         if (res == -1) {
 939                 /* clear everything that we might have written previously */
 940                 ctx->reply_clear(ctx->reply);
 941         }
 942
 943         (*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
 944
 945         evrpc_request_wrapper_free(ctx);
 946
 947         /* the http layer owned the original request structure, but if we
 948          * got paused, we asked for ownership and need to free it here. */
 949         if (req != NULL && evhttp_request_is_owned(req))
 950                 evhttp_request_free(req);
 951
 952         /* see if we can schedule another request */
 953         evrpc_pool_schedule(pool);
 954 }
 955
 956 static void
 957 evrpc_pool_schedule(struct evrpc_pool *pool)
 958 {
 959         struct evrpc_request_wrapper *ctx = TAILQ_FIRST(&pool->requests);
 960         struct evhttp_connection *evcon;
 961
 962         /* if no requests are pending, we have no work */
 963         if (ctx == NULL)
 964                 return;
 965
 966         if ((evcon = evrpc_pool_find_connection(pool)) != NULL) {
 967                 TAILQ_REMOVE(&pool->requests, ctx, next);
 968                 evrpc_schedule_request(evcon, ctx);
 969         }
 970 }
 971
 972 static void
 973 evrpc_request_timeout(evutil_socket_t fd, short what, void *arg)
 974 {
 975         struct evrpc_request_wrapper *ctx = arg;
 976         struct evhttp_connection *evcon = ctx->evcon;
 977         EVUTIL_ASSERT(evcon != NULL);
 978
 979         evhttp_connection_fail(evcon, EVCON_HTTP_TIMEOUT);
 980 }
 981
 982 /*
 983  * frees potential meta data associated with a request.
 984  */
 985
 986 static void
 987 evrpc_meta_data_free(struct evrpc_meta_list *meta_data)
 988 {
 989         struct evrpc_meta *entry;
 990         EVUTIL_ASSERT(meta_data != NULL);
 991
 992         while ((entry = TAILQ_FIRST(meta_data)) != NULL) {
 993                 TAILQ_REMOVE(meta_data, entry, next);
 994                 mm_free(entry->key);
 995                 mm_free(entry->data);
 996                 mm_free(entry);
 997         }
 998 }
 999
1000 static struct evrpc_hook_meta *
1001 evrpc_hook_meta_new(void)
1002 {
1003         struct evrpc_hook_meta *ctx;
1004         ctx = mm_malloc(sizeof(struct evrpc_hook_meta));
1005         EVUTIL_ASSERT(ctx != NULL);
1006
1007         TAILQ_INIT(&ctx->meta_data);
1008         ctx->evcon = NULL;
1009
1010         return (ctx);
1011 }
1012
1013 static void
1014 evrpc_hook_associate_meta(struct evrpc_hook_meta **pctx,
1015     struct evhttp_connection *evcon)
1016 {
1017         struct evrpc_hook_meta *ctx = *pctx;
1018         if (ctx == NULL)
1019                 *pctx = ctx = evrpc_hook_meta_new();
1020         ctx->evcon = evcon;
1021 }
1022
1023 static void
1024 evrpc_hook_context_free(struct evrpc_hook_meta *ctx)
1025 {
1026         evrpc_meta_data_free(&ctx->meta_data);
1027         mm_free(ctx);
1028 }
1029
1030 /* Adds meta data */
1031 void
1032 evrpc_hook_add_meta(void *ctx, const char *key,
1033     const void *data, size_t data_size)
1034 {
1035         struct evrpc_request_wrapper *req = ctx;
1036         struct evrpc_hook_meta *store = NULL;
1037         struct evrpc_meta *meta = NULL;
1038
1039         if ((store = req->hook_meta) == NULL)
1040                 store = req->hook_meta = evrpc_hook_meta_new();
1041
1042         meta = mm_malloc(sizeof(struct evrpc_meta));
1043         EVUTIL_ASSERT(meta != NULL);
1044         meta->key = mm_strdup(key);
1045         EVUTIL_ASSERT(meta->key != NULL);
1046         meta->data_size = data_size;
1047         meta->data = mm_malloc(data_size);
1048         EVUTIL_ASSERT(meta->data != NULL);
1049         memcpy(meta->data, data, data_size);
1050
1051         TAILQ_INSERT_TAIL(&store->meta_data, meta, next);
1052 }
1053
1054 int
1055 evrpc_hook_find_meta(void *ctx, const char *key, void **data, size_t *data_size)
1056 {
1057         struct evrpc_request_wrapper *req = ctx;
1058         struct evrpc_meta *meta = NULL;
1059
1060         if (req->hook_meta == NULL)
1061                 return (-1);
1062
1063         TAILQ_FOREACH(meta, &req->hook_meta->meta_data, next) {
1064                 if (strcmp(meta->key, key) == 0) {
1065                         *data = meta->data;
1066                         *data_size = meta->data_size;
1067                         return (0);
1068                 }
1069         }
1070
1071         return (-1);
1072 }
1073
1074 struct evhttp_connection *
1075 evrpc_hook_get_connection(void *ctx)
1076 {
1077         struct evrpc_request_wrapper *req = ctx;
1078         return (req->hook_meta != NULL ? req->hook_meta->evcon : NULL);
1079 }
1080
1081 int
1082 evrpc_send_request_generic(struct evrpc_pool *pool,
1083     void *request, void *reply,
1084     void (*cb)(struct evrpc_status *, void *, void *, void *),
1085     void *cb_arg,
1086     const char *rpcname,
1087     void (*req_marshal)(struct evbuffer *, void *),
1088     void (*rpl_clear)(void *),
1089     int (*rpl_unmarshal)(void *, struct evbuffer *))
1090 {
1091         struct evrpc_status status;
1092         struct evrpc_request_wrapper *ctx;
1093         ctx = evrpc_make_request_ctx(pool, request, reply,
1094             rpcname, req_marshal, rpl_clear, rpl_unmarshal, cb, cb_arg);
1095         if (ctx == NULL)
1096                 goto error;
1097         return (evrpc_make_request(ctx));
1098 error:
1099         memset(&status, 0, sizeof(status));
1100         status.error = EVRPC_STATUS_ERR_UNSTARTED;
1101         (*(cb))(&status, request, reply, cb_arg);
1102         return (-1);
1103 }
1104
1105 /** Takes a request object and fills it in with the right magic */
1106 static struct evrpc *
1107 evrpc_register_object(const char *name,
1108     void *(*req_new)(void*), void *req_new_arg, void (*req_free)(void *),
1109     int (*req_unmarshal)(void *, struct evbuffer *),
1110     void *(*rpl_new)(void*), void *rpl_new_arg, void (*rpl_free)(void *),
1111     int (*rpl_complete)(void *),
1112     void (*rpl_marshal)(struct evbuffer *, void *))
1113 {
1114         struct evrpc* rpc = (struct evrpc *)mm_calloc(1, sizeof(struct evrpc));
1115         if (rpc == NULL)
1116                 return (NULL);
1117         rpc->uri = mm_strdup(name);
1118         if (rpc->uri == NULL) {
1119                 mm_free(rpc);
1120                 return (NULL);
1121         }
1122         rpc->request_new = req_new;
1123         rpc->request_new_arg = req_new_arg;
1124         rpc->request_free = req_free;
1125         rpc->request_unmarshal = req_unmarshal;
1126         rpc->reply_new = rpl_new;
1127         rpc->reply_new_arg = rpl_new_arg;
1128         rpc->reply_free = rpl_free;
1129         rpc->reply_complete = rpl_complete;
1130         rpc->reply_marshal = rpl_marshal;
1131         return (rpc);
1132 }
1133
1134 int
1135 evrpc_register_generic(struct evrpc_base *base, const char *name,
1136     void (*callback)(struct evrpc_req_generic *, void *), void *cbarg,
1137     void *(*req_new)(void *), void *req_new_arg, void (*req_free)(void *),
1138     int (*req_unmarshal)(void *, struct evbuffer *),
1139     void *(*rpl_new)(void *), void *rpl_new_arg, void (*rpl_free)(void *),
1140     int (*rpl_complete)(void *),
1141     void (*rpl_marshal)(struct evbuffer *, void *))
1142 {
1143         struct evrpc* rpc =
1144             evrpc_register_object(name, req_new, req_new_arg, req_free, req_unmarshal,
1145                 rpl_new, rpl_new_arg, rpl_free, rpl_complete, rpl_marshal);
1146         if (rpc == NULL)
1147                 return (-1);
1148         evrpc_register_rpc(base, rpc,
1149             (void (*)(struct evrpc_req_generic*, void *))callback, cbarg);
1150         return (0);
1151 }
1152
1153 /** accessors for obscure and undocumented functionality */
1154 struct evrpc_pool *
1155 evrpc_request_get_pool(struct evrpc_request_wrapper *ctx)
1156 {
1157         return (ctx->pool);
1158 }
1159
1160 void
1161 evrpc_request_set_pool(struct evrpc_request_wrapper *ctx,
1162     struct evrpc_pool *pool)
1163 {
1164         ctx->pool = pool;
1165 }
1166
1167 void
1168 evrpc_request_set_cb(struct evrpc_request_wrapper *ctx,
1169     void (*cb)(struct evrpc_status*, void *request, void *reply, void *arg),
1170     void *cb_arg)
1171 {
1172         ctx->cb = cb;
1173         ctx->cb_arg = cb_arg;
1174 }