From 52422cebf0e513f7c6f25a3b44a35d05a7fe2c13 Mon Sep 17 00:00:00 2001 From: Artem Prilutskiy Date: Sat, 6 Aug 2016 14:30:27 +0300 Subject: [PATCH] Added implementation of RingBuffer to compensate existing issues on RA side --- CronosAgent.c | 82 ++++++++++++++++++++++++++++++++++++++++++++------- Makefile | 5 ++-- RingBuffer.c | 47 +++++++++++++++++++++++++++++ RingBuffer.h | 37 +++++++++++++++++++++++ Version.h | 2 +- 5 files changed, 159 insertions(+), 14 deletions(-) create mode 100644 RingBuffer.c create mode 100644 RingBuffer.h diff --git a/CronosAgent.c b/CronosAgent.c index d2f8cf3..b0e6185 100644 --- a/CronosAgent.c +++ b/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); - timerfd_settime(timerHandle, 0, &timerInterval, NULL); + timerInterval.it_value.tv_sec = REWIND_KEEP_ALIVE_INTERVAL; + 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__ diff --git a/Makefile b/Makefile index 25aa0b7..d9b8ee6 100644 --- a/Makefile +++ b/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: diff --git a/RingBuffer.c b/RingBuffer.c new file mode 100644 index 0000000..60f9c0c --- /dev/null +++ b/RingBuffer.c @@ -0,0 +1,47 @@ +#include "RingBuffer.h" + +#include +#include + +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; +} diff --git a/RingBuffer.h b/RingBuffer.h new file mode 100644 index 0000000..748609d --- /dev/null +++ b/RingBuffer.h @@ -0,0 +1,37 @@ +#ifndef RINGBUFFER_H +#define RINGBUFFER_H + +#include +#include +#include + +#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 diff --git a/Version.h b/Version.h index e44780b..3149507 100644 --- a/Version.h +++ b/Version.h @@ -1 +1 @@ -#define VERSION 20160804 +#define VERSION 20160806