Make sure handler.flush() doesn't deadlock.

Currently it deadlocks during process termination, when
atexit first calls handler.close() and then logging.shutdown(),
that in turn calls handler.flush() without arguments.

handler.close() kills the worker, and then handler.flush() forever waits for the dead worker to send the messages from the queue.

After this change, the deadlock is still possible if something concurrently closes the handler from another thread during the flush. However, this scenario is much less likely.

contrib/opencensus-ext-azure/opencensus/ext/azure/log_exporter/__init__.py

@@ -106,6 +106,20 @@ def log_record_to_envelope(self, record):
         raise NotImplementedError  # pragma: NO COVER
 
     def flush(self, timeout=None):
+        if self._queue.is_empty():
+            return
+
+        # We must check the worker thread is alive, because otherwise flush
+        # is useless. Also, it would deadlock if not timeout is given, and the
+        # queue isn't empty.
+        # This is a very possible scenario during process termination, when
+        # atexit first calls handler.close() and then logging.shutdown(),
+        # that in turn calls handler.flush() without arguments.
+        if not self._worker.is_alive():
+            logger.warning("Can't flush %s, worker thread is dead. ",
+                           "Pending messages will be lost.", self)
+            return
+
         self._queue.flush(timeout=timeout)
 
 

opencensus/common/schedule/__init__.py

@@ -112,6 +112,9 @@ def _gets(self, count, timeout):
     def gets(self, count, timeout):
         return tuple(self._gets(count, timeout))
 
+    def is_empty(self):
+        return not self._queue.qsize()
+
     def flush(self, timeout=None):
         if self._queue.qsize() == 0:
             return 0
@@ -124,7 +127,7 @@ def flush(self, timeout=None):
             return
         elapsed_time = time.time() - start_time
         wait_time = timeout and max(timeout - elapsed_time, 0)
-        if event.wait(timeout):
+        if event.wait(wait_time):
             return time.time() - start_time  # time taken to flush
 
     def put(self, item, block=True, timeout=None):