d8: Fix some TSAN bugs

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 1591 matching lines @@
 
 
 void SerializationDataQueue::Enqueue(SerializationData* data) {
   base::LockGuard<base::Mutex> lock_guard(&mutex_);
   data_.Add(data);
 }
 
 
 bool SerializationDataQueue::Dequeue(SerializationData** data) {
   base::LockGuard<base::Mutex> lock_guard(&mutex_);
+  *data = NULL;
   if (data_.is_empty()) return false;
   *data = data_.Remove(0);
   return true;
 }
 
 
 bool SerializationDataQueue::IsEmpty() {
   base::LockGuard<base::Mutex> lock_guard(&mutex_);
   return data_.is_empty();
 }
@@ skipping 33 matching lines @@
 
 void Worker::PostMessage(SerializationData* data) {
   in_queue_.Enqueue(data);
   in_semaphore_.Signal();
 }
 
 
 SerializationData* Worker::GetMessage() {
   SerializationData* data = NULL;
   while (!out_queue_.Dequeue(&data)) {
-    if (base::NoBarrier_Load(&state_) != RUNNING) break;
+    // If the worker is no longer running, and there are no messages in the
+    // queue, don't expect any more messages from it.
+    if (base::NoBarrier_Load(&state_) > RUNNING) break;
     out_semaphore_.Wait();
   }
 
   return data;
 }
 
 
 void Worker::Terminate() {
-  if (base::NoBarrier_CompareAndSwap(&state_, RUNNING, TERMINATED) == RUNNING) {
+  if (base::NoBarrier_CompareAndSwap(&state_, RUNNING, TERMINATING) ==
+      RUNNING) {
     // Post NULL to wake the Worker thread message loop.
     PostMessage(NULL);
-    thread_->Join();
   }
 }
 
 
+void Worker::WaitForThread() {
+  Terminate();
+
+  State expected_state = TERMINATING;
+  while (true) {
+    State actual_state = static_cast<State>(
+        base::NoBarrier_CompareAndSwap(&state_, expected_state, JOINING));
+
+    if (actual_state == expected_state) break;
+
+    switch (actual_state) {
+      case IDLE:
+      case RUNNING:
+        // We terminated above, we should never be IDLE or RUNNING.
+        UNREACHABLE();
+        return;
+
+      case TERMINATING:
+      case STOPPED:

[Jarin, 2015/07/10 10:43:17]

Why do we want to retry when STOPPED?

Actually, I am even confused why we are trying to go through the TERMINATING
state in WaitForThread. Why (instead of calling Terminate) don't you CAS
RUNNING->JOINING?

It would be helpful if you could describe what was wrong with the original state
machine.

[binji, 2015/07/10 17:40:46]

You're right, this is too complicated. I refactored this into a much simpler
change.

Basically, the problem was that the pthread object would leak if
Worker.terminate() was never called. Really all that needed to be done was call
base::Thread::Join(), but that crashes if you call it twice. I wanted to catch
that programmer error, but adding it to the state_ variable made it too complex.

+        // Try again.
+        expected_state = actual_state;
+        break;
+
+      case JOINING:

[Jarin, 2015/07/10 10:43:17]

Is not this only called when the shell finishes, so there should not be anyone
else joining?

[binji, 2015/07/10 17:40:46]

True, this was just caution to prevent multiple joiners (which crashes). I've
refactored this into a much simpler change.

+        // Someone else is joining. Bail.
+        return;
+
+      case JOINED:
+        // Tried to join twice.
+        UNREACHABLE();
+        return;
+    }
+  }
+
+  thread_->Join();
+  base::NoBarrier_Store(&state_, JOINED);
+}
+
+
 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);
     {
       HandleScope scope(isolate);
       PerIsolateData data(isolate);
       Local<Context> context = Shell::CreateEvaluationContext(isolate);
@@ skipping 42 matching lines @@
               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();
-  }
+  base::NoBarrier_Store(&state_, STOPPED);
+  // 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 377 matching lines @@
       result = True(isolate);
       break;
     case kSerializationTagFalse:
       result = False(isolate);
       break;
     case kSerializationTagNumber:
       result = Number::New(isolate, data.Read<double>(offset));
       break;
     case kSerializationTagString: {
       int length = data.Read<int>(offset);
-      static char s_buffer[128];

[binji, 2015/07/09 00:13:12]

I'm pretty ashamed of this one... :-|

-      char* p = s_buffer;
-      bool allocated = false;
-      if (length > static_cast<int>(sizeof(s_buffer))) {
-        p = new char[length];
-        allocated = true;
-      }
+      char* p = new char[length];

[Jarin, 2015/07/10 10:43:17]

How about using std::vector<char> here, instead of the naked pointer?

[binji, 2015/07/10 17:40:46]

Done (though I was not sure that std containers were allowed in this file, it
seems like they were explicitly avoided).

       data.ReadMemory(p, length, offset);
       MaybeLocal<String> str =
           String::NewFromUtf8(isolate, p, String::kNormalString, length);
       if (!str.IsEmpty()) result = str.ToLocalChecked();
-      if (allocated) delete[] p;
+      delete[] p;
       break;
     }
     case kSerializationTagArray: {
       uint32_t length = data.Read<uint32_t>(offset);
       Handle<Array> array = Array::New(isolate, length);
       for (uint32_t i = 0; i < length; ++i) {
         Local<Value> element_value;
         CHECK(DeserializeValue(isolate, data, offset).ToLocal(&element_value));
         array->Set(i, element_value);
       }
@@ skipping 51 matching lines @@
   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();
+    worker->WaitForThread();
     delete worker;
   }
 
   // Now that all workers are terminated, we can re-enable Worker creation.

[Jarin, 2015/07/10 10:43:17]

I know this has not be introduced by this CL, but why do we want to re-enable
worker creation? At this point, the process is about to die, no?

   {
     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());
   }
@@ skipping 192 matching lines @@
 }
 
 }  // namespace v8
 
 
 #ifndef GOOGLE3
 int main(int argc, char* argv[]) {
   return v8::Shell::Main(argc, argv);
 }
 #endif