content: /*
   * 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;
      }
  
      /**
       * Updates the last item found in the queue that matches the given item, or adds it to the end
       * of the queue if no such item is found.
       */
      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}.
       */
      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.");
              }
          }
      }
  }
  