pausing an rpc via a hook needs to deal with the fact that http callb…

…acks free the request after they return; provide a way for a callback to take ownership of the request structure; the user then needs to explicitly free it.

svn:r620

ChangeLog

@@ -31,6 +31,7 @@ Changes in current version:
  o event_rpcgen now allows creating integer arrays
  o support string arrays in event_rpcgen
  o change evrpc hooking to allow pausing of RPCs; this will make it possible for the hook to do some meaning ful work; this is not backwards compatible.
+ o allow an http request callback to take ownership of a request structure
 
 Changes in 1.4.0:
  o allow \r or \n individually to separate HTTP headers instead of the standard "\r\n"; from Charles Kerr.

evhttp.h

@@ -182,6 +182,7 @@ struct {
 	int flags;
 #define EVHTTP_REQ_OWN_CONNECTION	0x0001
 #define EVHTTP_PROXY_REQUEST		0x0002
+#define EVHTTP_USER_OWNED		0x0004
 
 	struct evkeyvalq *input_headers;
 	struct evkeyvalq *output_headers;
@@ -235,6 +236,16 @@ void evhttp_request_set_chunked_cb(struct evhttp_request *,
 /** Frees the request object and removes associated events. */
 void evhttp_request_free(struct evhttp_request *req);
 
+/** Takes ownership of the request object
+ *
+ * Can be used in a request callback to keep onto the request until
+ * evhttp_request_free() is explicitly called by the user.
+ */
+void evhttp_request_own(struct evhttp_request *req);
+
+/** Returns 1 if the request is owned by the user */
+int evhttp_request_is_owned(struct evhttp_request *req);
+
 /**
  * A connection object that can be used to for making HTTP requests.  The
  * connection object tries to establish the connection when it is given an

evrpc.c

@@ -175,11 +175,12 @@ evrpc_process_hooks(struct evrpc_hook_list *head, void *ctx,
 {
 	struct evrpc_hook *hook;
 	TAILQ_FOREACH(hook, head, next) {
-		if (hook->process(ctx, req, evbuf, hook->process_arg) == -1)
-			return (-1);
+		int res = hook->process(ctx, req, evbuf, hook->process_arg);
+		if (res != EVRPC_CONTINUE)
+			return (res);
 	}
 
-	return (0);
+	return (EVRPC_CONTINUE);
 }
 
 static void evrpc_pool_schedule(struct evrpc_pool *pool);
@@ -761,7 +762,6 @@ evrpc_reply_done(struct evhttp_request *req, void *arg)
 	/* cancel any timeout we might have scheduled */
 	event_del(&ctx->ev_timeout);
 
-	/* if we get paused we also need to know the request */
 	ctx->req = req;
 
 	/* we need to get the reply now */
@@ -780,12 +780,22 @@ evrpc_reply_done(struct evhttp_request *req, void *arg)
 		evrpc_reply_done_closure(ctx, hook_res);
 		return;
 	case EVRPC_PAUSE:
+		/*
+		 * if we get paused we also need to know the request.
+		 * unfortunately, the underlying layer is going to free it.
+		 * we need to request ownership explicitly
+		 */
+		if (req != NULL)
+			evhttp_request_own(req);
+
 		evrpc_pause_request(pool, ctx, evrpc_reply_done_closure);
 		return;
 	default:
 		assert(hook_res == EVRPC_TERMINATE ||
 		    hook_res == EVRPC_CONTINUE || hook_res == EVRPC_PAUSE);
 	}
+
+	/* http request is being freed by underlying layer */
 }
 
 static void
@@ -820,7 +830,10 @@ evrpc_reply_done_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
 
 	evrpc_request_wrapper_free(ctx);
 
-	/* the http layer owns the request structure */
+	/* the http layer owned the orignal request structure, but if we
+	 * got paused, we asked for ownership and need to free it here. */
+	if (req != NULL && evhttp_request_is_owned(req))
+		evhttp_request_free(req);
 
 	/* see if we can schedule another request */
 	evrpc_pool_schedule(pool);

http.c

@@ -674,8 +674,10 @@ evhttp_connection_done(struct evhttp_connection *evcon)
 	/* notify the user of the request */
 	(*req->cb)(req, req->cb_arg);
 
-	/* if this was an outgoing request, we own and it's done. so free it */
-	if (con_outgoing) {
+	/* if this was an outgoing request, we own and it's done. so free it.
+	 * unless the callback specifically requested to own the request.
+	 */
+	if (con_outgoing && ((req->flags & EVHTTP_USER_OWNED) == 0)) {
 		evhttp_request_free(req);
 	}
 }
@@ -2210,6 +2212,18 @@ evhttp_request_free(struct evhttp_request *req)
 	event_free(req);
 }
 
+void
+evhttp_request_own(struct evhttp_request *req)
+{
+	req->flags |= EVHTTP_USER_OWNED;
+}
+
+int
+evhttp_request_is_owned(struct evhttp_request *req)
+{
+	return (req->flags & EVHTTP_USER_OWNED) != 0;
+}
+
 void
 evhttp_request_set_chunked_cb(struct evhttp_request *req,
     void (*cb)(struct evhttp_request *, void *))

test/regress_rpc.c

@@ -579,10 +579,13 @@ struct _rpc_hook_ctx {
 	void *ctx;
 };
 
+static int hook_pause_cb_called;
+
 static void
 rpc_hook_pause_cb(int fd, short what, void *arg)
 {
 	struct _rpc_hook_ctx *ctx = arg;
+ 	++hook_pause_cb_called;
 	evrpc_resume_request(ctx->vbase, ctx->ctx, EVRPC_CONTINUE);
 }
 
@@ -631,31 +634,19 @@ rpc_basic_client_with_pause(void)
 
 	kill = kill_new();
 
-	EVRPC_MAKE_REQUEST(Message, pool, msg, kill,  GotKillCb, NULL);
+	EVRPC_MAKE_REQUEST(Message, pool, msg, kill, GotKillCb, NULL);
 
 	test_ok = 0;
 
 	event_dispatch();
 
-	if (test_ok != 1) {
-		fprintf(stdout, "FAILED (1)\n");
+	if (test_ok != 1 || hook_pause_cb_called != 4) {
+		fprintf(stdout, "FAILED\n");
 		exit(1);
 	}
 
-	/* we do it twice to make sure that reuse works correctly */
-	kill_clear(kill);
-
-	EVRPC_MAKE_REQUEST(Message, pool, msg, kill,  GotKillCb, NULL);
-
-	event_dispatch();
-
 	rpc_teardown(base);
 
-	if (test_ok != 2) {
-		fprintf(stdout, "FAILED (2)\n");
-		exit(1);
-	}
-
 	fprintf(stdout, "OK\n");
 
 	msg_free(msg);