PeerConnection(java): account for thread shutdown vagaries.

Android's JVM requires threads to detach before they exit, but ONLY if
they needed to AttachCurrentThread.  Conversly, threads that were
attached by the JVM (e.g. the result of making a native call from Java)
must NOT be detached by the application.  This is bug 2441.

The fix for the above is to only pthread_setspecific() for threads that
Attach(), not for already-attached threads.  To ensure that we only
detach Attached threads, added a GetEnv() call to ThreadDestructor(),
which revealed that Oracle's JVM can overly-eagerly clear TLS accounting
data, effectively detaching threads without their consent at shutdown.
Work around this with a specific check.

To guard against (some) regression, added a variant of PeerConnectionTest
that runs on a non-main thread.  This revealed a bug in LinuxDeviceManager
which implicitly assumes its talk_base::Thread has already been
initialized.  Fixed that here too.

BUG=2441
R=henrike@webrtc.org

Review URL: https://webrtc-codereview.appspot.com/8759004

git-svn-id: http://webrtc.googlecode.com/svn/trunk@5567 4adac7df-926f-26a2-2b94-8c16560cd09d

talk/app/webrtc/java/jni/peerconnection_jni.cc

@@ -152,7 +152,10 @@ namespace {
 static JavaVM* g_jvm = NULL;  // Set in JNI_OnLoad().
 
 static pthread_once_t g_jni_ptr_once = PTHREAD_ONCE_INIT;
-static pthread_key_t g_jni_ptr;  // Key for per-thread JNIEnv* data.
+// Key for per-thread JNIEnv* data.  Non-NULL in threads attached to |g_jvm| by
+// AttachCurrentThreadIfNeeded(), NULL in unattached threads and threads that
+// were attached by the JVM because of a Java->native call.
+static pthread_key_t g_jni_ptr;
 
 // Return thread ID as a string.
 static std::string GetThreadId() {
@@ -170,9 +173,32 @@ static std::string GetThreadName() {
   return std::string(name);
 }
 
-static void ThreadDestructor(void* unused) {
+// Return a |JNIEnv*| usable on this thread or NULL if this thread is detached.
+static JNIEnv* GetEnv() {
+  void* env = NULL;
+  jint status = g_jvm->GetEnv(&env, JNI_VERSION_1_6);
+  CHECK(((env != NULL) && (status == JNI_OK)) ||
+            ((env == NULL) && (status == JNI_EDETACHED)),
+        "Unexpected GetEnv return: " << status << ":" << env);
+  return reinterpret_cast<JNIEnv*>(env);
+}
+
+static void ThreadDestructor(void* prev_jni_ptr) {
+  // This function only runs on threads where |g_jni_ptr| is non-NULL, meaning
+  // we were responsible for originally attaching the thread, so are responsible
+  // for detaching it now.  However, because some JVM implementations (notably
+  // Oracle's http://goo.gl/eHApYT) also use the pthread_key_create mechanism,
+  // the JVMs accounting info for this thread may already be wiped out by the
+  // time this is called. Thus it may appear we are already detached even though
+  // it was our responsibility to detach!  Oh well.
+  if (!GetEnv())
+    return;
+
+  CHECK(GetEnv() == prev_jni_ptr,
+        "Detaching from another thread: " << prev_jni_ptr << ":" << GetEnv());
   jint status = g_jvm->DetachCurrentThread();
   CHECK(status == JNI_OK, "Failed to detach thread: " << status);
+  CHECK(!GetEnv(), "Detaching was a successful no-op???");
 }
 
 static void CreateJNIPtrKey() {
@@ -180,28 +206,29 @@ static void CreateJNIPtrKey() {
         "pthread_key_create");
 }
 
-// Deal with difference in signatures between Oracle's jni.h and Android's.
+// Return a |JNIEnv*| usable on this thread.  Attaches to |g_jvm| if necessary.
 static JNIEnv* AttachCurrentThreadIfNeeded() {
-  CHECK(!pthread_once(&g_jni_ptr_once, &CreateJNIPtrKey),
+  JNIEnv* jni = GetEnv();
-        "pthread_once");
+  if (jni)
-  JNIEnv* jni = reinterpret_cast<JNIEnv*>(pthread_getspecific(g_jni_ptr));
+    return jni;
-  if (jni == NULL) {
+  CHECK(!pthread_getspecific(g_jni_ptr), "TLS has a JNIEnv* but not attached?");
+
+  char* name = strdup((GetThreadName() + " - " + GetThreadId()).c_str());
+  JavaVMAttachArgs args;
+  args.version = JNI_VERSION_1_6;
+  args.name = name;
+  args.group = NULL;
+  // Deal with difference in signatures between Oracle's jni.h and Android's.
 #ifdef _JAVASOFT_JNI_H_  // Oracle's jni.h violates the JNI spec!
-    void* env;
+  void* env = NULL;
 #else
-    JNIEnv* env;
+  JNIEnv* env = NULL;
 #endif
-    char* name = strdup((GetThreadName() + " - " + GetThreadId()).c_str());
+  CHECK(!g_jvm->AttachCurrentThread(&env, &args), "Failed to attach thread");
-    JavaVMAttachArgs args;
+  free(name);
-    args.version = JNI_VERSION_1_6;
+  CHECK(env, "AttachCurrentThread handed back NULL!");
-    args.name = name;
+  jni = reinterpret_cast<JNIEnv*>(env);
-    args.group = NULL;
+  CHECK(!pthread_setspecific(g_jni_ptr, jni), "pthread_setspecific");
-    CHECK(!g_jvm->AttachCurrentThread(&env, &args), "Failed to attach thread");
-    free(name);
-    CHECK(env, "AttachCurrentThread handed back NULL!");
-    jni = reinterpret_cast<JNIEnv*>(env);
-    CHECK(!pthread_setspecific(g_jni_ptr, jni), "pthread_setspecific");
-  }
   return jni;
 }
 
@@ -1710,6 +1737,8 @@ extern "C" jint JNIEXPORT JNICALL JNI_OnLoad(JavaVM *jvm, void *reserved) {
   g_jvm = jvm;
   CHECK(g_jvm, "JNI_OnLoad handed NULL?");
 
+  CHECK(!pthread_once(&g_jni_ptr_once, &CreateJNIPtrKey), "pthread_once");
+
   CHECK(talk_base::InitializeSSL(), "Failed to InitializeSSL()");
 
   JNIEnv* jni;
@@ -1725,6 +1754,7 @@ extern "C" void JNIEXPORT JNICALL JNI_OnUnLoad(JavaVM *jvm, void *reserved) {
   delete g_class_reference_holder;
   g_class_reference_holder = NULL;
   CHECK(talk_base::CleanupSSL(), "Failed to CleanupSSL()");
+  g_jvm = NULL;
 }
 
 static DataChannelInterface* ExtractNativeDC(JNIEnv* jni, jobject j_dc) {

talk/app/webrtc/javatests/src/org/webrtc/PeerConnectionTest.java

@@ -505,6 +505,28 @@ public class PeerConnectionTest extends TestCase {
 
   @Test
   public void testCompleteSession() throws Exception {
+    doTest();
+  }
+
+  @Test
+  public void testCompleteSessionOnNonMainThread() throws Exception {
+    final Exception[] exceptionHolder = new Exception[1];
+    Thread nonMainThread = new Thread("PeerConnectionTest-nonMainThread") {
+        @Override public void run() {
+          try {
+            doTest();
+          } catch (Exception e) {
+            exceptionHolder[0] = e;
+          }
+        }
+      };
+    nonMainThread.start();
+    nonMainThread.join();
+    if (exceptionHolder[0] != null)
+      throw exceptionHolder[0];
+  }
+
+  private void doTest() throws Exception {
     CountDownLatch testDone = new CountDownLatch(1);
     System.gc();  // Encourage any GC-related threads to start up.
     //TreeSet<String> threadsBeforeTest = allThreads();

talk/media/devices/linuxdevicemanager.cc

@@ -302,6 +302,12 @@ LinuxDeviceWatcher::LinuxDeviceWatcher(DeviceManagerInterface* dm)
 LinuxDeviceWatcher::~LinuxDeviceWatcher() {
 }
 
+static talk_base::PhysicalSocketServer* CurrentSocketServer() {
+  talk_base::SocketServer* ss =
+      talk_base::ThreadManager::Instance()->WrapCurrentThread()->socketserver();
+  return reinterpret_cast<talk_base::PhysicalSocketServer*>(ss);
+}
+
 bool LinuxDeviceWatcher::Start() {
   // We deliberately return true in the failure paths here because libudev is
   // not a critical component of a Linux system so it may not be present/usable,
@@ -341,16 +347,14 @@ bool LinuxDeviceWatcher::Start() {
     LOG_ERR(LS_ERROR) << "udev_monitor_enable_receiving()";
     return true;
   }
-  static_cast<talk_base::PhysicalSocketServer*>(
+  CurrentSocketServer()->Add(this);
-      talk_base::Thread::Current()->socketserver())->Add(this);
   registered_ = true;
   return true;
 }
 
 void LinuxDeviceWatcher::Stop() {
   if (registered_) {
-    static_cast<talk_base::PhysicalSocketServer*>(
+    CurrentSocketServer()->Remove(this);
-        talk_base::Thread::Current()->socketserver())->Remove(this);
     registered_ = false;
   }
   if (udev_monitor_) {
@@ -380,8 +384,7 @@ void LinuxDeviceWatcher::OnEvent(uint32 ff, int err) {
     LOG_ERR(LS_WARNING) << "udev_monitor_receive_device()";
     // Stop listening to avoid potential livelock (an fd with EOF in it is
     // always considered readable).
-    static_cast<talk_base::PhysicalSocketServer*>(
+    CurrentSocketServer()->Remove(this);
-        talk_base::Thread::Current()->socketserver())->Remove(this);
     registered_ = false;
     return;
   }