Stop scheduling unnecessary flush operations

Each UsageTracker's log call will schedule a flush on a background
thread. Flush has a synchronized block - if a single flush
takes a long time, all other threads will wait for it to finish.

This can result in thousands of background threads waiting to
acquire the lock.

All these threads are unnecessary to start if flush is already
happening.

The fix contains two improvements:

* New logs will only schedule flush operation if no other call scheduled
  it yet and flush is not currently running

* tryFlush skips flushing if one is already happening

Both tryFlush and flush will schedule another flush at the end if there
is any item pending to be flushed. This avoids concurrency problem
where it would be possible to add item to the pending queue with no
pending flush, keeping the event way too long in the queue.

flush() is still blocking operation as it is required to properly
handle pending events on app close.

Fixes: 295909822
Test: Manual, triggered high use of usage tracker while monitoring
  number of thread from the thread pool
Change-Id: I4df060b2774ffae77fc394252eb0e9b674e01f65
(cherry picked from commit 22f97b11f5603c3f82b52aaef985688aa052a262)

1 files changed, 51 insertions, 7 deletions

diff --git a/tracker/src/main/java/com/android/tools/analytics/JournalingUsageTracker.kt b/tracker/src/main/java/com/android/tools/analytics/JournalingUsageTracker.kt
index 881b3a2..e9008a9 100755
--- a/tracker/src/main/java/com/android/tools/analytics/JournalingUsageTracker.kt
+++ b/tracker/src/main/java/com/android/tools/analytics/JournalingUsageTracker.kt
@@ -33,6 +33,9 @@ import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.ScheduledFuture
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
 
 /**
  * a UsageTracker that uses a spool to journal the logs tracked. This tracker can be used in both
@@ -65,6 +68,9 @@ class JournalingUsageTracker
     private val spoolLocation: Path
 ) : UsageTrackerWriter() {
 
+    // lock for blocking flush calls. To avoid deadlocks, the order of
+    // locking is: flushLock (if needed), gate.
+    private val flushLock = ReentrantLock()
     private val gate = Any()
     private var lock: FileLock? = null
     private var channel: FileChannel? = null
@@ -75,14 +81,9 @@ class JournalingUsageTracker
 
     @Volatile
     private var state = State.Open
+    private val flushScheduled = AtomicBoolean(false)
     private val pendingEvents: Queue<ClientAnalytics.LogEvent.Builder> =
         ConcurrentLinkedQueue<ClientAnalytics.LogEvent.Builder>()
-    private val quietFlushRunnable = Runnable {
-        try {
-            flush()
-        } catch (ignored: IOException) {
-        }
-    }
 
     private enum class State {
         Open,
@@ -179,7 +180,7 @@ class JournalingUsageTracker
             return
         }
         pendingEvents.add(logEvent)
-        scheduler.submit(quietFlushRunnable)
+        scheduleFlush()
     }
 
     /**
@@ -188,6 +189,49 @@ class JournalingUsageTracker
      * @throws IOException on failure, and the UsageTracker will be closed in those cases.
      */
     override fun flush() {
+        flushLock.withLock {
+            flushImpl()
+        }
+        // If there are new events to handle, schedule a new flush event
+        if (pendingEvents.isNotEmpty()) {
+            scheduleFlush()
+        }
+    }
+
+    /**
+    Schedules a flush if one is not running and not scheduled yet.
+     */
+    private fun scheduleFlush() {
+        if (!flushLock.isLocked && flushScheduled.compareAndSet(false, true)) {
+            scheduler.submit {
+                try {
+                    tryFlush()
+                } finally {
+                    flushScheduled.set(false)
+                }
+            }
+        }
+    }
+
+    /**
+    Triggers flush if one is not currently running.
+     */
+    private fun tryFlush() {
+        if (!flushLock.tryLock())
+            return
+
+        try {
+            flushImpl()
+        } finally {
+            flushLock.unlock()
+        }
+        // If there are new events to handle, schedule a new flush event
+        if (pendingEvents.isNotEmpty()) {
+            scheduleFlush()
+        }
+    }
+
+    private fun flushImpl() {
         while (true) {
             synchronized(gate) {
                 val logEvent = pendingEvents.poll() ?: return