Added implementation of RingBuffer to compensate existing issues on RA side

CronosAgent.c

@@ -53,6 +53,13 @@
 #include "Rewind.h"
 #include "Version.h"
 
+#include "RingBuffer.h"
+
+// #include "DMR.h"
+#ifndef DMR_H
+#define TDMA_FRAME_DURATION  60
+#endif
+
 // #include "ASNTools.h"
 #ifndef ASNTOOLS_H
 #define ASN_SEQUENCE      0x10
@@ -117,7 +124,7 @@
 #define MODE_SYSLOG  (1 << 1)
 #define MODE_DAEMON  (1 << 2)
 
-#define EVENT_LIST_LENGTH  (4 + 1 + 4)
+#define EVENT_LIST_LENGTH  (4 + 2 + 4)
 
 #define BUFFER_SIZE        2048
 #define EXPIRATION_TIME    60
@@ -359,15 +366,22 @@ int main(int argc, const char* argv[])
   
   // Initialize timer handle
 
-  int timerHandle;
+  int timerHandle1;
+  int timerHandle2;
   struct itimerspec timerInterval;
-
   memset(&timerInterval, 0, sizeof(timerInterval));
-  timerInterval.it_interval.tv_sec = REWIND_KEEP_ALIVE_INTERVAL;
-  timerInterval.it_value.tv_sec = REWIND_KEEP_ALIVE_INTERVAL;
 
-  timerHandle = timerfd_create(CLOCK_MONOTONIC, 0);
+  timerInterval.it_value.tv_sec = REWIND_KEEP_ALIVE_INTERVAL;
-  timerfd_settime(timerHandle, 0, &timerInterval, NULL);
+  timerInterval.it_interval.tv_sec = REWIND_KEEP_ALIVE_INTERVAL;
+
+  timerHandle1 = timerfd_create(CLOCK_MONOTONIC, 0);
+  timerfd_settime(timerHandle1, 0, &timerInterval, NULL);
+
+  timerInterval.it_interval.tv_sec = 0;
+  timerInterval.it_interval.tv_nsec = TDMA_FRAME_DURATION * 1000000;
+
+  timerHandle2 = timerfd_create(CLOCK_MONOTONIC, 0);
+  timerfd_settime(timerHandle2, 0, &timerInterval, NULL);
 
   // Initialize signal handle
 
@@ -407,7 +421,11 @@ int main(int argc, const char* argv[])
   epoll_ctl(pollHandle, EPOLL_CTL_ADD, event.data.fd, &event);
 
   event.events = EPOLLIN;
-  event.data.fd = timerHandle;
+  event.data.fd = timerHandle1;
+  epoll_ctl(pollHandle, EPOLL_CTL_ADD, event.data.fd, &event);
+
+  event.events = EPOLLIN;
+  event.data.fd = timerHandle2;
   epoll_ctl(pollHandle, EPOLL_CTL_ADD, event.data.fd, &event);
 
   event.events = EPOLLIN;
@@ -453,6 +471,9 @@ int main(int argc, const char* argv[])
   EV_SET(change, 1, EVFILT_TIMER, EV_ADD | EV_ENABLE, NOTE_SECONDS, REWIND_KEEP_ALIVE_INTERVAL, 0);
   change ++;
 
+  EV_SET(change, 2, EVFILT_TIMER, EV_ADD | EV_ENABLE, NOTE_USECONDS, TDMA_FRAME_DURATION, 0);
+  change ++;
+
   EV_SET(change, SIGINT, EVFILT_SIGNAL, EV_ADD | EV_ENABLE, 0, 0, 0);
   change ++;
 
@@ -488,6 +509,9 @@ int main(int argc, const char* argv[])
   size_t passwordLength = strlen(serverPassword);
   time_t watchDog = now.tv_sec + EXPIRATION_TIME;
 
+  struct RingBuffer ringBuffers[2];
+  memset(ringBuffers, 0, sizeof(ringBuffers));
+
   uint32_t sequenceNumbers[] = 
   {
     0,
@@ -547,6 +571,21 @@ int main(int argc, const char* argv[])
 
           if (type == REWIND_TYPE_EXTERNAL_SERVER)
           {
+            uint16_t flags  = le16toh(incomingBuffer->flags);
+            uint32_t number = le32toh(incomingBuffer->number);
+
+            if (flags & REWIND_FLAG_REAL_TIME_1)
+            {
+              PushData(ringBuffers + 0, number, incomingBuffer->data, length);
+              continue;
+            }
+
+            if (flags & REWIND_FLAG_REAL_TIME_2)
+            {
+              PushData(ringBuffers + 1, number, incomingBuffer->data, length);
+              continue;
+            }
+
             repeaterSocketAddress.sin_port = htons(KAIROS_HAM_DEFAULT_PORT);
             sendto(mediaHandle, incomingBuffer->data, length, 0, (struct sockaddr*)&repeaterSocketAddress, sizeof(struct sockaddr_in));
             continue;
@@ -733,7 +772,7 @@ int main(int argc, const char* argv[])
       // Handle timer to transmit keep-alive
 
 #ifdef __linux__
-      if (event->data.fd == timerHandle)
+      if (event->data.fd == timerHandle1)
       {
         uint64_t information;
         read(timerHandle, &information, sizeof(information));
@@ -741,7 +780,8 @@ int main(int argc, const char* argv[])
         clock_gettime(CLOCK_MONOTONIC, &now);
 #endif
 #ifdef __MACH__
-      if (event->filter == EVFILT_TIMER)
+      if ((event->ident  == 1) &&
+          (event->filter == EVFILT_TIMER))
       {
         clock_get_time(clockService, &now);
 #endif
@@ -769,6 +809,25 @@ int main(int argc, const char* argv[])
         continue;
       }
 
+      // Handle timer to process RingBuffer
+
+#ifdef __linux__
+      if (event->data.fd == timerHandle2)
+      {
+        uint64_t information;
+        read(timerHandle, &information, sizeof(information));
+#endif
+#ifdef __MACH__
+      if ((event->ident  == 2) &&
+          (event->filter == EVFILT_TIMER))
+      {
+#endif
+        repeaterSocketAddress.sin_port = htons(KAIROS_HAM_DEFAULT_PORT);
+        ProcessBuffer(ringBuffers + 0, mediaHandle, &repeaterSocketAddress);
+        ProcessBuffer(ringBuffers + 1, mediaHandle, &repeaterSocketAddress);
+        continue;
+      }
+
       // Handle signal from the kernel
 
 #ifdef __linux__
@@ -816,7 +875,8 @@ int main(int argc, const char* argv[])
 
 #ifdef __linux__
   close(pollHandle);
-  close(timerHandle);
+  close(timerHandle1);
+  close(timerHandle2);
   close(signalHandle);
 #endif
 #ifdef __MACH__

Makefile

@@ -21,7 +21,9 @@ ifeq ($(USE_OPENSSL), yes)
 endif
 endif
 
-OBJECTS = CronosAgent.o
+OBJECTS = \
+  RingBuffer.o \
+  CronosAgent.o
 
 ifneq ($(USE_OPENSSL), yes)
   OBJECTS += sha256.o
@@ -62,7 +64,6 @@ clean:
 	rm -f *.d
 
 version:
-	# echo "#define VERSION $(shell svn info | grep -E "^Revision:" | grep -o -E "[0-9]+")" > Version.h
 	echo "#define VERSION $(shell date -u +%Y%m%d)" > Version.h
 
 debian-package:

RingBuffer.c

@@ -0,0 +1,47 @@
+#include "RingBuffer.h"
+
+#include <string.h>
+#include <unistd.h>
+
+void PushData(struct RingBuffer* buffer, uint32_t number, uint8_t* data, size_t length)
+{
+  number %= BUFFER_LENGTH;
+
+  buffer->delay += (buffer->flags == 0);  // Postpone processing if buffer empty
+  buffer->flags |= (1 << number);         // Set processing flag for the record
+
+  struct BufferRecord* record = buffer->records + number;
+
+  record->length = length;
+  memcpy(record->data, data, length);
+}
+
+void ProcessBuffer(struct RingBuffer* buffer, int handle, struct sockaddr_in* address)
+{
+  if (buffer->flags == 0)
+  {
+  	// Nothing to process
+  	return;
+  }
+
+  if (buffer->delay > 0)
+  {
+  	// Processing was postponed
+    buffer->delay --;
+    return;
+  }
+
+  size_t mask = (1 << buffer->pointer);
+
+  if (buffer->flags & mask)
+  {
+  	// Transmit scheduled data
+    struct BufferRecord* record = buffer->records + buffer->pointer;
+    sendto(handle, record->data, record->length, 0, (struct sockaddr*)address, sizeof(struct sockaddr_in));
+    // Clear processing flag
+    buffer->flags ^= mask;
+  }
+
+  buffer->pointer ++;
+  buffer->pointer %= BUFFER_LENGTH;
+}

RingBuffer.h

@@ -0,0 +1,37 @@
+#ifndef RINGBUFFER_H
+#define RINGBUFFER_H
+
+#include <stdint.h>
+#include <stddef.h>
+#include <arpa/inet.h>
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#define DATA_LENGTH    256
+#define BUFFER_LENGTH  8
+
+struct BufferRecord
+{
+  size_t length;
+  uint8_t data[DATA_LENGTH];
+};
+
+struct RingBuffer
+{
+  size_t flags;
+  size_t delay;
+  size_t pointer;
+  struct BufferRecord records[BUFFER_LENGTH];
+};
+
+void PushData(struct RingBuffer* buffer, uint32_t number, uint8_t* data, size_t length);
+void ProcessBuffer(struct RingBuffer* buffer, int handle, struct sockaddr_in* address);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

Version.h

@@ -1 +1 @@
-#define VERSION 20160804
+#define VERSION 20160806