Add multithread support

Now there is one thread per probe doing the data capture.

Signed-off-by: Daniel Lezcano <daniel.lezcano@linaro.org>

2 files changed, 97 insertions, 64 deletions

diff --git a/Makefile b/Makefile
index e98bd5d..7853f96 100644
--- a/Makefile
+++ b/Makefile
@@ -4,7 +4,7 @@ SAMPLING=-l 100
 TIMESTAMP=-T
 
 # Compilation variables
-CFLAGS?=-g -Wall
+CFLAGS?=-g -Wall -pthread
 CC=gcc
 SRC=$(wildcard *.c)
 OBJS=$(SRC:%.c=%.o)
diff --git a/aep.c b/aep.c
index 576d213..c8a0b7b 100644
--- a/aep.c
+++ b/aep.c
@@ -9,6 +9,7 @@
 #include <string.h>
 #include <unistd.h>
 #include <arpa/inet.h>
+#include <pthread.h>
 #include <signal.h>
 #include <sys/epoll.h>
 #include <sys/ioctl.h>
@@ -342,7 +343,7 @@ static int aep_write_char(int fd, char c)
 	return 0;
 }
 
-int aep_stop(int fd)
+static int aep_stop(int fd)
 {
 	return aep_write_char(fd, AEP_STOP);
 }
@@ -433,7 +434,7 @@ int aep_config(int fd)
 	return 0;
 }
 
-int aep_start(int fd)
+static int aep_start(int fd)
 {
 	char ack;
 
@@ -468,9 +469,7 @@ static inline float aep_char2float(unsigned char *value)
 /*
  * We can't read more the 10000 samples at once
  */
-static unsigned char buffer[10000 * sizeof(struct aep_frame)];
-
-int aep_read_frame(struct aep_device *dev, struct aep_frame *frame, int nr_frame)
+static int aep_read_frame(struct aep_device *dev, struct aep_frame *frame, int nr_frame, unsigned char *buffer)
 {
 	ssize_t len, frame_size;
 	ssize_t nr_bytes, bytes_read;
@@ -638,11 +637,11 @@ static void sighandler(int sig)
 	intr = 1;
 }
 
-static void aep_print(struct aep_options *opt, struct aep_device *dev, struct aep_frame *frame)
+static void aep_print(struct aep_options *opt, struct aep_device *dev,
+		      struct aep_frame *frame, char *buffer)
 {
 	int i, j;
 	float values[AEP_MAX_PROBES][AEP_NR_CHANNELS][AEP_CHANNEL_SIZE];
-	static char buffer[1024];
 	size_t len = 0;
 
 	if (opt->average && !aep_throttle(dev, opt))
@@ -675,21 +674,50 @@ static void aep_print(struct aep_options *opt, struct aep_device *dev, struct ae
 	}
 }
 
-int main(int argc, char *argv[])
+struct aep_thread_arg {
+	struct aep_options *aep_opt;
+	int index;
+};
+
+static void *aep_capture_thread(void *data)
 {
-	int i, j, epfd;
+	int i, epfd;
 	int frame_read, nr_frame = 10000;
+	unsigned char *buffer;
+	char *output;
+
 	struct aep_frame frame[nr_frame];
-	struct aep_options aep_opt;
-	struct aep_device aep_dev[AEP_MAX_PROBES];
-	struct epoll_event eevt[AEP_MAX_PROBES];
+	struct aep_thread_arg *arg = data;
+	struct aep_options *aep_opt = arg->aep_opt;
+	struct aep_device aep_dev;
+	struct epoll_event eevt;
+
+	buffer = malloc(10000 * sizeof(struct aep_frame));
+	if (!buffer) {
+		fprintf(stderr, "Failed to allocate memory buffer\n");
+		return NULL;
+	}
 
-	setbuf(stdout, NULL);
+	output = malloc(1024);
+	if (!output) {
+		fprintf(stderr, "Failed to allocate memory for output\n");
+		return;
+	}
 
-	if (aep_getopt(argc, argv, &aep_opt)) {
-		fprintf(stderr, "Failed to get options\n");
+	memset(&aep_dev, 0, sizeof(aep_dev));
+	memset(&eevt, 0, sizeof(eevt));
+
+	aep_dev.fd = open(aep_opt->probe[arg->index], O_RDWR);
+	if (aep_dev.fd < 0) {
+		fprintf(stderr, "Failed to open '%s': %m\n", aep_opt->probe[arg->index]);
 		return 1;
 	}
+	
+	aep_dev.index = arg->index;
+	aep_dev.name = aep_opt->name[arg->index];
+
+	eevt.events = EPOLLIN | EPOLLET;
+	eevt.data.ptr = &aep_dev;
 
 	epfd = epoll_create(1);
 	if (epfd < 0) {
@@ -697,37 +725,19 @@ int main(int argc, char *argv[])
 		return 1;
 	}
 
-	memset(aep_dev, 0, sizeof(aep_dev) / sizeof(aep_dev[0]));
-	memset(eevt, 0, sizeof(eevt) / sizeof(eevt[0]));
-
-	for (i = 0; i < aep_opt.nrprobe; i++) {
-
-		aep_dev[i].fd = open(aep_opt.probe[i], O_RDWR);
-		if (aep_dev[i].fd < 0) {
-			fprintf(stderr, "Failed to open '%s': %m\n", aep_opt.probe[i]);
-			return 1;
-		}
-
-		aep_dev[i].index = i;
-		aep_dev[i].name = aep_opt.name[i];
-
-		eevt[i].events = EPOLLIN | EPOLLET;
-		eevt[i].data.ptr = &aep_dev[i];
-
-		if (epoll_ctl(epfd, EPOLL_CTL_ADD, aep_dev[i].fd, &eevt[i]) < 0) {
-			fprintf(stderr, "Failed to add fd to epoll: %m\n");
-			return 1;
-		}
+	if (epoll_ctl(epfd, EPOLL_CTL_ADD, aep_dev.fd, &eevt) < 0) {
+		fprintf(stderr, "Failed to add fd to epoll: %m\n");
+		return 1;
+	}
 
-		if (aep_config(aep_dev[i].fd)) {
-			fprintf(stderr, "Failed to configure probe\n");
-			return 1;
-		}
+	if (aep_config(aep_dev.fd)) {
+		fprintf(stderr, "Failed to configure probe\n");
+		return 1;
+	}
 
-		if (aep_start(aep_dev[i].fd)) {
-			fprintf(stderr, "Failed to start probe\n");
-			return 1;
-		}
+	if (aep_start(aep_dev.fd)) {
+		fprintf(stderr, "Failed to start probe\n");
+		return 1;
 	}
 
 	signal(SIGINT, sighandler);
@@ -737,7 +747,7 @@ again:
 		int nr_fds;
 		struct aep_device *dev;
 
-		nr_fds = epoll_wait(epfd, eevt, aep_opt.nrprobe, -1);
+		nr_fds = epoll_wait(epfd, &eevt, 1, -1);
 		if (nr_fds < 0) {
 			if (errno == EINTR)
 				goto again;
@@ -745,34 +755,57 @@ again:
 			return 2;
 		}
 
-		for (j = 0; j < nr_fds; j++) {
+		if (!(eevt.events & EPOLLIN))
+			continue;
 
-			if (!(eevt[j].events & EPOLLIN))
-				continue;
+		dev = (struct aep_device *)eevt.data.ptr;
+
+		frame_read = aep_read_frame(dev, frame, nr_frame, buffer);
+		if (frame_read < 0) {
+			fprintf(stderr, "Failed to read frames\n");
+			return 2;
+		}
 
-			dev = (struct aep_device *)eevt[j].data.ptr;
+		for (i = 0; i < frame_read; i++, dev->total_frame++)
+			aep_print(aep_opt, dev, &frame[i], output);
+	}
 
-			frame_read = aep_read_frame(dev, frame, nr_frame);
-			if (frame_read < 0) {
-				fprintf(stderr, "Failed to read frames\n");
-				return 2;
-			}
+	aep_stop(aep_dev.fd);
 
-			for (i = 0; i < frame_read; i++, dev->total_frame++)
-				aep_print(&aep_opt, dev, &frame[i]);
-		}
+	if (aep_dev.frame_missed)
+		fprintf(stderr, "%s missed %ld frames\n", aep_dev.name,
+			aep_dev.frame_missed);
+
+	return NULL;
+}
+
+int main(int argc, char *argv[])
+{
+	int i;
+	struct aep_options aep_opt;
+	struct aep_thread_arg *arg;
+
+	pthread_t threads[AEP_MAX_PROBES];
+
+	setbuf(stdout, NULL);
+
+	if (aep_getopt(argc, argv, &aep_opt)) {
+		fprintf(stderr, "Failed to get options\n");
+		return 1;
 	}
 
 	for (i = 0; i < aep_opt.nrprobe; i++) {
 
-		aep_stop(aep_dev[i].fd);
-
-		if (aep_dev[i].frame_missed)
-			fprintf(stderr, "%s missed %ld frames\n",
-				aep_dev[i].name,
-				aep_dev[i].frame_missed);
+		arg = malloc(sizeof(*arg));
+		arg->aep_opt = &aep_opt;
+		arg->index = i;
 
+		pthread_create(&threads[i], NULL,
+			       aep_capture_thread, arg);
 	}
 
+	for (i = 0; i < aep_opt.nrprobe; i++)
+		pthread_join(threads[i], NULL);
+
 	return 0;
 }