Batch event retrieval in cgo for performance enhancement, getting by …

…interval
KindlingProject · Aug 21, 2023 · ccdfa69 · ccdfa69
1 parent d569bf8
commit ccdfa69
Show file tree

Hide file tree

Showing 6 changed files with 34 additions and 20 deletions.
diff --git a/collector/pkg/component/receiver/cgoreceiver/cgo_func.h b/collector/pkg/component/receiver/cgoreceiver/cgo_func.h
@@ -9,8 +9,7 @@
 extern "C" {
 #endif
 int runForGo();
-int getKindlingEvent(void **kindlingEvent);
-int getKindlingEvents(void** kindlingEvent);
+int getEventsByInterval(int interval, void** kindlingEvent, void* count);
 void suppressEventsCommForGo(char *comm);
 void subEventForGo(char* eventName, char* category, void *params);
 int startProfile();

diff --git a/collector/pkg/component/receiver/cgoreceiver/cgoreceiver.go b/collector/pkg/component/receiver/cgoreceiver/cgoreceiver.go
@@ -82,16 +82,18 @@ func (r *CgoReceiver) Start() error {
 }
 
 func (r *CgoReceiver) startGetEvents() {
-    var pKindlingEvents [10]unsafe.Pointer
+	var count = 0
+    var pKindlingEvents [1000]unsafe.Pointer
     r.shutdownWG.Add(1)
+
     for {
         select {
         case <-r.stopCh:
             r.shutdownWG.Done()
             return
         default:
-            count := int(C.getKindlingEvents(&pKindlingEvents[0]))
-            for i := 0; i < count; i++ {
+            evt_count := int(C.getEventsByInterval(C.int(100000000), &pKindlingEvents[0], (unsafe.Pointer)(&count)))
+            for i := 0; i < evt_count; i++ {
                 event := convertEvent((*CKindlingEventForGo)(pKindlingEvents[i]))
                 r.eventChannel <- event
                 r.stats.add(event.Name, 1)

diff --git a/probe/src/cgo/cgo_func.cpp b/probe/src/cgo/cgo_func.cpp
@@ -8,8 +8,9 @@
 
 int runForGo() { return init_probe(); }
 
-int getKindlingEvent(void** kindlingEvent) { return getEvent(kindlingEvent); }
-int getKindlingEvents(void** kindlingEvents) { return get_events(kindlingEvents); }
+int getEventsByInterval(int interval, void** kindlingEvent, void* count) { 
+    return get_events_by_interval((uint64_t)interval, kindlingEvent, count); 
+}
 
 int startProfile() { return start_profile(); }
 int stopProfile() { return stop_profile(); }

diff --git a/probe/src/cgo/cgo_func.h b/probe/src/cgo/cgo_func.h
@@ -9,8 +9,7 @@
 extern "C" {
 #endif
 int runForGo();
-int getKindlingEvent(void** kindlingEvent);
-int getKindlingEvents(void** kindlingEvent);
+int getEventsByInterval(int interval, void** kindlingEvent, void* count);
 void suppressEventsCommForGo(char *comm);
 void subEventForGo(char* eventName, char* category, void* params);
 int startProfile();

diff --git a/probe/src/cgo/kindling.cpp b/probe/src/cgo/kindling.cpp
@@ -44,6 +44,7 @@ char* duration_char = new char[32];
 char* span_char = new char[1024];
 
 int16_t event_filters[1024][16];
+uint64_t receiver_ts = 0;
 
 void init_sub_label() {
   for (auto e : kindling_to_sysdig) {
@@ -89,16 +90,22 @@ void sub_event(char* eventName, char* category, event_params_for_subscribe param
   }
 }
 
-#define MAX_EVENTS 10  
-int get_events(void** kindlingEvents) {
-    int count = 0;
-    for (int i = 0; i < MAX_EVENTS; i++) {
-        if (getEvent(&kindlingEvents[i]) != 1) {
-            break;
+#define MAX_EVENTS 998
+int get_events_by_interval(uint64_t interval, void** kindlingEvents, void* count) {
+  uint64_t tmp_ts = receiver_ts;
+  int i = 0;
+    while (true) {
+      if (getEvent(interval, &kindlingEvents[i], (int*)count) == 1){
+        if(i >= MAX_EVENTS) {
+          break;
         }
-        count++;
+        i++;
+      }
+      if (tmp_ts != receiver_ts) {
+        break;
+      }
     }
-    return count;  
+  return i;  
 }
 
 void suppress_events_comm(string comm) {
@@ -173,7 +180,7 @@ int init_probe() {
   return 0;
 }
 
-int getEvent(void** pp_kindling_event) {
+int getEvent(uint64_t interval, void** pp_kindling_event, int* event_count) {
   int32_t res;
   sinsp_evt* ev;
   res = inspector->next(&ev);
@@ -204,6 +211,7 @@ int getEvent(void** pp_kindling_event) {
   }
   uint16_t kindling_category = get_kindling_category(ev);
   uint16_t ev_type = ev->get_type();
+  int evtcnt = *event_count;
 
   print_event(ev);
   if (ev_type != PPME_CPU_ANALYSIS_E && is_profile_debug && threadInfo->m_tid == debug_tid &&
@@ -450,6 +458,11 @@ int getEvent(void** pp_kindling_event) {
   }
   strcpy(p_kindling_event->context.tinfo.comm, tmp_comm);
   strcpy(p_kindling_event->context.tinfo.containerId, (char*)threadInfo->m_container_id.data());
+  evtcnt++;
+  *event_count = evtcnt;
+  if (ev->get_ts() - receiver_ts >= interval || *event_count > 998) {
+    receiver_ts = ev->get_ts();
+  }
   return 1;
 }
 

diff --git a/probe/src/cgo/kindling.h b/probe/src/cgo/kindling.h
@@ -24,9 +24,9 @@ void stop_perf();
 
 void exipre_window_cache();
 
-int getEvent(void** kindlingEvent);
+int getEvent(uint64_t interval, void** kindlingEvent, int* event_count);
 
-int get_events(void** kindlingEvents);
+int get_events_by_interval(uint64_t interval, void** kindlingEvent, void* count);
 
 uint16_t get_kindling_category(sinsp_evt* sEvt);