Implement launch bounds logic in Android (1/3)
Abstract threading logic that can be shared among persisters into a
common class. Later we'll introduce a new persister that stores mapping
from components to launch params (such as launch bounds and display
unique ID). That persister could reuse this class for threading which
could avoid creating another thread for persistence.
Also added some unit tests for the threading logic.
Bug: 113252871
Test: Manually test that recent tasks can still be restored across
reboots. atest PersisterThreadingTests. TaskPersisterTest is broken in
ToT.
Change-Id: I2b48593f38efcc205c2a213dbd93607c9588b12c
diff --git a/services/core/java/com/android/server/am/PersisterQueue.java b/services/core/java/com/android/server/am/PersisterQueue.java
new file mode 100644
index 0000000..60ea0fa
--- /dev/null
+++ b/services/core/java/com/android/server/am/PersisterQueue.java
@@ -0,0 +1,277 @@
+/*
+ * Copyright (C) 2018 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.android.server.am;
+
+import android.os.Process;
+import android.os.SystemClock;
+import android.util.Slog;
+
+import com.android.internal.annotations.VisibleForTesting;
+
+import java.util.ArrayList;
+import java.util.function.Predicate;
+
+/**
+ * The common threading logic for persisters to use so that they can run in the same threads.
+ * Methods in this class are synchronized on its instance, so caller could also synchronize on
+ * its instance to perform modifications in items.
+ */
+class PersisterQueue {
+ private static final String TAG = "PersisterQueue";
+ private static final boolean DEBUG = false;
+
+ /** When not flushing don't write out files faster than this */
+ private static final long INTER_WRITE_DELAY_MS = 500;
+
+ /**
+ * When not flushing delay this long before writing the first file out. This gives the next task
+ * being launched a chance to load its resources without this occupying IO bandwidth.
+ */
+ private static final long PRE_TASK_DELAY_MS = 3000;
+
+ /** The maximum number of entries to keep in the queue before draining it automatically. */
+ private static final int MAX_WRITE_QUEUE_LENGTH = 6;
+
+ /** Special value for mWriteTime to mean don't wait, just write */
+ private static final long FLUSH_QUEUE = -1;
+
+ /** An {@link WriteQueueItem} that doesn't do anything. Used to trigger {@link
+ * Listener#onPreProcessItem}. */
+ static final WriteQueueItem EMPTY_ITEM = () -> { };
+
+ private final long mInterWriteDelayMs;
+ private final long mPreTaskDelayMs;
+ private final LazyTaskWriterThread mLazyTaskWriterThread;
+ private final ArrayList<WriteQueueItem> mWriteQueue = new ArrayList<>();
+
+ private final ArrayList<Listener> mListeners = new ArrayList<>();
+
+ /**
+ * Value determines write delay mode as follows: < 0 We are Flushing. No delays between writes
+ * until the image queue is drained and all tasks needing persisting are written to disk. There
+ * is no delay between writes. == 0 We are Idle. Next writes will be delayed by
+ * #PRE_TASK_DELAY_MS. > 0 We are Actively writing. Next write will be at this time. Subsequent
+ * writes will be delayed by #INTER_WRITE_DELAY_MS.
+ */
+ private long mNextWriteTime = 0;
+
+ PersisterQueue() {
+ this(INTER_WRITE_DELAY_MS, PRE_TASK_DELAY_MS);
+ }
+
+ /** Used for tests to reduce waiting time. */
+ @VisibleForTesting
+ PersisterQueue(long interWriteDelayMs, long preTaskDelayMs) {
+ if (interWriteDelayMs < 0 || preTaskDelayMs < 0) {
+ throw new IllegalArgumentException("Both inter-write delay and pre-task delay need to"
+ + "be non-negative. inter-write delay: " + interWriteDelayMs
+ + "ms pre-task delay: " + preTaskDelayMs);
+ }
+ mInterWriteDelayMs = interWriteDelayMs;
+ mPreTaskDelayMs = preTaskDelayMs;
+ mLazyTaskWriterThread = new LazyTaskWriterThread("LazyTaskWriterThread");
+ }
+
+ synchronized void startPersisting() {
+ if (!mLazyTaskWriterThread.isAlive()) {
+ mLazyTaskWriterThread.start();
+ }
+ }
+
+ /** Stops persisting thread. Should only be used in tests. */
+ @VisibleForTesting
+ void stopPersisting() throws InterruptedException {
+ if (!mLazyTaskWriterThread.isAlive()) {
+ return;
+ }
+
+ synchronized (this) {
+ mLazyTaskWriterThread.interrupt();
+ }
+ mLazyTaskWriterThread.join();
+ }
+
+ synchronized void addItem(WriteQueueItem item, boolean flush) {
+ mWriteQueue.add(item);
+
+ if (flush || mWriteQueue.size() > MAX_WRITE_QUEUE_LENGTH) {
+ mNextWriteTime = FLUSH_QUEUE;
+ } else if (mNextWriteTime == 0) {
+ mNextWriteTime = SystemClock.uptimeMillis() + mPreTaskDelayMs;
+ }
+ notify();
+ }
+
+ synchronized <T extends WriteQueueItem> T findLastItem(Predicate<T> predicate, Class<T> clazz) {
+ for (int i = mWriteQueue.size() - 1; i >= 0; --i) {
+ WriteQueueItem writeQueueItem = mWriteQueue.get(i);
+ if (clazz.isInstance(writeQueueItem)) {
+ T item = clazz.cast(writeQueueItem);
+ if (predicate.test(item)) {
+ return item;
+ }
+ }
+ }
+
+ return null;
+ }
+
+ synchronized <T extends WriteQueueItem> void removeItems(Predicate<T> predicate,
+ Class<T> clazz) {
+ for (int i = mWriteQueue.size() - 1; i >= 0; --i) {
+ WriteQueueItem writeQueueItem = mWriteQueue.get(i);
+ if (clazz.isInstance(writeQueueItem)) {
+ T item = clazz.cast(writeQueueItem);
+ if (predicate.test(item)) {
+ if (DEBUG) Slog.d(TAG, "Removing " + item + " from write queue.");
+ mWriteQueue.remove(i);
+ }
+ }
+ }
+ }
+
+ synchronized void flush() {
+ mNextWriteTime = FLUSH_QUEUE;
+ notifyAll();
+ do {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ }
+ } while (mNextWriteTime == FLUSH_QUEUE);
+ }
+
+ void yieldIfQueueTooDeep() {
+ boolean stall = false;
+ synchronized (this) {
+ if (mNextWriteTime == FLUSH_QUEUE) {
+ stall = true;
+ }
+ }
+ if (stall) {
+ Thread.yield();
+ }
+ }
+
+ void addListener(Listener listener) {
+ mListeners.add(listener);
+ }
+
+ private void processNextItem() throws InterruptedException {
+ // This part is extracted into a method so that the GC can clearly see the end of the
+ // scope of the variable 'item'. If this part was in the loop in LazyTaskWriterThread, the
+ // last item it processed would always "leak".
+ // See https://b.corp.google.com/issues/64438652#comment7
+
+ // If mNextWriteTime, then don't delay between each call to saveToXml().
+ final WriteQueueItem item;
+ synchronized (this) {
+ if (mNextWriteTime != FLUSH_QUEUE) {
+ // The next write we don't have to wait so long.
+ mNextWriteTime = SystemClock.uptimeMillis() + mInterWriteDelayMs;
+ if (DEBUG) {
+ Slog.d(TAG, "Next write time may be in " + mInterWriteDelayMs
+ + " msec. (" + mNextWriteTime + ")");
+ }
+ }
+
+ while (mWriteQueue.isEmpty()) {
+ if (mNextWriteTime != 0) {
+ mNextWriteTime = 0; // idle.
+ notify(); // May need to wake up flush().
+ }
+ // Make sure we exit this thread correctly when interrupted before going to
+ // indefinite wait.
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException();
+ }
+ if (DEBUG) Slog.d(TAG, "LazyTaskWriter: waiting indefinitely.");
+ wait();
+ // Invariant: mNextWriteTime is either FLUSH_QUEUE or PRE_WRITE_DELAY_MS
+ // from now.
+ }
+ item = mWriteQueue.remove(0);
+
+ long now = SystemClock.uptimeMillis();
+ if (DEBUG) {
+ Slog.d(TAG, "LazyTaskWriter: now=" + now + " mNextWriteTime=" + mNextWriteTime
+ + " mWriteQueue.size=" + mWriteQueue.size());
+ }
+ while (now < mNextWriteTime) {
+ if (DEBUG) {
+ Slog.d(TAG, "LazyTaskWriter: waiting " + (mNextWriteTime - now));
+ }
+ wait(mNextWriteTime - now);
+ now = SystemClock.uptimeMillis();
+ }
+
+ // Got something to do.
+ }
+
+ item.process();
+ }
+
+ interface WriteQueueItem {
+ void process();
+ }
+
+ interface Listener {
+ /**
+ * Called before {@link PersisterQueue} tries to process next item.
+ *
+ * Note if the queue is empty, this callback will be called before the indefinite wait. This
+ * will be called once when {@link PersisterQueue} starts the internal thread before the
+ * indefinite wait.
+ *
+ * This callback is called w/o locking the instance of {@link PersisterQueue}.
+ *
+ * @param queueEmpty {@code true} if the queue is empty, which indicates {@link
+ * PersisterQueue} is likely to enter indefinite wait; or {@code false} if there is still
+ * item to process.
+ */
+ void onPreProcessItem(boolean queueEmpty);
+ }
+
+ private class LazyTaskWriterThread extends Thread {
+
+ private LazyTaskWriterThread(String name) {
+ super(name);
+ }
+
+ @Override
+ public void run() {
+ Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
+ try {
+ while (true) {
+ final boolean probablyDone;
+ synchronized (PersisterQueue.this) {
+ probablyDone = mWriteQueue.isEmpty();
+ }
+
+ for (int i = mListeners.size() - 1; i >= 0; --i) {
+ mListeners.get(i).onPreProcessItem(probablyDone);
+ }
+
+ processNextItem();
+ }
+ } catch (InterruptedException e) {
+ Slog.e(TAG, "Persister thread is exiting. Should never happen in prod, but"
+ + "it's OK in tests.");
+ }
+ }
+ }
+}