TSG-14342 tsg-service-chaining-engine在空闲时调用marsio_poll_wait让出cpu供其他进程使用v1.0.7-20230328

1 files changed, 34 insertions, 39 deletions

diff --git a/platform/src/main.cpp b/platform/src/main.cpp
index 21af7f4..364955d 100644
--- a/platform/src/main.cpp
+++ b/platform/src/main.cpp
@@ -56,33 +56,39 @@ static void *worker_thread_cycle(void *arg)
 {
     struct thread_ctx *thread_ctx = (struct thread_ctx *)arg;
     struct packet_io *handle = thread_ctx->ref_io;
-    int n_packet_recv;
+    struct sce_ctx *sce_ctx = thread_ctx->ref_sce_ctx;
+    struct timestamp *ts = sce_ctx->ts;
 
+    int timeout_ms = 0;
+    int n_pkt_recv_from_nf = 0;
+    int n_pkt_recv_from_endp = 0;
     char thread_name[16];
+    uint64_t sf_metrics_last_send_ts = timestamp_get_msec(ts);
+    uint64_t sf_metrics_send_interval = sf_metrics_get_interval(thread_ctx->sf_metrics) * 1000;
+
     snprintf(thread_name, sizeof(thread_name), "sce:worker-%d", thread_ctx->thread_index);
     prctl(PR_SET_NAME, (unsigned long long)thread_name, NULL, NULL, NULL);
 
-    char affinity[32] = {0};
-    if (thread_ctx->cpu_mask >= 0)
+    if (packet_io_thread_init(handle, thread_ctx) != 0)
     {
-        thread_set_affinity(thread_ctx->cpu_mask);
-        snprintf(affinity, sizeof(affinity), "affinity cpu%d", thread_ctx->cpu_mask);
+        goto error_out;
     }
 
-    LOG_INFO("%s: worker thread %d %s is running", LOG_TAG_SCE, thread_ctx->thread_index, thread_ctx->cpu_mask >= 0 ? affinity : "");
+    LOG_INFO("%s: worker thread %d is running", LOG_TAG_SCE, thread_ctx->thread_index);
 
     while (1)
     {
-        n_packet_recv = packet_io_polling_nf_interface(handle, thread_ctx->thread_index, thread_ctx);
-        if (n_packet_recv)
+        n_pkt_recv_from_nf = packet_io_polling_nf_interface(handle, thread_ctx->thread_index, thread_ctx);
+        n_pkt_recv_from_endp = packet_io_polling_endpoint(handle, thread_ctx->thread_index, thread_ctx);
+        if (n_pkt_recv_from_nf == 0 && n_pkt_recv_from_endp == 0)
         {
-            // LOG_INFO("%s: worker thread %d recv %03d packets from nf_interface", LOG_TAG_SCE, thread_ctx->thread_index, n_packet_recv);
-        }
+            timeout_ms = sf_metrics_last_send_ts + sf_metrics_send_interval - timestamp_get_msec(ts);
+            if (timeout_ms <= 0)
+            {
+                timeout_ms = 0;
+            }
 
-        n_packet_recv = packet_io_polling_endpoint(handle, thread_ctx->thread_index, thread_ctx);
-        if (n_packet_recv)
-        {
-            // LOG_INFO("%s: worker thread %d recv %03d packets from endpoint", LOG_TAG_SCE, thread_ctx->thread_index, n_packet_recv);
+            packet_io_thread_wait(handle, thread_ctx, timeout_ms);
         }
 
         if (__atomic_fetch_add(&thread_ctx->session_table_need_reset, 0, __ATOMIC_RELAXED) > 0)
@@ -91,14 +97,15 @@ static void *worker_thread_cycle(void *arg)
             __atomic_fetch_and(&thread_ctx->session_table_need_reset, 0, __ATOMIC_RELAXED);
         }
 
-        if (__atomic_fetch_add(&thread_ctx->sf_metrics_need_send, 0, __ATOMIC_RELAXED) > 0)
+        if (timestamp_get_msec(ts) - sf_metrics_last_send_ts >= sf_metrics_send_interval)
         {
             sf_metrics_send(thread_ctx->sf_metrics);
             sf_metrics_reset(thread_ctx->sf_metrics);
-            __atomic_fetch_and(&thread_ctx->sf_metrics_need_send, 0, __ATOMIC_RELAXED);
+            sf_metrics_last_send_ts = timestamp_get_msec(ts);
         }
     }
 
+error_out:
     LOG_ERROR("%s: worker thread %d exiting", LOG_TAG_SCE, thread_ctx->thread_index);
     return (void *)NULL;
 }
@@ -106,6 +113,9 @@ static void *worker_thread_cycle(void *arg)
 int main(int argc, char **argv)
 {
     const char *profile = "./conf/sce.conf";
+    uint64_t ts_update_interval = 0;
+    uint64_t g_metrics_last_send_ts = 0;
+    uint64_t g_metrics_send_interval = 0;
 
     int opt = 0;
     while ((opt = getopt(argc, argv, "vh")) != -1)
@@ -157,8 +167,6 @@ int main(int argc, char **argv)
         ctx->work_threads[i].ref_enforcer = ctx->enforcer;
         ctx->work_threads[i].ref_sce_ctx = ctx;
         ctx->work_threads[i].session_table_need_reset = 0;
-        ctx->work_threads[i].sf_metrics_need_send = 0;
-        ctx->work_threads[i].cpu_mask = ctx->enable_cpu_affinity ? ctx->cpu_affinity_mask[i] : -1;
     }
 
     for (int i = 0; i < ctx->nr_worker_threads; i++)
@@ -171,34 +179,21 @@ int main(int argc, char **argv)
         }
     }
 
-    struct timespec current_time;
-    struct timespec g_metrics_last_send_time;
-    struct timespec sf_metrics_last_send_time;
-
-    clock_gettime(CLOCK_MONOTONIC, &current_time);
-    clock_gettime(CLOCK_MONOTONIC, &g_metrics_last_send_time);
-    clock_gettime(CLOCK_MONOTONIC, &sf_metrics_last_send_time);
+    timestamp_update(ctx->ts);
+    ts_update_interval = timestamp_update_interval_ms(ctx->ts);
+    g_metrics_last_send_ts = timestamp_get_msec(ctx->ts);
+    g_metrics_send_interval = ctx->metrics->config.statsd_cycle * 1000;
 
     while (1)
     {
-        if (current_time.tv_sec - g_metrics_last_send_time.tv_sec >= ctx->metrics->config.statsd_cycle)
+        if (timestamp_get_msec(ctx->ts) - g_metrics_last_send_ts >= g_metrics_send_interval)
         {
-            clock_gettime(CLOCK_MONOTONIC, &g_metrics_last_send_time);
             global_metrics_dump(ctx->metrics);
+            g_metrics_last_send_ts = timestamp_get_msec(ctx->ts);
         }
 
-        if (current_time.tv_sec - sf_metrics_last_send_time.tv_sec >= sf_metrics_get_interval(ctx->work_threads[0].sf_metrics))
-        {
-            clock_gettime(CLOCK_MONOTONIC, &sf_metrics_last_send_time);
-            for (int i = 0; i < ctx->nr_worker_threads; i++)
-            {
-                struct thread_ctx *thread_ctx = &ctx->work_threads[i];
-                __atomic_fetch_add(&thread_ctx->sf_metrics_need_send, 1, __ATOMIC_RELAXED);
-            }
-        }
-
-        sleep(MIN(ctx->metrics->config.statsd_cycle, sf_metrics_get_interval(ctx->work_threads[0].sf_metrics)));
-        clock_gettime(CLOCK_MONOTONIC, &current_time);
+        usleep(ts_update_interval * 1000);
+        timestamp_update(ctx->ts);
     }
 
 error_out: