Improve paging performance on slower devices.

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationDataSource.java

@@ -13,10 +13,12 @@ import org.thoughtcrime.securesms.database.MmsSmsDatabase;
 import org.thoughtcrime.securesms.database.model.MessageRecord;
 import org.thoughtcrime.securesms.logging.Log;
 import org.thoughtcrime.securesms.util.Util;
+import org.thoughtcrime.securesms.util.concurrent.SignalExecutors;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
+import java.util.concurrent.Executor;
 
 /**
  * Core data source for loading an individual conversation.
@@ -25,11 +27,17 @@ class ConversationDataSource extends PositionalDataSource<MessageRecord> {
 
   private static final String TAG = Log.tag(ConversationDataSource.class);
 
+  public static final Executor EXECUTOR = SignalExecutors.newFixedLifoThreadExecutor("signal-conversation", 1, 1);
+
   private final Context             context;
   private final long                threadId;
   private final DataUpdatedCallback dataUpdateCallback;
 
-  private ConversationDataSource(@NonNull Context context, long threadId, @NonNull DataUpdatedCallback dataUpdateCallback) {
+  private ConversationDataSource(@NonNull Context context,
+                                 long threadId,
+                                 @NonNull Invalidator invalidator,
+                                 @NonNull DataUpdatedCallback dataUpdateCallback)
+  {
     this.context            = context;
     this.threadId           = threadId;
     this.dataUpdateCallback = dataUpdateCallback;
@@ -42,6 +50,8 @@ class ConversationDataSource extends PositionalDataSource<MessageRecord> {
       }
     };
 
+    invalidator.observe(this::invalidate);
+
     context.getContentResolver().registerContentObserver(DatabaseContentProviders.Conversation.getUriForThread(threadId), true, contentObserver);
   }
 
@@ -52,12 +62,16 @@ class ConversationDataSource extends PositionalDataSource<MessageRecord> {
     MmsSmsDatabase      db      = DatabaseFactory.getMmsSmsDatabase(context);
     List<MessageRecord> records = new ArrayList<>(params.requestedLoadSize);
 
+    if (!isInvalid()) {
       try (MmsSmsDatabase.Reader reader = db.readerFor(db.getConversation(threadId, params.requestedStartPosition, params.requestedLoadSize))) {
         MessageRecord record;
         while ((record = reader.getNext()) != null && !isInvalid()) {
           records.add(record);
         }
       }
+    } else {
+      Log.i(TAG, "[Initial Load] Invalidated before we could even query!");
+    }
 
     int effectiveCount = records.size() + params.requestedStartPosition;
     int totalCount     = db.getConversationCount(threadId);
@@ -85,12 +99,16 @@ class ConversationDataSource extends PositionalDataSource<MessageRecord> {
     MmsSmsDatabase      db      = DatabaseFactory.getMmsSmsDatabase(context);
     List<MessageRecord> records = new ArrayList<>(params.loadSize);
 
+    if (!isInvalid()) {
       try (MmsSmsDatabase.Reader reader = db.readerFor(db.getConversation(threadId, params.startPosition, params.loadSize))) {
         MessageRecord record;
         while ((record = reader.getNext()) != null && !isInvalid()) {
           records.add(record);
         }
       }
+    } else {
+      Log.i(TAG, "[Update] Invalidated before we could even query!");
+    }
 
     callback.onResult(records);
     Util.runOnMain(dataUpdateCallback::onDataUpdated);
@@ -111,21 +129,44 @@ class ConversationDataSource extends PositionalDataSource<MessageRecord> {
     void onDataUpdated();
   }
 
+  static class Invalidator {
+    private boolean  invalidated;
+    private Runnable callback;
+
+    synchronized void invalidate() {
+      invalidated = true;
+
+      if (callback != null) {
+        callback.run();
+      }
+    }
+
+    private synchronized void observe(@NonNull Runnable callback) {
+      if (invalidated) {
+        callback.run();
+      } else {
+        this.callback = callback;
+      }
+    }
+  }
+
   static class Factory extends DataSource.Factory<Integer, MessageRecord> {
 
     private final Context             context;
     private final long                threadId;
+    private final Invalidator         invalidator;
     private final DataUpdatedCallback callback;
 
-    Factory(Context context, long threadId, @NonNull DataUpdatedCallback callback) {
+    Factory(Context context, long threadId, @NonNull Invalidator invalidator, @NonNull DataUpdatedCallback callback) {
       this.context     = context;
       this.threadId    = threadId;
+      this.invalidator = invalidator;
       this.callback    = callback;
     }
 
     @Override
     public @NonNull DataSource<Integer, MessageRecord> create() {
-      return new ConversationDataSource(context, threadId, callback);
+      return new ConversationDataSource(context, threadId, invalidator, callback);
     }
   }
 }

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationViewModel.java

@@ -15,6 +15,7 @@ import androidx.paging.DataSource;
 import androidx.paging.LivePagedListBuilder;
 import androidx.paging.PagedList;
 
+import org.thoughtcrime.securesms.conversation.ConversationDataSource.Invalidator;
 import org.thoughtcrime.securesms.database.model.MessageRecord;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.logging.Log;
@@ -40,6 +41,7 @@ class ConversationViewModel extends ViewModel {
   private final LiveData<PagedList<MessageRecord>> messages;
   private final LiveData<ConversationData>         conversationMetadata;
   private final List<Runnable>                     onNextMessageLoad;
+  private final Invalidator                        invalidator;
 
   private int jumpToPosition;
 
@@ -50,15 +52,16 @@ class ConversationViewModel extends ViewModel {
     this.recentMedia            = new MutableLiveData<>();
     this.threadId               = new MutableLiveData<>();
     this.onNextMessageLoad      = new CopyOnWriteArrayList<>();
+    this.invalidator            = new Invalidator();
 
     LiveData<Pair<Long, PagedList<MessageRecord>>> messagesForThreadId = Transformations.switchMap(threadId, thread -> {
-      DataSource.Factory<Integer, MessageRecord> factory = new ConversationDataSource.Factory(context, thread, this::onMessagesUpdated);
+      DataSource.Factory<Integer, MessageRecord> factory = new ConversationDataSource.Factory(context, thread, invalidator, this::onMessagesUpdated);
       PagedList.Config                           config  = new PagedList.Config.Builder()
                                                                                .setPageSize(25)
                                                                                .setInitialLoadSizeHint(25)
                                                                                .build();
 
-      return Transformations.map(new LivePagedListBuilder<>(factory, config).setFetchExecutor(SignalExecutors.BOUNDED)
+      return Transformations.map(new LivePagedListBuilder<>(factory, config).setFetchExecutor(ConversationDataSource.EXECUTOR)
                                                                             .setInitialLoadKey(Math.max(jumpToPosition, 0))
                                                                             .build(),
                                  input -> new Pair<>(thread, input));
@@ -110,6 +113,12 @@ class ConversationViewModel extends ViewModel {
     onNextMessageLoad.add(runnable);
   }
 
+  @Override
+  protected void onCleared() {
+    super.onCleared();
+    invalidator.invalidate();
+  }
+
   private void onMessagesUpdated() {
     for (Runnable runnable : onNextMessageLoad) {
       runnable.run();

app/src/main/java/org/thoughtcrime/securesms/util/concurrent/SignalExecutors.java

@@ -2,6 +2,10 @@ package org.thoughtcrime.securesms.util.concurrent;
 
 import androidx.annotation.NonNull;
 
+import com.google.android.gms.common.util.concurrent.NumberedThreadFactory;
+
+import org.thoughtcrime.securesms.util.LinkedBlockingLifoQueue;
+
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -13,7 +17,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 public class SignalExecutors {
 
   public static final ExecutorService UNBOUNDED = Executors.newCachedThreadPool(new NumberedThreadFactory("signal-unbounded"));
-  public static final ExecutorService BOUNDED   = Executors.newFixedThreadPool(Math.max(2, Math.min(Runtime.getRuntime().availableProcessors() - 1, 4)), new NumberedThreadFactory("signal-bounded"));
+  public static final ExecutorService BOUNDED   = Executors.newFixedThreadPool(getIdealThreadCount(), new NumberedThreadFactory("signal-bounded"));
   public static final ExecutorService SERIAL    = Executors.newSingleThreadExecutor(new NumberedThreadFactory("signal-serial"));
 
   public static ExecutorService newCachedSingleThreadExecutor(final String name) {
@@ -22,6 +26,21 @@ public class SignalExecutors {
     return executor;
   }
 
+  /**
+   * Returns an executor that prioritizes newer work. This is the opposite of a traditional executor,
+   * which processor work in FIFO order.
+   */
+  public static ExecutorService newFixedLifoThreadExecutor(String name, int minThreads, int maxThreads) {
+    return new ThreadPoolExecutor(minThreads, maxThreads, 0, TimeUnit.MILLISECONDS, new LinkedBlockingLifoQueue<>(), new NumberedThreadFactory(name));
+  }
+
+  /**
+   * Returns an 'ideal' thread count based on the number of available processors.
+   */
+  public static int getIdealThreadCount() {
+    return Math.max(2, Math.min(Runtime.getRuntime().availableProcessors() - 1, 4));
+  }
+
   private static class NumberedThreadFactory implements ThreadFactory {
 
     private final String        baseName;