8366082: Improve queue size computation in CPU-time sampler

src/hotspot/share/jfr/periodic/sampling/jfrCPUTimeThreadSampler.cpp

@@ -25,7 +25,6 @@
 #include "jfr/periodic/sampling/jfrCPUTimeThreadSampler.hpp"
 #include "logging/log.hpp"
 
-
 #if defined(LINUX)
 #include "jfr/periodic/sampling/jfrThreadSampling.hpp"
 #include "jfr/support/jfrThreadLocal.hpp"
@@ -67,7 +66,7 @@ static JavaThread* get_java_thread_if_valid() {
 }
 
 JfrCPUTimeTraceQueue::JfrCPUTimeTraceQueue(u4 capacity) :
-   _data(nullptr), _capacity(capacity), _head(0), _lost_samples(0) {
+   _data(nullptr), _capacity(capacity), _head(0), _lost_samples(0), _lost_samples_due_to_queue_full(0) {
   if (capacity != 0) {
     _data = JfrCHeapObj::new_array<JfrCPUTimeSampleRequest>(capacity);
   }
@@ -110,10 +109,13 @@ void JfrCPUTimeTraceQueue::set_size(u4 size) {
 }
 
 u4 JfrCPUTimeTraceQueue::capacity() const {
-  return _capacity;
+  return Atomic::load_acquire(&_capacity);
 }
 
 void JfrCPUTimeTraceQueue::set_capacity(u4 capacity) {
+  if (capacity == Atomic::load(&_capacity)) {
+    return;
+  }
   _head = 0;
   if (_data != nullptr) {
     assert(_capacity != 0, "invariant");
@@ -124,7 +126,7 @@ void JfrCPUTimeTraceQueue::set_capacity(u4 capacity) {
   } else {
     _data = nullptr;
   }
-  _capacity = capacity;
+  Atomic::release_store(&_capacity, capacity);
 }
 
 bool JfrCPUTimeTraceQueue::is_empty() const {
@@ -140,28 +142,51 @@ void JfrCPUTimeTraceQueue::increment_lost_samples() {
   Atomic::inc(&_lost_samples);
 }
 
+void JfrCPUTimeTraceQueue::increment_lost_samples_due_to_queue_full() {
+  Atomic::inc(&_lost_samples_due_to_queue_full);
+}
+
 u4 JfrCPUTimeTraceQueue::get_and_reset_lost_samples() {
   return Atomic::xchg(&_lost_samples, (u4)0);
 }
 
-void JfrCPUTimeTraceQueue::resize(u4 capacity) {
-  if (capacity != _capacity) {
-    set_capacity(capacity);
-  }
+u4 JfrCPUTimeTraceQueue::get_and_reset_lost_samples_due_to_queue_full() {
+  return Atomic::xchg(&_lost_samples_due_to_queue_full, (u4)0);
 }
 
-void JfrCPUTimeTraceQueue::resize_for_period(u4 period_millis) {
-  u4 capacity = CPU_TIME_QUEUE_CAPACITY;
-  if (period_millis > 0 && period_millis < 10) {
-    capacity = (u4) ((double) capacity * 10 / period_millis);
-  }
-  resize(capacity);
+void JfrCPUTimeTraceQueue::init() {
+  set_capacity(JfrCPUTimeTraceQueue::CPU_TIME_QUEUE_INITIAL_CAPACITY);
 }
 
 void JfrCPUTimeTraceQueue::clear() {
   Atomic::release_store(&_head, (u4)0);
 }
 
+void JfrCPUTimeTraceQueue::resize_if_needed() {
+  u4 lost_samples_due_to_queue_full = get_and_reset_lost_samples_due_to_queue_full();
+  if (lost_samples_due_to_queue_full == 0) {
+    return;
+  }
+  u4 capacity = Atomic::load(&_capacity);
+  if (capacity < CPU_TIME_QUEUE_MAX_CAPACITY) {
+    float ratio = (float)lost_samples_due_to_queue_full / (float)capacity;
+    int factor = 1;
+    if (ratio > 8) { // idea is to quickly scale the queue in the worst case
+      factor = ratio;
+    } else if (ratio > 2) {
+      factor = 8;
+    } else if (ratio > 0.5) {
+      factor = 4;
+    } else if (ratio > 0.01) {
+      factor = 2;
+    }
+    if (factor > 1) {
+      u4 new_capacity = MIN2(CPU_TIME_QUEUE_MAX_CAPACITY, capacity * factor);
+      set_capacity(new_capacity);
+    }
+  }
+}
+
 // A throttle is either a rate or a fixed period
 class JfrCPUSamplerThrottle {
 
@@ -205,6 +230,8 @@ class JfrCPUSamplerThread : public NonJavaThread {
   volatile bool _is_async_processing_of_cpu_time_jfr_requests_triggered;
   volatile bool _warned_about_timer_creation_failure;
   volatile bool _signal_handler_installed;
+  DEBUG_ONLY(volatile bool _out_of_stack_walking_enabled;)
+  DEBUG_ONLY(volatile u8 _out_of_stack_walking_iterations;)
 
   static const u4 STOP_SIGNAL_BIT = 0x80000000;
 
@@ -250,6 +277,16 @@ class JfrCPUSamplerThread : public NonJavaThread {
   void stop_timer();
 
   void trigger_async_processing_of_cpu_time_jfr_requests();
+
+  #ifdef ASSERT
+  void set_out_of_stack_walking_enabled(bool runnable) {
+    Atomic::release_store(&_out_of_stack_walking_enabled, runnable);
+  }
+
+  u8 out_of_stack_walking_iterations() const {
+    return Atomic::load(&_out_of_stack_walking_iterations);
+  }
+  #endif
 };
 
 JfrCPUSamplerThread::JfrCPUSamplerThread(JfrCPUSamplerThrottle& throttle) :
@@ -275,7 +312,7 @@ void JfrCPUSamplerThread::on_javathread_create(JavaThread* thread) {
   }
   JfrThreadLocal* tl = thread->jfr_thread_local();
   assert(tl != nullptr, "invariant");
-  tl->cpu_time_jfr_queue().resize_for_period(_current_sampling_period_ns / 1000000);
+  tl->cpu_time_jfr_queue().init();
   timer_t timerid;
   if (create_timer_for_thread(thread, timerid)) {
     tl->set_cpu_timer(&timerid);
@@ -294,12 +331,14 @@ void JfrCPUSamplerThread::on_javathread_terminate(JavaThread* thread) {
   if (timer == nullptr) {
     return; // no timer was created for this thread
   }
+  tl->acquire_cpu_time_jfr_dequeue_lock();
   tl->unset_cpu_timer();
   tl->deallocate_cpu_time_jfr_queue();
   s4 lost_samples = tl->cpu_time_jfr_queue().lost_samples();
   if (lost_samples > 0) {
     JfrCPUTimeThreadSampling::send_lost_event(JfrTicks::now(), JfrThreadLocal::thread_id(thread), lost_samples);
   }
+  tl->release_cpu_time_jfr_queue_lock();
 }
 
 void JfrCPUSamplerThread::start_thread() {
@@ -352,10 +391,12 @@ void JfrCPUSamplerThread::run() {
       recompute_period_if_needed();
       last_recompute_check = os::javaTimeNanos();
     }
-
-    if (Atomic::cmpxchg(&_is_async_processing_of_cpu_time_jfr_requests_triggered, true, false)) {
-      stackwalk_threads_in_native();
-    }
+    DEBUG_ONLY(if (Atomic::load_acquire(&_out_of_stack_walking_enabled)) {)
+      if (Atomic::cmpxchg(&_is_async_processing_of_cpu_time_jfr_requests_triggered, true, false)) {
+        DEBUG_ONLY(Atomic::inc(&_out_of_stack_walking_iterations);)
+        stackwalk_threads_in_native();
+      }
+    DEBUG_ONLY(})
     os::naked_sleep(100);
   }
 }
@@ -545,6 +586,21 @@ void JfrCPUTimeThreadSampling::handle_timer_signal(siginfo_t* info, void* contex
   _sampler->decrement_signal_handler_count();
 }
 
+#ifdef ASSERT
+void JfrCPUTimeThreadSampling::set_out_of_stack_walking_enabled(bool runnable) {
+  if (_instance != nullptr && _instance->_sampler != nullptr) {
+    _instance->_sampler->set_out_of_stack_walking_enabled(runnable);
+  }
+}
+
+u8 JfrCPUTimeThreadSampling::out_of_stack_walking_iterations() {
+  if (_instance != nullptr && _instance->_sampler != nullptr) {
+    return _instance->_sampler->out_of_stack_walking_iterations();
+  }
+  return 0;
+}
+#endif
+
 void JfrCPUSamplerThread::sample_thread(JfrSampleRequest& request, void* ucontext, JavaThread* jt, JfrThreadLocal* tl, JfrTicks& now) {
   JfrSampleRequestBuilder::build_cpu_time_sample_request(request, ucontext, jt, jt->jfr_thread_local(), now);
 }
@@ -590,6 +646,7 @@ void JfrCPUSamplerThread::handle_timer_signal(siginfo_t* info, void* context) {
     }
   } else {
     queue.increment_lost_samples();
+    queue.increment_lost_samples_due_to_queue_full();
   }
 
   if (jt->thread_state() == _thread_in_native) {
@@ -807,4 +864,10 @@ void JfrCPUTimeThreadSampling::on_javathread_create(JavaThread* thread) {
 void JfrCPUTimeThreadSampling::on_javathread_terminate(JavaThread* thread) {
 }
 
+#ifdef ASSERT
+static void set_out_of_stack_walking_enabled(bool runnable) {
+  warn();
+}
+#endif
+
 #endif // defined(LINUX) && defined(INCLUDE_JFR)

src/hotspot/share/jfr/periodic/sampling/jfrCPUTimeThreadSampler.hpp

@@ -50,12 +50,15 @@ class JfrCPUTimeTraceQueue {
   static const u4 CPU_TIME_QUEUE_CAPACITY = 500;
 
   JfrCPUTimeSampleRequest* _data;
-  u4 _capacity;
+  volatile u4 _capacity;
   // next unfilled index
   volatile u4 _head;
 
   volatile u4 _lost_samples;
+  volatile u4 _lost_samples_due_to_queue_full;
 
+  static const u4 CPU_TIME_QUEUE_INITIAL_CAPACITY = 20;
+  static const u4 CPU_TIME_QUEUE_MAX_CAPACITY     = 2000;
 public:
   JfrCPUTimeTraceQueue(u4 capacity);
 
@@ -81,12 +84,17 @@ class JfrCPUTimeTraceQueue {
 
   void increment_lost_samples();
 
+  void increment_lost_samples_due_to_queue_full();
+
   // returns the previous lost samples count
   u4 get_and_reset_lost_samples();
 
-  void resize(u4 capacity);
+  u4 get_and_reset_lost_samples_due_to_queue_full();
+
+  void resize_if_needed();
 
-  void resize_for_period(u4 period_millis);
+  // init the queue capacity
+  void init();
 
   void clear();
 
@@ -130,6 +138,10 @@ class JfrCPUTimeThreadSampling : public JfrCHeapObj {
   static void send_lost_event(const JfrTicks& time, traceid tid, s4 lost_samples);
 
   static void trigger_async_processing_of_cpu_time_jfr_requests();
+
+  DEBUG_ONLY(static void set_out_of_stack_walking_enabled(bool runnable);)
+
+  DEBUG_ONLY(static u8 out_of_stack_walking_iterations();)
 };
 
 #else
@@ -150,6 +162,8 @@ class JfrCPUTimeThreadSampling : public JfrCHeapObj {
 
   static void on_javathread_create(JavaThread* thread);
   static void on_javathread_terminate(JavaThread* thread);
+  DEBUG_ONLY(static void set_out_of_stack_walking_enabled(bool runnable));
+  DEBUG_ONLY(static u8 out_of_stack_walking_iterations();)
 };
 
 #endif // defined(LINUX)

src/hotspot/share/jfr/periodic/sampling/jfrThreadSampling.cpp

@@ -368,6 +368,7 @@ static void drain_enqueued_cpu_time_requests(const JfrTicks& now, JfrThreadLocal
   tl->set_has_cpu_time_jfr_requests(false);
   if (queue.lost_samples() > 0) {
     JfrCPUTimeThreadSampling::send_lost_event( now, JfrThreadLocal::thread_id(jt), queue.get_and_reset_lost_samples());
+    queue.resize_if_needed();
   }
   if (lock) {
     tl->release_cpu_time_jfr_queue_lock();

src/hotspot/share/jfr/support/jfrThreadLocal.cpp

@@ -656,7 +656,7 @@ JfrCPUTimeTraceQueue& JfrThreadLocal::cpu_time_jfr_queue() {
 }
 
 void JfrThreadLocal::deallocate_cpu_time_jfr_queue() {
-  cpu_time_jfr_queue().resize(0);
+  cpu_time_jfr_queue().set_capacity(0);
 }
 
 void JfrThreadLocal::set_do_async_processing_of_cpu_time_jfr_requests(bool wants) {

src/hotspot/share/prims/whitebox.cpp

@@ -46,6 +46,7 @@
 #include "gc/shared/concurrentGCBreakpoints.hpp"
 #include "gc/shared/gcConfig.hpp"
 #include "gc/shared/genArguments.hpp"
+#include "jfr/periodic/sampling/jfrCPUTimeThreadSampler.hpp"
 #include "jvm.h"
 #include "jvmtifiles/jvmtiEnv.hpp"
 #include "logging/log.hpp"
@@ -2657,6 +2658,32 @@ WB_ENTRY(void, WB_WaitUnsafe(JNIEnv* env, jobject wb, jint time))
     os::naked_short_sleep(time);
 WB_END
 
+WB_ENTRY(void, WB_BusyWait(JNIEnv* env, jobject wb, jint time))
+  ThreadToNativeFromVM  ttn(thread);
+  u8 start = os::current_thread_cpu_time();
+  u8 target_duration = time * (u8)1000000;
+  while (os::current_thread_cpu_time() - start < target_duration) {
+    for (volatile int i = 0; i < 1000000; i++);
+  }
+WB_END
+
+WB_ENTRY(jboolean, WB_CPUSamplerSetOutOfStackWalking(JNIEnv* env, jobject wb, jboolean enable))
+  #if defined(ASSERT) && INCLUDE_JFR && defined(LINUX)
+    JfrCPUTimeThreadSampling::set_out_of_stack_walking_enabled(enable == JNI_TRUE);
+    return JNI_TRUE;
+  #else
+    return JNI_FALSE;
+  #endif
+WB_END
+
+WB_ENTRY(jlong, WB_CPUSamplerOutOfStackWalkingIterations(JNIEnv* env, jobject wb))
+  #if defined(ASSERT) && INCLUDE_JFR && defined(LINUX)
+    return (jlong)JfrCPUTimeThreadSampling::out_of_stack_walking_iterations();
+  #else
+    return 0;
+  #endif
+WB_END
+
 WB_ENTRY(jstring, WB_GetLibcName(JNIEnv* env, jobject o))
   ThreadToNativeFromVM ttn(thread);
   jstring info_string = env->NewStringUTF(XSTR(LIBC));
@@ -3010,6 +3037,9 @@ static JNINativeMethod methods[] = {
 
   {CC"isJVMTIIncluded", CC"()Z",                      (void*)&WB_IsJVMTIIncluded},
   {CC"waitUnsafe", CC"(I)V",                          (void*)&WB_WaitUnsafe},
+  {CC"busyWait", CC"(I)V",                            (void*)&WB_BusyWait},
+  {CC"cpuSamplerSetOutOfStackWalking", CC"(Z)Z",      (void*)&WB_CPUSamplerSetOutOfStackWalking},
+  {CC"cpuSamplerOutOfStackWalkingIterations", CC"()J",(void*)&WB_CPUSamplerOutOfStackWalkingIterations},
   {CC"getLibcName",     CC"()Ljava/lang/String;",     (void*)&WB_GetLibcName},
 
   {CC"pinObject",       CC"(Ljava/lang/Object;)V",    (void*)&WB_PinObject},