From: Andre Noll <maan@systemlinux.org>
Date: Sun, 12 Aug 2007 21:26:27 +0000 (+0200)
Subject: dccp_send.c: Implement chunk queuing for dccp.
X-Git-Tag: v0.2.17~28
X-Git-Url: http://git.tue.mpg.de/?a=commitdiff_plain;h=bd38899d7d8cd4043bfee85fcdb55159ca5ba1a2;p=paraslash.git

dccp_send.c: Implement chunk queuing for dccp.

As Gerrit mentioned, it might be a good idea to use chunk
queueing also for dccp. This patch uses the new cq_xxx()
abstractions in the same way as the http sender.

Not heavily tested yet.
---

diff --git a/dccp_send.c b/dccp_send.c
index bdc7681e..2c8f800f 100644
--- a/dccp_send.c
+++ b/dccp_send.c
@@ -21,6 +21,7 @@
 #include "string.h"
 #include "fd.h"
 #include "close_on_fork.h"
+#include "chunk_queue.h"
 #include "server.cmdline.h"
 
 /** the list of connected clients **/
@@ -28,6 +29,9 @@ static struct list_head clients;
 static int listen_fd = -1;
 static struct sender *self;
 
+/** Maximal number of bytes in a chunk queue. */
+#define DCCP_MAX_PENDING_BYTES 40000
+
 /** describes one connected client */
 struct dccp_client {
 	/** the dccp socket */
@@ -38,6 +42,8 @@ struct dccp_client {
 	struct list_head node;
 	/** non-zero if audio file header has been sent */
 	int header_sent;
+	/** The list of pending chunks for this client. */
+	struct chunk_queue *cq;
 };
 
 static void dccp_pre_select( int *max_fileno, fd_set *rfds,
@@ -67,6 +73,7 @@ static void dccp_post_select(fd_set *rfds, __a_unused fd_set *wfds)
 	para_list_add(&dc->node, &clients);
 	add_close_on_fork_list(dc->fd);
 	mark_fd_nonblock(dc->fd);
+	dc->cq = cq_new(DCCP_MAX_PENDING_BYTES);
 }
 
 static int dccp_open(void)
@@ -90,8 +97,10 @@ static int dccp_open(void)
 	if (ret < 0)
 		return ret;
 	ret = listen(listen_fd, 0);
-	if (ret < 0)
+	if (ret < 0) {
+		PARA_ERROR_LOG("%s\n", strerror(errno));
 		return -E_DCCP_LISTEN;
+	}
 	PARA_DEBUG_LOG("listening on fd %d\n", listen_fd);
 	add_close_on_fork_list(listen_fd);
 	mark_fd_nonblock(listen_fd);
@@ -104,10 +113,15 @@ static void dccp_shutdown_client(struct dccp_client *dc)
 		dc->fd);
 	close(dc->fd);
 	del_close_on_fork_list(dc->fd);
+	cq_destroy(dc->cq);
 	list_del(&dc->node);
 	free(dc);
 }
 
+/*
+ * ret: Negative on errors, zero if nothing was written and write would block,
+ * number of bytes written else.
+ */
 static int dccp_write(int fd, const char *buf, size_t len)
 {
 	size_t written = 0;
@@ -119,17 +133,51 @@ static int dccp_write(int fd, const char *buf, size_t len)
 		 * Error handling: CCID3 has a sending wait queue which fills up and is
 		 * emptied asynchronously. The EAGAIN case means that there is currently
 		 * no space in the wait queue, but this can change at any moment and is
-		 * thus not an error condition. Keep polling until an entry becomes free.
+		 * thus not an error condition.
 		 */
 		if (ret < 0 && errno == EAGAIN)
-			continue;
-		if (ret < 0)
+			return written;
+		if (ret < 0) {
+			PARA_ERROR_LOG("%s\n", strerror(errno));
 			return -E_DCCP_WRITE;
+		}
 		written += ret;
 	}
 	return written;
 }
 
+static int queue_chunk_or_shutdown(struct dccp_client *dc, long unsigned chunk_num,
+	size_t sent)
+{
+	int ret = cq_enqueue(dc->cq, chunk_num, sent);
+	if (ret < 0) {
+		PARA_NOTICE_LOG("enqueue error\n");
+		dccp_shutdown_client(dc);
+	}
+	return ret;
+}
+
+static int send_queued_chunks(struct dccp_client *dc)
+{
+	struct queued_chunk *qc;
+	while ((qc = cq_peek(dc->cq))) {
+		char *buf;
+		size_t len;
+		int ret = write_ok(dc->fd);
+		if (ret <= 0)
+			return ret? -E_WRITE_OK : 0;
+		cq_get(qc, &buf, &len);
+		ret = dccp_write(dc->fd, buf, len);
+		if (ret < 0)
+			return ret;
+		cq_update(dc->cq, ret);
+		if (ret != len)
+			return 1;
+		cq_dequeue(dc->cq);
+	}
+	return 1;
+}
+
 static void dccp_send(long unsigned current_chunk,
 		__a_unused long unsigned chunks_sent, const char *buf, size_t len)
 {
@@ -142,39 +190,28 @@ static void dccp_send(long unsigned current_chunk,
 		return;
 
 	list_for_each_entry_safe(dc, tmp, &clients, node) {
-		ret = write_ok(dc->fd);
-		if (ret < 0) {
-			dccp_shutdown_client(dc);
-			continue;
-		}
-		if (!ret)
-			continue;
 		if (!dc->header_sent && current_chunk) {
 			header_buf = vss_get_header(&header_len);
 			if (header_buf && header_len > 0) {
-				ret = dccp_write(dc->fd, header_buf, header_len);
-				if (ret != header_len) {
-					int err = errno;
-					PARA_ERROR_LOG("header write: %d/%u (%s)\n",
-						ret, header_len, ret < 0?
-						strerror(err) : "");
-					dccp_shutdown_client(dc);
-					continue;
-				}
-				dc->header_sent = 1;
-				ret = write_ok(dc->fd);
-				if (ret < 0) {
-					dccp_shutdown_client(dc);
-					continue;
-				}
-				if (!ret)
+				if (queue_chunk_or_shutdown(dc, -1U, 0) < 0)
 					continue;
 			}
+			dc->header_sent = 1;
+		}
+		ret = send_queued_chunks(dc);
+		if (ret < 0) {
+			dccp_shutdown_client(dc);
+			continue;
 		}
 //		PARA_DEBUG_LOG("writing %d bytes to fd %d\n", len, dc->fd);
 		ret = dccp_write(dc->fd, buf, len);
-		if (ret != len)
+		if (ret < 0) {
+			PARA_NOTICE_LOG("%s\n", PARA_STRERROR(-ret));
 			dccp_shutdown_client(dc);
+			continue;
+		}
+		if (ret != len)
+			queue_chunk_or_shutdown(dc, current_chunk, ret);
 	}
 }