[oilpan]: Make parking threads for GC timeout in the case parking exceeds 100 MS

Source/platform/heap/ThreadState.cpp

 /*
  * Copyright (C) 2013 Google Inc. 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
@@ skipping 78 matching lines @@
 uint8_t ThreadState::s_mainThreadStateStorage[sizeof(ThreadState)];
 SafePointBarrier* ThreadState::s_safePointBarrier = 0;
 bool ThreadState::s_inGC = false;
 
 static Mutex& threadAttachMutex()
 {
     AtomicallyInitializedStatic(Mutex&, mutex = *new Mutex);
     return mutex;
 }
 
+static double lockingTimeout()
+{
+    // Wait time for parking all threads is at most 10 MS.
+    return 0.01;
+}
+
+
 typedef void (*PushAllRegistersCallback)(SafePointBarrier*, ThreadState*, intptr_t*);
 extern "C" void pushAllRegisters(SafePointBarrier*, ThreadState*, PushAllRegistersCallback);
 
 class SafePointBarrier {
 public:
     SafePointBarrier() : m_canResume(1), m_unparkedThreadCount(0) { }
     ~SafePointBarrier() { }
 
     // Request other attached threads that are not at safe points to park themselves on safepoints.
-    void parkOthers()
+    bool parkOthers()
     {
         ASSERT(ThreadState::current()->isAtSafePoint());
 
         // Lock threadAttachMutex() to prevent threads from attaching.
         threadAttachMutex().lock();
 
         ThreadState::AttachedThreadStateSet& threads = ThreadState::attachedThreads();
 
         MutexLocker locker(m_mutex);
         atomicAdd(&m_unparkedThreadCount, threads.size());
         releaseStore(&m_canResume, 0);
 
         ThreadState* current = ThreadState::current();
         for (ThreadState::AttachedThreadStateSet::iterator it = threads.begin(), end = threads.end(); it != end; ++it) {
             if (*it == current)
                 continue;
 
             const Vector<ThreadState::Interruptor*>& interruptors = (*it)->interruptors();
             for (size_t i = 0; i < interruptors.size(); i++)
                 interruptors[i]->requestInterrupt();
         }
 
-        while (acquireLoad(&m_unparkedThreadCount) > 0)
-            m_parked.wait(m_mutex);
+        while (acquireLoad(&m_unparkedThreadCount) > 0) {
+            double expirationTime = currentTime() + lockingTimeout();
+            if (!m_parked.timedWait(m_mutex, expirationTime)) {
+                // One of the other threads did not return to a safepoint within the maximum
+                // time we allow for threads to be parked. Abandon the GC and resume the
+                // currently parked threads.
+                resumeOthers(true);

[haraken, 2014/05/01 04:03:56]

Nit: It would be more consistent if you can remove resumeOthers() from here and
let ~GCScope() call resumeOthers().

GCScope() {
  if (ThreadState::stopThreads()) {
    m_parkedAllThreads = true;
    m_state->enterGC();
  }
}

~GCScope() {
  if (m_parkedAllThreads) {
    m_state->leaveGC();
  }
  ThreadState::resumeThreads();
}

Then you can remove the |bool barrierLocked| flag from resumeOthers().

[wibling-chromium, 2014/05/01 06:51:48]

Yes, that would be nice. However I am worried that this would mean we release
m_mutex after timing out which will allow a thread to become parked in the
meantime and signal the mutex since it was the last remaining thread. By doing
it this way I am reasonably confident that cannot happen.

[haraken, 2014/05/01 07:33:50]

Yes, but is it problematic that we allow another thread to get parked while
releasing the mutex?

Probably I think it's problematic in the current code because line 154 and line
155 are not in the mutex. If we move the line 154 and 155 into the mutex, there
is no problem in allowing another thread to get parked while releasing the
mutex. (Then the code logic becomes clearer: we need to acquire the mutex
whenever we touch m_unparkedThreadCount and m_canResume.)

[wibling-chromium, 2014/05/01 08:25:18]

No, that should be okay. However previously we would never signal m_parked
unless someone were in parkOthers and had added threads.size() to the unparked
count. Now we do have a window where a thread could signal even though
parkOthers had failed (ie. timed out). That signal would now be lingering on
until the next parkOther's call. It might not be a big deal since we would
likely just loop once more in the parkOthers while(m_unparkedThreadCount > 0)
loop and find that threads are not yet parked.

That said I find the current code and API simpler to understand. As it is now
the SafePointBarrier API for parkOthers is to call resumeOthers only if it
succeeds and there are no interim side effects if it fails.
It is also easier to verify that the effects of parkOthers are negated since we
do it locally in the SafePointBarrier.
The accesses in 154 and 155 are both atomic so they should be sane given.
Strictly speaking this is not changing as part of my change. It was the case
before as well and from what I can tell is fine with the current usage.

[haraken, 2014/05/01 08:36:31]

Thanks for the clarification, makes sense!

+                return false;
+            }
+        }
+        return true;
     }
 
-    void resumeOthers()
+    void resumeOthers(bool barrierLocked = false)
     {
         ThreadState::AttachedThreadStateSet& threads = ThreadState::attachedThreads();
         atomicSubtract(&m_unparkedThreadCount, threads.size());

[haraken, 2014/05/01 04:03:56]

I guess this will confuse m_unparkedThreadCount. At this point, we need to
subtract X from m_unparkedThreadCount, where X is the number of threads that are
parking at this point.

Before this CL, it's guaranteed that m_unparkedThreadCount == 0 and X ==
threads.size(). Thus we can subtract threads.size() from m_unparkedThreadCount.

After this CL, it's not guaranteed that m_unparkedThreadCount == 0 nor X ==
threads.size(). Thus subtracting threads.size() from m_unparkedThreadCount will
confuse the count.

[wibling-chromium, 2014/05/01 06:51:48]

I initially read the code the same way, but parkOthers specifically add
threads.size() to the count. resumeOthers subtracts it which means they cancel
each other out. Similarly checkAndPark (specifically doPark call both decrement
and increment which should also cancel each other out. Finally enterSafePoint
(specifically doEnterSafePoint) and leaveSafePoint also both does a decrement
and an increment which cancels each other out.
As such I believe subtracting threads.size() is the correct thing to do even if
all threads have not been parked.

[haraken, 2014/05/01 07:33:50]

Thanks for the clarification. Your explanation sounds reasonable.

         releaseStore(&m_canResume, 1);
-        {
+
+        // FIXME: Resumed threads will all contend for m_mutex just to unlock it
+        // later which is a waste of resources.
+        if (UNLIKELY(barrierLocked)) {
+            m_resume.broadcast();
+        } else {
             // FIXME: Resumed threads will all contend for
             // m_mutex just to unlock it later which is a waste of
             // resources.
             MutexLocker locker(m_mutex);
             m_resume.broadcast();
         }
 
         ThreadState* current = ThreadState::current();
         for (ThreadState::AttachedThreadStateSet::iterator it = threads.begin(), end = threads.end(); it != end; ++it) {
             if (*it == current)
                 continue;
 
             const Vector<ThreadState::Interruptor*>& interruptors = (*it)->interruptors();
             for (size_t i = 0; i < interruptors.size(); i++)
                 interruptors[i]->clearInterrupt();
         }
 
         threadAttachMutex().unlock();
         ASSERT(ThreadState::current()->isAtSafePoint());
     }
 
+    void checkAndPark(ThreadState* state)

[zerny-chromium, 2014/05/01 07:05:38]

Nit: avoid reordering/formatting for unchanged code.

[wibling-chromium, 2014/05/01 08:25:18]

Done. I will revert that and do it in a separate change.

+    {
+        ASSERT(!state->isSweepInProgress());
+        if (!acquireLoad(&m_canResume)) {
+            pushAllRegisters(this, state, parkAfterPushRegisters);
+            state->performPendingSweep();
+        }
+    }
+
+    void enterSafePoint(ThreadState* state)
+    {
+        ASSERT(!state->isSweepInProgress());
+        pushAllRegisters(this, state, enterSafePointAfterPushRegisters);
+    }
+
+    void leaveSafePoint(ThreadState* state)
+    {
+        if (atomicIncrement(&m_unparkedThreadCount) > 0)
+            checkAndPark(state);
+    }
+
+private:
     void doPark(ThreadState* state, intptr_t* stackEnd)
     {
         state->recordStackEnd(stackEnd);
         MutexLocker locker(m_mutex);
         if (!atomicDecrement(&m_unparkedThreadCount))
             m_parked.signal();
         while (!acquireLoad(&m_canResume))
             m_resume.wait(m_mutex);
         atomicIncrement(&m_unparkedThreadCount);
     }
 
-    void checkAndPark(ThreadState* state)
+    static void parkAfterPushRegisters(SafePointBarrier* barrier, ThreadState* state, intptr_t* stackEnd)
     {
-        ASSERT(!state->isSweepInProgress());
-        if (!acquireLoad(&m_canResume)) {
-            pushAllRegisters(this, state, parkAfterPushRegisters);
-            state->performPendingSweep();
-        }
+        barrier->doPark(state, stackEnd);
     }
 
     void doEnterSafePoint(ThreadState* state, intptr_t* stackEnd)
     {
         state->recordStackEnd(stackEnd);
         state->copyStackUntilSafePointScope();
         // m_unparkedThreadCount tracks amount of unparked threads. It is
         // positive if and only if we have requested other threads to park
         // at safe-points in preparation for GC. The last thread to park
         // itself will make the counter hit zero and should notify GC thread
         // that it is safe to proceed.
         // If no other thread is waiting for other threads to park then
         // this counter can be negative: if N threads are at safe-points
         // the counter will be -N.
         if (!atomicDecrement(&m_unparkedThreadCount)) {
             MutexLocker locker(m_mutex);
             m_parked.signal(); // Safe point reached.
         }
     }
 
-    void enterSafePoint(ThreadState* state)
-    {
-        ASSERT(!state->isSweepInProgress());
-        pushAllRegisters(this, state, enterSafePointAfterPushRegisters);
-    }
-
-    void leaveSafePoint(ThreadState* state)
-    {
-        if (atomicIncrement(&m_unparkedThreadCount) > 0)
-            checkAndPark(state);
-    }
-
-private:
-    static void parkAfterPushRegisters(SafePointBarrier* barrier, ThreadState* state, intptr_t* stackEnd)
-    {
-        barrier->doPark(state, stackEnd);
-    }
-
     static void enterSafePointAfterPushRegisters(SafePointBarrier* barrier, ThreadState* state, intptr_t* stackEnd)
     {
         barrier->doEnterSafePoint(state, stackEnd);
     }
 
     volatile int m_canResume;
     volatile int m_unparkedThreadCount;
     Mutex m_mutex;
     ThreadCondition m_parked;
     ThreadCondition m_resume;
@@ skipping 466 matching lines @@
     if (isConsistentForGC()) {
         HeapStats scannedStats;
         scannedStats.clear();
         for (int i = 0; i < NumberOfHeaps; i++)
             m_heaps[i]->getScannedStats(scannedStats);
         ASSERT(scannedStats == stats);
     }
 #endif
 }
 
-void ThreadState::stopThreads()
+bool ThreadState::stopThreads()
 {
-    s_safePointBarrier->parkOthers();
+    return s_safePointBarrier->parkOthers();
 }
 
 void ThreadState::resumeThreads()
 {
     s_safePointBarrier->resumeOthers();
 }
 
 void ThreadState::safePoint(StackState stackState)
 {
     checkThread();
@@ skipping 124 matching lines @@
     state->safePoint(HeapPointersOnStack);
 }
 
 ThreadState::AttachedThreadStateSet& ThreadState::attachedThreads()
 {
     DEFINE_STATIC_LOCAL(AttachedThreadStateSet, threads, ());
     return threads;
 }
 
 }