Debugging and improvements

Makefile

@@ -6,7 +6,8 @@ CFLAGS=
 MACROS=\
 -DZMQ_TOKEN=\"$(ZMQ_TOKEN)\" \
 -DZMQ_MSG_SIZE=$(ZMQ_MSG_SIZE) \
--DLOG_DELAY=$(LOG_DELAY)
+-DLOG_DELAY=$(LOG_DELAY) \
+-DMAX_QUEUE=$(MAX_QUEUE)
 
 all: publisher subscriber
 

config.mk

@@ -29,3 +29,8 @@ LOG_INTERVAL=20
 # It allows you to filter the messages on the publisher
 # if you are using multiple monitoring nodes (multiple publishers)
 KEY=node1
+# MAX_QUEUE Maximum number of queues to use per publisher.
+# It allows the zmq thread to send the power measurements to the subscriber meanwhile
+# other measurements are collected. If only 1 queue is used, power measurements can
+# potentially be missing since parallelism is broken.
+MAX_QUEUE=2

src/publisher.c

@@ -4,9 +4,12 @@
 #include <pthread.h>
 #include <signal.h>
 #include <stdio.h>
+#include <stdlib.h>
 #include <sys/stat.h>
 #include <unistd.h>
 #include <zmq.h>
+#include <fcntl.h>
+
 
 #ifndef LOG_DELAY
 #define LOG_DELAY 0
@@ -82,9 +85,9 @@ int main(int argc, char *argv[]) {
 
   //----- Init logging variables
   pthread_t zmq_thread;
-  FILE *regptr, *logptr;
+  FILE *logptr;
   char logfilepath[STATIC_LEN] = "";
-  regptr = fopen(regpower, "r");
+  int regfd = open(regpower, O_RDONLY);
   char buffer[STATIC_LEN];
   time_t interval;
   struct timespec power_ts;
@@ -116,14 +119,14 @@ int main(int argc, char *argv[]) {
       if (__stop)
         break;
       // Read power:
-      fgets(buffer, STATIC_LEN, regptr);
+      read(regfd, buffer, STATIC_LEN);
       // Get power measurement timestamp:
       clock_gettime(CLOCK_REALTIME, &power_ts);
       // Write measurement into msg buffer:
       char line[MAX_RECORD_LEN];
       if ((queues[queue_id].size + MAX_RECORD_LEN) > ZMQ_MSG_SIZE) {
         printf(
-            "To many measurements to publish. Please increase ZMQ_MSG_SIZE\n");
+            "Too many measurements to publish. Please increase ZMQ_MSG_SIZE\n");
       } else {
         sprintf(queues[queue_id].msg + queues[queue_id].size, "%ld,%ld,%d\n",
                 power_ts.tv_sec, power_ts.tv_nsec, atoi(buffer));
@@ -131,7 +134,7 @@ int main(int argc, char *argv[]) {
             strlen(queues[queue_id].msg + queues[queue_id].size);
       }
       // Reset power register file:
-      fseek(regptr, 0, SEEK_SET);
+       lseek(regfd,0,SEEK_SET);
 #if LOG_DELAY > 0
       usleep(LOG_DELAY * 1000);
 #endif
@@ -140,7 +143,7 @@ int main(int argc, char *argv[]) {
   }
 
   //----- Cleaning
-  fclose(regptr);
+  close(regfd);
   pthread_join(zmq_thread, NULL);
   zmq_close(zmq_publisher);
   zmq_ctx_destroy(zmq_context);

src/utils.h

@@ -11,9 +11,12 @@
 #define INA260_POWER_REGISTER "registers/power"
 #define STATIC_LEN 255
 #define MAX_RECORD_LEN 100
-#define MAX_QUEUES 1
 #define CSV_HEADER "timestamp,nsecs,power"
 
+#ifndef MAX_QUEUES
+#define MAX_QUEUES 1
+#endif
+
 #ifndef ZMQ_TOKEN
 #define ZMQ_TOKEN "ina260-zmq-publisher"
 #endif