Re-reland "Flush parallel recompilation queues on context dispose notification"

src/optimizing-compiler-thread.cc

 // Copyright 2012 the V8 project authors. All rights reserved.
 // Redistribution and use in source and binary forms, with or without
 // modification, are permitted provided that the following conditions are
 // met:
 //
 //     * Redistributions of source code must retain the above copyright
 //       notice, this list of conditions and the following disclaimer.
 //     * Redistributions in binary form must reproduce the above
 //       copyright notice, this list of conditions and the following
 //       disclaimer in the documentation and/or other materials provided
@@ skipping 42 matching lines @@
 
   while (true) {
     input_queue_semaphore_->Wait();
     Logger::TimerEventScope timer(
         isolate_, Logger::TimerEventScope::v8_recompile_parallel);
 
     if (FLAG_parallel_recompilation_delay != 0) {
       OS::Sleep(FLAG_parallel_recompilation_delay);
     }
 
-    if (Acquire_Load(&stop_thread_)) {
-      stop_semaphore_->Signal();
-      if (FLAG_trace_parallel_recompilation) {
-        time_spent_total_ = OS::Ticks() - epoch;
-      }
-      return;
+    switch (static_cast<StopFlag>(Acquire_Load(&stop_thread_))) {
+      case CONTINUE:
+        break;
+      case STOP:
+        if (FLAG_trace_parallel_recompilation) {
+          time_spent_total_ = OS::Ticks() - epoch;
+        }
+        stop_semaphore_->Signal();
+        return;
+      case FLUSH:
+        // Reset input queue semaphore.
+        delete input_queue_semaphore_;
+        input_queue_semaphore_ = OS::CreateSemaphore(0);
+        // Signal for main thread to start flushing.
+        stop_semaphore_->Signal();
+        // Return to start of consumer loop.
+        continue;
     }
 
     int64_t compiling_start = 0;
     if (FLAG_trace_parallel_recompilation) compiling_start = OS::Ticks();
 
     CompileNext();
 
     if (FLAG_trace_parallel_recompilation) {
       time_spent_compiling_ += OS::Ticks() - compiling_start;
     }
@@ skipping 16 matching lines @@
   // are always also queued.
   ScopedLock mark_and_queue(install_mutex_);
   { Heap::RelocationLock relocation_lock(isolate_->heap());
     AllowHandleDereference ahd;
     optimizing_compiler->info()->closure()->MarkForInstallingRecompiledCode();
   }
   output_queue_.Enqueue(optimizing_compiler);
 }
 
 
+void OptimizingCompilerThread::FlushQueue(
+    UnboundQueue<OptimizingCompiler*>* queue,
+    bool restore_function_code) {
+  ASSERT(!IsOptimizerThread());
+  OptimizingCompiler* optimizing_compiler;
+  // The optimizing compiler is allocated in the CompilationInfo's zone.
+  while (queue->Dequeue(&optimizing_compiler)) {
+    CompilationInfo* info = optimizing_compiler->info();
+    if (restore_function_code) {
+      Handle<JSFunction> function = info->closure();
+      function->ReplaceCode(function->shared()->code());
+    }
+    delete info;
+  }
+}
+
+
+void OptimizingCompilerThread::Flush() {
+  ASSERT(!IsOptimizerThread());
+  Release_Store(&stop_thread_, static_cast<AtomicWord>(FLUSH));

[Michael Starzinger, 2013/08/06 17:57:56]

This is still racey, the FLUSH command might be executed by the compiler thread
before the input_queue_semaphore below has been signaled. Which means the below
signal would hit the new semaphore, not the old one. Consider the following
ordering, where "M" is the main thread and "C" is the compiler thread.

M: stop_thread = FLUSH
C: input_queue_semaphore = new Semaphore
C: input_queue_semaphore->Wait()
M: input_queue_semaphore->Signal()
M: input_queue->flush()
M: stop_thread = CONTINUE
C: CompileNext(), but input_queue is empty

Resetting the input_queue_semaphore is mighty dangerous.

+  input_queue_semaphore_->Signal();
+  stop_semaphore_->Wait();
+
+  // At this point, the optimizing thread is idling.  It is waiting for
+  // the input semaphore, which has just been reset.  It will not continue
+  // as long as the input semaphore is not signaled (by the main thread).
+  // Relying the input semaphore, we don't need any explicit memory barrier.
+  FlushQueue(&input_queue_, true);
+  FlushQueue(&output_queue_, true);
+  NoBarrier_Store(&queue_length_, static_cast<AtomicWord>(0));
+  NoBarrier_Store(&stop_thread_, static_cast<AtomicWord>(CONTINUE));
+}
+
+
 void OptimizingCompilerThread::Stop() {
   ASSERT(!IsOptimizerThread());
-  Release_Store(&stop_thread_, static_cast<AtomicWord>(true));
+  Release_Store(&stop_thread_, static_cast<AtomicWord>(STOP));
   input_queue_semaphore_->Signal();
   stop_semaphore_->Wait();
 
   if (FLAG_parallel_recompilation_delay != 0) {
     // Barrier when loading queue length is not necessary since the write
     // happens in CompileNext on the same thread.
     while (NoBarrier_Load(&queue_length_) > 0) CompileNext();
     InstallOptimizedFunctions();
   } else {
-    OptimizingCompiler* optimizing_compiler;
-    // The optimizing compiler is allocated in the CompilationInfo's zone.
-    while (input_queue_.Dequeue(&optimizing_compiler)) {
-      delete optimizing_compiler->info();
-    }
-    while (output_queue_.Dequeue(&optimizing_compiler)) {
-      delete optimizing_compiler->info();
-    }
+    FlushQueue(&input_queue_, false);
+    FlushQueue(&output_queue_, false);
   }
 
   if (FLAG_trace_parallel_recompilation) {
     double compile_time = static_cast<double>(time_spent_compiling_);
     double total_time = static_cast<double>(time_spent_total_);
     double percentage = (compile_time * 100) / total_time;
     PrintF("  ** Compiler thread did %.2f%% useful work\n", percentage);
   }
 
   Join();
@@ skipping 28 matching lines @@
 #ifdef DEBUG
 bool OptimizingCompilerThread::IsOptimizerThread() {
   if (!FLAG_parallel_recompilation) return false;
   ScopedLock lock(thread_id_mutex_);
   return ThreadId::Current().ToInteger() == thread_id_;
 }
 #endif
 
 
 } }  // namespace v8::internal