refactor: performance improvements in batch message processing, synchronized cache access and audible message notifications.

Increase audible timeout on DefaultMessageNotifier.java, don't send in-thread notification based on last audible notification. Create a batch message receive job to handle up to 20 chunked messages at a time per job instead of singular or open group poll amount Remove synchronized access to recipient cache and replace with a concurrent cache that's lock free from perf tracing monitor contention
2025-08-11 19:07:40 +00:00 · 2021-09-29 15:29:24 +10:00
parent 5a290ddf68
commit e036344c76
13 changed files with 700 additions and 52 deletions
--- a/libsession/src/main/java/org/session/libsession/database/StorageProtocol.kt
+++ b/libsession/src/main/java/org/session/libsession/database/StorageProtocol.kt
@@ -5,7 +5,6 @@ import android.net.Uri
 import org.session.libsession.messaging.contacts.Contact
 import org.session.libsession.messaging.jobs.AttachmentUploadJob
 import org.session.libsession.messaging.jobs.Job
-import org.session.libsession.messaging.jobs.MessageReceiveJob
 import org.session.libsession.messaging.jobs.MessageSendJob
 import org.session.libsession.messaging.messages.control.ConfigurationMessage
 import org.session.libsession.messaging.messages.visible.Attachment
@@ -43,7 +42,7 @@ interface StorageProtocol {
    fun getAllPendingJobs(type: String): Map<String,Job?>
    fun getAttachmentUploadJob(attachmentID: Long): AttachmentUploadJob?
    fun getMessageSendJob(messageSendJobID: String): MessageSendJob?
-    fun getMessageReceiveJob(messageReceiveJobID: String): MessageReceiveJob?
+    fun getMessageReceiveJob(messageReceiveJobID: String): Job?
    fun resumeMessageSendJobIfNeeded(messageSendJobID: String)
    fun isJobCanceled(job: Job): Boolean

--- a/libsession/src/main/java/org/session/libsession/messaging/jobs/BatchMessageReceiveJob.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/jobs/BatchMessageReceiveJob.kt
@@ -0,0 +1,120 @@
+package org.session.libsession.messaging.jobs
+
+import com.google.protobuf.ByteString
+import nl.komponents.kovenant.Promise
+import nl.komponents.kovenant.task
+import org.session.libsession.messaging.sending_receiving.MessageReceiver
+import org.session.libsession.messaging.sending_receiving.handle
+import org.session.libsession.messaging.utilities.Data
+import org.session.libsignal.protos.UtilProtos
+import org.session.libsignal.utilities.Log
+
+data class MessageReceiveParameters(
+    val data: ByteArray,
+    val serverHash: String? = null,
+    val openGroupMessageServerID: Long? = null
+)
+
+class BatchMessageReceiveJob(
+    val messages: List<MessageReceiveParameters>,
+    val openGroupID: String? = null
+) : Job {
+
+    override var delegate: JobDelegate? = null
+    override var id: String? = null
+    override var failureCount: Int = 0
+    override val maxFailureCount: Int = 10
+    // Failure Exceptions must be retryable if they're a  MessageReceiver.Error
+    val failures = mutableListOf<MessageReceiveParameters>()
+
+    companion object {
+        const val TAG = "BatchMessageReceiveJob"
+        const val KEY = "BatchMessageReceiveJob"
+
+        // Keys used for database storage
+        private val NUM_MESSAGES_KEY = "numMessages"
+        private val DATA_KEY = "data"
+        private val SERVER_HASH_KEY = "serverHash"
+        private val OPEN_GROUP_MESSAGE_SERVER_ID_KEY = "openGroupMessageServerID"
+        private val OPEN_GROUP_ID_KEY = "open_group_id"
+    }
+
+    override fun execute() {
+        executeAsync().get()
+    }
+
+    fun executeAsync(): Promise<Unit, Exception> {
+        return task {
+            messages.forEach { messageParameters ->
+                val (data, serverHash, openGroupMessageServerID) = messageParameters
+                try {
+                    val (message, proto) = MessageReceiver.parse(data, openGroupMessageServerID)
+                    message.serverHash = serverHash
+                    MessageReceiver.handle(message, proto, this.openGroupID)
+                } catch (e: Exception) {
+                    Log.e(TAG, "Couldn't receive message.", e)
+                    if (e is MessageReceiver.Error && !e.isRetryable) {
+                        Log.e(TAG, "Message failed permanently",e)
+                    } else {
+                        Log.e(TAG, "Message failed",e)
+                        failures += messageParameters
+                    }
+                }
+            }
+            if (failures.isEmpty()) {
+                handleSuccess()
+            } else {
+                handleFailure()
+            }
+        }
+    }
+
+    private fun handleSuccess() {
+        this.delegate?.handleJobSucceeded(this)
+    }
+
+    private fun handleFailure() {
+        this.delegate?.handleJobFailed(this, Exception("One or more jobs resulted in failure"))
+    }
+
+    override fun serialize(): Data {
+        val arraySize = messages.size
+        val dataArrays = UtilProtos.ByteArrayList.newBuilder()
+            .addAllContent(messages.map(MessageReceiveParameters::data).map(ByteString::copyFrom))
+            .build()
+        val serverHashes = messages.map { it.serverHash.orEmpty() }
+        val openGroupServerIds = messages.map { it.openGroupMessageServerID ?: -1L }
+        return Data.Builder()
+            .putInt(NUM_MESSAGES_KEY, arraySize)
+            .putByteArray(DATA_KEY, dataArrays.toByteArray())
+            .putString(OPEN_GROUP_ID_KEY, openGroupID)
+            .putLongArray(OPEN_GROUP_MESSAGE_SERVER_ID_KEY, openGroupServerIds.toLongArray())
+            .putStringArray(SERVER_HASH_KEY, serverHashes.toTypedArray())
+            .build()
+    }
+
+    override fun getFactoryKey(): String = KEY
+
+    class Factory : Job.Factory<BatchMessageReceiveJob> {
+        override fun create(data: Data): BatchMessageReceiveJob {
+            val numMessages = data.getInt(NUM_MESSAGES_KEY)
+            val dataArrays = data.getByteArray(DATA_KEY)
+            val contents =
+                UtilProtos.ByteArrayList.parseFrom(dataArrays).contentList.map(ByteString::toByteArray)
+            val serverHashes =
+                if (data.hasStringArray(SERVER_HASH_KEY)) data.getStringArray(SERVER_HASH_KEY) else arrayOf()
+            val openGroupMessageServerIDs = data.getLongArray(OPEN_GROUP_MESSAGE_SERVER_ID_KEY)
+            val openGroupID = data.getStringOrDefault(OPEN_GROUP_ID_KEY, null)
+
+            val parameters = (0 until numMessages).map { index ->
+                val data = contents[index]
+                val serverHash = serverHashes[index].let { if (it.isEmpty()) null else it }
+                val serverId = openGroupMessageServerIDs[index].let { if (it == -1L) null else it }
+                MessageReceiveParameters(data, serverHash, serverId)
+            }
+
+            return BatchMessageReceiveJob(parameters, openGroupID)
+        }
+    }
+
+}
--- a/libsession/src/main/java/org/session/libsession/messaging/jobs/JobQueue.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/jobs/JobQueue.kt
@@ -5,7 +5,6 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import org.session.libsession.messaging.MessagingModuleConfiguration
 import org.session.libsignal.utilities.Log
-import java.lang.IllegalStateException
 import java.util.*
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.Executors
@@ -51,7 +50,7 @@ class JobQueue : JobDelegate {
                    when (job) {
                        is NotifyPNServerJob, is AttachmentUploadJob, is MessageSendJob -> txQueue.send(job)
                        is AttachmentDownloadJob -> attachmentQueue.send(job)
-                        is MessageReceiveJob, is TrimThreadJob -> rxQueue.send(job)
+                        is MessageReceiveJob, is BatchMessageReceiveJob, is TrimThreadJob -> rxQueue.send(job)
                        else -> throw IllegalStateException("Unexpected job type.")
                    }
                }
@@ -128,7 +127,8 @@ class JobQueue : JobDelegate {
            AttachmentDownloadJob.KEY,
            MessageReceiveJob.KEY,
            MessageSendJob.KEY,
-            NotifyPNServerJob.KEY
+            NotifyPNServerJob.KEY,
+            BatchMessageReceiveJob.KEY
        )
        allJobTypes.forEach { type ->
            resumePendingJobs(type)
@@ -152,6 +152,11 @@ class JobQueue : JobDelegate {
            Log.i("Loki", "Message send job waiting for attachment upload to finish.")
            return
        }
+        // Batch message receive job, re-queue non-permanently failed jobs
+        if (job is BatchMessageReceiveJob) {
+            val replacementParameters = job.failures
+        }
+
        // Regular job failure
        job.failureCount += 1
        if (job.failureCount >= job.maxFailureCount) {
--- a/libsession/src/main/java/org/session/libsession/messaging/jobs/MessageReceiveJob.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/jobs/MessageReceiveJob.kt
@@ -17,8 +17,6 @@ class MessageReceiveJob(val data: ByteArray, val serverHash: String? = null, val
        val TAG = MessageReceiveJob::class.simpleName
        val KEY: String = "MessageReceiveJob"

-        private val RECEIVE_LOCK = Object()
-
        // Keys used for database storage
        private val DATA_KEY = "data"
        private val SERVER_HASH_KEY = "serverHash"
@@ -36,9 +34,7 @@ class MessageReceiveJob(val data: ByteArray, val serverHash: String? = null, val
            val isRetry: Boolean = failureCount != 0
            val (message, proto) = MessageReceiver.parse(this.data, this.openGroupMessageServerID)
            message.serverHash = serverHash
-            synchronized(RECEIVE_LOCK) { // FIXME: Do we need this?
-                MessageReceiver.handle(message, proto, this.openGroupID)
-            }
+            MessageReceiver.handle(message, proto, this.openGroupID)
            this.handleSuccess()
            deferred.resolve(Unit)
        } catch (e: Exception) {
@@ -82,11 +78,15 @@ class MessageReceiveJob(val data: ByteArray, val serverHash: String? = null, val
    class Factory: Job.Factory<MessageReceiveJob> {

        override fun create(data: Data): MessageReceiveJob {
+            val dataArray = data.getByteArray(DATA_KEY)
+            val serverHash = data.getStringOrDefault(SERVER_HASH_KEY, null)
+            val openGroupMessageServerID = data.getLongOrDefault(OPEN_GROUP_MESSAGE_SERVER_ID_KEY, -1).let { if (it == -1L) null else it }
+            val openGroupID = data.getStringOrDefault(OPEN_GROUP_ID_KEY, null)
            return MessageReceiveJob(
-                data.getByteArray(DATA_KEY),
-                data.getString(SERVER_HASH_KEY),
-                data.getLong(OPEN_GROUP_MESSAGE_SERVER_ID_KEY),
-                data.getString(OPEN_GROUP_ID_KEY)
+                dataArray,
+                serverHash,
+                openGroupMessageServerID,
+                openGroupID
            )
        }
    }
--- a/libsession/src/main/java/org/session/libsession/messaging/jobs/SessionJobManagerFactories.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/jobs/SessionJobManagerFactories.kt
@@ -11,7 +11,8 @@ class SessionJobManagerFactories {
                MessageReceiveJob.KEY to MessageReceiveJob.Factory(),
                MessageSendJob.KEY to MessageSendJob.Factory(),
                NotifyPNServerJob.KEY to NotifyPNServerJob.Factory(),
-                TrimThreadJob.KEY to TrimThreadJob.Factory()
+                TrimThreadJob.KEY to TrimThreadJob.Factory(),
+                BatchMessageReceiveJob.KEY to BatchMessageReceiveJob.Factory()
            )
        }
    }
--- a/libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/OpenGroupPollerV2.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/OpenGroupPollerV2.kt
@@ -3,17 +3,12 @@ package org.session.libsession.messaging.sending_receiving.pollers
 import nl.komponents.kovenant.Promise
 import nl.komponents.kovenant.functional.map
 import org.session.libsession.messaging.MessagingModuleConfiguration
-import org.session.libsession.messaging.jobs.JobQueue
-import org.session.libsession.messaging.jobs.MessageReceiveJob
-import org.session.libsession.messaging.jobs.TrimThreadJob
+import org.session.libsession.messaging.jobs.*
 import org.session.libsession.messaging.open_groups.OpenGroupAPIV2
 import org.session.libsession.messaging.open_groups.OpenGroupMessageV2
-import org.session.libsession.messaging.sending_receiving.MessageReceiver
-import org.session.libsession.messaging.sending_receiving.handle
 import org.session.libsession.utilities.Address
 import org.session.libsession.utilities.GroupUtil
 import org.session.libsignal.protos.SignalServiceProtos
-import org.session.libsignal.utilities.Log
 import org.session.libsignal.utilities.successBackground
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.ScheduledFuture
@@ -27,7 +22,7 @@ class OpenGroupPollerV2(private val server: String, private val executorService:
    private var future: ScheduledFuture<*>? = null

    companion object {
-        private val pollInterval: Long = 4 * 1000
+        private const val pollInterval: Long = 4000L
        const val maxInactivityPeriod = 14 * 24 * 60 * 60 * 1000
    }

@@ -66,21 +61,22 @@ class OpenGroupPollerV2(private val server: String, private val executorService:
        val threadId = storage.getThreadId(Address.fromSerialized(groupID)) ?: -1
        val threadExists = threadId >= 0
        if (!hasStarted || !threadExists) { return }
-        messages.sortedBy { it.serverID!! }.forEach { message ->
-            try {
-                val senderPublicKey = message.sender!!
-                val builder = SignalServiceProtos.Envelope.newBuilder()
-                builder.type = SignalServiceProtos.Envelope.Type.SESSION_MESSAGE
-                builder.source = senderPublicKey
-                builder.sourceDevice = 1
-                builder.content = message.toProto().toByteString()
-                builder.timestamp = message.sentTimestamp
-                val envelope = builder.build()
-                val (parsedMessage, content) = MessageReceiver.parse(envelope.toByteArray(), message.serverID)
-                MessageReceiver.handle(parsedMessage, content, openGroupID)
-            } catch (e: Exception) {
-                Log.e("Loki", "Exception parsing message", e)
+        val envelopes = messages.sortedBy { it.serverID!! }.map { message ->
+            val senderPublicKey = message.sender!!
+            val builder = SignalServiceProtos.Envelope.newBuilder()
+            builder.type = SignalServiceProtos.Envelope.Type.SESSION_MESSAGE
+            builder.source = senderPublicKey
+            builder.sourceDevice = 1
+            builder.content = message.toProto().toByteString()
+            builder.timestamp = message.sentTimestamp
+            builder.build() to message.serverID
+        }
+
+        envelopes.chunked(20).forEach { list ->
+            val parameters = list.map { (message, serverId) ->
+                MessageReceiveParameters(message.toByteArray(), openGroupMessageServerID = serverId)
            }
+            JobQueue.shared.add(BatchMessageReceiveJob(parameters, openGroupID))
        }

        val currentLastMessageServerID = storage.getLastMessageServerID(room, server) ?: 0
@@ -88,9 +84,6 @@ class OpenGroupPollerV2(private val server: String, private val executorService:
        if (actualMax > 0) {
            storage.setLastMessageServerID(room, server, actualMax)
        }
-        if (messages.isNotEmpty()) {
-            JobQueue.shared.add(TrimThreadJob(threadId))
-        }
    }

    private fun handleDeletedMessages(room: String, openGroupID: String, deletions: List<OpenGroupAPIV2.MessageDeletion>) {
--- a/libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt
+++ b/libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt
@@ -3,12 +3,13 @@ package org.session.libsession.messaging.sending_receiving.pollers
 import nl.komponents.kovenant.*
 import nl.komponents.kovenant.functional.bind
 import org.session.libsession.messaging.MessagingModuleConfiguration
+import org.session.libsession.messaging.jobs.BatchMessageReceiveJob
 import org.session.libsession.messaging.jobs.JobQueue
-import org.session.libsession.messaging.jobs.MessageReceiveJob
+import org.session.libsession.messaging.jobs.MessageReceiveParameters
 import org.session.libsession.snode.SnodeAPI
 import org.session.libsession.snode.SnodeModule
-import org.session.libsignal.utilities.Snode
 import org.session.libsignal.utilities.Log
+import org.session.libsignal.utilities.Snode
 import java.security.SecureRandom
 import java.util.*

@@ -91,10 +92,14 @@ class Poller {
                task { Unit } // The long polling connection has been canceled; don't recurse
            } else {
                val messages = SnodeAPI.parseRawMessagesResponse(rawResponse, snode, userPublicKey)
-                messages.forEach { (envelope, serverHash) ->
-                    val job = MessageReceiveJob(envelope.toByteArray(), serverHash)
+                val parameters = messages.map { (envelope, serverHash) ->
+                    MessageReceiveParameters(envelope.toByteArray(), serverHash = serverHash)
+                }
+                parameters.chunked(20).forEach { chunk ->
+                    val job = BatchMessageReceiveJob(chunk)
                    JobQueue.shared.add(job)
                }
+
                poll(snode, deferred)
            }
        }
--- a/libsession/src/main/java/org/session/libsession/utilities/TextSecurePreferences.kt
+++ b/libsession/src/main/java/org/session/libsession/utilities/TextSecurePreferences.kt
@@ -363,7 +363,7 @@ object TextSecurePreferences {

    @JvmStatic
    fun getDirectCaptureCameraId(context: Context): Int {
-        return getIntegerPreference(context, DIRECT_CAPTURE_CAMERA_ID, Camera.CameraInfo.CAMERA_FACING_FRONT)
+        return getIntegerPreference(context, DIRECT_CAPTURE_CAMERA_ID, Camera.CameraInfo.CAMERA_FACING_BACK)
    }

    @JvmStatic
--- a/libsession/src/main/java/org/session/libsession/utilities/recipients/RecipientProvider.java
+++ b/libsession/src/main/java/org/session/libsession/utilities/recipients/RecipientProvider.java
@@ -29,7 +29,6 @@ import org.session.libsession.utilities.Address;
 import org.session.libsession.utilities.GroupRecord;
 import org.session.libsession.utilities.ListenableFutureTask;
 import org.session.libsession.utilities.MaterialColor;
-import org.session.libsession.utilities.SoftHashMap;
 import org.session.libsession.utilities.TextSecurePreferences;
 import org.session.libsession.utilities.Util;
 import org.session.libsession.utilities.recipients.Recipient.RecipientSettings;
@@ -43,6 +42,7 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;

 class RecipientProvider {
@@ -222,17 +222,17 @@ class RecipientProvider {

  private static class RecipientCache {

-    private final Map<Address,Recipient> cache = new SoftHashMap<>(1000);
+    private final Map<Address,Recipient> cache = new ConcurrentHashMap<>(1000);

-    public synchronized Recipient get(Address address) {
+    public Recipient get(Address address) {
      return cache.get(address);
    }

-    public synchronized void set(Address address, Recipient recipient) {
+    public void set(Address address, Recipient recipient) {
      cache.put(address, recipient);
    }

-    public synchronized boolean remove(Address address) {
+    public boolean remove(Address address) {
      return cache.remove(address) != null;
    }