Fix cluster-fuzz regression when getting message from Worker

src/d8.cc

 // Copyright 2012 the V8 project authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 
 // Defined when linking against shared lib on Windows.
 #if defined(USING_V8_SHARED) && !defined(V8_SHARED)
 #define V8_SHARED
 #endif
 
@@ skipping 187 matching lines @@
 
 #ifndef V8_SHARED
 CounterMap* Shell::counter_map_;
 base::OS::MemoryMappedFile* Shell::counters_file_ = NULL;
 CounterCollection Shell::local_counters_;
 CounterCollection* Shell::counters_ = &local_counters_;
 base::Mutex Shell::context_mutex_;
 const base::TimeTicks Shell::kInitialTicks =
     base::TimeTicks::HighResolutionNow();
 Persistent<Context> Shell::utility_context_;
+base::Mutex Shell::workers_mutex_;
+bool Shell::allow_new_workers_ = true;
 i::List<Worker*> Shell::workers_;
 i::List<SharedArrayBuffer::Contents> Shell::externalized_shared_contents_;
 #endif  // !V8_SHARED
 
 Persistent<Context> Shell::evaluation_context_;
 ArrayBuffer::Allocator* Shell::array_buffer_allocator;
 ShellOptions Shell::options;
 const char* Shell::kPrompt = "d8> ";
 
 #ifndef V8_SHARED
@@ skipping 468 matching lines @@
 
 #ifndef V8_SHARED
 void Shell::WorkerNew(const v8::FunctionCallbackInfo<v8::Value>& args) {
   Isolate* isolate = args.GetIsolate();
   HandleScope handle_scope(isolate);
   if (args.Length() < 1 || !args[0]->IsFunction()) {
     Throw(args.GetIsolate(), "1st argument must be function");
     return;
   }
 
-  Worker* worker = new Worker;
-  args.This()->SetInternalField(0, External::New(isolate, worker));
-  workers_.Add(worker);
+  {
+    base::LockGuard<base::Mutex> lock_guard(&workers_mutex_);
+    if (!allow_new_workers_) return;
 
-  String::Utf8Value function_string(args[0]->ToString());
-  worker->StartExecuteInThread(isolate, *function_string);
+    Worker* worker = new Worker;
+    args.This()->SetInternalField(0, External::New(isolate, worker));
+    workers_.Add(worker);
+
+    String::Utf8Value function_string(args[0]->ToString());
+    worker->StartExecuteInThread(isolate, *function_string);
+  }
 }
 
 
 void Shell::WorkerPostMessage(const v8::FunctionCallbackInfo<v8::Value>& args) {
   Isolate* isolate = args.GetIsolate();
   HandleScope handle_scope(isolate);
   Local<Context> context = isolate->GetCurrentContext();
 
   if (args.Length() < 1) {
     Throw(isolate, "Invalid argument");
@@ skipping 74 matching lines @@
   HandleScope handle_scope(isolate);
   Local<Value> this_value = args.This()->GetInternalField(0);
   if (!this_value->IsExternal()) {
     Throw(isolate, "this is not a Worker");
     return;
   }
 
   Worker* worker =
       static_cast<Worker*>(Local<External>::Cast(this_value)->Value());
   worker->Terminate();
-  workers_.RemoveElement(worker);
-  delete worker;
 }
 #endif  // !V8_SHARED
 
 
 void Shell::Quit(const v8::FunctionCallbackInfo<v8::Value>& args) {
   int exit_code = args[0]->Int32Value();
   OnExit(args.GetIsolate());
   exit(exit_code);
 }
 
@@ skipping 805 matching lines @@
 void SerializationDataQueue::Clear() {
   base::LockGuard<base::Mutex> lock_guard(&mutex_);
   for (int i = 0; i < data_.length(); ++i) {
     delete data_[i];
   }
   data_.Clear();
 }
 
 
 Worker::Worker()
-    : in_semaphore_(0), out_semaphore_(0), thread_(NULL), script_(NULL) {}
+    : in_semaphore_(0),
+      out_semaphore_(0),
+      thread_(NULL),
+      script_(NULL),
+      state_(IDLE) {}
 
 
 Worker::~Worker() { Cleanup(); }
 
 
 void Worker::StartExecuteInThread(Isolate* isolate,
                                   const char* function_string) {
-  if (thread_) {
-    Throw(isolate, "Only one worker allowed");
-    return;
-  }
-
+  DCHECK(base::NoBarrier_Load(&state_) == IDLE);
   static const char format[] = "(%s).call(this);";
   size_t len = strlen(function_string) + sizeof(format);
 
   script_ = new char[len + 1];
   i::Vector<char> vec(script_, static_cast<int>(len + 1));
   i::SNPrintF(vec, format, function_string);
 
+  base::NoBarrier_Store(&state_, RUNNING);
   thread_ = new WorkerThread(this);
   thread_->Start();
 }
 
 
 void Worker::PostMessage(SerializationData* data) {
   in_queue_.Enqueue(data);
   in_semaphore_.Signal();
 }
 
 
 SerializationData* Worker::GetMessage() {
-  SerializationData* data;
+  SerializationData* data = NULL;
   while (!out_queue_.Dequeue(&data)) {
+    if (base::NoBarrier_Load(&state_) != RUNNING) break;
     out_semaphore_.Wait();
   }
 
   return data;
 }
 
 
 void Worker::Terminate() {
-  if (thread_ == NULL) return;
-  PostMessage(NULL);
-  thread_->Join();
-  Cleanup();
+  if (base::NoBarrier_CompareAndSwap(&state_, RUNNING, TERMINATED) == RUNNING) {
+    // Post NULL to wake the Worker thread message loop.
+    PostMessage(NULL);
+    thread_->Join();
+  }
 }
 
 
 void Worker::ExecuteInThread() {
   Isolate::CreateParams create_params;
   create_params.array_buffer_allocator = Shell::array_buffer_allocator;
   Isolate* isolate = Isolate::New(create_params);
   {
     Isolate::Scope iscope(isolate);
     {
@@ skipping 12 matching lines @@
         Handle<Function> postmessage_fun;
         if (postmessage_fun_template->GetFunction(context)
                 .ToLocal(&postmessage_fun)) {
           global->Set(String::NewFromUtf8(isolate, "postMessage"),
                       postmessage_fun);
         }
 
         // First run the script
         Handle<String> file_name = String::NewFromUtf8(isolate, "unnamed");
         Handle<String> source = String::NewFromUtf8(isolate, script_);
-        Shell::ExecuteString(isolate, source, file_name, false, true);
-
-        // Get the message handler
-        Handle<Value> onmessage =
-            global->Get(String::NewFromUtf8(isolate, "onmessage"));
-        if (onmessage->IsFunction()) {
-          Handle<Function> onmessage_fun = Handle<Function>::Cast(onmessage);
-          // Now wait for messages
-          bool done = false;
-          while (!done) {
-            in_semaphore_.Wait();
-            SerializationData* data;
-            if (!in_queue_.Dequeue(&data)) continue;
-            if (data == NULL) {
-              done = true;
-              break;
+        if (Shell::ExecuteString(isolate, source, file_name, true, true)) {
+          // Get the message handler
+          Handle<Value> onmessage =
+              global->Get(String::NewFromUtf8(isolate, "onmessage"));
+          if (onmessage->IsFunction()) {
+            Handle<Function> onmessage_fun = Handle<Function>::Cast(onmessage);
+            // Now wait for messages
+            bool done = false;
+            while (!done) {
+              in_semaphore_.Wait();
+              SerializationData* data;
+              if (!in_queue_.Dequeue(&data)) continue;
+              if (data == NULL) {
+                done = true;
+                break;
+              }
+              int offset = 0;
+              Local<Value> data_value;
+              if (Shell::DeserializeValue(isolate, *data, &offset)
+                      .ToLocal(&data_value)) {
+                Handle<Value> argv[] = {data_value};
+                (void)onmessage_fun->Call(context, global, 1, argv);
+              }
+              delete data;
             }
-            int offset = 0;
-            Local<Value> data_value;
-            if (Shell::DeserializeValue(isolate, *data, &offset)
-                    .ToLocal(&data_value)) {
-              Handle<Value> argv[] = {data_value};
-              (void)onmessage_fun->Call(context, global, 1, argv);
-            }
-            delete data;
           }
         }
       }
     }
     Shell::CollectGarbage(isolate);
   }
   isolate->Dispose();
+
+  if (base::NoBarrier_CompareAndSwap(&state_, RUNNING, TERMINATED) == RUNNING) {
+    // Post NULL to wake the thread waiting on GetMessage() if there is one.
+    out_queue_.Enqueue(NULL);
+    out_semaphore_.Signal();
+  }
 }
 
 
 void Worker::Cleanup() {
   delete thread_;
   thread_ = NULL;
   delete[] script_;
   script_ = NULL;
   in_queue_.Clear();
   out_queue_.Clear();
@@ skipping 442 matching lines @@
     }
     default:
       UNREACHABLE();
   }
 
   return scope.Escape(result);
 }
 
 
 void Shell::CleanupWorkers() {
-  for (int i = 0; i < workers_.length(); ++i) {
-    Worker* worker = workers_[i];
+  // Make a copy of workers_, because we don't want to call Worker::Terminate
+  // while holding the workers_mutex_ lock. Otherwise, if a worker is about to
+  // create a new Worker, it would deadlock.
+  i::List<Worker*> workers_copy;
+  {
+    base::LockGuard<base::Mutex> lock_guard(&workers_mutex_);
+    allow_new_workers_ = false;
+    workers_copy.AddAll(workers_);
+    workers_.Clear();
+  }
+
+  for (int i = 0; i < workers_copy.length(); ++i) {
+    Worker* worker = workers_copy[i];
     worker->Terminate();
     delete worker;
   }
-  workers_.Clear();
+
+  // Now that all workers are terminated, we can re-enable Worker creation.
+  {
+    base::LockGuard<base::Mutex> lock_guard(&workers_mutex_);
+    allow_new_workers_ = true;
+  }
 
   for (int i = 0; i < externalized_shared_contents_.length(); ++i) {
     const SharedArrayBuffer::Contents& contents =
         externalized_shared_contents_[i];
     Shell::array_buffer_allocator->Free(contents.Data(), contents.ByteLength());
   }
   externalized_shared_contents_.Clear();
 }
 
 
@@ skipping 188 matching lines @@
 }
 
 }  // namespace v8
 
 
 #ifndef GOOGLE3
 int main(int argc, char* argv[]) {
   return v8::Shell::Main(argc, argv);
 }
 #endif