refactor server/client discovery to use TLVs.

Signed-off-by: Gediminas Jakutis <gediminas@varciai.lt>

1 files changed, 75 insertions, 45 deletions

diff --git a/src/daemon/net.c b/src/daemon/net.c
index 62172fe..640bd28 100644
--- a/src/daemon/net.c
+++ b/src/daemon/net.c
@@ -19,6 +19,7 @@
  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  */
 
+#include <rin/time.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
@@ -34,6 +35,7 @@
 #include <limits.h>
 #include "settings.h"
 #include "net.h"
+#include "utils.h"
 
 struct netstate {
 	struct timespec lastreply;
@@ -43,8 +45,10 @@ struct netstate {
 	pthread_t listner;
 	char *data;
 	size_t bufsize;
+	ssize_t recvsize;
 	unsigned int available;
-	int status;
+	int sock_status;
+	int buffer_status;
 	pthread_mutex_t mutex;
 	struct sockaddr_in clientaddr;
 };
@@ -98,7 +102,8 @@ int net_init(const unsigned short int port)
 	state[i].data = malloc(MTU);
 	memset(state[i].data, 0, MTU);
 	state[i].bufsize = MTU;
-	state[i].status = NONEWDATA;
+	state[i].sock_status = NONEWDATA;
+	state[i].buffer_status = NONEWDATA;
 	pthread_mutex_init(&state[i].mutex, NULL);
 
 	if (pthread_create(&state[i].listner, NULL, dolisten, state + i)) {
@@ -137,13 +142,14 @@ int net_close(int nd)
 	return ret;
 }
 
-int net_getlastdata(int nd, char ** const data)
+int net_getlastdata(int nd, char ** const data, size_t *recvsize)
 {
+	size_t size;
 	int ret;
 
 	if (nd > count || nd < 1 || state[--nd].available) {
 		ret = ERROR;
-	} else if (!(ret = state[nd].status)) {
+	} else if (!(ret = state[nd].buffer_status)) {
 		if (!data) {
 			ret = ERROR;
 			goto out;
@@ -151,10 +157,13 @@ int net_getlastdata(int nd, char ** const data)
 			*data = malloc(MTU);
 		}
 
+		size = state[nd].recvsize;
 		memset(*data, 0, MTU);
-		memcpy(*data, state[nd].data, MTU);
+		memcpy(*data, state[nd].data, size);
+		*recvsize = size;
 	}
 
+	state[nd].buffer_status = NONEWDATA;
 out:
 	return ret;
 }
@@ -162,27 +171,11 @@ out:
 int net_send(int nd, const char * const buf, size_t buf_size)
 {
 	int ret = A_OK;
-	ssize_t sent;
-	struct timespec now;
 
-	if (nd > count || nd < 1 || state[--nd].available || state[nd].status != A_OK) {
+	if (nd > count || nd < 1 || state[nd - 1].available) {
 		ret = ERROR;
 	} else {
-		if (setting_verbose() >= INFO) {
-			clock_gettime(CLOCK_MONOTONIC_RAW, &now);
-			fprintf(stderr, "Sending DATA, timestap: %li \n", now.tv_sec);
-		}
-
-		sent = sendto(	state[nd].sock, buf, buf_size, MSG_DONTWAIT,
-				(struct sockaddr *) &state[nd].clientaddr,
-				sizeof(state[nd].clientaddr));
-
-		if (sent == -1) {
-			ret = DEAD;
-			if (setting_verbose() >= ERR) {
-				fprintf(stderr, "Sending packet to %s failed.\n", inet_ntoa(state[nd].clientaddr.sin_addr));
-			}
-		}
+		ret = net_send_addr(nd, buf, buf_size, &state[nd - 1].clientaddr);
 	}
 
 	return ret;
@@ -194,13 +187,12 @@ int net_send_addr(int nd, const char * const buf, size_t buf_size, const struct
 	ssize_t sent;
 	struct timespec now;
 
-
 	if (nd > count || nd < 1 || state[--nd].available) {
 		ret = ERROR;
 	} else {
 		if (setting_verbose() >= INFO) {
 			clock_gettime(CLOCK_MONOTONIC_RAW, &now);
-			fprintf(stderr, "Sending DATA, timestap: %li \n", now.tv_sec);
+			fprintf(stderr, "Sending packet to %s, timestap: %li \n", inet_ntoa(state[nd].clientaddr.sin_addr), now.tv_sec);
 		}
 
 		sent = sendto(	state[nd].sock, buf, buf_size, MSG_DONTWAIT,
@@ -218,46 +210,78 @@ int net_send_addr(int nd, const char * const buf, size_t buf_size, const struct
 	return ret;
 }
 
+void net_flush_read_buffer(int nd)
+{
+	struct netstate *st;
+	int cancelstate;
+	struct timespec now;
+	struct timespec later;
+	struct timespec delta;
+
+	if (setting_verbose() >= INFO) {
+		fprintf(stderr, "acquiring net mutex\n");
+	}
+
+	if (!(nd > count || nd < 1 || state[--nd].available || state[nd].sock_status != A_OK)) {
+		st = state + nd;
+		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cancelstate);
+		pthread_mutex_lock(&st->mutex);
+
+		if (setting_verbose() >= INFO) {
+			clock_gettime(CLOCK_MONOTONIC_RAW, &now);
+			fprintf(stderr, "flushing netsocket\n");
+		}
+
+		do {
+			st->sock_status = getpacket(st->data, st->bufsize, &st->recvsize, st->sock, &st->clientaddr);
+		} while(st->sock_status);
+
+		if (setting_verbose() >= INFO) {
+			clock_gettime(CLOCK_MONOTONIC_RAW, &later);
+			delta = rin_time_sub(&later, &now);
+			fprintf(stderr, "finished flushing netsocket in %lis%lins\n", delta.tv_sec, delta.tv_nsec);
+		}
+
+		pthread_mutex_unlock(&st->mutex);
+		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &cancelstate);
+		pthread_testcancel();
+	}
+}
+
 static void *dolisten(void * state)
 {
 	struct timespec now;
-	struct timespec wait = {0, 10 * 1000 * 1000}; /* 10 ms */
+	struct timespec wait = {0, 1 * 1000 * 1000}; /* 1 ms */
 	struct netstate *st;
-	ssize_t recvbufsize;
 	int cancelstate;
 	int oldstatus = DEAD;
-	char *ipstring;
 
 	st = state;
 
 	do {
 		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cancelstate);
 		pthread_mutex_lock(&st->mutex);
-		clock_gettime(CLOCK_MONOTONIC_RAW, &now);
-		st->status = getpacket(st->data, st->bufsize, &recvbufsize, st->sock, &st->clientaddr);
+		st->sock_status = getpacket(st->data, st->bufsize, &st->recvsize, st->sock, &st->clientaddr);
 
-		if (!st->status) {
+		clock_gettime(CLOCK_MONOTONIC_RAW, &now);
+		if (!st->sock_status) {
 			st->lastreply = now;
+			st->buffer_status = A_OK;
 		}
 
 		/* no packets in five seconds */
-		if ((now.tv_sec - st->lastreply.tv_sec) >= 5) {
-			st->status = DEAD;
-		} else {
-			st->status = A_OK;
-		}
+		if ((now.tv_sec - st->lastreply.tv_sec) >= 20) {
+			st->sock_status = DEAD;
+		};
 
-		if (oldstatus != st->status)
-		{
-			oldstatus = st->status;
-			if (st->status == DEAD) {
+		if (oldstatus != st->sock_status && st->sock_status != NONEWDATA) {
+			oldstatus = st->sock_status;
+			if (st->sock_status == DEAD) {
 				/* this timestamp is arbitraty */
 				setting_verbose() >= INFO ? fprintf(stderr, "Connection with the client has been lost. Last reply since: %li \n", st->lastreply.tv_sec) : 0;
-			} else if (st->status == A_OK) {
-				ipstring = inet_ntoa(st->clientaddr.sin_addr);
-				setting_verbose() >= INFO ? fprintf(stderr, "Successful incoming connection from %s\n", ipstring) : 0;
 			}
 		}
+
 		pthread_mutex_unlock(&st->mutex);
 		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &cancelstate);
 		pthread_testcancel();
@@ -273,6 +297,7 @@ static int getpacket(char *data, size_t buffsize, ssize_t *recvbufsize, int sock
 {
 	static struct pollfd pfd = {0};
 	int ret;
+	char *ipstring;
 	socklen_t sender_len = sizeof(*sender);
 
 	if (!pfd.fd) {
@@ -288,10 +313,15 @@ static int getpacket(char *data, size_t buffsize, ssize_t *recvbufsize, int sock
 		ret = NONEWDATA;
 	} else if (pfd.revents & POLLIN) {
 		*recvbufsize = recvfrom(sock, data, buffsize, MSG_DONTWAIT, (struct sockaddr *) sender, &sender_len);
-		ret = A_OK;
+		if (*recvbufsize < 0) {
+			ret = ERR;
+		} else {
+			ipstring = inet_ntoa(sender->sin_addr);
+			setting_verbose() >= INFO ? fprintf(stderr, "packet from %s received\n", ipstring) : 0;
+			ret = A_OK;
+		}
 	} else {
 		ret = NONEWDATA;
 	}
-
 	return ret;
 }