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