8343938: TestStressBailout triggers "Should not be locked when freed" assert

Reviewed-by: kvn, thartmann

src/hotspot/share/compiler/compileBroker.cpp

@@ -365,13 +365,24 @@ void CompileQueue::free_all() {
   while (next != nullptr) {
     CompileTask* current = next;
     next = current->next();
+    bool found_waiter = false;
     {
-      // Wake up thread that blocks on the compile task.
       MutexLocker ct_lock(current->lock());
-      current->lock()->notify();
+      assert(current->waiting_for_completion_count() <= 1, "more than one thread are waiting for task");
+      if (current->waiting_for_completion_count() > 0) {
+        // If another thread waits for this task, we must wake them up
+        // so they will stop waiting and free the task.
+        current->lock()->notify();
+        found_waiter = true;
+      }
+    }
+    if (!found_waiter) {
+      // If no one was waiting for this task, we need to free it ourselves. In this case, the task
+      // is also certainly unlocked, because, again, there is no waiter.
+      // Otherwise, by convention, it's the waiters responsibility to free the task.
+      // Put the task back on the freelist.
+      CompileTask::free(current);
     }
-    // Put the task back on the freelist.
-    CompileTask::free(current);
   }
   _first = nullptr;
   _last = nullptr;
@@ -1722,9 +1733,11 @@ void CompileBroker::wait_for_completion(CompileTask* task) {
   {
     MonitorLocker ml(thread, task->lock());
     free_task = true;
+    task->inc_waiting_for_completion();
     while (!task->is_complete() && !is_compilation_disabled_forever()) {
       ml.wait();
     }
+    task->dec_waiting_for_completion();
   }
 
   if (free_task) {

src/hotspot/share/compiler/compileTask.cpp

@@ -106,6 +106,8 @@ void CompileTask::initialize(int compile_id,
   _comp_level = comp_level;
   _num_inlined_bytecodes = 0;
 
+  _waiting_count = 0;
+
   _is_complete = false;
   _is_success = false;
 

src/hotspot/share/compiler/compileTask.hpp

@@ -98,6 +98,7 @@ class CompileTask : public CHeapObj<mtCompiler> {
   // Compilation state for a blocking JVMCI compilation
   JVMCICompileState*   _blocking_jvmci_compile_state;
 #endif
+  int                  _waiting_count;  // See waiting_for_completion_count()
   int                  _comp_level;
   int                  _num_inlined_bytecodes;
   CompileTask*         _next, *_prev;
@@ -174,6 +175,23 @@ class CompileTask : public CHeapObj<mtCompiler> {
 
   Monitor*     lock() const                      { return _lock; }
 
+  // See how many threads are waiting for this task. Must have lock to read this.
+  int waiting_for_completion_count() {
+    assert(_lock->owned_by_self(), "must have lock to use waiting_for_completion_count()");
+    return _waiting_count;
+  }
+  // Indicates that a thread is waiting for this task to complete. Must have lock to use this.
+  void inc_waiting_for_completion() {
+    assert(_lock->owned_by_self(), "must have lock to use inc_waiting_for_completion()");
+    _waiting_count++;
+  }
+  // Indicates that a thread stopped waiting for this task to complete. Must have lock to use this.
+  void dec_waiting_for_completion() {
+    assert(_lock->owned_by_self(), "must have lock to use dec_waiting_for_completion()");
+    assert(_waiting_count > 0, "waiting count is not positive");
+    _waiting_count--;
+  }
+
   void         mark_complete()                   { _is_complete = true; }
   void         mark_success()                    { _is_success = true; }
   void         mark_started(jlong time)          { _time_started = time; }