From 92940b605473582b4371eb36ec52c9c8b561aa7d Mon Sep 17 00:00:00 2001
From: Petr Baudis <pasky@ucw.cz>
Date: Sun, 27 Nov 2011 01:24:52 +0100
Subject: [PATCH] Connection communication in separate thread

---
 Makefile      |  5 ++--
 agent.cc      |  6 +++--
 connection.cc | 66 +++++++++++++++++++++++++++++++++++++++++++++++++--
 connection.h  | 22 ++++++++++++++++-
 4 files changed, 92 insertions(+), 7 deletions(-)

diff --git a/Makefile b/Makefile
index 82f1b8b..df473ef 100644
--- a/Makefile
+++ b/Makefile
@@ -1,9 +1,10 @@
-CFLAGS=-Wall -g
+CFLAGS=-Wall -g -pthread
+LDFLAGS=-pthread
 
 OBJS=main.o map.o agent.o connection.o
 
 brmlife: $(OBJS)
-	$(CXX) -o $@ $^
+	$(CXX) $(LDFLAGS) -o $@ $^
 
 
 clean:
diff --git a/agent.cc b/agent.cc
index 05b7ad4..348c948 100644
--- a/agent.cc
+++ b/agent.cc
@@ -51,6 +51,8 @@ agent::on_senses_update(void)
 agent::~agent()
 {
 	tile->on_agent_leave(*this);
-	if (conn)
-		delete conn;
+	if (conn) {
+		conn->cancel();
+		conn = NULL;
+	}
 };
diff --git a/connection.cc b/connection.cc
index dcd9696..fdc2221 100644
--- a/connection.cc
+++ b/connection.cc
@@ -4,6 +4,8 @@
 #include <iostream>
 #include <unistd.h>
 
+#include <pthread.h>
+
 #include "agent.h"
 #include "connection.h"
 
@@ -12,6 +14,66 @@ connection::senses(int tick_id, char around[4])
 {
 	char buf[1024];
 	snprintf(buf, sizeof(buf), "tick %d\naround %c%c%c%c\n\n", tick_id, around[0], around[1], around[2], around[3]);
-	if (write(fd, buf, strlen(buf)) < (ssize_t) strlen(buf))
-		error = true;
+
+	pthread_mutex_lock(&buf_lock);
+	out_buf.append(buf);
+	pthread_mutex_unlock(&buf_lock);
+}
+
+void
+connection::cancel(void)
+{
+	pthread_cond_signal(&cancel_cond);
+}
+
+
+void *
+conn_thread_worker(void *ctx)
+{
+	class connection *conn = (class connection *) ctx;
+	conn->thread_loop();
+	return NULL;
+}
+
+void
+connection::spawn_thread(void)
+{
+	pthread_mutex_init(&buf_lock, NULL);
+	pthread_cond_init(&cancel_cond, NULL);
+	pthread_mutex_init(&cancel_lock, NULL);
+
+	pthread_t tid;
+	pthread_create(&tid, NULL, conn_thread_worker, (void *) this);
+	pthread_detach(tid);
+}
+
+void
+connection::thread_loop(void)
+{
+	while (!error) {
+		/* Repeatedly try to write and read stuff. */
+		std::string buf;
+		ssize_t len;
+
+		pthread_mutex_lock(&buf_lock);
+		buf = out_buf;
+		pthread_mutex_unlock(&buf_lock);
+
+		len = write(fd, buf.c_str(), buf.size());
+		if (len < 0) {
+			pthread_mutex_lock(&cancel_lock);
+			error = true;
+		} else {
+			pthread_mutex_lock(&buf_lock);
+			out_buf.erase(0, len);
+			pthread_mutex_unlock(&buf_lock);
+		}
+
+		/* TODO: The reading. ;-) */
+
+		usleep(10000); // XXX: Signal-oriented instead.
+	}
+
+	pthread_cond_wait(&cancel_cond, &cancel_lock);
+	delete this;
 }
diff --git a/connection.h b/connection.h
index b0e27c6..272c517 100644
--- a/connection.h
+++ b/connection.h
@@ -1,6 +1,9 @@
 #ifndef BRMLIFE__CONNECTION_H
 #define BRMLIFE__CONNECTION_H
 
+#include <string>
+
+#include <pthread.h>
 #include <unistd.h>
 
 #include "map.h"
@@ -10,7 +13,11 @@ public:
 	int fd;
 	bool error;
 
-	connection(int fd_) : fd(fd_) {}
+	connection(int fd_)
+	: fd(fd_), error(false)
+	{
+		spawn_thread();
+	}
 
 	~connection()
 	{
@@ -18,6 +25,19 @@ public:
 	}
 
 	void senses(int tick_id, char around[4]);
+
+	void cancel(void);
+
+private:
+	std::string out_buf;
+	pthread_mutex_t buf_lock;
+
+	pthread_cond_t cancel_cond;
+	pthread_mutex_t cancel_lock;
+
+	void spawn_thread(void);
+	friend void *conn_thread_worker(void *ctx);
+	void thread_loop(void);
 };
 
 #endif