libevent/evrpc.c

   1 /*
   2  * Copyright (c) 2000-2007 Niels Provos <provos@citi.umich.edu>
   3  * Copyright (c) 2007-2010 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 = rpc_state->rpc;
 342         struct evhttp_request *req = rpc_state->http_req;
 343
 344         if (hook_res == EVRPC_TERMINATE)
 345                 goto error;
 346
 347         /* let's check that we can parse the request */
 348         rpc_state->request = rpc->request_new(rpc->request_new_arg);
 349         if (rpc_state->request == NULL)
 350                 goto error;
 351
 352         if (rpc->request_unmarshal(
 353                     rpc_state->request, req->input_buffer) == -1) {
 354                 /* we failed to parse the request; that's a bummer */
 355                 goto error;
 356         }
 357
 358         /* at this point, we have a well formed request, prepare the reply */
 359
 360         rpc_state->reply = rpc->reply_new(rpc->reply_new_arg);
 361         if (rpc_state->reply == NULL)
 362                 goto error;
 363
 364         /* give the rpc to the user; they can deal with it */
 365         rpc->cb(rpc_state, rpc->cb_arg);
 366
 367         return;
 368
 369 error:
 370         if (rpc_state != NULL)
 371                 evrpc_reqstate_free(rpc_state);
 372         evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
 373         return;
 374 }
 375
 376
 377 void
 378 evrpc_reqstate_free(struct evrpc_req_generic* rpc_state)
 379 {
 380         struct evrpc *rpc;
 381         EVUTIL_ASSERT(rpc_state != NULL);
 382         rpc = rpc_state->rpc;
 383
 384         /* clean up all memory */
 385         if (rpc_state->hook_meta != NULL)
 386                 evrpc_hook_context_free(rpc_state->hook_meta);
 387         if (rpc_state->request != NULL)
 388                 rpc->request_free(rpc_state->request);
 389         if (rpc_state->reply != NULL)
 390                 rpc->reply_free(rpc_state->reply);
 391         if (rpc_state->rpc_data != NULL)
 392                 evbuffer_free(rpc_state->rpc_data);
 393         mm_free(rpc_state);
 394 }
 395
 396 static void
 397 evrpc_request_done_closure(void *, enum EVRPC_HOOK_RESULT);
 398
 399 void
 400 evrpc_request_done(struct evrpc_req_generic *rpc_state)
 401 {
 402         struct evhttp_request *req = rpc_state->http_req;
 403         struct evrpc *rpc = rpc_state->rpc;
 404
 405         if (rpc->reply_complete(rpc_state->reply) == -1) {
 406                 /* the reply was not completely filled in.  error out */
 407                 goto error;
 408         }
 409
 410         if ((rpc_state->rpc_data = evbuffer_new()) == NULL) {
 411                 /* out of memory */
 412                 goto error;
 413         }
 414
 415         /* serialize the reply */
 416         rpc->reply_marshal(rpc_state->rpc_data, rpc_state->reply);
 417
 418         if (TAILQ_FIRST(&rpc->base->output_hooks) != NULL) {
 419                 int hook_res;
 420
 421                 evrpc_hook_associate_meta(&rpc_state->hook_meta, req->evcon);
 422
 423                 /* do hook based tweaks to the request */
 424                 hook_res = evrpc_process_hooks(&rpc->base->output_hooks,
 425                     rpc_state, req, rpc_state->rpc_data);
 426                 switch (hook_res) {
 427                 case EVRPC_TERMINATE:
 428                         goto error;
 429                 case EVRPC_PAUSE:
 430                         if (evrpc_pause_request(rpc->base, rpc_state,
 431                                 evrpc_request_done_closure) == -1)
 432                                 goto error;
 433                         return;
 434                 case EVRPC_CONTINUE:
 435                         break;
 436                 default:
 437                         EVUTIL_ASSERT(hook_res == EVRPC_TERMINATE ||
 438                             hook_res == EVRPC_CONTINUE ||
 439                             hook_res == EVRPC_PAUSE);
 440                 }
 441         }
 442
 443         evrpc_request_done_closure(rpc_state, EVRPC_CONTINUE);
 444         return;
 445
 446 error:
 447         if (rpc_state != NULL)
 448                 evrpc_reqstate_free(rpc_state);
 449         evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
 450         return;
 451 }
 452
 453 void *
 454 evrpc_get_request(struct evrpc_req_generic *req)
 455 {
 456         return req->request;
 457 }
 458
 459 void *
 460 evrpc_get_reply(struct evrpc_req_generic *req)
 461 {
 462         return req->reply;
 463 }
 464
 465 static void
 466 evrpc_request_done_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
 467 {
 468         struct evrpc_req_generic *rpc_state = arg;
 469         struct evhttp_request *req = rpc_state->http_req;
 470
 471         if (hook_res == EVRPC_TERMINATE)
 472                 goto error;
 473
 474         /* on success, we are going to transmit marshaled binary data */
 475         if (evhttp_find_header(req->output_headers, "Content-Type") == NULL) {
 476                 evhttp_add_header(req->output_headers,
 477                     "Content-Type", "application/octet-stream");
 478         }
 479         evhttp_send_reply(req, HTTP_OK, "OK", rpc_state->rpc_data);
 480
 481         evrpc_reqstate_free(rpc_state);
 482
 483         return;
 484
 485 error:
 486         if (rpc_state != NULL)
 487                 evrpc_reqstate_free(rpc_state);
 488         evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
 489         return;
 490 }
 491
 492
 493 /* Client implementation of RPC site */
 494
 495 static int evrpc_schedule_request(struct evhttp_connection *connection,
 496     struct evrpc_request_wrapper *ctx);
 497
 498 struct evrpc_pool *
 499 evrpc_pool_new(struct event_base *base)
 500 {
 501         struct evrpc_pool *pool = mm_calloc(1, sizeof(struct evrpc_pool));
 502         if (pool == NULL)
 503                 return (NULL);
 504
 505         TAILQ_INIT(&pool->connections);
 506         TAILQ_INIT(&pool->requests);
 507
 508         TAILQ_INIT(&pool->paused_requests);
 509
 510         TAILQ_INIT(&pool->input_hooks);
 511         TAILQ_INIT(&pool->output_hooks);
 512
 513         pool->base = base;
 514         pool->timeout = -1;
 515
 516         return (pool);
 517 }
 518
 519 static void
 520 evrpc_request_wrapper_free(struct evrpc_request_wrapper *request)
 521 {
 522         if (request->hook_meta != NULL)
 523                 evrpc_hook_context_free(request->hook_meta);
 524         mm_free(request->name);
 525         mm_free(request);
 526 }
 527
 528 void
 529 evrpc_pool_free(struct evrpc_pool *pool)
 530 {
 531         struct evhttp_connection *connection;
 532         struct evrpc_request_wrapper *request;
 533         struct evrpc_hook_ctx *pause;
 534         struct evrpc_hook *hook;
 535         int r;
 536
 537         while ((request = TAILQ_FIRST(&pool->requests)) != NULL) {
 538                 TAILQ_REMOVE(&pool->requests, request, next);
 539                 evrpc_request_wrapper_free(request);
 540         }
 541
 542         while ((pause = TAILQ_FIRST(&pool->paused_requests)) != NULL) {
 543                 TAILQ_REMOVE(&pool->paused_requests, pause, next);
 544                 mm_free(pause);
 545         }
 546
 547         while ((connection = TAILQ_FIRST(&pool->connections)) != NULL) {
 548                 TAILQ_REMOVE(&pool->connections, connection, next);
 549                 evhttp_connection_free(connection);
 550         }
 551
 552         while ((hook = TAILQ_FIRST(&pool->input_hooks)) != NULL) {
 553                 r = evrpc_remove_hook(pool, EVRPC_INPUT, hook);
 554                 EVUTIL_ASSERT(r);
 555         }
 556
 557         while ((hook = TAILQ_FIRST(&pool->output_hooks)) != NULL) {
 558                 r = evrpc_remove_hook(pool, EVRPC_OUTPUT, hook);
 559                 EVUTIL_ASSERT(r);
 560         }
 561
 562         mm_free(pool);
 563 }
 564
 565 /*
 566  * Add a connection to the RPC pool.   A request scheduled on the pool
 567  * may use any available connection.
 568  */
 569
 570 void
 571 evrpc_pool_add_connection(struct evrpc_pool *pool,
 572     struct evhttp_connection *connection)
 573 {
 574         EVUTIL_ASSERT(connection->http_server == NULL);
 575         TAILQ_INSERT_TAIL(&pool->connections, connection, next);
 576
 577         /*
 578          * associate an event base with this connection
 579          */
 580         if (pool->base != NULL)
 581                 evhttp_connection_set_base(connection, pool->base);
 582
 583         /*
 584          * unless a timeout was specifically set for a connection,
 585          * the connection inherits the timeout from the pool.
 586          */
 587         if (connection->timeout == -1)
 588                 connection->timeout = pool->timeout;
 589
 590         /*
 591          * if we have any requests pending, schedule them with the new
 592          * connections.
 593          */
 594
 595         if (TAILQ_FIRST(&pool->requests) != NULL) {
 596                 struct evrpc_request_wrapper *request =
 597                     TAILQ_FIRST(&pool->requests);
 598                 TAILQ_REMOVE(&pool->requests, request, next);
 599                 evrpc_schedule_request(connection, request);
 600         }
 601 }
 602
 603 void
 604 evrpc_pool_remove_connection(struct evrpc_pool *pool,
 605     struct evhttp_connection *connection)
 606 {
 607         TAILQ_REMOVE(&pool->connections, connection, next);
 608 }
 609
 610 void
 611 evrpc_pool_set_timeout(struct evrpc_pool *pool, int timeout_in_secs)
 612 {
 613         struct evhttp_connection *evcon;
 614         TAILQ_FOREACH(evcon, &pool->connections, next) {
 615                 evcon->timeout = timeout_in_secs;
 616         }
 617         pool->timeout = timeout_in_secs;
 618 }
 619
 620
 621 static void evrpc_reply_done(struct evhttp_request *, void *);
 622 static void evrpc_request_timeout(evutil_socket_t, short, void *);
 623
 624 /*
 625  * Finds a connection object associated with the pool that is currently
 626  * idle and can be used to make a request.
 627  */
 628 static struct evhttp_connection *
 629 evrpc_pool_find_connection(struct evrpc_pool *pool)
 630 {
 631         struct evhttp_connection *connection;
 632         TAILQ_FOREACH(connection, &pool->connections, next) {
 633                 if (TAILQ_FIRST(&connection->requests) == NULL)
 634                         return (connection);
 635         }
 636
 637         return (NULL);
 638 }
 639
 640 /*
 641  * Prototypes responsible for evrpc scheduling and hooking
 642  */
 643
 644 static void evrpc_schedule_request_closure(void *ctx, enum EVRPC_HOOK_RESULT);
 645
 646 /*
 647  * We assume that the ctx is no longer queued on the pool.
 648  */
 649 static int
 650 evrpc_schedule_request(struct evhttp_connection *connection,
 651     struct evrpc_request_wrapper *ctx)
 652 {
 653         struct evhttp_request *req = NULL;
 654         struct evrpc_pool *pool = ctx->pool;
 655         struct evrpc_status status;
 656
 657         if ((req = evhttp_request_new(evrpc_reply_done, ctx)) == NULL)
 658                 goto error;
 659
 660         /* serialize the request data into the output buffer */
 661         ctx->request_marshal(req->output_buffer, ctx->request);
 662
 663         /* we need to know the connection that we might have to abort */
 664         ctx->evcon = connection;
 665
 666         /* if we get paused we also need to know the request */
 667         ctx->req = req;
 668
 669         if (TAILQ_FIRST(&pool->output_hooks) != NULL) {
 670                 int hook_res;
 671
 672                 evrpc_hook_associate_meta(&ctx->hook_meta, connection);
 673
 674                 /* apply hooks to the outgoing request */
 675                 hook_res = evrpc_process_hooks(&pool->output_hooks,
 676                     ctx, req, req->output_buffer);
 677
 678                 switch (hook_res) {
 679                 case EVRPC_TERMINATE:
 680                         goto error;
 681                 case EVRPC_PAUSE:
 682                         /* we need to be explicitly resumed */
 683                         if (evrpc_pause_request(pool, ctx,
 684                                 evrpc_schedule_request_closure) == -1)
 685                                 goto error;
 686                         return (0);
 687                 case EVRPC_CONTINUE:
 688                         /* we can just continue */
 689                         break;
 690                 default:
 691                         EVUTIL_ASSERT(hook_res == EVRPC_TERMINATE ||
 692                             hook_res == EVRPC_CONTINUE ||
 693                             hook_res == EVRPC_PAUSE);
 694                 }
 695         }
 696
 697         evrpc_schedule_request_closure(ctx, EVRPC_CONTINUE);
 698         return (0);
 699
 700 error:
 701         memset(&status, 0, sizeof(status));
 702         status.error = EVRPC_STATUS_ERR_UNSTARTED;
 703         (*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
 704         evrpc_request_wrapper_free(ctx);
 705         return (-1);
 706 }
 707
 708 static void
 709 evrpc_schedule_request_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
 710 {
 711         struct evrpc_request_wrapper *ctx = arg;
 712         struct evhttp_connection *connection = ctx->evcon;
 713         struct evhttp_request *req = ctx->req;
 714         struct evrpc_pool *pool = ctx->pool;
 715         struct evrpc_status status;
 716         char *uri = NULL;
 717         int res = 0;
 718
 719         if (hook_res == EVRPC_TERMINATE)
 720                 goto error;
 721
 722         uri = evrpc_construct_uri(ctx->name);
 723         if (uri == NULL)
 724                 goto error;
 725
 726         if (pool->timeout > 0) {
 727                 /*
 728                  * a timeout after which the whole rpc is going to be aborted.
 729                  */
 730                 struct timeval tv;
 731                 evutil_timerclear(&tv);
 732                 tv.tv_sec = pool->timeout;
 733                 evtimer_add(&ctx->ev_timeout, &tv);
 734         }
 735
 736         /* start the request over the connection */
 737         res = evhttp_make_request(connection, req, EVHTTP_REQ_POST, uri);
 738         mm_free(uri);
 739
 740         if (res == -1)
 741                 goto error;
 742
 743         return;
 744
 745 error:
 746         memset(&status, 0, sizeof(status));
 747         status.error = EVRPC_STATUS_ERR_UNSTARTED;
 748         (*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
 749         evrpc_request_wrapper_free(ctx);
 750 }
 751
 752 /* we just queue the paused request on the pool under the req object */
 753 static int
 754 evrpc_pause_request(void *vbase, void *ctx,
 755     void (*cb)(void *, enum EVRPC_HOOK_RESULT))
 756 {
 757         struct _evrpc_hooks *base = vbase;
 758         struct evrpc_hook_ctx *pause = mm_malloc(sizeof(*pause));
 759         if (pause == NULL)
 760                 return (-1);
 761
 762         pause->ctx = ctx;
 763         pause->cb = cb;
 764
 765         TAILQ_INSERT_TAIL(&base->pause_requests, pause, next);
 766         return (0);
 767 }
 768
 769 int
 770 evrpc_resume_request(void *vbase, void *ctx, enum EVRPC_HOOK_RESULT res)
 771 {
 772         struct _evrpc_hooks *base = vbase;
 773         struct evrpc_pause_list *head = &base->pause_requests;
 774         struct evrpc_hook_ctx *pause;
 775
 776         TAILQ_FOREACH(pause, head, next) {
 777                 if (pause->ctx == ctx)
 778                         break;
 779         }
 780
 781         if (pause == NULL)
 782                 return (-1);
 783
 784         (*pause->cb)(pause->ctx, res);
 785         TAILQ_REMOVE(head, pause, next);
 786         mm_free(pause);
 787         return (0);
 788 }
 789
 790 int
 791 evrpc_make_request(struct evrpc_request_wrapper *ctx)
 792 {
 793         struct evrpc_pool *pool = ctx->pool;
 794
 795         /* initialize the event structure for this rpc */
 796         evtimer_assign(&ctx->ev_timeout, pool->base, evrpc_request_timeout, ctx);
 797
 798         /* we better have some available connections on the pool */
 799         EVUTIL_ASSERT(TAILQ_FIRST(&pool->connections) != NULL);
 800
 801         /*
 802          * if no connection is available, we queue the request on the pool,
 803          * the next time a connection is empty, the rpc will be send on that.
 804          */
 805         TAILQ_INSERT_TAIL(&pool->requests, ctx, next);
 806
 807         evrpc_pool_schedule(pool);
 808
 809         return (0);
 810 }
 811
 812
 813 struct evrpc_request_wrapper *
 814 evrpc_make_request_ctx(
 815         struct evrpc_pool *pool, void *request, void *reply,
 816         const char *rpcname,
 817         void (*req_marshal)(struct evbuffer*, void *),
 818         void (*rpl_clear)(void *),
 819         int (*rpl_unmarshal)(void *, struct evbuffer *),
 820         void (*cb)(struct evrpc_status *, void *, void *, void *),
 821         void *cbarg)
 822 {
 823         struct evrpc_request_wrapper *ctx = (struct evrpc_request_wrapper *)
 824             mm_malloc(sizeof(struct evrpc_request_wrapper));
 825         if (ctx == NULL)
 826                 return (NULL);
 827
 828         ctx->pool = pool;
 829         ctx->hook_meta = NULL;
 830         ctx->evcon = NULL;
 831         ctx->name = mm_strdup(rpcname);
 832         if (ctx->name == NULL) {
 833                 mm_free(ctx);
 834                 return (NULL);
 835         }
 836         ctx->cb = cb;
 837         ctx->cb_arg = cbarg;
 838         ctx->request = request;
 839         ctx->reply = reply;
 840         ctx->request_marshal = req_marshal;
 841         ctx->reply_clear = rpl_clear;
 842         ctx->reply_unmarshal = rpl_unmarshal;
 843
 844         return (ctx);
 845 }
 846
 847 static void
 848 evrpc_reply_done_closure(void *, enum EVRPC_HOOK_RESULT);
 849
 850 static void
 851 evrpc_reply_done(struct evhttp_request *req, void *arg)
 852 {
 853         struct evrpc_request_wrapper *ctx = arg;
 854         struct evrpc_pool *pool = ctx->pool;
 855         int hook_res = EVRPC_CONTINUE;
 856
 857         /* cancel any timeout we might have scheduled */
 858         event_del(&ctx->ev_timeout);
 859
 860         ctx->req = req;
 861
 862         /* we need to get the reply now */
 863         if (req == NULL) {
 864                 evrpc_reply_done_closure(ctx, EVRPC_CONTINUE);
 865                 return;
 866         }
 867
 868         if (TAILQ_FIRST(&pool->input_hooks) != NULL) {
 869                 evrpc_hook_associate_meta(&ctx->hook_meta, ctx->evcon);
 870
 871                 /* apply hooks to the incoming request */
 872                 hook_res = evrpc_process_hooks(&pool->input_hooks,
 873                     ctx, req, req->input_buffer);
 874
 875                 switch (hook_res) {
 876                 case EVRPC_TERMINATE:
 877                 case EVRPC_CONTINUE:
 878                         break;
 879                 case EVRPC_PAUSE:
 880                         /*
 881                          * if we get paused we also need to know the
 882                          * request.  unfortunately, the underlying
 883                          * layer is going to free it.  we need to
 884                          * request ownership explicitly
 885                          */
 886                         if (req != NULL)
 887                                 evhttp_request_own(req);
 888
 889                         evrpc_pause_request(pool, ctx,
 890                             evrpc_reply_done_closure);
 891                         return;
 892                 default:
 893                         EVUTIL_ASSERT(hook_res == EVRPC_TERMINATE ||
 894                             hook_res == EVRPC_CONTINUE ||
 895                             hook_res == EVRPC_PAUSE);
 896                 }
 897         }
 898
 899         evrpc_reply_done_closure(ctx, hook_res);
 900
 901         /* http request is being freed by underlying layer */
 902 }
 903
 904 static void
 905 evrpc_reply_done_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
 906 {
 907         struct evrpc_request_wrapper *ctx = arg;
 908         struct evhttp_request *req = ctx->req;
 909         struct evrpc_pool *pool = ctx->pool;
 910         struct evrpc_status status;
 911         int res = -1;
 912
 913         memset(&status, 0, sizeof(status));
 914         status.http_req = req;
 915
 916         /* we need to get the reply now */
 917         if (req == NULL) {
 918                 status.error = EVRPC_STATUS_ERR_TIMEOUT;
 919         } else if (hook_res == EVRPC_TERMINATE) {
 920                 status.error = EVRPC_STATUS_ERR_HOOKABORTED;
 921         } else {
 922                 res = ctx->reply_unmarshal(ctx->reply, req->input_buffer);
 923                 if (res == -1)
 924                         status.error = EVRPC_STATUS_ERR_BADPAYLOAD;
 925         }
 926
 927         if (res == -1) {
 928                 /* clear everything that we might have written previously */
 929                 ctx->reply_clear(ctx->reply);
 930         }
 931
 932         (*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
 933
 934         evrpc_request_wrapper_free(ctx);
 935
 936         /* the http layer owned the original request structure, but if we
 937          * got paused, we asked for ownership and need to free it here. */
 938         if (req != NULL && evhttp_request_is_owned(req))
 939                 evhttp_request_free(req);
 940
 941         /* see if we can schedule another request */
 942         evrpc_pool_schedule(pool);
 943 }
 944
 945 static void
 946 evrpc_pool_schedule(struct evrpc_pool *pool)
 947 {
 948         struct evrpc_request_wrapper *ctx = TAILQ_FIRST(&pool->requests);
 949         struct evhttp_connection *evcon;
 950
 951         /* if no requests are pending, we have no work */
 952         if (ctx == NULL)
 953                 return;
 954
 955         if ((evcon = evrpc_pool_find_connection(pool)) != NULL) {
 956                 TAILQ_REMOVE(&pool->requests, ctx, next);
 957                 evrpc_schedule_request(evcon, ctx);
 958         }
 959 }
 960
 961 static void
 962 evrpc_request_timeout(evutil_socket_t fd, short what, void *arg)
 963 {
 964         struct evrpc_request_wrapper *ctx = arg;
 965         struct evhttp_connection *evcon = ctx->evcon;
 966         EVUTIL_ASSERT(evcon != NULL);
 967
 968         evhttp_connection_fail(evcon, EVCON_HTTP_TIMEOUT);
 969 }
 970
 971 /*
 972  * frees potential meta data associated with a request.
 973  */
 974
 975 static void
 976 evrpc_meta_data_free(struct evrpc_meta_list *meta_data)
 977 {
 978         struct evrpc_meta *entry;
 979         EVUTIL_ASSERT(meta_data != NULL);
 980
 981         while ((entry = TAILQ_FIRST(meta_data)) != NULL) {
 982                 TAILQ_REMOVE(meta_data, entry, next);
 983                 mm_free(entry->key);
 984                 mm_free(entry->data);
 985                 mm_free(entry);
 986         }
 987 }
 988
 989 static struct evrpc_hook_meta *
 990 evrpc_hook_meta_new(void)
 991 {
 992         struct evrpc_hook_meta *ctx;
 993         ctx = mm_malloc(sizeof(struct evrpc_hook_meta));
 994         EVUTIL_ASSERT(ctx != NULL);
 995
 996         TAILQ_INIT(&ctx->meta_data);
 997         ctx->evcon = NULL;
 998
 999         return (ctx);
1000 }
1001
1002 static void
1003 evrpc_hook_associate_meta(struct evrpc_hook_meta **pctx,
1004     struct evhttp_connection *evcon)
1005 {
1006         struct evrpc_hook_meta *ctx = *pctx;
1007         if (ctx == NULL)
1008                 *pctx = ctx = evrpc_hook_meta_new();
1009         ctx->evcon = evcon;
1010 }
1011
1012 static void
1013 evrpc_hook_context_free(struct evrpc_hook_meta *ctx)
1014 {
1015         evrpc_meta_data_free(&ctx->meta_data);
1016         mm_free(ctx);
1017 }
1018
1019 /* Adds meta data */
1020 void
1021 evrpc_hook_add_meta(void *ctx, const char *key,
1022     const void *data, size_t data_size)
1023 {
1024         struct evrpc_request_wrapper *req = ctx;
1025         struct evrpc_hook_meta *store = NULL;
1026         struct evrpc_meta *meta = NULL;
1027
1028         if ((store = req->hook_meta) == NULL)
1029                 store = req->hook_meta = evrpc_hook_meta_new();
1030
1031         meta = mm_malloc(sizeof(struct evrpc_meta));
1032         EVUTIL_ASSERT(meta != NULL);
1033         meta->key = mm_strdup(key);
1034         EVUTIL_ASSERT(meta->key != NULL);
1035         meta->data_size = data_size;
1036         meta->data = mm_malloc(data_size);
1037         EVUTIL_ASSERT(meta->data != NULL);
1038         memcpy(meta->data, data, data_size);
1039
1040         TAILQ_INSERT_TAIL(&store->meta_data, meta, next);
1041 }
1042
1043 int
1044 evrpc_hook_find_meta(void *ctx, const char *key, void **data, size_t *data_size)
1045 {
1046         struct evrpc_request_wrapper *req = ctx;
1047         struct evrpc_meta *meta = NULL;
1048
1049         if (req->hook_meta == NULL)
1050                 return (-1);
1051
1052         TAILQ_FOREACH(meta, &req->hook_meta->meta_data, next) {
1053                 if (strcmp(meta->key, key) == 0) {
1054                         *data = meta->data;
1055                         *data_size = meta->data_size;
1056                         return (0);
1057                 }
1058         }
1059
1060         return (-1);
1061 }
1062
1063 struct evhttp_connection *
1064 evrpc_hook_get_connection(void *ctx)
1065 {
1066         struct evrpc_request_wrapper *req = ctx;
1067         return (req->hook_meta != NULL ? req->hook_meta->evcon : NULL);
1068 }
1069
1070 int
1071 evrpc_send_request_generic(struct evrpc_pool *pool,
1072     void *request, void *reply,
1073     void (*cb)(struct evrpc_status *, void *, void *, void *),
1074     void *cb_arg,
1075     const char *rpcname,
1076     void (*req_marshal)(struct evbuffer *, void *),
1077     void (*rpl_clear)(void *),
1078     int (*rpl_unmarshal)(void *, struct evbuffer *))
1079 {
1080         struct evrpc_status status;
1081         struct evrpc_request_wrapper *ctx;
1082         ctx = evrpc_make_request_ctx(pool, request, reply,
1083             rpcname, req_marshal, rpl_clear, rpl_unmarshal, cb, cb_arg);
1084         if (ctx == NULL)
1085                 goto error;
1086         return (evrpc_make_request(ctx));
1087 error:
1088         memset(&status, 0, sizeof(status));
1089         status.error = EVRPC_STATUS_ERR_UNSTARTED;
1090         (*(cb))(&status, request, reply, cb_arg);
1091         return (-1);
1092 }
1093
1094 /** Takes a request object and fills it in with the right magic */
1095 static struct evrpc *
1096 evrpc_register_object(const char *name,
1097     void *(*req_new)(void*), void *req_new_arg, void (*req_free)(void *),
1098     int (*req_unmarshal)(void *, struct evbuffer *),
1099     void *(*rpl_new)(void*), void *rpl_new_arg, void (*rpl_free)(void *),
1100     int (*rpl_complete)(void *),
1101     void (*rpl_marshal)(struct evbuffer *, void *))
1102 {
1103         struct evrpc* rpc = (struct evrpc *)mm_calloc(1, sizeof(struct evrpc));
1104         if (rpc == NULL)
1105                 return (NULL);
1106         rpc->uri = mm_strdup(name);
1107         if (rpc->uri == NULL) {
1108                 mm_free(rpc);
1109                 return (NULL);
1110         }
1111         rpc->request_new = req_new;
1112         rpc->request_new_arg = req_new_arg;
1113         rpc->request_free = req_free;
1114         rpc->request_unmarshal = req_unmarshal;
1115         rpc->reply_new = rpl_new;
1116         rpc->reply_new_arg = rpl_new_arg;
1117         rpc->reply_free = rpl_free;
1118         rpc->reply_complete = rpl_complete;
1119         rpc->reply_marshal = rpl_marshal;
1120         return (rpc);
1121 }
1122
1123 int
1124 evrpc_register_generic(struct evrpc_base *base, const char *name,
1125     void (*callback)(struct evrpc_req_generic *, void *), void *cbarg,
1126     void *(*req_new)(void *), void *req_new_arg, void (*req_free)(void *),
1127     int (*req_unmarshal)(void *, struct evbuffer *),
1128     void *(*rpl_new)(void *), void *rpl_new_arg, void (*rpl_free)(void *),
1129     int (*rpl_complete)(void *),
1130     void (*rpl_marshal)(struct evbuffer *, void *))
1131 {
1132         struct evrpc* rpc =
1133             evrpc_register_object(name, req_new, req_new_arg, req_free, req_unmarshal,
1134                 rpl_new, rpl_new_arg, rpl_free, rpl_complete, rpl_marshal);
1135         if (rpc == NULL)
1136                 return (-1);
1137         evrpc_register_rpc(base, rpc,
1138             (void (*)(struct evrpc_req_generic*, void *))callback, cbarg);
1139         return (0);
1140 }
1141
1142 /** accessors for obscure and undocumented functionality */
1143 struct evrpc_pool *
1144 evrpc_request_get_pool(struct evrpc_request_wrapper *ctx)
1145 {
1146         return (ctx->pool);
1147 }
1148
1149 void
1150 evrpc_request_set_pool(struct evrpc_request_wrapper *ctx,
1151     struct evrpc_pool *pool)
1152 {
1153         ctx->pool = pool;
1154 }
1155
1156 void
1157 evrpc_request_set_cb(struct evrpc_request_wrapper *ctx,
1158     void (*cb)(struct evrpc_status*, void *request, void *reply, void *arg),
1159     void *cb_arg)
1160 {
1161         ctx->cb = cb;
1162         ctx->cb_arg = cb_arg;
1163 }