From 1ba21da6cbc63c0c549fb92731e25bedc482eb51 Mon Sep 17 00:00:00 2001 From: Filip Wandzio Date: Thu, 4 Sep 2025 22:25:39 +0200 Subject: Unify the directory, add new analysis methods, unify the code style Signed-off-by: Filip Wandzio --- analysis/rtt/src/logger.c | 86 +++++++++++++++++++++++++++ analysis/rtt/src/logger.h | 9 +++ analysis/rtt/src/main.c | 28 +++++++++ analysis/rtt/src/mqtt_client.c | 48 +++++++++++++++ analysis/rtt/src/mqtt_client.h | 16 +++++ analysis/rtt/src/mqtt_rtt_logger.c | 117 +++++++++++++++++++++++++++++++++++++ 6 files changed, 304 insertions(+) create mode 100644 analysis/rtt/src/logger.c create mode 100644 analysis/rtt/src/logger.h create mode 100644 analysis/rtt/src/main.c create mode 100644 analysis/rtt/src/mqtt_client.c create mode 100644 analysis/rtt/src/mqtt_client.h create mode 100644 analysis/rtt/src/mqtt_rtt_logger.c (limited to 'analysis/rtt/src') diff --git a/analysis/rtt/src/logger.c b/analysis/rtt/src/logger.c new file mode 100644 index 0000000..ed5f6e5 --- /dev/null +++ b/analysis/rtt/src/logger.c @@ -0,0 +1,86 @@ + +#include "logger.h" +#include +#include +#include +#include +#include + +#define WINDOW_SEC 1 +#define MAX_TIMES 1000 + +static FILE *csv_file = NULL; +static double msg_times[MAX_TIMES]; +static int msg_count = 0; + +static long long current_time_ms() { + struct timeval tv; + gettimeofday(&tv, NULL); + return ((long long)tv.tv_sec * 1000) + (tv.tv_usec / 1000); +} + +static double current_time_sec() { + struct timeval tv; + gettimeofday(&tv, NULL); + return (double)tv.tv_sec + (tv.tv_usec / 1e6); +} + +static int update_throughput_window(double now) { + int i, new_count = 0; + for (i = 0; i < msg_count; i++) { + if (now - msg_times[i] < WINDOW_SEC) { + msg_times[new_count++] = msg_times[i]; + } + } + msg_times[new_count++] = now; + msg_count = new_count; + return msg_count; +} + +void logger_init(const char *filename) { + csv_file = fopen(filename, "w"); + if (!csv_file) { + perror("Cannot open CSV file"); + exit(EXIT_FAILURE); + } + fprintf(csv_file, + "timestamp,sent_ms,received_ms,rtt_ms,throughput_msg_per_s\n"); + fflush(csv_file); +} + +void logger_cleanup() { + if (csv_file) { + fclose(csv_file); + csv_file = NULL; + } +} + +void logger_handle_message(const void *payload, int payloadlen) { + char *payload_str = malloc(payloadlen + 1); + if (!payload_str) + return; + memcpy(payload_str, payload, payloadlen); + payload_str[payloadlen] = '\0'; + + long long sent_ms = atoll(payload_str); + free(payload_str); + + long long received_ms = current_time_ms(); + long long rtt = received_ms - sent_ms; + + double now_sec = current_time_sec(); + int throughput = update_throughput_window(now_sec); + + time_t now = time(NULL); + struct tm *tm_info = localtime(&now); + char iso_time[32]; + strftime(iso_time, sizeof(iso_time), "%Y-%m-%dT%H:%M:%S", tm_info); + + if (csv_file) { + fprintf(csv_file, "%s,%lld,%lld,%lld,%d\n", iso_time, sent_ms, received_ms, + rtt, throughput); + fflush(csv_file); + } + + printf("RTT: %lld ms | Throughput: %d msg/s\n", rtt, throughput); +} diff --git a/analysis/rtt/src/logger.h b/analysis/rtt/src/logger.h new file mode 100644 index 0000000..f1602c9 --- /dev/null +++ b/analysis/rtt/src/logger.h @@ -0,0 +1,9 @@ + +#ifndef LOGGER_H +#define LOGGER_H + +void logger_init(const char *filename); +void logger_cleanup(); +void logger_handle_message(const void *payload, int payloadlen); + +#endif // LOGGER_H diff --git a/analysis/rtt/src/main.c b/analysis/rtt/src/main.c new file mode 100644 index 0000000..13f7716 --- /dev/null +++ b/analysis/rtt/src/main.c @@ -0,0 +1,28 @@ + +#include "logger.h" +#include "mqtt_client.h" +#include + +#define BROKER_ADDRESS "192.168.1.101" +#define BROKER_PORT 1883 +#define TOPIC "device/echo/in" +#define CSV_FILE "rtt_throughput_log.csv" + +int main() { + mqtt_client_t client; + + logger_init(CSV_FILE); + + if (mqtt_client_init(&client, BROKER_ADDRESS, BROKER_PORT, TOPIC) != 0) { + fprintf(stderr, "MQTT initialization error\n"); + return 1; + } + + printf("Connection established.Waiting for messages...\n"); + mqtt_client_loop(&client); + + logger_cleanup(); + mqtt_client_cleanup(&client); + + return 0; +} diff --git a/analysis/rtt/src/mqtt_client.c b/analysis/rtt/src/mqtt_client.c new file mode 100644 index 0000000..43a06d8 --- /dev/null +++ b/analysis/rtt/src/mqtt_client.c @@ -0,0 +1,48 @@ +#include "mqtt_client.h" +#include "logger.h" +#include +#include + +#define WINDOW_SEC 1 + +static void on_message(struct mosquitto *mosq, void *userdata, + const struct mosquitto_message *msg); + +int mqtt_client_init(mqtt_client_t *client, const char *broker_address, + int port, const char *topic) { + mosquitto_lib_init(); + client->mosq = mosquitto_new(NULL, true, NULL); + if (!client->mosq) { + fprintf(stderr, "Cannot create MQTT client\n"); + return 1; + } + + mosquitto_message_callback_set(client->mosq, on_message); + + if (mosquitto_connect(client->mosq, broker_address, port, 60) != + MOSQ_ERR_SUCCESS) { + fprintf(stderr, "Nie można połączyć się z brokerem MQTT\n"); + return 1; + } + + if (mosquitto_subscribe(client->mosq, NULL, topic, 0) != MOSQ_ERR_SUCCESS) { + fprintf(stderr, "Cannot subscribe to topic\n"); + return 1; + } + + return 0; +} + +void mqtt_client_cleanup(mqtt_client_t *client) { + mosquitto_destroy(client->mosq); + mosquitto_lib_cleanup(); +} + +void mqtt_client_loop(mqtt_client_t *client) { + mosquitto_loop_forever(client->mosq, -1, 1); +} + +static void on_message(struct mosquitto *mosq, void *userdata, + const struct mosquitto_message *msg) { + logger_handle_message(msg->payload, msg->payloadlen); +} diff --git a/analysis/rtt/src/mqtt_client.h b/analysis/rtt/src/mqtt_client.h new file mode 100644 index 0000000..781e742 --- /dev/null +++ b/analysis/rtt/src/mqtt_client.h @@ -0,0 +1,16 @@ + +#ifndef MQTT_CLIENT_H +#define MQTT_CLIENT_H + +#include + +typedef struct { + struct mosquitto *mosq; +} mqtt_client_t; + +int mqtt_client_init(mqtt_client_t *client, const char *broker_address, + int port, const char *topic); +void mqtt_client_cleanup(mqtt_client_t *client); +void mqtt_client_loop(mqtt_client_t *client); + +#endif // MQTT_CLIENT_H diff --git a/analysis/rtt/src/mqtt_rtt_logger.c b/analysis/rtt/src/mqtt_rtt_logger.c new file mode 100644 index 0000000..60e5d2d --- /dev/null +++ b/analysis/rtt/src/mqtt_rtt_logger.c @@ -0,0 +1,117 @@ +// #include +// #include +// #include +// #include +// #include +// #include +// #include +// +// #define BROKER_ADDRESS "192.168.1.101" +// #define BROKER_PORT 1883 +// #define TOPIC "device/echo/in" +// #define CSV_FILE "rtt_throughput_log.csv" +// #define WINDOW_SEC 1 +// #define MAX_TIMES 1000 +// +// // Kolejka czasów przyjścia wiadomości (dla throughput) +// double msg_times[MAX_TIMES]; +// int msg_count = 0; +// +// // Zwraca aktualny czas w milisekundach +// long long current_time_ms() { +// struct timeval tv; +// gettimeofday(&tv, NULL); +// return ((long long)tv.tv_sec * 1000) + (tv.tv_usec / 1000); +// } +// +// // Zwraca aktualny czas w sekundach z ułamkiem +// double current_time_sec() { +// struct timeval tv; +// gettimeofday(&tv, NULL); +// return (double)tv.tv_sec + (tv.tv_usec / 1e6); +// } +// +// // Aktualizuje throughput window i zwraca liczbę wiadomości w oknie +// int update_throughput_window(double now) { +// int i, new_count = 0; +// for (i = 0; i < msg_count; i++) { +// if (now - msg_times[i] < WINDOW_SEC) { +// msg_times[new_count++] = msg_times[i]; +// } +// } +// msg_times[new_count++] = now; +// msg_count = new_count; +// return msg_count; +// } +// +// // Callback po odebraniu wiadomości +// void on_message(struct mosquitto *mosq, void *userdata, +// const struct mosquitto_message *msg) { +// char *payload = (char *)msg->payload; +// long long sent_ms = atoll(payload); +// long long received_ms = current_time_ms(); +// long long rtt = received_ms - sent_ms; +// +// double now_sec = current_time_sec(); +// int throughput = update_throughput_window(now_sec); +// +// // Timestamp ISO 8601 +// time_t now = time(NULL); +// struct tm *tm_info = localtime(&now); +// char iso_time[32]; +// strftime(iso_time, sizeof(iso_time), "%Y-%m-%dT%H:%M:%S", tm_info); +// +// // Zapisz do CSV +// FILE *f = fopen(CSV_FILE, "a"); +// if (f) { +// fprintf(f, "%s,%lld,%lld,%lld,%d\n", iso_time, sent_ms, received_ms, rtt, +// throughput); +// fclose(f); +// } +// +// // Wydruk +// printf("RTT: %lld ms | Throughput: %d msg/s\n", rtt, throughput); +// } +// +// int main() { +// printf("Start programu\n"); +// +// FILE *f = fopen(CSV_FILE, "w"); +// if (!f) { +// perror("Nie można otworzyć pliku CSV"); +// return 1; +// } +// fprintf(f, "timestamp,sent_ms,received_ms,rtt_ms,throughput_msg_per_s\n"); +// fclose(f); +// printf("Plik CSV utworzony\n"); +// +// mosquitto_lib_init(); +// struct mosquitto *mosq = mosquitto_new(NULL, true, NULL); +// if (!mosq) { +// fprintf(stderr, "Błąd tworzenia klienta MQTT\n"); +// return 1; +// } +// printf("Klient MQTT utworzony\n"); +// +// mosquitto_message_callback_set(mosq, on_message); +// +// int rc = mosquitto_connect(mosq, BROKER_ADDRESS, BROKER_PORT, 60); +// if (rc != MOSQ_ERR_SUCCESS) { +// fprintf(stderr, "Nie można połączyć się z brokerem MQTT, kod błędu: +// %d\n", +// rc); +// return 1; +// } +// printf("Połączono z brokerem MQTT\n"); +// +// mosquitto_subscribe(mosq, NULL, TOPIC, 0); +// printf("Subskrypcja tematu %s ustawiona\n", TOPIC); +// +// printf("Oczekiwanie na wiadomości...\n"); +// mosquitto_loop_forever(mosq, -1, 1); +// +// mosquitto_destroy(mosq); +// mosquitto_lib_cleanup(); +// +// return 0; +// } -- cgit v1.2.3