1 files changed, 92 insertions, 28 deletions

diff --git a/src/dispatch.c b/src/dispatch.c
index 8a9c981..7ab93f0 100644
--- a/src/dispatch.c
+++ b/src/dispatch.c
@@ -74,9 +74,12 @@ struct dispatch_client {
 	char *client_query;
 	ssize_t client_query_size;
 	void *client_query_cookie;
-	/* The reply to the client, when we're sending one. */
-	char client_outbuf[4096];
-	ssize_t client_outbuf_used;
+	/* The outgoing replies to the client, when we're sending one. */
+	char *client_outbuf;
+	ssize_t client_outbuf_size, client_outbuf_used;
+	/* Working space for a client reply. */
+	char *client_workbuf;
+	ssize_t client_workbuf_size;
 	/* This is a linked list. */
 	struct dispatch_client *client_next;
 };
@@ -87,6 +90,8 @@ struct dispatch_client_data {
 		int client_fd;
 		struct sockaddr client_addr;
 		socklen_t client_addrlen;
+		char *reply_buf;
+		size_t reply_buf_size;
 	} dgram;
 };
 
@@ -95,33 +100,42 @@ static bool_t
 dispatch_reply_fragment_dgram(struct plugin_state *state,
 			      struct dispatch_client_data *cdata,
 			      struct rpc_msg *reply,
-			      XDR *reply_xdrs, char *reply_buf,
+			      XDR *reply_xdrs,
 			      bool_t first_fragment, bool_t last_fragment)
 {
-	xdr_replymsg(reply_xdrs, reply);
-	if (!first_fragment || !last_fragment) {
-		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
-				"trying to sending datagram reply (%d bytes), "
-				"even though the reply is not suitable for "
-				"transmission as a datagram\n",
-				xdr_getpos(reply_xdrs));
+	if (xdr_replymsg(reply_xdrs, reply)) {
+		if (!first_fragment || !last_fragment) {
+			slapi_log_error(SLAPI_LOG_PLUGIN,
+					state->plugin_desc->spd_id,
+					"trying to sending datagram reply (%d "
+					"bytes), even though the reply is not "
+					"suitable for transmission as a "
+					"datagram\n", xdr_getpos(reply_xdrs));
+		} else {
+			slapi_log_error(SLAPI_LOG_PLUGIN,
+					state->plugin_desc->spd_id,
+					"sending datagram reply (%d bytes)\n",
+					xdr_getpos(reply_xdrs));
+		}
+		sendto(cdata->dgram.client_fd,
+		       cdata->dgram.reply_buf, xdr_getpos(reply_xdrs),
+		       0,
+		       &cdata->dgram.client_addr,
+		       cdata->dgram.client_addrlen);
 	} else {
-		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
-				"sending datagram reply (%d bytes)\n",
-				xdr_getpos(reply_xdrs));
+		slapi_log_error(SLAPI_LOG_PLUGIN,
+				state->plugin_desc->spd_id,
+				"error sending datagram reply -- too large?\n");
 	}
-	sendto(cdata->dgram.client_fd, reply_buf, xdr_getpos(reply_xdrs),
-	       0, &cdata->dgram.client_addr, cdata->dgram.client_addrlen);
 	return TRUE;
 }
 static void
 dispatch_reply_dgram(struct plugin_state *state,
 		     struct dispatch_client_data *cdata,
-		     struct rpc_msg *reply,
-		     XDR *reply_xdrs, char *reply_buf)
+		     struct rpc_msg *reply, XDR *reply_xdrs)
 {
 	dispatch_reply_fragment_dgram(state, cdata,
-				      reply, reply_xdrs, reply_buf,
+				      reply, reply_xdrs,
 				      TRUE, TRUE);
 }
 
@@ -129,11 +143,11 @@ static bool_t
 dispatch_reply_fragment_connected(struct plugin_state *state,
 				  struct dispatch_client_data *cdata,
 				  struct rpc_msg *reply,
-				  XDR *reply_xdrs, char *reply_buf,
+				  XDR *reply_xdrs,
 				  bool_t first_fragment, bool_t last_fragment)
 {
 	uint32_t len;
-	size_t next_size;
+	ssize_t next_size;
 	/* Record reply - first fragment. */
 	if (first_fragment) {
 		xdr_replymsg(reply_xdrs, reply);
@@ -141,10 +155,20 @@ dispatch_reply_fragment_connected(struct plugin_state *state,
 	/* If we don't have space for the data, stop now. */
 	next_size = cdata->connected->client_outbuf_used + 4 +
 		    xdr_getpos(reply_xdrs);
-	if (next_size > sizeof(cdata->connected->client_outbuf)) {
+	if (next_size > cdata->connected->client_outbuf_size) {
+		slapi_log_error(SLAPI_LOG_PLUGIN,
+				state->plugin_desc->spd_id,
+				"failed to queue stream reply (4+%d bytes)!\n",
+				xdr_getpos(reply_xdrs));
+		return FALSE;
+	}
+	/* If we already have data in the buffer, and this would put us over
+	 * the target size, then punt it until next time. */
+	if ((cdata->connected->client_outbuf_used > 4) &&
+	    (next_size > DEFAULT_TARGET_REPLY_SIZE)) {
 		slapi_log_error(SLAPI_LOG_PLUGIN,
 				state->plugin_desc->spd_id,
-				"failed to queue stream reply (4+%d bytes)!n",
+				"saving stream reply (4+%d bytes) for later\n",
 				xdr_getpos(reply_xdrs));
 		return FALSE;
 	}
@@ -156,7 +180,7 @@ dispatch_reply_fragment_connected(struct plugin_state *state,
 	       &len, 4);
 	memcpy(cdata->connected->client_outbuf +
 	       cdata->connected->client_outbuf_used + 4,
-	       reply_buf,
+	       cdata->connected->client_workbuf,
 	       xdr_getpos(reply_xdrs));
 	cdata->connected->client_outbuf_used += (4 + xdr_getpos(reply_xdrs));
 	slapi_log_error(SLAPI_LOG_PLUGIN,
@@ -171,10 +195,10 @@ static void
 dispatch_reply_connected(struct plugin_state *state,
 			 struct dispatch_client_data *cdata,
 			 struct rpc_msg *reply,
-			 XDR *reply_xdrs, char *reply_buf)
+			 XDR *reply_xdrs)
 {
 	dispatch_reply_fragment_connected(state, cdata,
-					  reply, reply_xdrs, reply_buf,
+					  reply, reply_xdrs,
 					  TRUE, TRUE);
 }
 
@@ -189,6 +213,14 @@ dispatch_dgram(struct plugin_state *state, int fd)
 	/* Read the request. */
 	cdata.dgram.client_fd = fd;
 	cdata.dgram.client_addrlen = sizeof(cdata.dgram.client_addr);
+	cdata.dgram.reply_buf = malloc(state->max_dgram_size);
+	cdata.dgram.reply_buf_size = state->max_dgram_size;
+	if (cdata.dgram.reply_buf == NULL) {
+		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
+				"error reading datagram request: "
+				"out of memory\n");
+		return;
+	}
 	reqsize = recvfrom(cdata.dgram.client_fd, dgram, sizeof(dgram), 0,
 			   &cdata.dgram.client_addr,
 			   &cdata.dgram.client_addrlen);
@@ -199,7 +231,11 @@ dispatch_dgram(struct plugin_state *state, int fd)
 	nis_process_request(state, dgram, reqsize,
 			    &dispatch_reply_fragment_dgram,
 			    &dispatch_reply_dgram,
-			    &cdata, NULL);
+			    &cdata,
+			    cdata.dgram.reply_buf,
+			    cdata.dgram.reply_buf_size,
+			    NULL);
+	free(cdata.dgram.reply_buf);
 }
 
 /* Set the client's record up to start reading a new query. */
@@ -222,7 +258,13 @@ client_set_closing(struct plugin_state *state, struct dispatch_client *client)
 	free(client->client_query);
 	client->client_query = NULL;
 	client->client_query_size = 0;
+	free(client->client_outbuf);
+	client->client_outbuf = NULL;
+	client->client_outbuf_size = 0;
 	client->client_outbuf_used = 0;
+	free(client->client_workbuf);
+	client->client_workbuf = NULL;
+	client->client_workbuf_size = 0;
 	client->client_state = client_closing;
 }
 
@@ -232,10 +274,22 @@ dispatch_accept_client(struct plugin_state *state, int fd)
 {
 	struct dispatch_client *client;
 	int flags;
+	char *outbuf, *workbuf;
 	client = malloc(sizeof(*client));
 	if (client == NULL) {
 		return NULL;
 	}
+	outbuf = malloc(state->max_dgram_size + state->max_value_size);
+	if (outbuf == NULL) {
+		free(client);
+		return NULL;
+	}
+	workbuf = malloc(state->max_dgram_size + state->max_value_size);
+	if (workbuf == NULL) {
+		free(outbuf);
+		free(client);
+		return NULL;
+	}
 	fd = accept(fd, &client->client_addr, &client->client_addrlen);
 	if (fd == -1) {
 		free(client);
@@ -249,6 +303,12 @@ dispatch_accept_client(struct plugin_state *state, int fd)
 			"new connected client on %d\n", fd);
 	memset(client, 0, sizeof(*client));
 	client->client_fd = fd;
+	client->client_outbuf = outbuf;
+	client->client_outbuf_size = state->max_dgram_size +
+				     state->max_value_size;
+	client->client_workbuf = workbuf;
+	client->client_workbuf_size = state->max_dgram_size +
+				      state->max_value_size;
 	client_set_reading(state, client);
 	return client;
 }
@@ -369,6 +429,8 @@ client_read(struct plugin_state *state, struct dispatch_client *client)
 					    &dispatch_reply_fragment_connected,
 					    &dispatch_reply_connected,
 					    &client_data,
+					    client->client_workbuf,
+					    client->client_workbuf_size,
 					    &client->client_query_cookie);
 			/* Decide what to do next. */
 			client_interpret_nis_result(state, client);
@@ -420,7 +482,7 @@ client_write(struct plugin_state *state, struct dispatch_client *client)
 			/* More to send, so ask for more reply data. */
 			slapi_log_error(SLAPI_LOG_PLUGIN,
 					state->plugin_desc->spd_id,
-					"waiting for more data for %d\n",
+					"fetching more data for %d\n",
 					client->client_fd);
 			client->client_outbuf_used = 0;
 			memset(&client_data, 0, sizeof(client_data));
@@ -431,6 +493,8 @@ client_write(struct plugin_state *state, struct dispatch_client *client)
 					    &dispatch_reply_fragment_connected,
 					    &dispatch_reply_connected,
 					    &client_data,
+					    client->client_workbuf,
+					    client->client_workbuf_size,
 					    &client->client_query_cookie);
 			client_interpret_nis_result(state, client);
 		}