services/core/java/com/android/server/wm/PersisterQueue.java

/*
 * 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.wm;

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;
    }

    /**
     *
     * @param item
     * @param flush
     * @param <T>
     */
    synchronized <T extends WriteQueueItem> void updateLastOrAddItem(T item, boolean flush) {
        final T itemToUpdate = findLastItem(item::matches, (Class<T>) item.getClass());
        if (itemToUpdate == null) {
            addItem(item, flush);
        } else {
            itemToUpdate.updateFrom(item);
        }

        yieldIfQueueTooDeep();
    }

    /**
     * Removes all items with which given predicate returns {@code true}.
     *
     * @param predicate the predicate
     * @param clazz
     * @param <T>
     */
    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);
    }

    @VisibleForTesting
    boolean removeListener(Listener listener) {
        return mListeners.remove(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<T extends WriteQueueItem<T>> {
        void process();

        default void updateFrom(T item) {}

        default boolean matches(T item) {
            return false;
        }
    }

    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.");
            }
        }
    }
}