add TaskBlock events for blocking intervals

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

@@ -386,6 +386,14 @@ Error Arguments::parse(const char *args) {
         _jvmtistacks = true;
       }
 
+      CASE("wallprecheck")
+      if (value != NULL) {
+        _wall_precheck = strcmp(value, "false") != 0 && strcmp(value, "0") != 0;
+      } else {
+        // No value means enable
+        _wall_precheck = true;
+      }
+
       CASE("wallsampler")
       if (value != NULL) {
           switch (value[0]) {

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

@@ -168,6 +168,7 @@ class Arguments {
   long _cpu;
   long _wall;
   bool _wall_collapsing;
+  bool _wall_precheck;
   int _wall_threads_per_tick;
   WallclockSampler _wallclock_sampler;
   long _memory;
@@ -205,6 +206,7 @@ class Arguments {
         _cpu(-1),
         _wall(-1),
         _wall_collapsing(false),
+        _wall_precheck(false),
         _wall_threads_per_tick(DEFAULT_WALL_THREADS_PER_TICK),
         _wallclock_sampler(ASGCT),
         _memory(-1),

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

@@ -232,19 +232,12 @@ CallTrace *CallTraceHashTable::findCallTrace(LongHashTable *table, u64 hash) {
 
   u32 slot = probe.slot();
   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) {
-      // Use acquireTrace() to pair with the RELEASE store in setTrace().
-      // If still PREPARING, treat as not found: callers will create a new entry.
+    u64 key_value = __atomic_load_n(&keys[slot], __ATOMIC_ACQUIRE);
+    if (key_value == hash) {
       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,13 +310,21 @@ 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':
@@ -329,6 +337,13 @@ void CodeCache::addImport(void **entry, const char *name) {
               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,18 +358,39 @@ 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);
           }
           break;
       case 'r':
           if (strcmp(name, "realloc") == 0) {
               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);
           }
           break;
       case 's':
           if (strcmp(name, "sigaction") == 0) {
               saveImport(im_sigaction, entry);
+          } else if (strcmp(name, "send") == 0) {
+              saveImport(im_send, entry);
+          } else if (strcmp(name, "select") == 0) {
+              saveImport(im_select, entry);
+          }
+          break;
+      case 'w':
+          if (strcmp(name, "write") == 0) {
+              saveImport(im_write, entry);
           }
           break;
     }
@@ -456,4 +492,3 @@ void CodeCache::setBuildId(const char* build_id, size_t build_id_len) {
     }
   }
 }
-

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

@@ -38,6 +38,21 @@ enum ImportId {
   im_posix_memalign,
   im_aligned_alloc,
   im_sigaction,
+  im_send,
+  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

@@ -65,6 +65,8 @@
   X(AGCT_NATIVE_NO_JAVA_CONTEXT, "agct_native_no_java_context")                \
   X(AGCT_BLOCKED_IN_VM, "agct_blocked_in_vm")                                  \
   X(SKIPPED_WALLCLOCK_UNWINDS, "skipped_wallclock_unwinds")                    \
+  X(WC_SIGNAL_SUPPRESSED_SAMPLED_RUN, "wc_signals_suppressed_sampled_run")     \
+  X(WC_SIGNAL_QUEUE_FULL, "wc_signals_queue_full")                             \
   X(UNWINDING_TIME_ASYNC, "unwinding_ticks_async")                             \
   X(UNWINDING_TIME_JVMTI, "unwinding_ticks_jvmti")                             \
   X(CALLTRACE_STORAGE_DROPPED, "calltrace_storage_dropped_traces")             \
@@ -114,6 +116,7 @@
   X(JVMTI_STACKS_REQUESTED, "jvmti_stacks_requested")                         \
   X(JVMTI_STACKS_FAILED_WRONG_PHASE, "jvmti_stacks_failed_wrong_phase")       \
   X(JVMTI_STACKS_FAILED_OTHER, "jvmti_stacks_failed_other")                  \
+  X(TASK_BLOCK_QUEUE_DROPPED, "task_block_queue_dropped")                    \
   X(JVMTI_STACKS_DROPPED_LOCK, "jvmti_stacks_dropped_lock")
 #define X_ENUM(a, b) a,
 typedef enum CounterId : int {

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

@@ -58,10 +58,11 @@ class ExecutionEvent : public Event {
   ExecutionMode _execution_mode;
   u64 _weight;
   u32 _call_trace_id;
+  u64 _sample_id;
 
   ExecutionEvent()
       : Event(), _thread_state(OSThreadState::RUNNABLE), _execution_mode(ExecutionMode::UNKNOWN),
-        _weight(1), _call_trace_id(0) {}
+        _weight(1), _call_trace_id(0), _sample_id(0) {}
 };
 
 class AllocEvent : public Event {
@@ -109,12 +110,18 @@ class WallClockEpochEvent {
   u32 _num_failed_samples;
   u32 _num_exited_threads;
   u32 _num_permission_denied;
+  u64 _num_suppressed_sampled_run;
+  u64 _num_task_block_emitted;
+  u64 _num_task_block_skipped_trace_context;
+  u64 _num_task_block_skipped_too_short;
 
   WallClockEpochEvent(u64 start_time)
       : _dirty(false), _start_time(start_time), _duration_millis(0),
         _num_samplable_threads(0), _num_successful_samples(0),
         _num_failed_samples(0), _num_exited_threads(0),
-        _num_permission_denied(0) {}
+        _num_permission_denied(0), _num_suppressed_sampled_run(0),
+        _num_task_block_emitted(0), _num_task_block_skipped_trace_context(0),
+        _num_task_block_skipped_too_short(0) {}
 
   bool hasChanged() { return _dirty; }
 
@@ -153,13 +160,45 @@ class WallClockEpochEvent {
     }
   }
 
+  void addNumSuppressedSampledRun(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_suppressed_sampled_run += n;
+    }
+  }
+
+  void addNumTaskBlockEmitted(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_task_block_emitted += n;
+    }
+  }
+
+  void addNumTaskBlockSkippedTraceContext(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_task_block_skipped_trace_context += n;
+    }
+  }
+
+  void addNumTaskBlockSkippedTooShort(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_task_block_skipped_too_short += n;
+    }
+  }
+
   void endEpoch(u64 millis) { _duration_millis = millis; }
 
   void clean() { _dirty = false; }
 
   void newEpoch(u64 start_time) {
     _dirty = false;
     _start_time = start_time;
+    _num_suppressed_sampled_run = 0;
+    _num_task_block_emitted = 0;
+    _num_task_block_skipped_trace_context = 0;
+    _num_task_block_skipped_too_short = 0;
   }
 };
 
@@ -184,4 +223,15 @@ typedef struct QueueTimeEvent {
   u32 _queueLength;
 } QueueTimeEvent;
 
+typedef struct TaskBlockEvent {
+  u64 _start;
+  u64 _end;
+  u64 _blocker;
+  u64 _unblockingSpanId;
+  Context _ctx;
+  u64 _anchorSampleId;
+  u64 _suppressedSampleCount;
+  OSThreadState _observedBlockingState;
+} TaskBlockEvent;
+
 #endif // _EVENT_H

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

@@ -1408,7 +1408,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));
@@ -1429,6 +1429,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);
 }
 
@@ -1747,11 +1749,28 @@ void Recording::recordMethodSample(Buffer *buf, int tid, u64 call_trace_id,
   buf->put8(static_cast<int>(event->_execution_mode));
   buf->putVar64(event->_weight);
   buf->putVar64(correlation_id);
+  buf->putVar64(event->_sample_id);
   writeCurrentContext(buf);
   writeEventSizePrefix(buf, start);
   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->_anchorSampleId);
+  buf->putVar64(event->_suppressedSampleCount);
+  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);
@@ -1762,6 +1781,10 @@ void Recording::recordWallClockEpoch(Buffer *buf, WallClockEpochEvent *event) {
   buf->putVar64(event->_num_failed_samples);
   buf->putVar64(event->_num_exited_threads);
   buf->putVar64(event->_num_permission_denied);
+  buf->putVar64(event->_num_suppressed_sampled_run);
+  buf->putVar64(event->_num_task_block_emitted);
+  buf->putVar64(event->_num_task_block_skipped_trace_context);
+  buf->putVar64(event->_num_task_block_skipped_too_short);
   writeEventSizePrefix(buf, start);
   flushIfNeeded(buf);
 }
@@ -2022,6 +2045,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) {

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

@@ -300,6 +300,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,
@@ -407,6 +408,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; }
 

ddprof-lib/src/main/cpp/hotspot/vmStructs.inline.h

@@ -9,12 +9,21 @@
 
 #include "hotspot/vmStructs.h"
 #include "jvmThread.h"
+#include "vmEntry.h"
 
 VMThread* VMThread::current() {
+    // JVMThread::current() is the native thread self pointer. On OpenJ9/Zing it
+    // is not a HotSpot JavaThread*; only HotSpot may reinterpret it as VMThread*.
+    if (!VM::isHotspot() || JVMThread::current() == nullptr) {
+        return nullptr;
+    }
     return VMThread::cast(JVMThread::current());
 }
 
 VMThread* VMThread::fromJavaThread(JNIEnv* env, jthread thread) {
+    if (!VM::isHotspot()) {
+        return nullptr;
+    }
     assert(_eetop != nullptr);
     if (_eetop != nullptr) {
         jlong eetop = env->GetLongField(thread, _eetop);