wn/iiotfeeder
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

queue and so on

Browse Source
This commit is contained in:
Wolfgang Hottgenroth 2021-10-17 14:41:21 +02:00
parent ffbaa822c0
commit a99c3b7121
Signed by: wn
GPG Key ID: E49AF3B9EF6DD469
10 changed files with 264 additions and 356 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored Normal file
View File

@ -0,0 +1 @@
build/

9
Makefile
View File

@ -1,9 +1,10 @@
BUILD_DIR = build
C_SOURCES = \
sha256.c \
sink20169.c \
logging.c
iiotfeeder.c \
logging.c \
config.c \
dbhandler.c
VERSION ?= $(shell git rev-parse --short=8 HEAD)
UNAME_S := $(shell uname -s)
@ -13,7 +14,7 @@ CFLAGS = $(shell pkg-config --cflags libpq libconfig) -I. \
-Wall -Werror -std=c99 \
-D$(UNAME_S)=1 -DVERSION="\"$(VERSION)\""
LDFLAGS = $(shell pkg-config --libs libpq libconfig)
TARGET = sink20169
TARGET = iiotfeeder
all: $(BUILD_DIR)/$(TARGET)

28
config.c Normal file
View File

@ -0,0 +1,28 @@
#include <stdint.h>
#include <stdbool.h>
#include <config.h>
#include <libconfig.h>
#include <logging.h>
bool verbose = false;
bool debug = false;
int initConfig(const char *configFilename, t_configHandle *configHandle) {
config_init(&(configHandle->cfg));
if (! config_read_file(&(configHandle->cfg), configFilename)) {
logmsg(LOG_ERR, "failed to read config file: %s:%d - %s\n",
config_error_file(&(configHandle->cfg)), config_error_line(&(configHandle->cfg)),
config_error_text(&(configHandle->cfg)));
config_destroy(&(configHandle->cfg));
return -1;
}
return 0;
}
void deinitConfig(t_configHandle *configHandle) {
config_destroy(&(configHandle->cfg));
}

16
config.h Normal file
View File

@ -0,0 +1,16 @@
#ifndef _CONFIG_H_
#define _CONFIG_H_
#include <libconfig.h>
#include <stdint.h>
typedef struct {
config_t cfg;
} t_configHandle;
int initConfig(const char *configFilename, t_configHandle *configHandle);
void deinitConfig(t_configHandle *configHandle);
#endif // _CONFIG_H_

92
dbhandler.c Normal file
View File

@ -0,0 +1,92 @@
#include <logging.h>
#include <dbhandler.h>
#include <config.h>
#include <libpq-fe.h>
#include <stdlib.h>
t_dbHandle *initDatabaseHandler(t_configHandle *configHandle) {
t_dbHandle *dbHandle = (t_dbHandle*) malloc(sizeof(t_dbHandle));
if (! dbHandle) {
logmsg(LOG_ERR, "out of memory when trying to allocate dbHandle");
return NULL;
}
dbHandle->configHandle = configHandle;
dbHandle->postgresqlConnInfo = NULL;
config_lookup_string(&(configHandle->cfg), "postgresqlConnInfo", &(dbHandle->postgresqlConnInfo));
if (! dbHandle->postgresqlConnInfo) {
logmsg(LOG_ERR, "no postgresql connInfo configured");
free(dbHandle);
dbHandle = NULL;
return NULL;
}
dbHandle->conn = NULL;
return dbHandle;
}
void deinitDatabaseHandler(t_dbHandle *handle) {
if (handle) {
PQfinish(handle->conn);
free(handle);
handle = NULL;
}
}
int openDatabase(t_dbHandle *handle) {
int res = 0;
if (! handle->conn) {
logmsg(LOG_DEBUG, "Opening connection to database");
handle->conn = PQconnectdb(handle->postgresqlConnInfo);
} else if (PQstatus(handle->conn) != CONNECTION_OK) {
logmsg(LOG_DEBUG, "Resetting connection to database");
PQreset(handle->conn);
}
if (PQstatus(handle->conn) != CONNECTION_OK) {
logmsg(LOG_ERR, "Connection to database failed: %s", PQerrorMessage(handle->conn));
res = -1;
}
return res;
}
int sendToDB(t_dbHandle *handle, const uint32_t experimentId, const uint32_t value,
const uint64_t timestamp) {
int retcode = 0;
if (0 == openDatabase(handle)) {
char stmt[256];
int res1 = snprintf(stmt, sizeof(stmt),
"INSERT INTO mainsfrequency (time, host, location, valid, freq) "
"VALUES(to_timestamp('%lu'), '%d', '%d')",
timestamp, experimentId, value);
if (res1 > sizeof(stmt)) {
logmsg(LOG_ERR, "stmt buffer to small");
retcode = -1;
} else {
logmsg(LOG_DEBUG, "Statement: %s", stmt);
PGresult *res2 = PQexec(handle->conn, stmt);
if (PQresultStatus(res2) != PGRES_COMMAND_OK) {
logmsg(LOG_ERR, "Failed to insert into database (%s), data lost",
PQresultErrorMessage(res2));
retcode = -2;
}
PQclear(res2);
}
} else {
logmsg(LOG_ERR, "No database connection available, data lost");
retcode = -1;
}
return retcode;
}

24
dbhandler.h Normal file
View File

@ -0,0 +1,24 @@
#ifndef _DBHANDLER_H_
#define _DBHANDLER_H_
#include <config.h>
#include <libpq-fe.h>
typedef struct {
t_configHandle *configHandle;
const char *postgresqlConnInfo;
PGconn *conn;
} t_dbHandle;
t_dbHandle *initDatabaseHandler(t_configHandle *configHandle);
void deinitDatabaseHandler(t_dbHandle *handle);
int sendToDB(t_dbHandle *handle, const uint32_t experimentId, const uint32_t value,
const uint64_t timestamp);
#endif // __DBHANDLER_H_

369
iiotfeeder.c
View File

@ -9,332 +9,22 @@
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <unistd.h>
#include <errno.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <string.h>
#include <sys/types.h>
#include <unistd.h>
#include <getopt.h>
#include <pwd.h>
#include <libconfig.h>
#include <libpq-fe.h>
#include <sinkStruct.h>
#include <logging.h>
#include <sha256.h>
#include <config.h>
#include <dbhandler.h>
const char DEFAULT_CONFIG_FILENAME[] = "./sink20169.cfg";
extern bool verbose;
extern bool debug;
const char DEFAULT_CONFIG_FILENAME[] = "./iiotfeeder.cfg";
typedef struct {
const char *deviceId;
const char *location;
const char *sharedSecret;
int inactive;
PGresult *deviceResult;
} t_device;
typedef struct {
config_t cfg;
uint16_t numOfDevices;
t_device *devices;
} t_configHandle;
#define NUM_OF_STMT_PARAMS 4
typedef struct {
t_configHandle *configHandle;
int receiveSockFd;
int32_t lowerBound;
int32_t upperBound;
const char *postgresqlConnInfo;
PGconn *conn;
t_device foundDevice;
} t_commonHandle;
bool verbose = false;
bool debug = false;
int openDatabaseConnection(t_commonHandle *handle) {
int res = 0;
if (! handle->conn) {
logmsg(LOG_DEBUG, "Opening connection to database");
handle->conn = PQconnectdb(handle->postgresqlConnInfo);
} else if (PQstatus(handle->conn) != CONNECTION_OK) {
logmsg(LOG_DEBUG, "Resetting connection to database");
PQreset(handle->conn);
}
if (PQstatus(handle->conn) != CONNECTION_OK) {
logmsg(LOG_ERR, "Connection to database failed: %s", PQerrorMessage(handle->conn));
res = -1;
}
return res;
}
int initConfig(const char *configFilename, t_configHandle *configHandle) {
config_init(&(configHandle->cfg));
if (! config_read_file(&(configHandle->cfg), configFilename)) {
logmsg(LOG_ERR, "failed to read config file: %s:%d - %s\n",
config_error_file(&(configHandle->cfg)), config_error_line(&(configHandle->cfg)),
config_error_text(&(configHandle->cfg)));
config_destroy(&(configHandle->cfg));
return -1;
}
return 0;
}
void deinitConfig(t_configHandle *configHandle) {
config_destroy(&(configHandle->cfg));
}
// When you got a result here, remember to free it using freeDevice
int findDevice(t_commonHandle *handle, char *deviceId) {
int retCode = 0;
// we already have found it
if (handle->foundDevice.deviceResult) {
return 0;
}
if (0 == openDatabaseConnection(handle)) {
char stmt[256];
int res1 = snprintf(stmt, sizeof(stmt),
"SELECT sharedsecret, active, location "
" FROM device_t "
" WHERE deviceid = '%s'",
deviceId);
if (res1 > sizeof(stmt)) {
logmsg(LOG_ERR, "stmt buffer to small");
retCode = -1;
} else {
logmsg(LOG_DEBUG, "Statement: %s", stmt);
PGresult *res2 = PQexec(handle->conn, stmt);
ExecStatusType execStatus = PQresultStatus(res2);
if (execStatus != PGRES_TUPLES_OK) {
logmsg(LOG_ERR, "findDevice query fails, database returns %s", PQresStatus(execStatus));
retCode = -2;
} else {
int ntuples = PQntuples(res2);
if (ntuples == 1) {
handle->foundDevice.deviceResult = res2;
handle->foundDevice.sharedSecret = PQgetvalue(res2, 0, 0);
handle->foundDevice.inactive = (strcmp(PQgetvalue(res2, 0, 1), "f") == 0);
handle->foundDevice.location = PQgetvalue(res2, 0, 2);
logmsg(LOG_DEBUG, "found sharedsecret is %s, inactive is %d, location is %s",
handle->foundDevice.sharedSecret, handle->foundDevice.inactive,
handle->foundDevice.location);
} else {
logmsg(LOG_ERR, "no device found");
PQclear(res2);
retCode = -3;
}
}
}
} else {
logmsg(LOG_ERR, "No database connection available, data lost");
retCode = -4;
}
return retCode;
}
void freeDevice(t_commonHandle *handle) {
if (handle->foundDevice.deviceResult) {
PQclear(handle->foundDevice.deviceResult);
handle->foundDevice.deviceResult = NULL;
handle->foundDevice.deviceId = NULL;
handle->foundDevice.sharedSecret = NULL;
handle->foundDevice.location = NULL;
logmsg(LOG_DEBUG, "device has been free");
}
}
int initReceiver(t_configHandle *configHandle, t_commonHandle *handle) {
handle->configHandle = configHandle;
struct sockaddr_in servaddr;
handle->receiveSockFd = socket(AF_INET, SOCK_DGRAM, 0);
if (handle->receiveSockFd == -1) {
logmsg(LOG_ERR, "failed to create receive socket: %d", errno);
return -1;
}
int receivePort = 20169;
config_lookup_int(&(configHandle->cfg), "receivePort", &receivePort);
if (receivePort < 1 || receivePort > 65535) {
logmsg(LOG_ERR, "illegal receive port configured");
return -2;
}
memset(&servaddr, 0, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(receivePort);
if (-1 == bind(handle->receiveSockFd, (const struct sockaddr *) &servaddr, sizeof(servaddr))) {
logmsg(LOG_ERR, "unable to bind receive: %d", errno);
return -3;
}
return 0;
}
void deinitReceiver(t_commonHandle *handle) {
close(handle->receiveSockFd);
}
int receiveAndVerifyMinuteBuffer(t_commonHandle *handle, t_minuteBuffer *buf) {
struct sockaddr_in cliaddr;
socklen_t cliaddrlen = sizeof(cliaddr);
int n = recvfrom(handle->receiveSockFd, buf->b, sizeof(buf->b), MSG_TRUNC,
(struct sockaddr *) &cliaddr, &cliaddrlen);
logmsg(LOG_INFO, "received %d octets from %d.%d.%d.%d",
n,
(cliaddr.sin_addr.s_addr & 0x0ff),
((cliaddr.sin_addr.s_addr >> 8) & 0x0ff),
((cliaddr.sin_addr.s_addr >> 16) & 0x0ff),
((cliaddr.sin_addr.s_addr >> 24) & 0x0ff));
if (n != sizeof(buf->b)) {
logmsg(LOG_INFO, "Illegal packet size: %d", n);
return -1;
}
if (0 != findDevice(handle, buf->s.deviceId)) {
logmsg(LOG_ERR, "Device %s not found", buf->s.deviceId);
return -4;
}
const char *sharedSecret = handle->foundDevice.sharedSecret;
uint8_t receivedHash[SHA256_BLOCK_SIZE];
memcpy(receivedHash, buf->s.hash, SHA256_BLOCK_SIZE);
memcpy(buf->s.hash, sharedSecret, SHA256_BLOCK_SIZE);
SHA256_CTX ctx;
uint8_t calculatedHash[SHA256_BLOCK_SIZE];
sha256_init(&ctx);
sha256_update(&ctx, buf->b, sizeof(buf->b));
sha256_final(&ctx, calculatedHash);
if (memcmp(receivedHash, calculatedHash, SHA256_BLOCK_SIZE) != 0) {
logmsg(LOG_INFO, "Invalid hash in msg for device %s", buf->s.deviceId);
return -5;
}
return 0;
}
int initForwarder(t_configHandle *configHandle, t_commonHandle *handle) {
handle->configHandle = configHandle;
handle->postgresqlConnInfo = NULL;
config_lookup_string(&(configHandle->cfg), "postgresqlConnInfo", &(handle->postgresqlConnInfo));
if (! handle->postgresqlConnInfo) {
logmsg(LOG_ERR, "no postgresql connInfo configured");
return -1;
}
handle->conn = NULL;
handle->lowerBound = 45000;
config_lookup_int(&(configHandle->cfg), "lowerBound", &(handle->lowerBound));
handle->upperBound = 55000;
config_lookup_int(&(configHandle->cfg), "upperBound", &(handle->upperBound));
logmsg(LOG_INFO, "lowerBound: %u, upperBound: %u", handle->lowerBound, handle->upperBound);
return 0;
}
void deinitForwarder(t_commonHandle *handle) {
PQfinish(handle->conn);
}
int sendToDB(t_commonHandle *handle, const char *location, const char *deviceId,
uint32_t frequency, uint8_t valid, uint64_t timestamp) {
int retcode = 0;
if (0 == openDatabaseConnection(handle)) {
int frequency_before_point = frequency / 1000;
int frequency_behind_point = frequency - (frequency_before_point * 1000);
char stmt[256];
int res1 = snprintf(stmt, sizeof(stmt),
"INSERT INTO mainsfrequency (time, host, location, valid, freq) "
"VALUES(to_timestamp("
#ifdef OpenBSD
"%llu"
#else
"%lu"
#endif
"), '%s', '%s', %d, %d.%03d)",
timestamp, deviceId, location, valid,
frequency_before_point, frequency_behind_point);
if (res1 > sizeof(stmt)) {
logmsg(LOG_ERR, "stmt buffer to small");
retcode = -1;
} else {
logmsg(LOG_DEBUG, "Statement: %s", stmt);
PGresult *res2 = PQexec(handle->conn, stmt);
if (PQresultStatus(res2) != PGRES_COMMAND_OK) {
logmsg(LOG_ERR, "Failed to insert into database (%s), data lost",
PQresultErrorMessage(res2));
retcode = -2;
}
PQclear(res2);
}
} else {
logmsg(LOG_ERR, "No database connection available, data lost");
retcode = -1;
}
return retcode;
}
int forwardMinuteBuffer(t_commonHandle *handle, t_minuteBuffer *buf) {
if (0 != findDevice(handle, buf->s.deviceId)) {
logmsg(LOG_ERR, "Device %s not found", buf->s.deviceId);
return -4;
}
t_device *device = &(handle->foundDevice);
const char *location = device->location;
logmsg(LOG_INFO, "D: %s, R: %u, P: %u, W: %u, V: %08x, L: %s",
buf->s.deviceId, buf->s.totalRunningHours, buf->s.totalPowercycles, buf->s.totalWatchdogResets,
buf->s.version, location);
int sendSuccess = 0;
for (uint8_t j = 0; j < SECONDS_PER_MINUTE; j++) {
uint64_t timestamp = buf->s.timestamp + j;
logmsg(LOG_DEBUG, "Time: %lu, Frequency: %u", timestamp, buf->s.frequency[j]);
if (device->inactive == 0) {
uint8_t valid = ((buf->s.frequency[j] >= handle->lowerBound) && (buf->s.frequency[j] <= handle->upperBound)) ? 1 : 0;
if (valid == 0) {
logmsg(LOG_INFO, "Out of range: Time: %lu, Frequency: %u", timestamp, buf->s.frequency[j]);
}
sendSuccess += sendToDB(handle, location, buf->s.deviceId, buf->s.frequency[j], valid, timestamp);
} else {
logmsg(LOG_DEBUG, "Inactive device, not sent to database");
}
}
if (device->inactive == 0) {
if (sendSuccess == 0) {
logmsg(LOG_INFO, "Successfully sent whole minute to database");
} else {
logmsg(LOG_INFO, "Errors when sending to database, see above");
}
} else {
logmsg(LOG_INFO, "Not sent to database, device is marked as inactive");
}
return 0;
}
void usage() {
printf("sinkserver for mainsfrequency counter implementations\n");
@ -357,12 +47,8 @@ void usage() {
}
int main(int argc, char **argv) {
t_configHandle configHandle;
t_commonHandle commonHandle;
commonHandle.foundDevice.deviceResult = NULL;
const char *configFilename = DEFAULT_CONFIG_FILENAME;
const char *dropPrivilegesToUser = NULL;
bool doFork = false;
@ -410,11 +96,6 @@ int main(int argc, char **argv) {
logmsg(LOG_INFO, "Version: " VERSION);
if (0 != initConfig(configFilename, &configHandle)) {
logmsg(LOG_ERR, "error when reading configuration");
exit(3);
}
if (doFork) {
int pid = fork();
if (pid == -1) {
@ -427,36 +108,22 @@ int main(int argc, char **argv) {
}
}
if (0 != initReceiver(&configHandle, &commonHandle)) {
logmsg(LOG_ERR, "error when initializing receiver");
t_configHandle configHandle;
if (0 != initConfig(configFilename, &configHandle)) {
logmsg(LOG_ERR, "error when reading configuration");
exit(3);
}
t_dbHandle *dbHandle = initDatabaseHandler(&configHandle);
if (! dbHandle) {
logmsg(LOG_ERR, "error when trying to initialize database connection");
exit(5);
}
if (0 != initForwarder(&configHandle, &commonHandle)) {
logmsg(LOG_ERR, "error when initializing forwarder");
exit(6);
}
while (1) {
t_minuteBuffer buf;
// this is relevant AFTER one or both of the following calls,
// is has an effect first in the second cycle through the loop
freeDevice(&commonHandle);
if (receiveAndVerifyMinuteBuffer(&commonHandle, &buf) < 0) {
logmsg(LOG_ERR, "error in receiveAndVerify");
continue;
}
if (forwardMinuteBuffer(&commonHandle, &buf) < 0) {
logmsg(LOG_ERR, "error in send");
}
}
deinitForwarder(&commonHandle);
deinitReceiver(&commonHandle);
deinitDatabaseHandler(dbHandle);
deinitConfig(&configHandle);
}

2
logging.c
View File

@ -48,7 +48,7 @@ void logmsg(int prio, const char* format, ...) {
printf("%s\n", buf);
}
openlog("counter", 0, facility);
openlog("iiotfeeder", 0, facility);
syslog(prio, "%s", buf);
closelog();
}

47
queue.c Normal file
View File

@ -0,0 +1,47 @@
#include <queue.h>
#include <stdlib.h>
#include <logging.h>
#include <pthread.h>
t_queue *initQueue() {
t_queue *queue = (t_queue*) malloc(sizeof(t_queue));
if (! queue) {
logmsg(LOG_ERR, "out of memory when initializing queue");
return NULL;
}
queue->root = (t_queueBlock*) malloc(sizeof(t_queueBlock));
if (! queue->root) {
logmsg(LOG_ERR, "out of memory when initializing root of queue");
free(queue);
queue = NULL;
return NULL;
}
queue->head = queue->root;
queue->head_lock = PTHREAD_MUTEX_INITIALIZER;
queue->tail = queue->root;
queue->tail_lock = PTHREAD_MUTEX_INITIALIZER;
queue->notEmptyLock = PTHREAD_MUTEX_INITIALIZER;
queue->notEmpty = PTHREAD_COND_INITIALIZER;
}
// allocate memory for the queueBlock on your own and fill in the payloadc
void putQueue(t_queue *queue, t_queueBlock *queueBlock);
// will block if queue is empty
// remember to free the queueBlock afterwards
t_queueBlock *getQueue(t_queue *queue) {
pthread_mutex_lock(&queue->notEmptyLock);
while (queue->head == queue->tail) {
pthread_cond_wait(&queue->notEmpty, &queue->notEmptyLock);
}
pthread_mutex_unlock(&queue->notEmpty);
t_queueBlock item = queue->head;
}

32
queue.h Normal file
View File

@ -0,0 +1,32 @@
#include <stdint.h>
#include <pthread.h>
typedef struct s_queueBlock {
void *payload;
struct s_queueBlock *next;
} t_queueBlock;
typedef struct {
t_queueBlock *root;
t_queueBlock *head;
pthread_mutex_t head_lock;
t_queueBlock *tail;
pthread_mutex_t tail_lock;
pthread_mutex_t notEmptyLock;
pthread_cond_t notEmpty;
} t_queue;
t_queue *initQueue();
// allocate memory for the queueBlock on your own and fill in the payloadc
void putQueue(t_queue *queue, t_queueBlock *queueBlock);
// will block if queue is empty
// remember to free the queueBlock afterwards
t_queueBlock *getQueue(t_queue *queue);