allow a configurable timeout for connections and RPCs.
svn:r274
/* Frees an http connection */
void evhttp_connection_free(struct evhttp_connection *evcon);
+/* Sets the timeout for events related to this connection */
+void evhttp_connection_set_timeout(struct evhttp_connection *evcon,
+ int timeout_in_secs);
+
/* The connection gets ownership of the request */
int evhttp_make_request(struct evhttp_connection *evcon,
struct evhttp_request *req,
struct evrpc_pool {
struct evconq connections;
+ int timeout;
+
TAILQ_HEAD(evrpc_requestq, evrpc_request_wrapper) requests;
};
}
-void evrpc_request_cb(struct evhttp_request *, void *);
+static void evrpc_pool_schedule(struct evrpc_pool *pool);
+static void evrpc_request_cb(struct evhttp_request *, void *);
void evrpc_request_done(struct evrpc_req_generic*);
/*
return (0);
}
-void
+static void
evrpc_request_cb(struct evhttp_request *req, void *arg)
{
struct evrpc *rpc = arg;
TAILQ_INIT(&pool->connections);
TAILQ_INIT(&pool->requests);
+ pool->timeout = -1;
+
return (pool);
}
assert(connection->http_server == NULL);
TAILQ_INSERT_TAIL(&pool->connections, connection, next);
+ /*
+ * unless a timeout was specifically set for a connection,
+ * the connection inherits the timeout from the pool.
+ */
+ if (connection->timeout == -1)
+ connection->timeout = pool->timeout;
+
/*
* if we have any requests pending, schedule them with the new
* connections.
}
}
+void
+evrpc_pool_set_timeout(struct evrpc_pool *pool, int timeout_in_secs)
+{
+ struct evhttp_connection *evcon;
+ TAILQ_FOREACH(evcon, &pool->connections, next) {
+ evcon->timeout = timeout_in_secs;
+ }
+ pool->timeout = timeout_in_secs;
+}
+
static void evrpc_reply_done(struct evhttp_request *, void *);
+static void evrpc_request_timeout(int, short, void *);
/*
* Finds a connection object associated with the pool that is currently
struct evrpc_request_wrapper *ctx)
{
struct evhttp_request *req = NULL;
+ struct evrpc_pool *pool = ctx->pool;
char *uri = NULL;
int res = 0;
if (uri == NULL)
goto error;
+ /* we need to know the connection that we might have to abort */
+ ctx->evcon = connection;
+
+ if (pool->timeout > 0) {
+ /*
+ * a timeout after which the whole rpc is going to be aborted.
+ */
+ struct timeval tv;
+ timerclear(&tv);
+ tv.tv_sec = pool->timeout;
+ evtimer_add(&ctx->ev_timeout, &tv);
+ }
+
/* start the request over the connection */
res = evhttp_make_request(connection, req, EVHTTP_REQ_POST, uri);
free(uri);
evrpc_make_request(struct evrpc_request_wrapper *ctx)
{
struct evrpc_pool *pool = ctx->pool;
- struct evhttp_connection *connection;
+
+ /* initialize the event structure for this rpc */
+ evtimer_set(&ctx->ev_timeout, evrpc_request_timeout, ctx);
/* we better have some available connections on the pool */
assert(TAILQ_FIRST(&pool->connections) != NULL);
-
- /* even if a connection might be available, we do FIFO */
- if (TAILQ_FIRST(&pool->requests) == NULL) {
- connection = evrpc_pool_find_connection(pool);
- if (connection != NULL)
- return evrpc_schedule_request(connection, ctx);
- }
-
/*
* if no connection is available, we queue the request on the pool,
* the next time a connection is empty, the rpc will be send on that.
*/
TAILQ_INSERT_TAIL(&pool->requests, ctx, next);
+
+ evrpc_pool_schedule(pool);
+
return (0);
}
evrpc_reply_done(struct evhttp_request *req, void *arg)
{
struct evrpc_request_wrapper *ctx = arg;
- int res;
+ struct evrpc_pool *pool = ctx->pool;
+ int res = -1;
+
+ /* cancel any timeout we might have scheduled */
+ event_del(&ctx->ev_timeout);
/* we need to get the reply now */
- res = ctx->reply_unmarshal(ctx->reply, req->input_buffer);
+ if (req != NULL)
+ res = ctx->reply_unmarshal(ctx->reply, req->input_buffer);
if (res == -1) {
/* clear everything that we might have written previously */
ctx->reply_clear(ctx->reply);
evrpc_request_wrapper_free(ctx);
/* the http layer owns the request structure */
+
+ /* see if we can schedule another request */
+ evrpc_pool_schedule(pool);
+}
+
+static void
+evrpc_pool_schedule(struct evrpc_pool *pool)
+{
+ struct evrpc_request_wrapper *ctx = TAILQ_FIRST(&pool->requests);
+ struct evhttp_connection *evcon;
+
+ /* if no requests are pending, we have no work */
+ if (ctx == NULL)
+ return;
+
+ if ((evcon = evrpc_pool_find_connection(pool)) != NULL) {
+ TAILQ_REMOVE(&pool->requests, ctx, next);
+ evrpc_schedule_request(evcon, ctx);
+ }
+}
+
+static void
+evrpc_request_timeout(int fd, short what, void *arg)
+{
+ struct evrpc_request_wrapper *ctx = arg;
+ struct evhttp_connection *evcon = ctx->evcon;
+ assert(evcon != NULL);
+
+ evhttp_connection_fail(evcon, EVCON_HTTP_TIMEOUT);
}
return (-1); \
} \
ctx->pool = pool; \
+ ctx->evcon = NULL; \
ctx->name = strdup(#rpcname); \
if (ctx->name == NULL) { \
free(ctx); \
/* pool on which this rpc request is being made */
struct evrpc_pool *pool;
+ /* connection on which the request is being sent */
+ struct evhttp_connection *evcon;
+
+ /* event for implementing request timeouts */
+ struct event ev_timeout;
+
/* the name of the rpc */
char *name;
void evrpc_pool_add_connection(struct evrpc_pool *,
struct evhttp_connection *);
+/*
+ * Sets the timeout in secs after which a request has to complete. The
+ * RPC is completely aborted if it does not complete by then. Setting
+ * the timeout to 0 means that it never timeouts and can be used to
+ * implement callback type RPCs.
+ *
+ * Any connection already in the pool will be updated with the new
+ * timeout. Connections added to the pool after set_timeout has be
+ * called receive the pool timeout only if no timeout has been set
+ * for the connection itself.
+ */
+void evrpc_pool_set_timeout(struct evrpc_pool *, int timeout_in_secs);
+
#endif /* _EVRPC_H_ */
int flags;
#define EVHTTP_CON_INCOMING 0x0001 /* only one request on it ever */
#define EVHTTP_CON_OUTGOING 0x0002 /* multiple requests possible */
+
+ int timeout; /* timeout in seconds for events */
enum evhttp_connection_state state;
return (method);
}
+static void
+evhttp_add_event(struct event *ev, int timeout, int default_timeout)
+{
+ if (timeout != 0) {
+ struct timeval tv;
+
+ timerclear(&tv);
+ tv.tv_sec = timeout != -1 ? timeout : default_timeout;
+ event_add(ev, &tv);
+ } else {
+ event_add(ev, NULL);
+ }
+}
+
void
evhttp_write_buffer(struct evhttp_connection *evcon,
void (*cb)(struct evhttp_connection *, void *), void *arg)
{
- struct timeval tv;
-
event_debug(("%s: preparing to write buffer\n", __func__));
/* Set call back */
/* xxx: maybe check if the event is still pending? */
event_set(&evcon->ev, evcon->fd, EV_WRITE, evhttp_write, evcon);
- timerclear(&tv);
- tv.tv_sec = HTTP_WRITE_TIMEOUT;
- event_add(&evcon->ev, &tv);
+ evhttp_add_event(&evcon->ev, evcon->timeout, HTTP_WRITE_TIMEOUT);
}
/*
evhttp_write(int fd, short what, void *arg)
{
struct evhttp_connection *evcon = arg;
- struct timeval tv;
int n;
if (what == EV_TIMEOUT) {
}
if (EVBUFFER_LENGTH(evcon->output_buffer) != 0) {
- timerclear(&tv);
- tv.tv_sec = HTTP_WRITE_TIMEOUT;
- event_add(&evcon->ev, &tv);
+ evhttp_add_event(&evcon->ev,
+ evcon->timeout, HTTP_WRITE_TIMEOUT);
return;
}
{
struct evhttp_connection *evcon = arg;
struct evhttp_request *req = TAILQ_FIRST(&evcon->requests);
- struct timeval tv;
int n;
if (what == EV_TIMEOUT) {
evhttp_connection_done(evcon);
return;
}
-
- timerclear(&tv);
- tv.tv_sec = HTTP_READ_TIMEOUT;
- event_add(&evcon->ev, &tv);
+
+ evhttp_add_event(&evcon->ev, evcon->timeout, HTTP_READ_TIMEOUT);
}
void
void
evhttp_get_body(struct evhttp_connection *evcon, struct evhttp_request *req)
{
- struct timeval tv;
const char *content_length;
const char *connection;
struct evkeyvalq *headers = req->input_headers;
}
event_set(&evcon->ev, evcon->fd, EV_READ, evhttp_read, evcon);
- timerclear(&tv);
- tv.tv_sec = HTTP_READ_TIMEOUT;
- event_add(&evcon->ev, &tv);
- return;
+ evhttp_add_event(&evcon->ev, evcon->timeout, HTTP_READ_TIMEOUT);
}
void
evhttp_read_header(int fd, short what, void *arg)
{
- struct timeval tv;
struct evhttp_connection *evcon = arg;
struct evhttp_request *req = TAILQ_FIRST(&evcon->requests);
int n, res;
return;
} else if (res == 0) {
/* Need more header lines */
- timerclear(&tv);
- tv.tv_sec = HTTP_READ_TIMEOUT;
- event_add(&evcon->ev, &tv);
+ evhttp_add_event(&evcon->ev,
+ evcon->timeout, HTTP_READ_TIMEOUT);
return;
}
evcon->fd = -1;
evcon->port = port;
+ evcon->timeout = -1;
+
if ((evcon->address = strdup(address)) == NULL) {
event_warn("%s: strdup failed", __func__);
goto error;
return (NULL);
}
+void
+evhttp_connection_set_timeout(struct evhttp_connection *evcon,
+ int timeout_in_secs)
+{
+ evcon->timeout = timeout_in_secs;
+}
+
int
evhttp_connection_connect(struct evhttp_connection *evcon)
{
- struct timeval tv;
-
if (evcon->state == EVCON_CONNECTING)
return (0);
/* Set up a callback for successful connection setup */
event_set(&evcon->ev, evcon->fd, EV_WRITE, evhttp_connectioncb, evcon);
- timerclear(&tv);
- tv.tv_sec = HTTP_CONNECT_TIMEOUT;
- event_add(&evcon->ev, &tv);
+ evhttp_add_event(&evcon->ev, evcon->timeout, HTTP_CONNECT_TIMEOUT);
evcon->state = EVCON_CONNECTING;
void
evhttp_start_read(struct evhttp_connection *evcon)
{
- struct timeval tv;
-
/* Set up an event to read the headers */
if (event_initialized(&evcon->ev))
event_del(&evcon->ev);
event_set(&evcon->ev, evcon->fd, EV_READ, evhttp_read_header, evcon);
-
- timerclear(&tv);
- tv.tv_sec = HTTP_READ_TIMEOUT;
- event_add(&evcon->ev, &tv);
+
+ evhttp_add_event(&evcon->ev, evcon->timeout, HTTP_READ_TIMEOUT);
}
void
goto done;
test_ok += 1;
+
done:
event_loopexit(NULL);
}
+static void
+GotKillCbTwo(struct msg *msg, struct kill *kill, void *arg)
+{
+ char *weapon;
+ char *action;
+
+ if (EVTAG_GET(kill, weapon, &weapon) == -1) {
+ fprintf(stderr, "get weapon\n");
+ goto done;
+ }
+ if (EVTAG_GET(kill, action, &action) == -1) {
+ fprintf(stderr, "get action\n");
+ goto done;
+ }
+
+ if (strcmp(weapon, "dagger"))
+ goto done;
+
+ if (strcmp(action, "wave around like an idiot"))
+ goto done;
+
+ test_ok += 1;
+
+done:
+ if (test_ok == 2)
+ event_loopexit(NULL);
+}
+
static void
rpc_basic_client(void)
{
evhttp_free(http);
}
+/*
+ * We are testing that the second requests gets send over the same
+ * connection after the first RPCs completes.
+ */
+static void
+rpc_basic_queued_client(void)
+{
+ short port;
+ struct evhttp *http = NULL;
+ struct evrpc_base *base = NULL;
+ struct evrpc_pool *pool = NULL;
+ struct msg *msg;
+ struct kill *kill_one, *kill_two;
+
+ fprintf(stdout, "Testing RPC (Queued) Client: ");
+
+ rpc_setup(&http, &port, &base);
+
+ pool = rpc_pool_with_connection(port);
+
+ /* set up the basic message */
+ msg = msg_new();
+ EVTAG_ASSIGN(msg, from_name, "niels");
+ EVTAG_ASSIGN(msg, to_name, "tester");
+
+ kill_one = kill_new();
+ kill_two = kill_new();
+
+ EVRPC_MAKE_REQUEST(Message, msg, kill_one, GotKillCbTwo, NULL);
+ EVRPC_MAKE_REQUEST(Message, msg, kill_two, GotKillCb, NULL);
+
+ test_ok = 0;
+
+ event_dispatch();
+
+ if (test_ok != 2) {
+ fprintf(stdout, "FAILED (1)\n");
+ exit(1);
+ }
+
+ fprintf(stdout, "OK\n");
+
+ msg_free(msg);
+ kill_free(kill_one);
+ kill_free(kill_two);
+
+ evrpc_pool_free(pool);
+ evhttp_free(http);
+}
+
+
void
rpc_suite(void)
{
rpc_basic_test();
rpc_basic_message();
rpc_basic_client();
+ rpc_basic_queued_client();
}
projects / libevent / commitdiff