add TaskBlock events for blocking intervals

ddprof-lib/src/main/cpp/callTraceHashTable.cpp

@@ -252,17 +252,14 @@ CallTrace *CallTraceHashTable::findCallTrace(LongHashTable *table, u64 hash) {
   while (true) {
     // Use atomic load: keys[] can be written concurrently via CAS in put()
     // when a table is promoted to prev but still has in-flight insertions.
-    u64 key = __atomic_load_n(&keys[slot], __ATOMIC_ACQUIRE);
-    if (key == hash) {
+    u64 key_value = __atomic_load_n(&keys[slot], __ATOMIC_ACQUIRE);
+    if (key_value == hash) {
       // Use acquireTrace() to pair with the RELEASE store in setTrace().
       // If still PREPARING, treat as not found: callers will create a new entry.
       CallTrace *trace = table->values()[slot].acquireTrace();
-      if (trace == CallTraceSample::PREPARING) {
-        return nullptr;
-      }
-      return trace;
+      return trace == CallTraceSample::PREPARING ? nullptr : trace;
     }
-    if (key == 0) {
+    if (key_value == 0) {
       return nullptr;
     }
     if (!probe.hasNext()) {

ddprof-lib/src/main/cpp/codeCache.cpp

@@ -310,25 +310,44 @@ void CodeCache::saveImport(ImportId id, void** entry) {
 void CodeCache::addImport(void **entry, const char *name) {
     switch (name[0]) {
       case 'a':
-          if (strcmp(name, "aligned_alloc") == 0) {
+          if (strcmp(name, "accept") == 0) {
+              saveImport(im_accept, entry);
+          } else if (strcmp(name, "accept4") == 0) {
+              saveImport(im_accept4, entry);
+          } else if (strcmp(name, "aligned_alloc") == 0) {
               saveImport(im_aligned_alloc, entry);
           }
           break;
       case 'c':
           if (strcmp(name, "calloc") == 0) {
               saveImport(im_calloc, entry);
+          } else if (strcmp(name, "close") == 0) {
+              saveImport(im_close, entry);
+          } else if (strcmp(name, "connect") == 0) {
+              saveImport(im_connect, entry);
           }
           break;
       case 'd':
           if (strcmp(name, "dlopen") == 0) {
               saveImport(im_dlopen, entry);
+          } else if (strcmp(name, "dup2") == 0) {
+              saveImport(im_dup2, entry);
+          } else if (strcmp(name, "dup3") == 0) {
+              saveImport(im_dup3, entry);
           }
           break;
       case 'f':
           if (strcmp(name, "free") == 0) {
               saveImport(im_free, entry);
           }
           break;
+      case 'e':
+          if (strcmp(name, "epoll_wait") == 0) {
+              saveImport(im_epoll_wait, entry);
+          } else if (strcmp(name, "epoll_pwait") == 0) {
+              saveImport(im_epoll_pwait, entry);
+          }
+          break;
       case 'm':
           if (strcmp(name, "malloc") == 0) {
               saveImport(im_malloc, entry);
@@ -343,6 +362,10 @@ void CodeCache::addImport(void **entry, const char *name) {
               saveImport(im_pthread_setspecific, entry);
           } else if (strcmp(name, "poll") == 0) {
               saveImport(im_poll, entry);
+          } else if (strcmp(name, "ppoll") == 0) {
+              saveImport(im_ppoll, entry);
+          } else if (strcmp(name, "pselect") == 0) {
+              saveImport(im_pselect, entry);
           } else if (strcmp(name, "posix_memalign") == 0) {
               saveImport(im_posix_memalign, entry);
           }
@@ -352,6 +375,10 @@ void CodeCache::addImport(void **entry, const char *name) {
               saveImport(im_realloc, entry);
           } else if (strcmp(name, "recv") == 0) {
               saveImport(im_recv, entry);
+          } else if (strcmp(name, "recvfrom") == 0) {
+              saveImport(im_recvfrom, entry);
+          } else if (strcmp(name, "recvmsg") == 0) {
+              saveImport(im_recvmsg, entry);
           } else if (strcmp(name, "read") == 0) {
               saveImport(im_read, entry);
           }
@@ -361,6 +388,8 @@ void CodeCache::addImport(void **entry, const char *name) {
               saveImport(im_send, entry);
           } else if (strcmp(name, "sigaction") == 0) {
               saveImport(im_sigaction, entry);
+          } else if (strcmp(name, "select") == 0) {
+              saveImport(im_select, entry);
           }
           break;
       case 'w':
@@ -467,4 +496,3 @@ void CodeCache::setBuildId(const char* build_id, size_t build_id_len) {
     }
   }
 }
-

ddprof-lib/src/main/cpp/codeCache.h

@@ -27,6 +27,8 @@ const int MAX_NATIVE_LIBS = 2048;
 
 enum ImportId {
   im_dlopen,
+  im_dup2,
+  im_dup3,
   im_pthread_create,
   im_pthread_exit,
   im_pthread_setspecific,
@@ -42,6 +44,17 @@ enum ImportId {
   im_recv,
   im_write,
   im_read,
+  im_close,
+  im_connect,
+  im_accept,
+  im_accept4,
+  im_recvfrom,
+  im_recvmsg,
+  im_epoll_wait,
+  im_epoll_pwait,
+  im_ppoll,
+  im_select,
+  im_pselect,
   NUM_IMPORTS
 };
 

ddprof-lib/src/main/cpp/counters.h

@@ -120,6 +120,13 @@
   X(JVMTI_STACKS_FAILED_OTHER, "jvmti_stacks_failed_other")                  \
   /* Delegated stacks dropped at slot-lock. Rec-lock drops from all recording  \
    * paths (delegated and direct) go into SAMPLES_DROPPED_REC_LOCK. */         \
+  X(TASK_BLOCK_EMITTED, "task_block_emitted")                                \
+  X(TASK_BLOCK_SKIPPED_TRACE_CONTEXT, "task_block_skipped_trace_context")    \
+  X(TASK_BLOCK_SKIPPED_TOO_SHORT, "task_block_skipped_too_short")            \
+  X(TASK_BLOCK_SKIPPED_NO_STACK_REFERENCE,                                   \
+    "task_block_skipped_no_stack_reference")                                 \
+  X(TASK_BLOCK_RECORD_FAILED, "task_block_record_failed")                    \
+  X(TASK_BLOCK_QUEUE_DROPPED, "task_block_queue_dropped")                    \
   X(JVMTI_STACKS_DROPPED_LOCK, "jvmti_stacks_dropped_lock")                   \
   X(SAMPLES_DROPPED_REC_LOCK, "samples_dropped_rec_lock")                     \
   X(SAMPLES_DROPPED_THREAD_LOCAL, "samples_dropped_thread_local")

ddprof-lib/src/main/cpp/event.h

@@ -57,7 +57,7 @@ class ExecutionEvent : public Event {
   OSThreadState _thread_state;
   ExecutionMode _execution_mode;
   u64 _weight;
-  u32 _call_trace_id;
+  u64 _call_trace_id;
 
   ExecutionEvent()
       : Event(), _thread_state(OSThreadState::RUNNABLE), _execution_mode(ExecutionMode::UNKNOWN),
@@ -206,4 +206,15 @@ typedef struct QueueTimeEvent {
   u32 _queueLength;
 } QueueTimeEvent;
 
+typedef struct TaskBlockEvent {
+  u64 _start;
+  u64 _end;
+  u64 _blocker;
+  u64 _unblockingSpanId;
+  Context _ctx;
+  u64 _callTraceId;
+  u64 _correlationId;
+  OSThreadState _observedBlockingState;
+} TaskBlockEvent;
+
 #endif // _EVENT_H

ddprof-lib/src/main/cpp/flightRecorder.cpp

@@ -1445,7 +1445,7 @@ void Recording::writeFrameTypes(Buffer *buf) {
 
 void Recording::writeThreadStates(Buffer *buf) {
   buf->putVar64(T_THREAD_STATE);
-  buf->put8(10);
+  buf->put8(11);
   buf->put8(static_cast<int>(OSThreadState::UNKNOWN));
   buf->putUtf8("UNKNOWN");
   buf->put8(static_cast<int>(OSThreadState::NEW));
@@ -1466,6 +1466,8 @@ void Recording::writeThreadStates(Buffer *buf) {
   buf->putUtf8("TERMINATED");
   buf->put8(static_cast<int>(OSThreadState::SYSCALL));
   buf->putUtf8("SYSCALL");
+  buf->put8(static_cast<int>(OSThreadState::IO_WAIT));
+  buf->putUtf8("IO_WAIT");
   flushIfNeeded(buf);
 }
 
@@ -1817,6 +1819,22 @@ void Recording::recordMethodSample(Buffer *buf, int tid, u64 call_trace_id,
   flushIfNeeded(buf);
 }
 
+void Recording::recordTaskBlock(Buffer *buf, int tid, TaskBlockEvent *event) {
+  int start = buf->skip(1);
+  buf->putVar64(T_TASK_BLOCK);
+  buf->putVar64(event->_start);
+  buf->putVar64(event->_end - event->_start);
+  buf->putVar64(tid);
+  buf->putVar64(event->_blocker);
+  buf->putVar64(event->_unblockingSpanId);
+  buf->putVar64(event->_callTraceId);
+  buf->putVar64(event->_correlationId);
+  buf->put8(static_cast<int>(event->_observedBlockingState));
+  writeContextSnapshot(buf, event->_ctx);
+  writeEventSizePrefix(buf, start);
+  flushIfNeeded(buf);
+}
+
 void Recording::recordWallClockEpoch(Buffer *buf, WallClockEpochEvent *event) {
   int start = buf->skip(1);
   buf->putVar64(T_WALLCLOCK_SAMPLE_EPOCH);
@@ -2082,6 +2100,20 @@ void FlightRecorder::recordQueueTime(int lock_index, int tid,
   }
 }
 
+bool FlightRecorder::recordTaskBlock(int lock_index, int tid,
+                                     TaskBlockEvent *event) {
+  OptionalSharedLockGuard locker(&_rec_lock);
+  if (locker.ownsLock()) {
+    Recording* rec = _rec;
+    if (rec != nullptr) {
+      Buffer *buf = rec->buffer(lock_index);
+      rec->recordTaskBlock(buf, tid, event);
+      return true;
+    }
+  }
+  return false;
+}
+
 void FlightRecorder::recordDatadogSetting(int lock_index, int length,
                                           const char *name, const char *value,
                                           const char *unit) {
@@ -2115,6 +2147,7 @@ bool FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id,
     Recording* rec = _rec;
     if (rec != nullptr) {
       RecordingBuffer *buf = rec->buffer(lock_index);
+      bool recorded = true;
       switch (event_type) {
       case BCI_CPU:
           rec->recordExecutionSample(buf, tid, call_trace_id, 0,
@@ -2144,11 +2177,14 @@ bool FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id,
           rec->recordNativeSocketSample(buf, tid, call_trace_id, (NativeSocketEvent *)event);
           break;
         default:
-          return false;
+          recorded = false;
+          break;
         }
-        rec->flushIfNeeded(buf);
-        rec->addThread(lock_index, tid);
-        return true;
+        if (recorded) {
+          rec->flushIfNeeded(buf);
+          rec->addThread(lock_index, tid);
+        }
+        return recorded;
       }
   } else {
     Counters::increment(SAMPLES_DROPPED_REC_LOCK);
@@ -2164,6 +2200,7 @@ bool FlightRecorder::recordEventDelegated(int lock_index, int tid,
     Recording* rec = _rec;
     if (rec != nullptr) {
       RecordingBuffer *buf = rec->buffer(lock_index);
+      bool recorded = true;
       switch (event_type) {
         case BCI_CPU:
           rec->recordExecutionSample(buf, tid, 0, correlation_id,
@@ -2175,11 +2212,14 @@ bool FlightRecorder::recordEventDelegated(int lock_index, int tid,
           break;
         default:
           // Delegation is only wired for CPU/wall samples in v1.
-          return false;
+          recorded = false;
+          break;
       }
-      rec->flushIfNeeded(buf);
-      rec->addThread(lock_index, tid);
-      return true;
+      if (recorded) {
+        rec->flushIfNeeded(buf);
+        rec->addThread(lock_index, tid);
+      }
+      return recorded;
     }
   } else {
     Counters::increment(SAMPLES_DROPPED_REC_LOCK);

ddprof-lib/src/main/cpp/flightRecorder.h

@@ -316,6 +316,7 @@ class Recording {
   void recordWallClockEpoch(Buffer *buf, WallClockEpochEvent *event);
   void recordTraceRoot(Buffer *buf, int tid, TraceRootEvent *event);
   void recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event);
+  void recordTaskBlock(Buffer *buf, int tid, TaskBlockEvent *event);
   void recordAllocation(RecordingBuffer *buf, int tid, u64 call_trace_id,
                         AllocEvent *event);
   void recordMallocSample(Buffer *buf, int tid, u64 call_trace_id,
@@ -424,6 +425,7 @@ class FlightRecorder {
   void wallClockEpoch(int lock_index, WallClockEpochEvent *event);
   void recordTraceRoot(int lock_index, int tid, TraceRootEvent *event);
   void recordQueueTime(int lock_index, int tid, QueueTimeEvent *event);
+  bool recordTaskBlock(int lock_index, int tid, TaskBlockEvent *event);
 
   bool active() const { return _rec != NULL; }