move mutex into ringbuffer

also make buffer void*

controller/controller.c

@@ -16,7 +16,7 @@
 
 #define IIC_DEVICE "/dev/i2c-7"
 #define BUFFER_BLOCKS 128
-#define BUFFER_BLOCK_SIZE 8
+#define BUFFER_DATA_SIZE 8
 
 #define HEADING_SIM_UP "sim/autopilot/heading_up"
 #define HEADING_SIM_DOWN "sim/autopilot/heading_down"
@@ -27,10 +27,13 @@
 
 #define ERROR_IIC_OPEN "open failed"
 
-pthread_mutex_t i2cDataWriteLock;
-pthread_mutex_t i2cDataReadLock;
-struct ringBuffer i2cDataWrite;
 struct ringBuffer i2cDataRead;
+struct ringBuffer i2cDataWrite;
+
+struct blockData {
+  char data[BUFFER_DATA_SIZE];
+  int triggerBit;
+};
 
 struct device {
   uint8_t address;       // I2C address of device
@@ -123,23 +126,25 @@ void sendI2CData(int i2c, struct device d, uint8_t triggerBit, char *data, uint8
 }
 
 int tempCounter = 0;
+int tempGauge = 0;
 void* mqttHandler(void* arg) {
-  char block[BUFFER_BLOCK_SIZE];
+  struct blockData *block = malloc(sizeof(struct blockData));
 
   while (1) {
-    sprintf(block, "%03d", tempCounter);
+    sprintf(block->data, "%03d", tempCounter);
-    pthread_mutex_lock(&i2cDataWriteLock);
+    block->triggerBit = TRIGGER_BIT_1;
     ringBufferWrite(&i2cDataWrite, block);
-    pthread_mutex_unlock(&i2cDataWriteLock);
     tempCounter++;
     sleep(1);
   }
+
+  free(block);
 }
 
 void* i2cHandler(void* arg) {
-  int i2c, tmp;
+  int i2c, tmp, buzy;
   __s32 i2cResponse;
-  char block[BUFFER_BLOCK_SIZE];
+  struct blockData *block = malloc(sizeof(struct blockData));
 
   i2c = open(IIC_DEVICE, O_RDWR);
   if (i2c < 0) {
@@ -147,6 +152,7 @@ void* i2cHandler(void* arg) {
   }
 
   while (1) {
+    buzy = 0;
     for (int i = 0; i < numDevices; i++) {
       setI2CAddress(i2c, devices[i].address);
       i2cResponse = i2c_smbus_read_byte_data(i2c, 0);
@@ -158,48 +164,49 @@ void* i2cHandler(void* arg) {
 
       if (i2cResponse == DATA_RESET_BYTE) {
         sendI2CInit(i2c, devices[i]);
-        sendI2CData(i2c, devices[i], TRIGGER_BIT_1, "111", 3);
+        buzy = 1;
-        sendI2CData(i2c, devices[i], TRIGGER_BIT_2, "222", 3);
       } else if (i2cResponse == DATA_STOP_BYTE) {
         // expect no forther data from device
         // send data to device
         do {
-          pthread_mutex_lock(&i2cDataWriteLock);
           tmp = ringBufferRead(&i2cDataWrite, block);
-          pthread_mutex_unlock(&i2cDataWriteLock);
           if (tmp == 0) {
             // we got something to send
-            sendI2CData(i2c, devices[i], TRIGGER_BIT_1, block, 3);
+            sendI2CData(i2c, devices[i], block->triggerBit, block->data, 3);
+            buzy = 1;
           }
         } while (tmp == 0);
-        usleep(50 * 1000);
         continue;
       } else {
         // real data
         _log("data: 0x%02x\n", i2cResponse);
-        /*
-        XXX: put data into cache and handle with another core / loop / fork / whatever
-        if (i2cResponse & TRIGGER_BIT_1) {
-          if (i2cResponse & HIGH_BIT_1) {
-            sendI2CData(i2c, devices[i], TRIGGER_BIT_1, "+1+", 3);
-          } else {
-            sendI2CData(i2c, devices[i], TRIGGER_BIT_1, "-1-", 3);
-          }
-        }
-
         if (i2cResponse & TRIGGER_BIT_2) {
           if (i2cResponse & HIGH_BIT_2) {
-            sendI2CData(i2c, devices[i], TRIGGER_BIT_2, "+2+", 3);
+            tempGauge++;
+            if (tempGauge > 999) {
+              tempGauge = 999;
+            }
           } else {
-            sendI2CData(i2c, devices[i], TRIGGER_BIT_2, "-2-", 3);
+            tempGauge--;
+            if (tempGauge < 0) {
+              tempGauge = 0;
+            }
           }
+          sprintf(block->data, "%03d", tempGauge);
+          block->triggerBit = TRIGGER_BIT_2;
+          ringBufferWrite(&i2cDataWrite, block);
         }
-        */
+        buzy = 1;
       }
     }
+
+    if (buzy == 0) {
+      usleep(50 * 1000);
+    }
   }
 
   close(i2c);
+  free(block);
   pthread_exit(NULL);
 }
 
@@ -207,16 +214,12 @@ int main() {
   pthread_t i2cThread, mqttThread;
 
   // create the ring buffers for i2c devices
-  ringBufferCreate(BUFFER_BLOCKS, BUFFER_BLOCK_SIZE, &i2cDataRead);
+  ringBufferCreate(BUFFER_BLOCKS, sizeof(struct blockData), &i2cDataRead);
-  ringBufferCreate(BUFFER_BLOCKS, BUFFER_BLOCK_SIZE, &i2cDataWrite);
+  ringBufferCreate(BUFFER_BLOCKS, sizeof(struct blockData), &i2cDataWrite);
 
   // alloc
   rrefSubscriptions = calloc(sizeof(struct device), numDevices);
 
-  // init mutex for ring buffer access
-  pthread_mutex_init(&i2cDataReadLock, NULL);
-  pthread_mutex_init(&i2cDataWriteLock, NULL);
-
   // start and join threads
   pthread_create(&i2cThread, NULL, i2cHandler, NULL);
   pthread_create(&mqttThread, NULL, mqttHandler, NULL);
@@ -224,8 +227,6 @@ int main() {
   pthread_join(mqttThread, NULL);
 
   // free resources
-  pthread_mutex_destroy(&i2cDataReadLock);
-  pthread_mutex_destroy(&i2cDataWriteLock);
   free(rrefSubscriptions);
   ringBufferDestroy(&i2cDataRead);
   ringBufferDestroy(&i2cDataWrite);

controller/ringbuffer.c

@@ -8,15 +8,15 @@
 
 void _ringBufferIncReader(struct ringBuffer *buf) {
   buf->reader += buf->blockSize;
-  if (buf->reader >= buf->blocks * buf->blockSize) {
+  if (buf->reader >= buf->buffer + buf->blocks * buf->blockSize) {
-    buf->reader = 0;
+    buf->reader = buf->buffer;
   }
 }
 
 void _ringBufferIncWriter(struct ringBuffer *buf) {
   buf->writer += buf->blockSize;
-  if (buf->writer >= buf->blocks * buf->blockSize) {
+  if (buf->writer == buf->buffer + buf->blocks * buf->blockSize) {
-    buf->writer = 0;
+    buf->writer = buf->buffer;
   }
 
   if (buf->writer == buf->reader) {
@@ -29,30 +29,37 @@ void _ringBufferIncWriter(struct ringBuffer *buf) {
 }
 
 void ringBufferCreate(int blocks, size_t blockSize, struct ringBuffer *out) {
-  out->buffer = malloc(blocks * blockSize * sizeof(char));
+  out->buffer = malloc(blocks * blockSize);
   out->blocks = blocks;
   out->blockSize = blockSize;
-  out->reader = 0;
+  out->reader = out->buffer;
-  out->writer = 0;
+  out->writer = out->buffer;
+  pthread_mutex_init(&out->mutex, NULL);
 }
 
 void ringBufferDestroy(struct ringBuffer *buf) {
   free(buf->buffer);
+  pthread_mutex_destroy(&buf->mutex);
 }
 
-int ringBufferRead(struct ringBuffer *buf, char *out) {
+int ringBufferRead(struct ringBuffer *buf, void *out) {
+  pthread_mutex_lock(&buf->mutex);
   if (buf->reader != buf->writer) {
     // we have data to read
-    memcpy(out, &buf->buffer[buf->reader], buf->blockSize * sizeof(char));
+    memcpy(out, buf->reader, buf->blockSize);
     _ringBufferIncReader(buf);
+    pthread_mutex_unlock(&buf->mutex);
     return 0;
   } else {
     // nothing to read
+    pthread_mutex_unlock(&buf->mutex);
     return 1;
   }
 }
 
-void ringBufferWrite(struct ringBuffer *buf, char *in) {
+void ringBufferWrite(struct ringBuffer *buf, void *in) {
-  memcpy(&buf->buffer[buf->writer], in, buf->blockSize * sizeof(char));
+  pthread_mutex_lock(&buf->mutex);
+  memcpy(buf->writer, in, buf->blockSize);
   _ringBufferIncWriter(buf);
+  pthread_mutex_unlock(&buf->mutex);
 }

controller/ringbuffer.h

@@ -1,12 +1,15 @@
+#include <pthread.h>
+
 struct ringBuffer {
-  char *buffer;
+  void *buffer;
   int blocks;
   size_t blockSize;
-  int reader;
+  void *reader;
-  int writer;
+  void *writer;
+  pthread_mutex_t mutex;
 };
 
 void ringBufferCreate(int blocks, size_t blockSize, struct ringBuffer *out);
 void ringBufferDestroy(struct ringBuffer *buf);
-int ringBufferRead(struct ringBuffer *buf, char *out);
+int ringBufferRead(struct ringBuffer *buf, void *out);
-void ringBufferWrite(struct ringBuffer *buf, char *in);
+void ringBufferWrite(struct ringBuffer *buf, void *in);