1 files changed, 89 insertions, 16 deletions

diff --git a/src/tools/oping/oping_server.c b/src/tools/oping/oping_server.c
index ec9e85ab..e98ca040 100644
--- a/src/tools/oping/oping_server.c
+++ b/src/tools/oping/oping_server.c
@@ -1,5 +1,5 @@
 /*
- * Ouroboros - Copyright (C) 2016 - 2021
+ * Ouroboros - Copyright (C) 2016 - 2026
  *
  * Ouroboros ping application
  *
@@ -36,6 +36,8 @@
  * OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 
+ #include <ouroboros/hash.h>
+
 void shutdown_server(int signo, siginfo_t * info, void * c)
 {
         (void) info;
@@ -67,7 +69,7 @@ void * cleaner_thread(void * o)
                                 time_t diff;
 
                                 pthread_mutex_lock(&server.lock);
-                                diff = ts_diff_ms(&server.times[i], &now);
+                                diff = ts_diff_ms(&now, &server.times[i]);
                                 pthread_mutex_unlock(&server.lock);
 
                                 if (diff > deadline_ms) {
@@ -87,12 +89,15 @@ void * server_thread(void *o)
         struct oping_msg * msg = (struct oping_msg *) buf;
         struct timespec now = {0, 0};
         struct timespec timeout = {0, 100 * MILLION};
+        struct timespec poll_timeout = {0, 0};
         int fd;
 
         (void) o;
 
         while (true) {
-                if (fevent(server.flows, server.fq, &timeout) == -ETIMEDOUT)
+                if (fevent(server.flows, server.fq,
+                           server.poll ? &poll_timeout : &timeout)
+                    == -ETIMEDOUT)
                         continue;
 
                 while ((fd = fqueue_next(server.fq)) >= 0) {
@@ -100,13 +105,15 @@ void * server_thread(void *o)
                         if (msg_len < 0)
                                 continue;
 
+                        if (!server.quiet)
+                                printf("Received %d bytes on fd %d.\n",
+                                       msg_len, fd);
+
                         if (ntohl(msg->type) != ECHO_REQUEST) {
                                 printf("Invalid message on fd %d.\n", fd);
                                 continue;
                         }
 
-                        printf("Received %d bytes on fd %d.\n", msg_len, fd);
-
                         clock_gettime(CLOCK_REALTIME, &now);
 
                         pthread_mutex_lock(&server.lock);
@@ -131,13 +138,16 @@ void * accept_thread(void * o)
 
         (void) o;
 
-        printf("Ouroboros ping server started.\n");
+        printf("Ouroboros ping server started.");
+        if (server.busy)
+                printf(" [busy-poll]");
+        printf("\n");
 
         while (true) {
                 fd = flow_accept(&qs, NULL);
                 if (fd < 0) {
-                        printf("Failed to accept flow.\n");
-                        break;
+                        printf("Failed to accept flow: %d \n", fd);
+                        continue;
                 }
 
                 printf("New flow %d.\n", fd);
@@ -151,9 +161,53 @@ void * accept_thread(void * o)
                 pthread_mutex_unlock(&server.lock);
 
                 fccntl(fd, FLOWSFLAGS,
-                       FLOWFRNOBLOCK | FLOWFRDWR | FLOWFRNOPART);
+                       FLOWFRNOBLOCK | FLOWFRDWR
+                       | FLOWFRNOPART);
+        }
+
+        return (void *) 0;
+}
+
+void * busy_thread(void * o)
+{
+        char buf[OPING_BUF_SIZE];
+        struct oping_msg * msg = (struct oping_msg *) buf;
+        int    fd;
+        int    msg_len;
+
+        (void) o;
+
+        /* Accept a single flow. */
+        fd = flow_accept(NULL, NULL);
+        if (fd < 0) {
+                printf("Failed to accept flow.\n");
+                return (void *) -1;
+        }
+
+        printf("New flow %d (busy-poll).\n", fd);
+
+        fccntl(fd, FLOWSFLAGS,
+               FLOWFRNOBLOCK | FLOWFRDWR
+               | FLOWFRNOPART);
+
+        while (true) {
+                msg_len = flow_read(fd, buf,
+                                    OPING_BUF_SIZE);
+                if (msg_len == -EAGAIN)
+                        continue;
+                if (msg_len < 0)
+                        break;
+
+                if (ntohl(msg->type) != ECHO_REQUEST)
+                        continue;
+
+                msg->type = htonl(ECHO_REPLY);
+
+                flow_write(fd, buf, msg_len);
         }
 
+        flow_dealloc(fd);
+
         return (void *) 0;
 }
 
@@ -183,20 +237,39 @@ int server_main(void)
                 return -1;
         }
 
+        if (pthread_mutex_init(&server.lock, NULL)) {
+                fqueue_destroy(server.fq);
+                fset_destroy(server.flows);
+                return -1;
+        }
+
+        memset(server.times, 0, sizeof(server.times));
+
         pthread_create(&server.cleaner_pt, NULL, cleaner_thread, NULL);
-        pthread_create(&server.accept_pt, NULL, accept_thread, NULL);
-        pthread_create(&server.server_pt, NULL, server_thread, NULL);
 
-        pthread_join(server.accept_pt, NULL);
+        if (server.busy) {
+                pthread_create(&server.server_pt, NULL,
+                               busy_thread, NULL);
+                pthread_join(server.server_pt, NULL);
+                pthread_cancel(server.cleaner_pt);
+        } else {
+                pthread_create(&server.accept_pt, NULL,
+                               accept_thread, NULL);
+                pthread_create(&server.server_pt, NULL,
+                               server_thread, NULL);
+                pthread_join(server.accept_pt, NULL);
+                pthread_cancel(server.server_pt);
+        }
 
-        pthread_cancel(server.server_pt);
         pthread_cancel(server.cleaner_pt);
 
-        fset_destroy(server.flows);
-        fqueue_destroy(server.fq);
-
+        /* Join cancellable threads before tearing down their fset. */
         pthread_join(server.server_pt, NULL);
         pthread_join(server.cleaner_pt, NULL);
 
+        pthread_mutex_destroy(&server.lock);
+        fset_destroy(server.flows);
+        fqueue_destroy(server.fq);
+
         return 0;
 }