ipc/chromium/src/third_party/libevent/evrpc.c

Wed, 31 Dec 2014 06:09:35 +0100

author
Michael Schloh von Bennewitz <michael@schloh.com>
date
Wed, 31 Dec 2014 06:09:35 +0100
changeset 0
6474c204b198
permissions
-rw-r--r--

Cloned upstream origin tor-browser at tor-browser-31.3.0esr-4.5-1-build1
revision ID fc1c9ff7c1b2defdbc039f12214767608f46423f for hacking purpose.

     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"
    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
    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>
    53 #include <sys/queue.h>
    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"
    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);
    77 	/* we rely on the tagging sub system */
    78 	evtag_init();
    80 	TAILQ_INIT(&base->registered_rpcs);
    81 	TAILQ_INIT(&base->input_hooks);
    82 	TAILQ_INIT(&base->output_hooks);
    84 	TAILQ_INIT(&base->paused_requests);
    86 	base->http_server = http_server;
    88 	return (base);
    89 }
    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;
    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 }
   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 	}
   138 	hook = mm_calloc(1, sizeof(struct evrpc_hook));
   139 	EVUTIL_ASSERT(hook != NULL);
   141 	hook->process = cb;
   142 	hook->process_arg = cb_arg;
   143 	TAILQ_INSERT_TAIL(head, hook, next);
   145 	return (hook);
   146 }
   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 	}
   160 	return (0);
   161 }
   163 /*
   164  * remove the hook specified by the handle
   165  */
   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 	}
   183 	return (evrpc_remove_hook_internal(head, handle));
   184 }
   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 	}
   197 	return (EVRPC_CONTINUE);
   198 }
   200 static void evrpc_pool_schedule(struct evrpc_pool *pool);
   201 static void evrpc_request_cb(struct evhttp_request *, void *);
   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  */
   209 static char *
   210 evrpc_construct_uri(const char *uri)
   211 {
   212 	char *constructed_uri;
   213 	size_t constructed_uri_len;
   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';
   223 	return (constructed_uri);
   224 }
   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);
   232 	rpc->base = base;
   233 	rpc->cb = cb;
   234 	rpc->cb_arg = cb_arg;
   236 	TAILQ_INSERT_TAIL(&base->registered_rpcs, rpc, next);
   238 	evhttp_set_cb(base->http_server,
   239 	    constructed_uri,
   240 	    evrpc_request_cb,
   241 	    rpc);
   243 	mm_free(constructed_uri);
   245 	return (0);
   246 }
   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;
   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);
   266 	registered_uri = evrpc_construct_uri(name);
   268 	/* remove the http server callback */
   269 	r = evhttp_del_cb(base->http_server, registered_uri);
   270 	EVUTIL_ASSERT(r == 0);
   272 	mm_free(registered_uri);
   274 	mm_free((char *)rpc->uri);
   275 	mm_free(rpc);
   276 	return (0);
   277 }
   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);
   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;
   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;
   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;
   301 	if (TAILQ_FIRST(&rpc->base->input_hooks) != NULL) {
   302 		int hook_res;
   304 		evrpc_hook_associate_meta(&rpc_state->hook_meta, req->evcon);
   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 	}
   327 	evrpc_request_cb_closure(rpc_state, EVRPC_CONTINUE);
   328 	return;
   330 error:
   331 	if (rpc_state != NULL)
   332 		evrpc_reqstate_free(rpc_state);
   333 	evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
   334 	return;
   335 }
   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;
   344 	EVUTIL_ASSERT(rpc_state);
   345 	rpc = rpc_state->rpc;
   346 	req = rpc_state->http_req;
   348 	if (hook_res == EVRPC_TERMINATE)
   349 		goto error;
   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;
   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 	}
   362 	/* at this point, we have a well formed request, prepare the reply */
   364 	rpc_state->reply = rpc->reply_new(rpc->reply_new_arg);
   365 	if (rpc_state->reply == NULL)
   366 		goto error;
   368 	/* give the rpc to the user; they can deal with it */
   369 	rpc->cb(rpc_state, rpc->cb_arg);
   371 	return;
   373 error:
   374 	if (rpc_state != NULL)
   375 		evrpc_reqstate_free(rpc_state);
   376 	evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
   377 	return;
   378 }
   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;
   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 }
   400 static void
   401 evrpc_request_done_closure(void *, enum EVRPC_HOOK_RESULT);
   403 void
   404 evrpc_request_done(struct evrpc_req_generic *rpc_state)
   405 {
   406 	struct evhttp_request *req;
   407 	struct evrpc *rpc;
   409 	EVUTIL_ASSERT(rpc_state);
   411 	req = rpc_state->http_req;
   412 	rpc = rpc_state->rpc;
   414 	if (rpc->reply_complete(rpc_state->reply) == -1) {
   415 		/* the reply was not completely filled in.  error out */
   416 		goto error;
   417 	}
   419 	if ((rpc_state->rpc_data = evbuffer_new()) == NULL) {
   420 		/* out of memory */
   421 		goto error;
   422 	}
   424 	/* serialize the reply */
   425 	rpc->reply_marshal(rpc_state->rpc_data, rpc_state->reply);
   427 	if (TAILQ_FIRST(&rpc->base->output_hooks) != NULL) {
   428 		int hook_res;
   430 		evrpc_hook_associate_meta(&rpc_state->hook_meta, req->evcon);
   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 	}
   452 	evrpc_request_done_closure(rpc_state, EVRPC_CONTINUE);
   453 	return;
   455 error:
   456 	if (rpc_state != NULL)
   457 		evrpc_reqstate_free(rpc_state);
   458 	evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
   459 	return;
   460 }
   462 void *
   463 evrpc_get_request(struct evrpc_req_generic *req)
   464 {
   465 	return req->request;
   466 }
   468 void *
   469 evrpc_get_reply(struct evrpc_req_generic *req)
   470 {
   471 	return req->reply;
   472 }
   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;
   482 	if (hook_res == EVRPC_TERMINATE)
   483 		goto error;
   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);
   492 	evrpc_reqstate_free(rpc_state);
   494 	return;
   496 error:
   497 	if (rpc_state != NULL)
   498 		evrpc_reqstate_free(rpc_state);
   499 	evhttp_send_error(req, HTTP_SERVUNAVAIL, NULL);
   500 	return;
   501 }
   504 /* Client implementation of RPC site */
   506 static int evrpc_schedule_request(struct evhttp_connection *connection,
   507     struct evrpc_request_wrapper *ctx);
   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);
   516 	TAILQ_INIT(&pool->connections);
   517 	TAILQ_INIT(&pool->requests);
   519 	TAILQ_INIT(&pool->paused_requests);
   521 	TAILQ_INIT(&pool->input_hooks);
   522 	TAILQ_INIT(&pool->output_hooks);
   524 	pool->base = base;
   525 	pool->timeout = -1;
   527 	return (pool);
   528 }
   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 }
   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;
   548 	while ((request = TAILQ_FIRST(&pool->requests)) != NULL) {
   549 		TAILQ_REMOVE(&pool->requests, request, next);
   550 		evrpc_request_wrapper_free(request);
   551 	}
   553 	while ((pause = TAILQ_FIRST(&pool->paused_requests)) != NULL) {
   554 		TAILQ_REMOVE(&pool->paused_requests, pause, next);
   555 		mm_free(pause);
   556 	}
   558 	while ((connection = TAILQ_FIRST(&pool->connections)) != NULL) {
   559 		TAILQ_REMOVE(&pool->connections, connection, next);
   560 		evhttp_connection_free(connection);
   561 	}
   563 	while ((hook = TAILQ_FIRST(&pool->input_hooks)) != NULL) {
   564 		r = evrpc_remove_hook(pool, EVRPC_INPUT, hook);
   565 		EVUTIL_ASSERT(r);
   566 	}
   568 	while ((hook = TAILQ_FIRST(&pool->output_hooks)) != NULL) {
   569 		r = evrpc_remove_hook(pool, EVRPC_OUTPUT, hook);
   570 		EVUTIL_ASSERT(r);
   571 	}
   573 	mm_free(pool);
   574 }
   576 /*
   577  * Add a connection to the RPC pool.   A request scheduled on the pool
   578  * may use any available connection.
   579  */
   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);
   588 	/*
   589 	 * associate an event base with this connection
   590 	 */
   591 	if (pool->base != NULL)
   592 		evhttp_connection_set_base(connection, pool->base);
   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;
   601 	/*
   602 	 * if we have any requests pending, schedule them with the new
   603 	 * connections.
   604 	 */
   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 }
   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 }
   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 }
   632 static void evrpc_reply_done(struct evhttp_request *, void *);
   633 static void evrpc_request_timeout(evutil_socket_t, short, void *);
   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 	}
   648 	return (NULL);
   649 }
   651 /*
   652  * Prototypes responsible for evrpc scheduling and hooking
   653  */
   655 static void evrpc_schedule_request_closure(void *ctx, enum EVRPC_HOOK_RESULT);
   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;
   668 	if ((req = evhttp_request_new(evrpc_reply_done, ctx)) == NULL)
   669 		goto error;
   671 	/* serialize the request data into the output buffer */
   672 	ctx->request_marshal(req->output_buffer, ctx->request);
   674 	/* we need to know the connection that we might have to abort */
   675 	ctx->evcon = connection;
   677 	/* if we get paused we also need to know the request */
   678 	ctx->req = req;
   680 	if (TAILQ_FIRST(&pool->output_hooks) != NULL) {
   681 		int hook_res;
   683 		evrpc_hook_associate_meta(&ctx->hook_meta, connection);
   685 		/* apply hooks to the outgoing request */
   686 		hook_res = evrpc_process_hooks(&pool->output_hooks,
   687 		    ctx, req, req->output_buffer);
   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 	}
   708 	evrpc_schedule_request_closure(ctx, EVRPC_CONTINUE);
   709 	return (0);
   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 }
   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;
   730 	if (hook_res == EVRPC_TERMINATE)
   731 		goto error;
   733 	uri = evrpc_construct_uri(ctx->name);
   734 	if (uri == NULL)
   735 		goto error;
   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 	}
   747 	/* start the request over the connection */
   748 	res = evhttp_make_request(connection, req, EVHTTP_REQ_POST, uri);
   749 	mm_free(uri);
   751 	if (res == -1)
   752 		goto error;
   754 	return;
   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 }
   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);
   773 	pause->ctx = ctx;
   774 	pause->cb = cb;
   776 	TAILQ_INSERT_TAIL(&base->pause_requests, pause, next);
   777 	return (0);
   778 }
   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;
   787 	TAILQ_FOREACH(pause, head, next) {
   788 		if (pause->ctx == ctx)
   789 			break;
   790 	}
   792 	if (pause == NULL)
   793 		return (-1);
   795 	(*pause->cb)(pause->ctx, res);
   796 	TAILQ_REMOVE(head, pause, next);
   797 	mm_free(pause);
   798 	return (0);
   799 }
   801 int
   802 evrpc_make_request(struct evrpc_request_wrapper *ctx)
   803 {
   804 	struct evrpc_pool *pool = ctx->pool;
   806 	/* initialize the event structure for this rpc */
   807 	evtimer_assign(&ctx->ev_timeout, pool->base, evrpc_request_timeout, ctx);
   809 	/* we better have some available connections on the pool */
   810 	EVUTIL_ASSERT(TAILQ_FIRST(&pool->connections) != NULL);
   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);
   818 	evrpc_pool_schedule(pool);
   820 	return (0);
   821 }
   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);
   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;
   855 	return (ctx);
   856 }
   858 static void
   859 evrpc_reply_done_closure(void *, enum EVRPC_HOOK_RESULT);
   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;
   868 	/* cancel any timeout we might have scheduled */
   869 	event_del(&ctx->ev_timeout);
   871 	ctx->req = req;
   873 	/* we need to get the reply now */
   874 	if (req == NULL) {
   875 		evrpc_reply_done_closure(ctx, EVRPC_CONTINUE);
   876 		return;
   877 	}
   879 	if (TAILQ_FIRST(&pool->input_hooks) != NULL) {
   880 		evrpc_hook_associate_meta(&ctx->hook_meta, ctx->evcon);
   882 		/* apply hooks to the incoming request */
   883 		hook_res = evrpc_process_hooks(&pool->input_hooks,
   884 		    ctx, req, req->input_buffer);
   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);
   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 	}
   910 	evrpc_reply_done_closure(ctx, hook_res);
   912 	/* http request is being freed by underlying layer */
   913 }
   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;
   924 	memset(&status, 0, sizeof(status));
   925 	status.http_req = req;
   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 	}
   938 	if (res == -1) {
   939 		/* clear everything that we might have written previously */
   940 		ctx->reply_clear(ctx->reply);
   941 	}
   943 	(*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
   945 	evrpc_request_wrapper_free(ctx);
   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);
   952 	/* see if we can schedule another request */
   953 	evrpc_pool_schedule(pool);
   954 }
   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;
   962 	/* if no requests are pending, we have no work */
   963 	if (ctx == NULL)
   964 		return;
   966 	if ((evcon = evrpc_pool_find_connection(pool)) != NULL) {
   967 		TAILQ_REMOVE(&pool->requests, ctx, next);
   968 		evrpc_schedule_request(evcon, ctx);
   969 	}
   970 }
   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);
   979 	evhttp_connection_fail(evcon, EVCON_HTTP_TIMEOUT);
   980 }
   982 /*
   983  * frees potential meta data associated with a request.
   984  */
   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);
   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 }
  1000 static struct evrpc_hook_meta *
  1001 evrpc_hook_meta_new(void)
  1003 	struct evrpc_hook_meta *ctx;
  1004 	ctx = mm_malloc(sizeof(struct evrpc_hook_meta));
  1005 	EVUTIL_ASSERT(ctx != NULL);
  1007 	TAILQ_INIT(&ctx->meta_data);
  1008 	ctx->evcon = NULL;
  1010 	return (ctx);
  1013 static void
  1014 evrpc_hook_associate_meta(struct evrpc_hook_meta **pctx,
  1015     struct evhttp_connection *evcon)
  1017 	struct evrpc_hook_meta *ctx = *pctx;
  1018 	if (ctx == NULL)
  1019 		*pctx = ctx = evrpc_hook_meta_new();
  1020 	ctx->evcon = evcon;
  1023 static void
  1024 evrpc_hook_context_free(struct evrpc_hook_meta *ctx)
  1026 	evrpc_meta_data_free(&ctx->meta_data);
  1027 	mm_free(ctx);
  1030 /* Adds meta data */
  1031 void
  1032 evrpc_hook_add_meta(void *ctx, const char *key,
  1033     const void *data, size_t data_size)
  1035 	struct evrpc_request_wrapper *req = ctx;
  1036 	struct evrpc_hook_meta *store = NULL;
  1037 	struct evrpc_meta *meta = NULL;
  1039 	if ((store = req->hook_meta) == NULL)
  1040 		store = req->hook_meta = evrpc_hook_meta_new();
  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);
  1051 	TAILQ_INSERT_TAIL(&store->meta_data, meta, next);
  1054 int
  1055 evrpc_hook_find_meta(void *ctx, const char *key, void **data, size_t *data_size)
  1057 	struct evrpc_request_wrapper *req = ctx;
  1058 	struct evrpc_meta *meta = NULL;
  1060 	if (req->hook_meta == NULL)
  1061 		return (-1);
  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);
  1071 	return (-1);
  1074 struct evhttp_connection *
  1075 evrpc_hook_get_connection(void *ctx)
  1077 	struct evrpc_request_wrapper *req = ctx;
  1078 	return (req->hook_meta != NULL ? req->hook_meta->evcon : NULL);
  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 *))
  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);
  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 *))
  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);
  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);
  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 *))
  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);
  1153 /** accessors for obscure and undocumented functionality */
  1154 struct evrpc_pool *
  1155 evrpc_request_get_pool(struct evrpc_request_wrapper *ctx)
  1157 	return (ctx->pool);
  1160 void
  1161 evrpc_request_set_pool(struct evrpc_request_wrapper *ctx,
  1162     struct evrpc_pool *pool)
  1164 	ctx->pool = pool;
  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)
  1172 	ctx->cb = cb;
  1173 	ctx->cb_arg = cb_arg;

mercurial