X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=dccp_send.c;h=3d2511f7ef755e759a2d2f486d4060a785cff225;hp=27343fbc0874e96865880eaa5866d1e65a1fa385;hb=7c11e1acf8efc93b36fcae77c3f0ce04ae491c23;hpb=83e1fb1efd6f9e25198792c182e18ec53165925b

diff --git a/dccp_send.c b/dccp_send.c
index 27343fbc..3d2511f7 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,
@@ -59,7 +65,7 @@ static void dccp_post_select(fd_set *rfds, __a_unused fd_set *wfds)
 	dc = para_calloc(sizeof(struct dccp_client));
 	ret = para_accept(listen_fd, &dc->addr, sizeof(struct sockaddr_in));
 	if (ret < 0) {
-		PARA_ERROR_LOG("%s", PARA_STRERROR(-ret));
+		PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
 		return;
 	}
 	PARA_NOTICE_LOG("connection from %s\n", inet_ntoa(dc->addr.sin_addr));
@@ -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,35 +113,67 @@ 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);
 }
 
-/** give up if write would block that many times */
-#define DCCP_WRITE_RETRIES 100
-
+/*
+ * ret: Negative on errors, zero if nothing was written and write returned
+ * EAGAIN, number of bytes written else.
+ */
 static int dccp_write(int fd, const char *buf, size_t len)
 {
-	size_t size, written = 0;
-	int ret, retries = 0;
-again:
-	size = PARA_MIN(1024, len - written);
-	ret = write(fd, buf + written, size);
+	size_t written = 0;
+	int ret = 0;
+
+	while (written < len) {
+		ret = write(fd, buf + written, PARA_MIN(1024, len - written));
+		/*
+		 * 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.
+		 */
+		if (ret < 0 && errno == EAGAIN)
+			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) {
-		if (errno != EAGAIN || retries++ > DCCP_WRITE_RETRIES)
-			goto err_out;
-		PARA_DEBUG_LOG("EAGAIN #%d@%zd/%zd\n", retries, written, len);
-		goto again;
+		PARA_NOTICE_LOG("enqueue error\n");
+		dccp_shutdown_client(dc);
 	}
-	retries = 0;
-	written += ret;
-	if (written >= len)
-		return written;
-	ret = write_ok(fd);
-	if (ret > 0)
-		goto again;
-err_out:
-	return -E_DCCP_WRITE;
+	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;
+		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,
@@ -141,45 +182,34 @@ static void dccp_send(long unsigned current_chunk,
 	struct dccp_client *dc, *tmp;
 	int ret;
 	char *header_buf;
-	unsigned header_len;
+	size_t header_len;
 
 	if (listen_fd < 0 || !len)
 		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);
 	}
 }