diff --git a/ddprof-lib/src/main/cpp/event.h b/ddprof-lib/src/main/cpp/event.h index e9363165f..0f96d251b 100644 --- a/ddprof-lib/src/main/cpp/event.h +++ b/ddprof-lib/src/main/cpp/event.h @@ -155,14 +155,27 @@ class WallClockEpochEvent { class TraceRootEvent { public: u64 _local_root_span_id; + u64 _parent_span_id; + u64 _start_ticks; u32 _label; u32 _operation; - TraceRootEvent(u64 local_root_span_id, u32 label, u32 operation) - : _local_root_span_id(local_root_span_id), _label(label), - _operation(operation){}; + TraceRootEvent(u64 local_root_span_id, u64 parent_span_id, u64 start_ticks, + u32 label, u32 operation) + : _local_root_span_id(local_root_span_id), + _parent_span_id(parent_span_id), _start_ticks(start_ticks), + _label(label), _operation(operation){}; }; +typedef struct TaskBlockEvent { + u64 _start_ticks; + u64 _end_ticks; + u64 _span_id; + u64 _root_span_id; + uintptr_t _blocker; + u64 _unblocking_span_id; +} TaskBlockEvent; + typedef struct QueueTimeEvent { u64 _start; u64 _end; @@ -171,6 +184,7 @@ typedef struct QueueTimeEvent { u32 _origin; u32 _queueType; u32 _queueLength; + u64 _submitting_span_id; } QueueTimeEvent; #endif // _EVENT_H diff --git a/ddprof-lib/src/main/cpp/flightRecorder.cpp b/ddprof-lib/src/main/cpp/flightRecorder.cpp index 4adc5f727..aa001778f 100644 --- a/ddprof-lib/src/main/cpp/flightRecorder.cpp +++ b/ddprof-lib/src/main/cpp/flightRecorder.cpp @@ -1531,13 +1531,63 @@ void Recording::recordTraceRoot(Buffer *buf, int tid, TraceRootEvent *event) { flushIfNeeded(buf); int start = buf->skip(1); buf->putVar64(T_ENDPOINT); - buf->putVar64(TSC::ticks()); - buf->put8(0); + buf->putVar64(event->_start_ticks); + buf->putVar64(TSC::ticks() - event->_start_ticks); buf->putVar32(tid); buf->put8(0); buf->putVar32(event->_label); buf->putVar32(event->_operation); buf->putVar64(event->_local_root_span_id); + buf->putVar64(event->_parent_span_id); + writeEventSizePrefix(buf, start); + flushIfNeeded(buf); +} + +void Recording::recordTaskBlock(Buffer *buf, int tid, TaskBlockEvent *event) { + flushIfNeeded(buf); + int start = buf->skip(1); + buf->putVar64(T_TASK_BLOCK); + buf->putVar64(event->_start_ticks); + buf->putVar64(event->_end_ticks - event->_start_ticks); + buf->putVar32(tid); + buf->put8(0); + buf->putVar64(event->_span_id); + buf->putVar64(event->_root_span_id); + buf->putVar64((u64)event->_blocker); + buf->putVar64(event->_unblocking_span_id); + writeEventSizePrefix(buf, start); + flushIfNeeded(buf); +} + +void Recording::recordSpanNode(Buffer *buf, int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId, + u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource) { + // Convert epoch nanoseconds to JFR ticks so that standard JFR tooling (JMC, Mission + // Control) can correlate SpanNode events with other events on the recording timeline. + // _start_time is in microseconds; multiply by 1000 to get the recording epoch in nanos. + u64 start_epoch_nanos = _start_time * 1000ULL; + // Use signed arithmetic: a span that started in a previous JFR chunk has + // startNanos < start_epoch_nanos. Unsigned subtraction would wrap around and + // produce a huge positive u64, making (u64)(negative_double) undefined behaviour. + // With signed delta the result is a negative tick offset, placing the event just + // before the chunk boundary, which is the correct behaviour for pre-chunk spans. + long long delta_nanos = (long long)startNanos - (long long)start_epoch_nanos; + long long delta_ticks = (long long)((double)delta_nanos * TSC::frequency() / NANOTIME_FREQ); + u64 startTicks = (u64)((long long)_start_ticks + delta_ticks); + u64 durationTicks = (u64)((double)durationNanos * TSC::frequency() / NANOTIME_FREQ); + + flushIfNeeded(buf); + int start = buf->skip(1); + buf->putVar64(T_SPAN_NODE); + buf->putVar64(startTicks); // startTime (F_TIME_TICKS) + buf->putVar64(durationTicks); // duration (F_DURATION_TICKS) + buf->putVar32(tid); // eventThread (F_CPOOL) + buf->putVar64(spanId); + buf->putVar64(parentSpanId); + buf->putVar64(rootSpanId); + buf->putVar64(startNanos); // startNanos — epoch ns, used by backend extractor + buf->putVar64(durationNanos); // durationNanos — ns + buf->putVar32(encodedOperation); + buf->putVar32(encodedResource); writeEventSizePrefix(buf, start); flushIfNeeded(buf); } @@ -1553,7 +1603,28 @@ void Recording::recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event) { buf->putVar64(event->_scheduler); buf->putVar64(event->_queueType); buf->putVar64(event->_queueLength); - writeContext(buf, Contexts::get()); + // The schema declares fields in this order: + // spanId, localRootSpanId, submittingSpanId, contextAttr[0..n] + // writeContext() would emit spanId + localRootSpanId + contextAttrs[0..n] in one shot, + // leaving no room to insert submittingSpanId between localRootSpanId and contextAttrs. + // Inline the context fields manually so submittingSpanId lands at the correct position. + Context &ctx = Contexts::get(); + u64 spanId = 0, rootSpanId = 0; + u64 stored = ctx.checksum; + if (stored != 0) { + spanId = ctx.spanId; + rootSpanId = ctx.rootSpanId; + if (stored != Contexts::checksum(spanId, rootSpanId)) { + spanId = 0; + rootSpanId = 0; + } + } + buf->putVar64(spanId); // schema pos: spanId + buf->putVar64(rootSpanId); // schema pos: localRootSpanId + buf->putVar64(event->_submitting_span_id); // schema pos: submittingSpanId (CORRECT position) + for (size_t i = 0; i < Profiler::instance()->numContextAttributes(); i++) { + buf->putVar32(ctx.get_tag(i).value); // schema pos: contextAttr[i] + } writeEventSizePrefix(buf, start); flushIfNeeded(buf); } @@ -1745,6 +1816,7 @@ void FlightRecorder::recordTraceRoot(int lock_index, int tid, if (rec != nullptr) { Buffer *buf = rec->buffer(lock_index); rec->recordTraceRoot(buf, tid, event); + rec->addThread(lock_index, tid); } } } @@ -1757,6 +1829,37 @@ void FlightRecorder::recordQueueTime(int lock_index, int tid, if (rec != nullptr) { Buffer *buf = rec->buffer(lock_index); rec->recordQueueTime(buf, tid, event); + rec->addThread(lock_index, tid); + } + } +} + +void 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); + rec->addThread(lock_index, tid); + } + } +} + +void FlightRecorder::recordSpanNode(int lock_index, int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId, + u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource) { + OptionalSharedLockGuard locker(&_rec_lock); + if (locker.ownsLock()) { + Recording* rec = _rec; + if (rec != nullptr) { + Buffer *buf = rec->buffer(lock_index); + rec->recordSpanNode(buf, tid, spanId, parentSpanId, rootSpanId, startNanos, durationNanos, encodedOperation, encodedResource); + // Register the emitting thread in the JFR thread CPOOL so that JMC can resolve + // the eventThread reference. Without this, threads that emit SpanNode events but + // have no CPU/wall profiling samples in the current chunk are absent from the + // CPOOL, causing IMCThread to be null in the backend and threadId=0 ("unknown span"). + rec->addThread(lock_index, tid); } } } diff --git a/ddprof-lib/src/main/cpp/flightRecorder.h b/ddprof-lib/src/main/cpp/flightRecorder.h index c1ab88262..3fa4fdf13 100644 --- a/ddprof-lib/src/main/cpp/flightRecorder.h +++ b/ddprof-lib/src/main/cpp/flightRecorder.h @@ -278,6 +278,9 @@ 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 recordSpanNode(Buffer *buf, int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId, + u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource); void recordAllocation(RecordingBuffer *buf, int tid, u64 call_trace_id, AllocEvent *event); void recordHeapLiveObject(Buffer *buf, int tid, u64 call_trace_id, @@ -344,6 +347,9 @@ 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); + void recordTaskBlock(int lock_index, int tid, TaskBlockEvent *event); + void recordSpanNode(int lock_index, int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId, + u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource); bool active() const { return _rec != NULL; } diff --git a/ddprof-lib/src/main/cpp/javaApi.cpp b/ddprof-lib/src/main/cpp/javaApi.cpp index a354274c6..b47cd159a 100644 --- a/ddprof-lib/src/main/cpp/javaApi.cpp +++ b/ddprof-lib/src/main/cpp/javaApi.cpp @@ -199,8 +199,8 @@ Java_com_datadoghq_profiler_JavaProfiler_filterThreadRemove0(JNIEnv *env, extern "C" DLLEXPORT jboolean JNICALL Java_com_datadoghq_profiler_JavaProfiler_recordTrace0( - JNIEnv *env, jclass unused, jlong rootSpanId, jstring endpoint, - jstring operation, jint sizeLimit) { + JNIEnv *env, jclass unused, jlong rootSpanId, jlong parentSpanId, + jlong startTicks, jstring endpoint, jstring operation, jint sizeLimit) { JniString endpoint_str(env, endpoint); u32 endpointLabel = Profiler::instance()->stringLabelMap()->bounded_lookup( endpoint_str.c_str(), endpoint_str.length(), sizeLimit); @@ -212,13 +212,41 @@ Java_com_datadoghq_profiler_JavaProfiler_recordTrace0( operationLabel = Profiler::instance()->contextValueMap()->bounded_lookup( operation_str.c_str(), operation_str.length(), 1 << 16); } - TraceRootEvent event(rootSpanId, endpointLabel, operationLabel); + TraceRootEvent event(rootSpanId, (u64)parentSpanId, (u64)startTicks, + endpointLabel, operationLabel); int tid = ProfiledThread::currentTid(); Profiler::instance()->recordTraceRoot(tid, &event); } return acceptValue; } +extern "C" DLLEXPORT void JNICALL +Java_com_datadoghq_profiler_JavaProfiler_recordTaskBlock0( + JNIEnv *env, jclass unused, jlong startTicks, jlong endTicks, + jlong spanId, jlong rootSpanId, jlong blocker, jlong unblockingSpanId) { + TaskBlockEvent event; + event._start_ticks = (u64)startTicks; + event._end_ticks = (u64)endTicks; + event._span_id = (u64)spanId; + event._root_span_id = (u64)rootSpanId; + event._blocker = (uintptr_t)blocker; + event._unblocking_span_id = (u64)unblockingSpanId; + int tid = ProfiledThread::currentTid(); + Profiler::instance()->recordTaskBlock(tid, &event); +} + +extern "C" DLLEXPORT void JNICALL +Java_com_datadoghq_profiler_JavaProfiler_recordSpanNode0( + JNIEnv *env, jclass unused, + jlong spanId, jlong parentSpanId, jlong rootSpanId, + jlong startNanos, jlong durationNanos, + jint encodedOperation, jint encodedResource) { + int tid = ProfiledThread::currentTid(); + Profiler::instance()->recordSpanNode(tid, (u64)spanId, (u64)parentSpanId, (u64)rootSpanId, + (u64)startNanos, (u64)durationNanos, + (u32)encodedOperation, (u32)encodedResource); +} + extern "C" DLLEXPORT jint JNICALL Java_com_datadoghq_profiler_JavaProfiler_registerConstant0(JNIEnv *env, jclass unused, @@ -291,7 +319,8 @@ static int dictionarizeClassName(JNIEnv* env, jstring className) { extern "C" DLLEXPORT void JNICALL Java_com_datadoghq_profiler_JavaProfiler_recordQueueEnd0( JNIEnv *env, jclass unused, jlong startTime, jlong endTime, jstring task, - jstring scheduler, jthread origin, jstring queueType, jint queueLength) { + jstring scheduler, jthread origin, jstring queueType, jint queueLength, + jlong submittingSpanId) { int tid = ProfiledThread::currentTid(); if (tid < 0) { return; @@ -321,6 +350,7 @@ Java_com_datadoghq_profiler_JavaProfiler_recordQueueEnd0( event._origin = origin_tid; event._queueType = queue_type_offset; event._queueLength = queueLength; + event._submitting_span_id = (u64)submittingSpanId; Profiler::instance()->recordQueueTime(tid, &event); } diff --git a/ddprof-lib/src/main/cpp/jfrMetadata.cpp b/ddprof-lib/src/main/cpp/jfrMetadata.cpp index 6991a8a12..15d185a89 100644 --- a/ddprof-lib/src/main/cpp/jfrMetadata.cpp +++ b/ddprof-lib/src/main/cpp/jfrMetadata.cpp @@ -176,7 +176,32 @@ void JfrMetadata::initialize( << field("stackTrace", T_STACK_TRACE, "Stack Trace", F_CPOOL) << field("endpoint", T_STRING, "Endpoint", F_CPOOL) << field("operation", T_ATTRIBUTE_VALUE, "Operation", F_CPOOL) - << field("localRootSpanId", T_LONG, "Local Root Span ID")) + << field("localRootSpanId", T_LONG, "Local Root Span ID") + << field("parentSpanId", T_LONG, "Parent Span ID")) + + << (type("datadog.TaskBlock", T_TASK_BLOCK, "Task Block") + << category("Datadog") + << field("startTime", T_LONG, "Start Time", F_TIME_TICKS) + << field("duration", T_LONG, "Duration", F_DURATION_TICKS) + << field("eventThread", T_THREAD, "Event Thread", F_CPOOL) + << field("stackTrace", T_STACK_TRACE, "Stack Trace", F_CPOOL) + << field("spanId", T_LONG, "Span ID") + << field("localRootSpanId", T_LONG, "Local Root Span ID") + << field("blocker", T_LONG, "Blocker Object Hash", F_UNSIGNED) + << field("unblockingSpanId", T_LONG, "Unblocking Span ID")) + + << (type("datadog.SpanNode", T_SPAN_NODE, "Span Node") + << category("Datadog") + << field("startTime", T_LONG, "Start Time", F_TIME_TICKS) + << field("duration", T_LONG, "Duration", F_DURATION_TICKS) + << field("eventThread", T_THREAD, "Event Thread", F_CPOOL) + << field("spanId", T_LONG, "Span ID") + << field("parentSpanId", T_LONG, "Parent Span ID") + << field("localRootSpanId", T_LONG, "Local Root Span ID") + << field("startNanos", T_LONG, "Start Time (epoch ns)") + << field("durationNanos", T_LONG, "Duration (ns)") + << field("encodedOperation", T_INT, "Encoded Operation Name") + << field("encodedResource", T_INT, "Encoded Resource Name")) << (type("datadog.QueueTime", T_QUEUE_TIME, "Queue Time") << category("Datadog") @@ -189,7 +214,8 @@ void JfrMetadata::initialize( << field("queueType", T_CLASS, "Queue Type", F_CPOOL) << field("queueLength", T_INT, "Queue Length on Entry") << field("spanId", T_LONG, "Span ID") - << field("localRootSpanId", T_LONG, "Local Root Span ID") || + << field("localRootSpanId", T_LONG, "Local Root Span ID") + << field("submittingSpanId", T_LONG, "Submitting Span ID") || contextAttributes) << (type("datadog.HeapUsage", T_HEAP_USAGE, "JVM Heap Usage") diff --git a/ddprof-lib/src/main/cpp/jfrMetadata.h b/ddprof-lib/src/main/cpp/jfrMetadata.h index 77da96d3f..bc2997e2e 100644 --- a/ddprof-lib/src/main/cpp/jfrMetadata.h +++ b/ddprof-lib/src/main/cpp/jfrMetadata.h @@ -78,6 +78,8 @@ enum JfrType { T_DATADOG_CLASSREF_CACHE = 124, T_DATADOG_COUNTER = 125, T_UNWIND_FAILURE = 126, + T_TASK_BLOCK = 127, + T_SPAN_NODE = 128, T_ANNOTATION = 200, T_LABEL = 201, T_CATEGORY = 202, diff --git a/ddprof-lib/src/main/cpp/profiler.cpp b/ddprof-lib/src/main/cpp/profiler.cpp index cbcfb7284..a3f58bcc6 100644 --- a/ddprof-lib/src/main/cpp/profiler.cpp +++ b/ddprof-lib/src/main/cpp/profiler.cpp @@ -940,6 +940,60 @@ void Profiler::recordQueueTime(int tid, QueueTimeEvent *event) { _locks[lock_index].unlock(); } +void JNICALL Profiler::MonitorContendedEnterCallback(jvmtiEnv *jvmti, JNIEnv *jni, + jthread thread, jobject object) { + ProfiledThread* thrd = ProfiledThread::current(); + if (thrd == nullptr) return; + Context& ctx = Contexts::get(); + if (ctx.spanId == 0) return; + thrd->_monitor_block.start_ticks = TSC::ticks(); + thrd->_monitor_block.span_id = ctx.spanId; + thrd->_monitor_block.root_span_id = ctx.rootSpanId; + thrd->_monitor_block.obj_addr = (uintptr_t)(void*)object; + thrd->_monitor_block.unblocking_span_id = VMThread::monitorOwnerSpanId((const void*)object); +} + +void JNICALL Profiler::MonitorContendedEnteredCallback(jvmtiEnv *jvmti, JNIEnv *jni, + jthread thread, jobject object) { + ProfiledThread* thrd = ProfiledThread::current(); + if (thrd == nullptr || thrd->_monitor_block.obj_addr == 0) return; + + TaskBlockEvent event; + event._start_ticks = thrd->_monitor_block.start_ticks; + event._end_ticks = TSC::ticks(); + event._span_id = thrd->_monitor_block.span_id; + event._root_span_id = thrd->_monitor_block.root_span_id; + event._blocker = thrd->_monitor_block.obj_addr; + event._unblocking_span_id = thrd->_monitor_block.unblocking_span_id; + + thrd->_monitor_block.obj_addr = 0; + + instance()->recordTaskBlock(ProfiledThread::currentTid(), &event); +} + +void Profiler::recordTaskBlock(int tid, TaskBlockEvent *event) { + u32 lock_index = getLockIndex(tid); + if (!_locks[lock_index].tryLock() && + !_locks[lock_index = (lock_index + 1) % CONCURRENCY_LEVEL].tryLock() && + !_locks[lock_index = (lock_index + 2) % CONCURRENCY_LEVEL].tryLock()) { + return; + } + _jfr.recordTaskBlock(lock_index, tid, event); + _locks[lock_index].unlock(); +} + +void Profiler::recordSpanNode(int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId, + u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource) { + u32 lock_index = getLockIndex(tid); + if (!_locks[lock_index].tryLock() && + !_locks[lock_index = (lock_index + 1) % CONCURRENCY_LEVEL].tryLock() && + !_locks[lock_index = (lock_index + 2) % CONCURRENCY_LEVEL].tryLock()) { + return; + } + _jfr.recordSpanNode(lock_index, tid, spanId, parentSpanId, rootSpanId, startNanos, durationNanos, encodedOperation, encodedResource); + _locks[lock_index].unlock(); +} + void Profiler::recordExternalSample(u64 weight, int tid, int num_frames, ASGCT_CallFrame *frames, bool truncated, jint event_type, Event *event) { diff --git a/ddprof-lib/src/main/cpp/profiler.h b/ddprof-lib/src/main/cpp/profiler.h index 285c64805..53e513490 100644 --- a/ddprof-lib/src/main/cpp/profiler.h +++ b/ddprof-lib/src/main/cpp/profiler.h @@ -375,6 +375,9 @@ class alignas(alignof(SpinLock)) Profiler { void recordWallClockEpoch(int tid, WallClockEpochEvent *event); void recordTraceRoot(int tid, TraceRootEvent *event); void recordQueueTime(int tid, QueueTimeEvent *event); + void recordTaskBlock(int tid, TaskBlockEvent *event); + void recordSpanNode(int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId, + u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource); void writeLog(LogLevel level, const char *message); void writeLog(LogLevel level, const char *message, size_t len); void writeDatadogProfilerSetting(int tid, int length, const char *name, @@ -420,6 +423,11 @@ class alignas(alignof(SpinLock)) Profiler { instance()->onThreadEnd(jvmti, jni, thread); } + static void JNICALL MonitorContendedEnterCallback(jvmtiEnv *jvmti, JNIEnv *jni, + jthread thread, jobject object); + static void JNICALL MonitorContendedEnteredCallback(jvmtiEnv *jvmti, JNIEnv *jni, + jthread thread, jobject object); + // Keep backward compatibility with the upstream async-profiler inline CodeCache* findLibraryByAddress(const void *address) { #ifdef DEBUG diff --git a/ddprof-lib/src/main/cpp/thread.cpp b/ddprof-lib/src/main/cpp/thread.cpp index 5457f3fab..b73a0dea4 100644 --- a/ddprof-lib/src/main/cpp/thread.cpp +++ b/ddprof-lib/src/main/cpp/thread.cpp @@ -133,6 +133,15 @@ ProfiledThread *ProfiledThread::currentSignalSafe() { return __atomic_load_n(&_tls_key_initialized, __ATOMIC_ACQUIRE) ? (ProfiledThread *)pthread_getspecific(_tls_key) : nullptr; } +ProfiledThread* ProfiledThread::findByTid(int tid) { + int size = __atomic_load_n(&_running_buffer_pos, __ATOMIC_ACQUIRE); + for (int i = 0; i < size; i++) { + ProfiledThread* t = _buffer[i]; + if (t != nullptr && t->_tid == tid) return t; + } + return nullptr; +} + int ProfiledThread::popFreeSlot() { int current_top; int new_top; diff --git a/ddprof-lib/src/main/cpp/thread.h b/ddprof-lib/src/main/cpp/thread.h index 4cb12d0ca..d991d0558 100644 --- a/ddprof-lib/src/main/cpp/thread.h +++ b/ddprof-lib/src/main/cpp/thread.h @@ -74,12 +74,25 @@ class ProfiledThread : public ThreadLocalData { ProfiledThread(int buffer_pos, int tid) : ThreadLocalData(), _pc(0), _sp(0), _span_id(0), _root_span_id(0), _crash_depth(0), _buffer_pos(buffer_pos), _tid(tid), _cpu_epoch(0), - _wall_epoch(0), _call_trace_id(0), _recording_epoch(0), _misc_flags(0), _filter_slot_id(-1), _ctx_tls_initialized(false), _crash_protection_active(false), _ctx_tls_ptr(nullptr) {}; + _wall_epoch(0), _call_trace_id(0), _recording_epoch(0), _misc_flags(0), _filter_slot_id(-1), _ctx_tls_initialized(false), _crash_protection_active(false), _ctx_tls_ptr(nullptr), _monitor_block{} {}; void releaseFromBuffer(); public: + // In-flight state for monitor contention tracking (MonitorContendedEnter → + // MonitorContendedEntered). Keyed on obj_addr == 0 meaning "no contention in + // progress". Object addresses are never 0 in the JVM. + struct MonitorBlockState { + u64 start_ticks; + u64 span_id; + u64 root_span_id; + uintptr_t obj_addr; + u64 unblocking_span_id; + }; + MonitorBlockState _monitor_block{}; + static ProfiledThread *forTid(int tid) { return new ProfiledThread(-1, tid); } + static ProfiledThread* findByTid(int tid); static ProfiledThread *inBuffer(int buffer_pos) { return new ProfiledThread(buffer_pos, 0); } diff --git a/ddprof-lib/src/main/cpp/vmEntry.cpp b/ddprof-lib/src/main/cpp/vmEntry.cpp index 272c6e053..0c040200a 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.cpp +++ b/ddprof-lib/src/main/cpp/vmEntry.cpp @@ -435,6 +435,8 @@ bool VM::initProfilerBridge(JavaVM *vm, bool attach) { callbacks.SampledObjectAlloc = ObjectSampler::SampledObjectAlloc; callbacks.GarbageCollectionFinish = LivenessTracker::GarbageCollectionFinish; callbacks.NativeMethodBind = VMStructs::NativeMethodBind; + callbacks.MonitorContendedEnter = Profiler::MonitorContendedEnterCallback; + callbacks.MonitorContendedEntered = Profiler::MonitorContendedEnteredCallback; _jvmti->SetEventCallbacks(&callbacks, sizeof(callbacks)); _jvmti->SetEventNotificationMode(JVMTI_ENABLE, JVMTI_EVENT_VM_DEATH, NULL); @@ -445,6 +447,8 @@ bool VM::initProfilerBridge(JavaVM *vm, bool attach) { JVMTI_EVENT_DYNAMIC_CODE_GENERATED, NULL); _jvmti->SetEventNotificationMode(JVMTI_ENABLE, JVMTI_EVENT_NATIVE_METHOD_BIND, NULL); + _jvmti->SetEventNotificationMode(JVMTI_ENABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTER, NULL); + _jvmti->SetEventNotificationMode(JVMTI_ENABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTERED, NULL); if (hotspot_version() == 0 || !CodeHeap::available()) { // Workaround for JDK-8173361: avoid CompiledMethodLoad events when possible diff --git a/ddprof-lib/src/main/cpp/vmStructs.cpp b/ddprof-lib/src/main/cpp/vmStructs.cpp index 00ed4f991..3d8700683 100644 --- a/ddprof-lib/src/main/cpp/vmStructs.cpp +++ b/ddprof-lib/src/main/cpp/vmStructs.cpp @@ -9,6 +9,8 @@ #include #include "vmStructs.h" #include "vmEntry.h" +#include "context.h" +#include "thread.h" #include "j9Ext.h" #include "jniHelper.h" #include "jvmHeap.h" @@ -675,6 +677,38 @@ int VMThread::osThreadId() { return -1; } +u64 VMThread::monitorOwnerSpanId(const void* object) { + if (_monitor_owner_offset == 0) return 0; + + // Read the mark word from the object header. + uintptr_t mark = (uintptr_t)SafeAccess::load((void**)object, nullptr); + + // At MonitorContendedEnter time the monitor must be inflated (MONITOR_BIT set + // in bits [1:0]). Guard defensively for any narrow races. + if ((mark & 0x3) != MONITOR_BIT) return 0; + + // Extract ObjectMonitor* (pointer stored in bits 63:2, 4-byte aligned). + const char* monitor = (const char*)(mark & ~(uintptr_t)0x3); + + // Read ObjectMonitor::_owner (JavaThread*). + const char* owner_thread = (const char*)SafeAccess::load( + (void**)(monitor + _monitor_owner_offset), nullptr); + if (owner_thread == nullptr) return 0; + + // JavaThread* -> OS thread ID via existing VMThread infrastructure. + int owner_tid = VMThread::cast(owner_thread)->osThreadId(); + if (owner_tid <= 0) return 0; + + // OS thread ID -> ProfiledThread* -> span ID. + ProfiledThread* owner = ProfiledThread::findByTid(owner_tid); + if (owner == nullptr || !owner->isContextTlsInitialized()) return 0; + + Context* ctx = owner->getContextTlsPtr(); + if (ctx == nullptr) return 0; + + return ctx->spanId; // volatile u64, safe to read cross-thread on x86/aarch64 +} + JNIEnv* VMThread::jni() { if (_env_offset < 0) { return VM::jni(); // fallback for non-HotSpot JVM diff --git a/ddprof-lib/src/main/cpp/vmStructs.h b/ddprof-lib/src/main/cpp/vmStructs.h index 9a51fdc08..6a47edd03 100644 --- a/ddprof-lib/src/main/cpp/vmStructs.h +++ b/ddprof-lib/src/main/cpp/vmStructs.h @@ -196,6 +196,9 @@ typedef void* address; field(_osthread_id_offset, offset, MATCH_SYMBOLS("_thread_id")) \ field_with_version(_osthread_state_offset, offset, 10, MAX_VERSION, MATCH_SYMBOLS("_state")) \ type_end() \ + type_begin(VMObjectMonitor, MATCH_SYMBOLS("ObjectMonitor")) \ + field(_monitor_owner_offset, offset, MATCH_SYMBOLS("_owner")) \ + type_end() \ type_begin(VMThreadShadow, MATCH_SYMBOLS("ThreadShadow")) \ field(_thread_exception_offset, offset, MATCH_SYMBOLS("_exception_file")) \ type_end() \ @@ -714,6 +717,11 @@ DECLARE(VMThread) int osThreadId(); + // Returns the span ID of the JavaThread currently owning the given monitor + // object, reading ObjectMonitor::_owner directly without a safepoint. + // Returns 0 if the owner cannot be determined. + static u64 monitorOwnerSpanId(const void* object); + JNIEnv* jni(); const void** vtable() { diff --git a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java index c49b4479e..1a2fb55f8 100644 --- a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java +++ b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java @@ -161,11 +161,26 @@ public String execute(String command) throws IllegalArgumentException, IllegalSt return execute0(command); } + /** + * Records the completion of the trace root with causal DAG metadata. + * @param rootSpanId the local root span ID + * @param parentSpanId the parent span ID (0 if none) + * @param startTicks TSC tick captured at span start via {@link #getCurrentTicks()} + * @param endpoint the endpoint/resource name + * @param operation the operation name + * @param sizeLimit max number of distinct endpoints to track + */ + public boolean recordTraceRoot(long rootSpanId, long parentSpanId, long startTicks, + String endpoint, String operation, int sizeLimit) { + return recordTrace0(rootSpanId, parentSpanId, startTicks, endpoint, operation, sizeLimit); + } + /** * Records the completion of the trace root */ + @Deprecated public boolean recordTraceRoot(long rootSpanId, String endpoint, String operation, int sizeLimit) { - return recordTrace0(rootSpanId, endpoint, operation, sizeLimit); + return recordTrace0(rootSpanId, 0L, 0L, endpoint, operation, sizeLimit); } /** @@ -173,7 +188,27 @@ public boolean recordTraceRoot(long rootSpanId, String endpoint, String operatio */ @Deprecated public boolean recordTraceRoot(long rootSpanId, String endpoint, int sizeLimit) { - return recordTrace0(rootSpanId, endpoint, null, sizeLimit); + return recordTrace0(rootSpanId, 0L, 0L, endpoint, null, sizeLimit); + } + + /** + * Records a blocking interval for the current span. + * @param startTicks TSC tick at block entry + * @param endTicks TSC tick at block exit + * @param spanId the span that was blocked + * @param rootSpanId the local root span ID + * @param blocker identity hash code of the blocking object + */ + public void recordTaskBlock(long startTicks, long endTicks, + long spanId, long rootSpanId, long blocker, long unblockingSpanId) { + recordTaskBlock0(startTicks, endTicks, spanId, rootSpanId, blocker, unblockingSpanId); + } + + public void recordSpanNode(long spanId, long parentSpanId, long rootSpanId, + long startNanos, long durationNanos, + int encodedOperation, int encodedResource) { + recordSpanNode0(spanId, parentSpanId, rootSpanId, startNanos, durationNanos, + encodedOperation, encodedResource); } /** @@ -279,8 +314,9 @@ public void recordQueueTime(long startTicks, Class scheduler, Class queueType, int queueLength, - Thread origin) { - recordQueueEnd0(startTicks, endTicks, task.getName(), scheduler.getName(), origin, queueType.getName(), queueLength); + Thread origin, + long submittingSpanId) { + recordQueueEnd0(startTicks, endTicks, task.getName(), scheduler.getName(), origin, queueType.getName(), queueLength, submittingSpanId); } /** @@ -323,7 +359,11 @@ private static ThreadContext initializeThreadContext() { private static native int getTid0(); - private static native boolean recordTrace0(long rootSpanId, String endpoint, String operation, int sizeLimit); + private static native boolean recordTrace0(long rootSpanId, long parentSpanId, long startTicks, String endpoint, String operation, int sizeLimit); + + private static native void recordTaskBlock0(long startTicks, long endTicks, long spanId, long rootSpanId, long blocker, long unblockingSpanId); + + private static native void recordSpanNode0(long spanId, long parentSpanId, long rootSpanId, long startNanos, long durationNanos, int encodedOperation, int encodedResource); private static native int registerConstant0(String value); @@ -335,7 +375,7 @@ private static ThreadContext initializeThreadContext() { private static native void recordSettingEvent0(String name, String value, String unit); - private static native void recordQueueEnd0(long startTicks, long endTicks, String task, String scheduler, Thread origin, String queueType, int queueLength); + private static native void recordQueueEnd0(long startTicks, long endTicks, String task, String scheduler, Thread origin, String queueType, int queueLength, long submittingSpanId); private static native long currentTicks0(); diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/queue/QueueTimeTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/queue/QueueTimeTest.java index f16912930..f023eb4da 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/queue/QueueTimeTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/queue/QueueTimeTest.java @@ -45,12 +45,60 @@ public void run() { profiler.setContext(1, 2); long now = profiler.getCurrentTicks(); if (profiler.isThresholdExceeded(9, start, now)) { - profiler.recordQueueTime(start, now, getClass(), QueueTimeTest.class, ArrayBlockingQueue.class, 10, origin); + profiler.recordQueueTime(start, now, getClass(), QueueTimeTest.class, ArrayBlockingQueue.class, 10, origin, 0L); } profiler.clearContext(); } } + /** + * Regression test for the QueueTime field serialization order bug: submittingSpanId was + * written before writeContext(), placing it at field position 9 (where the schema expects + * the consuming spanId). This caused the backend to read completely wrong span IDs. + *

+ * Verifies that all three span-ID fields (spanId, localRootSpanId, submittingSpanId) round-trip + * correctly when they are all distinct and non-zero. + */ + @Test + public void testQueueTimeFieldOrder() throws Exception { + IAttribute submittingSpanIdAttr = attr("submittingSpanId", "", "", NUMBER); + + Thread origin = Thread.currentThread(); + origin.setName("origin-field-order"); + // Use distinct, non-zero values so any field-order swap is detectable: + // consuming spanId=7, rootSpanId=42 (set via setContext), submittingSpanId=99. + long start = profiler.getCurrentTicks(); + Runnable worker = () -> { + profiler.setContext(7, 42); + long now = profiler.getCurrentTicks(); + profiler.recordQueueTime(start, now, QueueTimeTest.class, QueueTimeTest.class, + ArrayBlockingQueue.class, 1, origin, 99L); + profiler.clearContext(); + }; + Thread thread = new Thread(worker, "destination-field-order"); + Thread.sleep(10); + thread.start(); + thread.join(); + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.QueueTime"); + boolean found = false; + for (IItemIterable it : events) { + IMemberAccessor spanIdAccessor = SPAN_ID.getAccessor(it.getType()); + IMemberAccessor rootSpanIdAccessor = LOCAL_ROOT_SPAN_ID.getAccessor(it.getType()); + IMemberAccessor submittingAccessor = submittingSpanIdAttr.getAccessor(it.getType()); + for (IItem item : it) { + if (spanIdAccessor.getMember(item).longValue() == 7) { + found = true; + assertEquals(7, spanIdAccessor.getMember(item).longValue(), "spanId must be the consuming span (not the submitting span)"); + assertEquals(42, rootSpanIdAccessor.getMember(item).longValue(), "localRootSpanId must be the root"); + assertEquals(99, submittingAccessor.getMember(item).longValue(), "submittingSpanId must be the submitting span (not the root)"); + } + } + } + assertTrue(found, "Expected at least one QueueTime event with spanId=7"); + } + @Test public void testRecordQueueTime() throws Exception { Thread origin = Thread.currentThread();