data client (WIP)

2 files changed, 171 insertions, 99 deletions

diff --git a/data/client.cpp b/data/client.cpp
index f0ffa6a..65494b0 100644
--- a/data/client.cpp
+++ b/data/client.cpp
@@ -1,16 +1,22 @@
+#include <arpa/inet.h>
 #include <GLFW/glfw3.h>
 #include <imgui.h>
 #include <imgui_impl_glfw.h>
 #include <imgui_impl_opengl3.h>
 #include <implot.h>
+#include <json-c/json.h>
 #include <limits.h>
 #include <math.h>
 #include <signal.h>
+#include <threads.h>
 
 #include <initializer_list>
 
 #include "logger.c"
 
+#define FETCH_INTERVAL 10
+#define FETCH_INTERVAL_SUBDIV 1000
+
 #define COLOR_BG ImVec4(0.f, 0.f, 0.f, 1.f)
 #define FONT_SIZE 12.f
 #define FONT_SOURCE "/usr/share/fonts/droid/DroidSansMono.ttf"
@@ -43,17 +49,13 @@ struct Plot {
     const char *const *cols;
 };
 
-// Globals
+// Window globals
 GLFWwindow *g_window;
 ImGuiIO *g_io;
 ImGuiStyle *g_imgui_style;
 ImPlotStyle *g_implot_style;
-int g_status;
-
-// Data col
-bool g_data_col_visible = true;
-float g_data_col_width;
 
+// Data globals
 const char *g_x_axes[] = {
     "rowid",
     "steps",
@@ -62,6 +64,7 @@ const char *g_x_axes[] = {
 #undef FOR_CORE
 };
 
+int g_status;
 int g_entries = DEFAULT_ENTRIES;
 int g_nth = DEFAULT_NTH;
 int g_x_axis = DEFAULT_X_AXIS;
@@ -70,9 +73,10 @@ int g_x_high = DEFAULT_X_HIGH;
 int g_hm_left = DEFAULT_HM_LEFT;
 int g_hm_pixel_count = DEFAULT_HM_PIXEL_COUNT;
 int g_hm_pixel_pow; // calculate on startup
-bool g_data_touched;
+int g_x_current;
+thrd_t g_data_fetching_thread;
 
-// Plots
+// Plot globals
 Plot g_plots[] = {
     Plot("cycl", "general", {
 #define FOR_CORE(i) "cycl_" #i,
@@ -114,7 +118,9 @@ Plot g_plots[] = {
 
 #define PLOT_COUNT (int)(sizeof(g_plots) / sizeof(g_plots[0]))
 
-// Layout
+// Layout globals
+float g_data_col_width;
+bool g_data_col_visible = true;
 int g_plot_cols = 2;
 int g_plot_col_selected;
 int g_plot_row_selected;
@@ -135,7 +141,7 @@ void data_clamp(int *field, int low, int high) {
     if (*field > high) *field = high;
 }
 
-void data_validate(void) {
+void data_touched(void) {
     data_clamp(&g_entries, 1, DEFAULT_ENTRIES);
     data_clamp(&g_nth, DEFAULT_NTH, INT_MAX);
     data_clamp(&g_x_low, DEFAULT_X_LOW, INT_MAX);
@@ -145,7 +151,7 @@ void data_validate(void) {
  #endif
     data_clamp(&g_hm_pixel_count, 1, DEFAULT_HM_PIXEL_COUNT);
     data_clamp(&g_hm_pixel_pow, 0, data_calc_max_hm_pixel_pow());
-    g_data_touched = false;
+    g_x_current = 0;
 }
 
 void data_reset_values(void) {
@@ -160,24 +166,75 @@ void data_reset_values(void) {
 }
 
 void data_reset_plot_cells(void) {
-    for (size_t i = 0; i < PLOT_MAX_COLS; i++) {
-        for (size_t j = 0; j < PLOT_COUNT; j++) {
+    for (int i = 0; i < PLOT_MAX_COLS; i++) {
+        for (int j = 0; j < PLOT_COUNT; j++) {
             g_plot_cells[i][j] = nullptr;
         }
     }
 }
 
+void data_fetch(void) {
+    struct json_object *request = json_object_new_object();
+    json_object_object_add(request, "request", json_object_new_string("data"));
+    json_object_object_add(request, "entries", json_object_new_int(g_entries));
+    json_object_object_add(request, "nth", json_object_new_int(g_nth));
+    json_object_object_add(request, "x-axis", json_object_new_string(g_x_axes[g_x_axis]));
+    json_object_object_add(request, "x-low", json_object_new_int(g_x_low));
+    json_object_object_add(request, "x-high", json_object_new_int(g_x_high));
+    json_object_object_add(request, "hm-left", json_object_new_int(g_hm_left));
+    json_object_object_add(request, "hm-pixel-count", json_object_new_int(g_hm_pixel_count));
+    json_object_object_add(request, "hm-pixel-pow", json_object_new_int(g_hm_pixel_pow));
+    json_object_object_add(request, "x-current", json_object_new_int(g_x_current));
+    const char *request_str = json_object_to_json_string(request);
+
+    log_info("Sending request to server: %s", request_str);
+    int socket_fd = socket(AF_INET, SOCK_STREAM, 0);
+    struct sockaddr_in socket_addr;
+    memset(&socket_addr, 0, sizeof(struct sockaddr_in));
+    socket_addr.sin_family = AF_INET;
+    socket_addr.sin_port = htons(PORT);
+    inet_pton(AF_INET, IP, &socket_addr.sin_addr);
+    if (connect(socket_fd, (struct sockaddr *)&socket_addr, sizeof(struct sockaddr_in))) assert(false);
+    json_object_to_fd(socket_fd, request, 0);
+    shutdown(socket_fd, SHUT_WR);
+
+    struct json_object *response = json_object_from_fd(socket_fd);
+    const char *response_str = json_object_to_json_string(response);
+    log_info("Server responded with: %s", response_str);
+
+    json_object_put(request);
+    json_object_put(response);
+}
+
+int data_fetching_thread(void *data) {
+    (void)data;
+
+    assert(g_status == STATUS_RUNNING || g_status == STATUS_STOPPING);
+
+    while (g_status == STATUS_RUNNING) {
+        data_fetch();
+
+        for (int i = 0; i < FETCH_INTERVAL_SUBDIV && g_status == STATUS_RUNNING; i++) {
+            usleep((FETCH_INTERVAL * 1000000) / FETCH_INTERVAL_SUBDIV);
+        }
+    }
+
+    assert(g_status == STATUS_STOPPING);
+    g_status = STATUS_STOPPED;
+    return 0;
+}
+
 void data_start_fetching(void) {
-    assert(g_status == STATUS_RUNNING);
     log_info("Starting data fetching thread");
-    // start data fetching thread
+    g_status = STATUS_RUNNING;
+    thrd_create(&g_data_fetching_thread, (thrd_start_t)data_fetching_thread, nullptr);
 }
 
 void data_stop_fetching(void) {
-    assert(g_status == STATUS_STOPPING);
+    assert(g_status == STATUS_RUNNING);
     log_info("Stopping data fetching thread");
-    // join data fetching thread (set STATUS_STOPPED from within thread)
-    g_status = STATUS_STOPPED;
+    g_status = STATUS_STOPPING;
+    thrd_join(g_data_fetching_thread, nullptr);
 }
 
 // ----------------------------------------------------------------------------
@@ -207,47 +264,30 @@ void gui_print_data_col(void) {
  #endif
     ImGui::LabelText("data-push", "%#lx", DATA_PUSH_INTERVAL);
 
-    switch (g_status) {
-    case STATUS_STOPPED:
-        ImGui::LabelText("status", "%s", "stopped");
-        break;
-    case STATUS_RUNNING:
-        ImGui::LabelText("status", "%s", "running");
-        break;
-    case STATUS_STOPPING:
-        ImGui::LabelText("status", "%s", "stopping");
-        break;
-    }
-
     ImGui::SeparatorText("Data fields");
 
     switch (g_status) {
     case STATUS_STOPPED:
-        if (ImGui::InputInt("entries", &g_entries, 0, 0, ImGuiInputTextFlags_CharsDecimal)) g_data_touched = true;
-        if (ImGui::InputInt("nth", &g_nth, 0, 0, ImGuiInputTextFlags_CharsDecimal)) g_data_touched = true;
+        if (ImGui::InputInt("entries", &g_entries, 0, 0, ImGuiInputTextFlags_CharsDecimal)) data_touched();
+        if (ImGui::InputInt("nth", &g_nth, 0, 0, ImGuiInputTextFlags_CharsDecimal)) data_touched();
 
         if (ImGui::BeginCombo("x-axis", g_x_axes[g_x_axis])) {
             for (int i = 0; i < CORES + 2; i++) {
                 if (ImGui::Selectable(g_x_axes[i], g_x_axis == i)) {
                     g_x_axis = i;
-                    g_data_touched = true;
+                    data_touched();
                 }
             }
 
             ImGui::EndCombo();
         }
 
-        if (ImGui::InputInt("x-low", &g_x_low, 0, 0, ImGuiInputTextFlags_CharsDecimal)) g_data_touched = true;
-        if (ImGui::InputInt("x-high", &g_x_high, 0, 0, ImGuiInputTextFlags_CharsDecimal)) g_data_touched = true;
-        if (ImGui::InputInt("hm-left", &g_hm_left, 0, 0, ImGuiInputTextFlags_CharsDecimal)) g_data_touched = true;
-        if (ImGui::InputInt("hm-pxl-count", &g_hm_pixel_count, 0, 0, ImGuiInputTextFlags_CharsDecimal)) g_data_touched = true;
-        if (ImGui::InputInt("hm-pxl-pow", &g_hm_pixel_pow, 0, 0, ImGuiInputTextFlags_CharsDecimal)) g_data_touched = true;
-
-        if (ImGui::Button("Run", ImVec2(-1.f, 0.f))) {
-            g_status = STATUS_RUNNING;
-            data_start_fetching();
-        }
-
+        if (ImGui::InputInt("x-low", &g_x_low, 0, 0, ImGuiInputTextFlags_CharsDecimal)) data_touched();
+        if (ImGui::InputInt("x-high", &g_x_high, 0, 0, ImGuiInputTextFlags_CharsDecimal)) data_touched();
+        if (ImGui::InputInt("hm-left", &g_hm_left, 0, 0, ImGuiInputTextFlags_CharsDecimal)) data_touched();
+        if (ImGui::InputInt("hm-pxl-count", &g_hm_pixel_count, 0, 0, ImGuiInputTextFlags_CharsDecimal)) data_touched();
+        if (ImGui::InputInt("hm-pxl-pow", &g_hm_pixel_pow, 0, 0, ImGuiInputTextFlags_CharsDecimal)) data_touched();
+        if (ImGui::Button("Run", ImVec2(-1.f, 0.f))) data_start_fetching();
         if (ImGui::Button("Reset", ImVec2(-1.f, 0.f))) data_reset_values();
 
         break;
@@ -263,10 +303,10 @@ void gui_print_data_col(void) {
         ImGui::LabelText("hm-pxl-pow", "%d", g_hm_pixel_pow);
 
         if (g_status == STATUS_RUNNING) {
-            if (ImGui::Button("Stop", ImVec2(-1.f, 0.f))) {
-                g_status = STATUS_STOPPING;
-                data_stop_fetching();
-            }
+            if (ImGui::Button("Stop", ImVec2(-1.f, 0.f))) data_stop_fetching();
+            ImGui::LabelText("##", "Running");
+        } else {
+            ImGui::LabelText("##", "Stopping");
         }
     }
 
@@ -311,11 +351,11 @@ void gui_print_plots(void) {
             }
 
             if (ImPlot::BeginPlot(g_plots[i].name)) {
-                int test_x[] = {0,1,2,3};
-                int test_y1[] = {1,2,3,4};
-                int test_y2[] = {2,4,8,16};
-                ImPlot::PlotLine("test1", test_x, test_y1, 4);
-                ImPlot::PlotLine("test2", test_x, test_y2, 4);
+                //int test_x[] = {0,1,2,3};
+                //int test_y1[] = {1,2,3,4};
+                //int test_y2[] = {2,4,8,16};
+                //ImPlot::PlotLine("test1", test_x, test_y1, 4);
+                //ImPlot::PlotLine("test2", test_x, test_y2, 4);
                 ImPlot::EndPlot();
             }
 
@@ -346,11 +386,11 @@ void gui_print_plot_maximized(void) {
     ImGui::Begin("plot-fullscreen", nullptr, WINDOW_STYLE);
 
     if (ImPlot::BeginPlot(g_plot_selected->name, viewport->Size)) {
-        int test_x[] = {0,1,2,3};
-        int test_y1[] = {1,2,3,4};
-        int test_y2[] = {2,4,8,16};
-        ImPlot::PlotLine("test1", test_x, test_y1, 4);
-        ImPlot::PlotLine("test2", test_x, test_y2, 4);
+        //int test_x[] = {0,1,2,3};
+        //int test_y1[] = {1,2,3,4};
+        //int test_y2[] = {2,4,8,16};
+        //ImPlot::PlotLine("test1", test_x, test_y1, 4);
+        //ImPlot::PlotLine("test2", test_x, test_y2, 4);
         ImPlot::EndPlot();
     }
 
@@ -373,6 +413,7 @@ void sig_handler(int signo) {
     (void)signo;
 
     log_warn("Signal received, will stop SALIS data client...");
+    if (g_status == STATUS_RUNNING) data_stop_fetching();
     glfwSetWindowShouldClose(g_window, GLFW_TRUE);
 }
 
@@ -513,10 +554,6 @@ int main(int argc, char **argv) {
 
         gui_print();
 
-        if (g_data_touched) {
-            data_validate();
-        }
-
         ImGui::Render();
         int display_w, display_h;
         glfwGetFramebufferSize(g_window, &display_w, &display_h);
diff --git a/data/server.c b/data/server.c
index 2d96b81..ea94364 100644
--- a/data/server.c
+++ b/data/server.c
@@ -1,6 +1,7 @@
 #include <arpa/inet.h>
 #include <json-c/json.h>
 #include <signal.h>
+#include <sqlite3.h>
 #include <string.h>
 #include <threads.h>
 
@@ -8,34 +9,46 @@
 
 #define BACKLOG 10
 
-struct Client {
+struct Socket {
     int fd;
     struct sockaddr_in addr;
 };
 
+sqlite3 *g_sim_db;
+void (*g_info)(const char *fmt, ...) = log_info;
+void (*g_warn)(const char *fmt, ...) = log_warn;
+#include "sql.c"
+
+// ----------------------------------------------------------------------------
+// SQL functions
+// ----------------------------------------------------------------------------
+
+// ----------------------------------------------------------------------------
+// Main functions
+// ----------------------------------------------------------------------------
 void sig_handler(int signo) {
     log_warn("Signal %d received, shutting down data server", signo);
     exit(0);
 }
 
-void respond_name(int client_fd) {
+void respond_name(int socket_fd) {
     log_info("Client requested simulation name");
 
     struct json_object *sim_name = json_object_new_object();
     json_object_object_add(sim_name, "name", json_object_new_string(NAME));
-    json_object_to_fd(client_fd, sim_name, JSON_C_TO_STRING_PRETTY);
+    json_object_to_fd(socket_fd, sim_name, 0);
     json_object_put(sim_name);
 }
 
-void respond_opts(int client_fd) {
+void respond_opts(int socket_fd) {
     log_info("Client requested simulation options");
 
     struct json_object *sim_opts = json_object_from_file(SIM_OPTS);
-    json_object_to_fd(client_fd, sim_opts, JSON_C_TO_STRING_PRETTY);
+    json_object_to_fd(socket_fd, sim_opts, 0);
     json_object_put(sim_opts);
 }
 
-void respond_hash(int client_fd) {
+void respond_hash(int socket_fd) {
     log_info("Client requested git hash");
 
     char buff[41] = { 0 };
@@ -45,23 +58,34 @@ void respond_hash(int client_fd) {
 
     struct json_object *git_hash = json_object_new_object();
     json_object_object_add(git_hash, "hash", json_object_new_string(buff));
-    json_object_to_fd(client_fd, git_hash, JSON_C_TO_STRING_PRETTY);
+    json_object_to_fd(socket_fd, git_hash, 0);
     json_object_put(git_hash);
 }
 
-void respond_data(int client_fd, struct json_object *request_json) {
-    assert(request_json);
-    (void)client_fd;
+void respond_data(int socket_fd, struct json_object *request) {
+    assert(request);
+
+    const char *request_str = json_object_to_json_string(request);
+    log_info("Client requested simulation data with the following parameters: %s", request_str);
+
+    struct json_object *response = json_object_new_object();
+    json_object_object_add(response, "response", json_object_new_string("hello!"));
+    const char *response_str = json_object_to_json_string(response);
+    log_info("Sending response to client: %s", response_str);
+    json_object_to_fd(socket_fd, response, 0);
+
+    shutdown(socket_fd, SHUT_WR);
+    json_object_put(response);
 }
 
-int handle_client(struct Client *client) {
-    assert(client);
+int handle_client(struct Socket *socket) {
+    assert(socket);
 
-    char client_ip[INET_ADDRSTRLEN];
-    inet_ntop(AF_INET, &client->addr.sin_addr, client_ip, INET_ADDRSTRLEN);
-    log_info("Client connected: %s:%d", client_ip, ntohs(client->addr.sin_port));
+    char socket_ip[INET_ADDRSTRLEN];
+    inet_ntop(AF_INET, &socket->addr.sin_addr, socket_ip, INET_ADDRSTRLEN);
+    log_info("Client connected: %s:%d", socket_ip, ntohs(socket->addr.sin_port));
 
-    struct json_object *request_json = json_object_from_fd(client->fd);
+    struct json_object *request_json = json_object_from_fd(socket->fd);
     struct json_object *request_str = NULL;
 
     if (!json_object_object_get_ex(request_json, "request", &request_str)) assert(false);
@@ -70,23 +94,23 @@ int handle_client(struct Client *client) {
     assert(request);
 
     if (!strcmp(request, "name")) {
-        respond_name(client->fd);
+        respond_name(socket->fd);
     } else if (!strcmp(request, "opts")) {
-        respond_opts(client->fd);
+        respond_opts(socket->fd);
     } else if (!strcmp(request, "hash")) {
-        respond_hash(client->fd);
+        respond_hash(socket->fd);
     } else if (!strcmp(request, "data")) {
-        respond_data(client->fd, request_json);
+        respond_data(socket->fd, request_json);
     } else {
         assert(false);
     }
 
     json_object_put(request_json);
 
-    log_info("Client disconnected: %s:%d", client_ip, ntohs(client->addr.sin_port));
-    close(client->fd);
+    log_info("Client disconnected: %s:%d", socket_ip, ntohs(socket->addr.sin_port));
+    close(socket->fd);
 
-    free(client);
+    free(socket);
     return 0;
 }
 
@@ -95,30 +119,41 @@ int main(void) {
     signal(SIGTERM, sig_handler);
 
     log_info("Initializing salis data server");
-    int opt = 1;
-    int server_fd = socket(AF_INET, SOCK_STREAM, 0);
-    setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+    log_info("Connecting to database in: %s", DATA_PUSH_PATH);
+    sqlite3_open(DATA_PUSH_PATH, &g_sim_db);
+    assert(g_sim_db);
+
+    // Install busy handler to retry transactions if DB is locked
+    sqlite3_busy_timeout(g_sim_db, DATA_PUSH_BUSY_TIMEOUT);
+
+    // Enable Write-Ahead Logging (WAL)
+    // This seems to help prevent DB locks when displaying live data.
+    // See: https://sqlite.org/wal.html
+    salis_exec_sql(0, NULL, NULL, "pragma journal_mode=wal;");
 
     log_info("Binding to port: %d", PORT);
-    struct sockaddr_in server_addr = { 0 };
-    server_addr.sin_family = AF_INET;
-    server_addr.sin_addr.s_addr = INADDR_ANY;
-    server_addr.sin_port = htons(PORT);
-    bind(server_fd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr_in));
+    int opt = 1;
+    int socket_fd = socket(AF_INET, SOCK_STREAM, 0);
+    setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+    struct sockaddr_in socket_addr = { 0 };
+    socket_addr.sin_family = AF_INET;
+    socket_addr.sin_addr.s_addr = INADDR_ANY;
+    socket_addr.sin_port = htons(PORT);
+    bind(socket_fd, (struct sockaddr *)&socket_addr, sizeof(struct sockaddr_in));
 
     log_info("Listening...");
-    listen(server_fd, BACKLOG);
+    listen(socket_fd, BACKLOG);
 
     while (true) {
-        struct Client *client = calloc(1, sizeof(struct Client));
-        socklen_t client_len = sizeof(struct sockaddr_in);
-        client->fd = accept(server_fd, (struct sockaddr *)&client->addr, &client_len);
+        struct Socket *socket = calloc(1, sizeof(struct Socket));
+        socklen_t socket_len = sizeof(struct sockaddr_in);
+        socket->fd = accept(socket_fd, (struct sockaddr *)&socket->addr, &socket_len);
 
         thrd_t thrd;
-        thrd_create(&thrd, (thrd_start_t)handle_client, client);
+        thrd_create(&thrd, (thrd_start_t)handle_client, socket);
         thrd_detach(thrd);
     }
 
-    close(server_fd);
+    close(socket_fd);
     return 0;
 }