feat: Add back the tox_loop implementation for low latency.

.circleci/cmake-tsan

@@ -22,6 +22,7 @@ cmake -B_build -H. -GNinja \
   -DSTRICT_ABI=ON \
   -DTEST_TIMEOUT_SECONDS=120 \
   -DUSE_IPV6=OFF \
+  -DUSE_LIBEV=ON \
   -DAUTOTEST=ON
 
 cd _build

.circleci/config.yml

@@ -68,6 +68,7 @@ jobs:
             cmake
             git
             libconfig-dev
+            libev-dev
             libgmock-dev
             libgtest-dev
             libopus-dev
@@ -120,6 +121,8 @@ jobs:
       - run: other/analysis/check_logger_levels
       - run: other/analysis/run-clang
       - run: other/analysis/run-gcc
+      - run: other/analysis/run-cppcheck
+      - run: other/analysis/run-clang-analyze
 
   clang-analyze:
     working_directory: ~/work

CMakeLists.txt

@@ -194,6 +194,12 @@ endif()
 # We don't transfer floats over the network, so we disable this functionality.
 add_definitions(-DCMP_NO_FLOAT=1)
 
+# TODO(iphydf): Check whether this is actually true.
+option(USE_LIBEV "Whether to use libev for tox_loop" OFF)
+if(USE_LIBEV)
+  add_definitions(-DHAVE_LIBEV=1)
+endif()
+
 ################################################################################
 #
 # :: Tox Core Library

auto_tests/CMakeLists.txt

@@ -79,6 +79,7 @@ auto_test(set_name)
 auto_test(set_status_message)
 auto_test(tox_dispatch)
 auto_test(tox_events)
+auto_test(tox_loop)
 auto_test(tox_many)
 auto_test(tox_many_tcp)
 auto_test(tox_strncasecmp)

auto_tests/tox_loop_test.c

@@ -0,0 +1,128 @@
+#include <pthread.h>
+#include <stdlib.h>
+#include <time.h>
+
+#include "../toxcore/tox.h"
+
+#include "check_compat.h"
+#include "../testing/misc_tools.h"
+
+/* The CI containers respond poorly to ::1 as a localhost address
+ * You're encouraged to -D FORCE_TESTS_IPV6 on a local test  */
+#ifdef TOX_LOCALHOST
+#undef TOX_LOCALHOST
+#endif
+#ifdef FORCE_TESTS_IPV6
+#define TOX_LOCALHOST "::1"
+#else
+#define TOX_LOCALHOST "127.0.0.1"
+#endif
+
+#ifdef TCP_RELAY_PORT
+#undef TCP_RELAY_PORT
+#endif
+#define TCP_RELAY_PORT 33431
+
+typedef struct Loop_Test {
+    int start_count;
+    int stop_count;
+    pthread_mutex_t mutex;
+    Tox *tox;
+} Loop_Test;
+
+static void tox_loop_cb_start(Tox *tox, void *data)
+{
+    Loop_Test *userdata = (Loop_Test *)data;
+    pthread_mutex_lock(&userdata->mutex);
+    ++userdata->start_count;
+}
+
+static void tox_loop_cb_stop(Tox *tox, void *data)
+{
+    Loop_Test *userdata = (Loop_Test *)data;
+    ++userdata->stop_count;
+    pthread_mutex_unlock(&userdata->mutex);
+}
+
+static void *tox_loop_worker(void *data)
+{
+    Loop_Test *userdata = (Loop_Test *)data;
+    Tox_Err_Loop err;
+    tox_loop(userdata->tox, userdata, &err);
+    ck_assert_msg(err == TOX_ERR_LOOP_OK, "tox_loop error: %d", err);
+    return nullptr;
+}
+
+static void test_tox_loop(void)
+{
+    pthread_t worker, worker_tcp;
+    Tox_Err_Options_New err_opts;
+    struct Tox_Options *opts = tox_options_new(&err_opts);
+    ck_assert_msg(err_opts == TOX_ERR_OPTIONS_NEW_OK, "tox_options_new: %d\n", err_opts);
+    tox_options_set_experimental_thread_safety(opts, true);
+
+    Loop_Test *userdata = (Loop_Test *)calloc(1, sizeof(Loop_Test));
+    ck_assert(userdata != nullptr);
+    uint8_t dpk[TOX_PUBLIC_KEY_SIZE];
+
+    userdata->start_count = 0;
+    userdata->stop_count = 0;
+    pthread_mutex_init(&userdata->mutex, nullptr);
+
+    tox_options_set_tcp_port(opts, TCP_RELAY_PORT);
+    Tox_Err_New err_new;
+    userdata->tox = tox_new(opts, &err_new);
+    ck_assert_msg(err_new == TOX_ERR_NEW_OK, "tox_new: %d\n", err_new);
+    tox_callback_loop_begin(userdata->tox, tox_loop_cb_start);
+    tox_callback_loop_end(userdata->tox, tox_loop_cb_stop);
+    pthread_create(&worker, nullptr, tox_loop_worker, userdata);
+
+    tox_self_get_dht_id(userdata->tox, dpk);
+
+    tox_options_default(opts);
+    tox_options_set_experimental_thread_safety(opts, true);
+    Loop_Test userdata_tcp;
+    userdata_tcp.start_count = 0;
+    userdata_tcp.stop_count = 0;
+    pthread_mutex_init(&userdata_tcp.mutex, nullptr);
+    userdata_tcp.tox = tox_new(opts, &err_new);
+    ck_assert_msg(err_new == TOX_ERR_NEW_OK, "tox_new: %d\n", err_new);
+    tox_callback_loop_begin(userdata_tcp.tox, tox_loop_cb_start);
+    tox_callback_loop_end(userdata_tcp.tox, tox_loop_cb_stop);
+    pthread_create(&worker_tcp, nullptr, tox_loop_worker, &userdata_tcp);
+
+    pthread_mutex_lock(&userdata_tcp.mutex);
+    Tox_Err_Bootstrap error;
+    ck_assert_msg(tox_add_tcp_relay(userdata_tcp.tox, TOX_LOCALHOST, TCP_RELAY_PORT, dpk, &error), "Add relay error, %i",
+                  error);
+    ck_assert_msg(tox_bootstrap(userdata_tcp.tox, TOX_LOCALHOST, 33445, dpk, &error), "Bootstrap error, %i", error);
+    pthread_mutex_unlock(&userdata_tcp.mutex);
+
+    c_sleep(1000);
+
+    tox_loop_stop(userdata->tox);
+    void *retval = nullptr;
+    pthread_join(worker, &retval);
+    ck_assert_msg((uintptr_t)retval == 0, "tox_loop didn't return 0");
+
+    tox_kill(userdata->tox);
+    ck_assert_msg(userdata->start_count == userdata->stop_count, "start and stop must match (start = %d, stop = %d)",
+                  userdata->start_count, userdata->stop_count);
+
+    tox_loop_stop(userdata_tcp.tox);
+    pthread_join(worker_tcp, &retval);
+    ck_assert_msg((uintptr_t)retval == 0, "tox_loop didn't return 0");
+
+    tox_kill(userdata_tcp.tox);
+    ck_assert_msg(userdata_tcp.start_count == userdata_tcp.stop_count, "start and stop must match (start = %d, stop = %d)",
+                  userdata_tcp.start_count, userdata_tcp.stop_count);
+
+    tox_options_free(opts);
+    free(userdata);
+}
+
+int main(int argc, char *argv[])
+{
+    test_tox_loop();
+    return 0;
+}

other/analysis/gen-file.sh

@@ -15,7 +15,7 @@ CPPFLAGS+=("-Itoxav")
 CPPFLAGS+=("-Itoxencryptsave")
 CPPFLAGS+=("-Ithird_party/cmp")
 
-LDFLAGS=("-lopus" "-lsodium" "-lvpx" "-lpthread" "-lconfig" "-lgmock" "-lgtest")
+LDFLAGS=("-lopus" "-lsodium" "-lvpx" "-lpthread" "-lconfig" "-lgmock" "-lgtest" "-lev")
 LDFLAGS+=("-fuse-ld=gold")
 LDFLAGS+=("-Wl,--detect-odr-violations")
 LDFLAGS+=("-Wl,--warn-common")

other/analysis/run-clang-tidy

@@ -79,6 +79,9 @@ CHECKS="$CHECKS,-clang-diagnostic-tautological-pointer-compare"
 # [unreadVariable]
 CHECKS="$CHECKS,-cppcoreguidelines-init-variables"
 
+# Used by libev.
+CHECKS="$CHECKS,-hicpp-no-assembler"
+
 # Short variable names are used quite a lot, and we don't consider them a
 # readability issue.
 CHECKS="$CHECKS,-readability-identifier-length"

other/analysis/variants.sh

@@ -1,3 +1,5 @@
 #!/bin/bash
 
-run
+run "$@"
+#run -DVANILLA_NACL -I/usr/include/sodium "$@"
+run -DHAVE_LIBEV "$@"

toxcore/BUILD.bazel

@@ -303,6 +303,7 @@ cc_library(
     name = "network",
     srcs = ["network.c"],
     hdrs = ["network.h"],
+    defines = ["HAVE_LIBEV"],
     visibility = [
         "//c-toxcore/auto_tests:__pkg__",
         "//c-toxcore/other:__pkg__",
@@ -319,6 +320,7 @@ cc_library(
         ":mem",
         ":mono_time",
         ":util",
+        "@ev",
         "@libsodium",
         "@psocket",
         "@pthread",

toxcore/LAN_discovery.c

@@ -204,7 +204,8 @@ static Broadcast_Info *fetch_broadcast_info(const Network *ns)
 
 #endif /* platforms */
 
-/** @brief Send packet to all IPv4 broadcast addresses
+/**
+ * @brief Send packet to all IPv4 broadcast addresses
  *
  * @retval true if sent to at least one broadcast target.
  * @retval false on failure to find any valid broadcast target.

toxcore/TCP_client.c

@@ -12,6 +12,11 @@
 #include <stdio.h>
 #include <string.h>
 
+#ifdef HAVE_LIBEV
+#include <ev.h>
+#endif
+
+#include "DHT.h"
 #include "TCP_common.h"
 #include "attributes.h"
 #include "ccompat.h"
@@ -30,9 +35,19 @@ typedef struct TCP_Client_Conn {
     uint32_t number;
 } TCP_Client_Conn;
 
+#ifdef HAVE_LIBEV
+typedef struct TCP_Client_Socket_Listener {
+    ev_io listener;
+    struct ev_loop *dispatcher;
+} TCP_Client_Socket_Listener;
+#endif
+
 struct TCP_Client_Connection {
     TCP_Connection con;
     TCP_Client_Status status;
+#ifdef HAVE_LIBEV
+    TCP_Client_Socket_Listener sock_listener;
+#endif
     uint8_t self_public_key[CRYPTO_PUBLIC_KEY_SIZE]; /* our public key */
     uint8_t public_key[CRYPTO_PUBLIC_KEY_SIZE]; /* public key of the server */
     IP_Port ip_port; /* The ip and port of the server */
@@ -85,6 +100,44 @@ TCP_Client_Status tcp_con_status(const TCP_Client_Connection *con)
 {
     return con->status;
 }
+
+#ifdef HAVE_LIBEV
+non_null()
+static bool tcp_con_ev_is_active(TCP_Client_Connection *con)
+{
+    return ev_is_active(&con->sock_listener.listener)
+           || ev_is_pending(&con->sock_listener.listener);
+}
+
+void tcp_con_ev_listen(TCP_Client_Connection *con, struct ev_loop *dispatcher, tcp_con_ev_listen_cb *callback,
+                       void *data)
+{
+    if (tcp_con_ev_is_active(con)) {
+        return;
+    }
+
+    con->sock_listener.dispatcher = dispatcher;
+    con->sock_listener.listener.data = data;
+
+    ev_io_init(&con->sock_listener.listener, callback, con->con.sock.sock, EV_READ);
+    ev_io_start(dispatcher, &con->sock_listener.listener);
+}
+
+void tcp_con_ev_stop(TCP_Client_Connection *con)
+{
+    if (!tcp_con_ev_is_active(con)) {
+        return;
+    }
+
+    ev_io_stop(con->sock_listener.dispatcher, &con->sock_listener.listener);
+}
+#else
+Socket tcp_con_sock(const TCP_Client_Connection *con)
+{
+    return con->con.sock;
+}
+#endif
+
 void *tcp_con_custom_object(const TCP_Client_Connection *con)
 {
     return con->custom_object;
@@ -1024,6 +1077,11 @@ void kill_tcp_connection(TCP_Client_Connection *tcp_connection)
 
     wipe_priority_list(tcp_connection->con.mem, tcp_connection->con.priority_queue_start);
     kill_sock(tcp_connection->con.ns, tcp_connection->con.sock);
+
+#ifdef HAVE_LIBEV
+    ev_io_stop(tcp_connection->sock_listener.dispatcher, &tcp_connection->sock_listener.listener);
+#endif
+
     crypto_memzero(tcp_connection, sizeof(TCP_Client_Connection));
     mem_delete(mem, tcp_connection);
 }

toxcore/TCP_client.h

@@ -9,6 +9,10 @@
 #ifndef C_TOXCORE_TOXCORE_TCP_CLIENT_H
 #define C_TOXCORE_TOXCORE_TCP_CLIENT_H
 
+#ifdef HAVE_LIBEV
+#include <ev.h>
+#endif
+
 #include "attributes.h"
 #include "crypto_core.h"
 #include "forwarding.h"
@@ -50,6 +54,20 @@ IP_Port tcp_con_ip_port(const TCP_Client_Connection *con);
 non_null()
 TCP_Client_Status tcp_con_status(const TCP_Client_Connection *con);
 
+// TODO(iphydf): This is exactly the same as in network.h. It should be factored
+// out and probably abstracted away from ev.h.
+#ifdef HAVE_LIBEV
+typedef void tcp_con_ev_listen_cb(struct ev_loop *dispatcher, ev_io *sock_listener, int events);
+non_null()
+void tcp_con_ev_listen(TCP_Client_Connection *con, struct ev_loop *dispatcher, tcp_con_ev_listen_cb *callback,
+                       void *data);
+non_null()
+void tcp_con_ev_stop(TCP_Client_Connection *con);
+#else
+non_null()
+Socket tcp_con_sock(const TCP_Client_Connection *con);
+#endif
+
 non_null()
 void *tcp_con_custom_object(const TCP_Client_Connection *con);
 non_null()

toxcore/TCP_connection.c

@@ -71,6 +71,37 @@ uint32_t tcp_connections_count(const TCP_Connections *tcp_c)
     return tcp_c->tcp_connections_length;
 }
 
+/**
+ * Return number of elements of TCP connection array.
+ *
+ * @param tcp_c struct containing TCP_con array.
+ *
+ * @return number of elements of TCP connection array.
+ */
+uint32_t tcp_connections_length(const TCP_Connections *tcp_c)
+{
+    return tcp_c->tcp_connections_length;
+}
+
+
+/**
+ * Return TCP connection stored at "idx" position.
+ *
+ * @param tcp_c struct containing TCP_con array.
+ * @param idx   index of TCP connection to return (values from 0 to `tcp_connections_length() - 1`).
+ *
+ * @return TCP connection stored at "idx" position, or NULL if errors occurred.
+ */
+const TCP_con *tcp_connections_connection_at(const TCP_Connections *tcp_c, uint32_t idx)
+{
+    if (idx >= tcp_c->tcp_connections_length) {
+        return nullptr;
+    }
+
+    return &tcp_c->tcp_connections[idx];
+}
+
+
 /** @brief Set the size of the array to num.
  *
  * @retval -1 if mem_vrealloc fails.