Fix realm scheduler leak (#1463)

Author: clementetb

CHANGELOG.md

@@ -9,6 +9,7 @@
 ### Fixed
 * Rare corruption causing 'Invalid streaming format cookie'-exception. Typically following compact, convert or copying to a new file. (Issue [#1440](https://github.com/realm/realm-kotlin/issues/1440))
 * Compiler error when using Kotlin 1.9.0 and backlinks. (Issue [#1469](https://github.com/realm/realm-kotlin/issues/1469))
+* Leaking `JVMScheduler` instances. In certain circumstances, it could lead to a JNI crash. (Issue [#1463](https://github.com/realm/realm-kotlin/pull/1463))
 * [Sync] Changing a subscriptions query type or query itself will now trigger the `WaitForSync.FIRST_TIME` behaviour, rather than only checking changes to the name. (Issues [#1466](https://github.com/realm/realm-kotlin/issues/1466))
 
 ### Compatibility

packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/NativePointer.kt

@@ -43,3 +43,14 @@ public interface NativePointer<T : CapiT> {
      */
     public fun isReleased(): Boolean
 }
+
+/**
+ * Deletes the underlying pointer after executing the lambda block.
+ */
+fun <T : CapiT, R> NativePointer<T>.use(
+    block: (NativePointer<T>) -> R,
+): R = try {
+    block(this)
+} finally {
+    release()
+}

packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -70,6 +70,7 @@ interface RealmQueryT : CapiT
 interface RealmCallbackTokenT : CapiT
 interface RealmNotificationTokenT : CapiT
 interface RealmChangesT : CapiT
+interface RealmSchedulerT : CapiT
 
 // Public type aliases binding to internal verbose type safe type definitions. This should allow us
 // to easily change implementation details later on.
@@ -88,6 +89,7 @@ typealias RealmQueryPointer = NativePointer<RealmQueryT>
 typealias RealmCallbackTokenPointer = NativePointer<RealmCallbackTokenT>
 typealias RealmNotificationTokenPointer = NativePointer<RealmNotificationTokenT>
 typealias RealmChangesPointer = NativePointer<RealmChangesT>
+typealias RealmSchedulerPointer = NativePointer<RealmSchedulerT>
 
 // Sync types
 // Pure marker interfaces corresponding to the C-API realm_x_t struct types
@@ -190,6 +192,8 @@ expect object RealmInterop {
     fun realm_config_set_in_memory(config: RealmConfigurationPointer, inMemory: Boolean)
     fun realm_schema_validate(schema: RealmSchemaPointer, mode: SchemaValidationMode): Boolean
 
+    fun realm_create_scheduler(): RealmSchedulerPointer
+    fun realm_create_scheduler(dispatcher: CoroutineDispatcher): RealmSchedulerPointer
     /**
      * Open a realm on the current thread.
      *
@@ -211,7 +215,7 @@ expect object RealmInterop {
      */
     // The dispatcher argument is only used on Native to build a core scheduler dispatching to the
     // dispatcher. The realm itself must also be opened on the same thread
-    fun realm_open(config: RealmConfigurationPointer, dispatcher: CoroutineDispatcher? = null): Pair<LiveRealmPointer, Boolean>
+    fun realm_open(config: RealmConfigurationPointer, scheduler: RealmSchedulerPointer): Pair<LiveRealmPointer, Boolean>
 
     // Opening a Realm asynchronously. Only supported for synchronized realms.
     fun realm_open_synchronized(config: RealmConfigurationPointer): RealmAsyncOpenTaskPointer

packages/cinterop/src/jvm/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -33,7 +33,6 @@ import io.realm.kotlin.internal.interop.sync.SyncSessionResyncMode
 import io.realm.kotlin.internal.interop.sync.SyncUserIdentity
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.launch
 import org.mongodb.kbson.ObjectId
 
@@ -180,22 +179,26 @@ actual object RealmInterop {
         realmc.realm_config_set_in_memory(config.cptr(), inMemory)
     }
 
-    actual fun realm_open(config: RealmConfigurationPointer, dispatcher: CoroutineDispatcher?): Pair<LiveRealmPointer, Boolean> {
+    actual fun realm_create_scheduler(): RealmSchedulerPointer =
+        LongPointerWrapper(realmc.realm_scheduler_make_default())
+
+    actual fun realm_create_scheduler(dispatcher: CoroutineDispatcher): RealmSchedulerPointer =
+        LongPointerWrapper(realmc.realm_create_scheduler(JVMScheduler(dispatcher)))
+
+    actual fun realm_open(
+        config: RealmConfigurationPointer,
+        scheduler: RealmSchedulerPointer,
+    ): Pair<LiveRealmPointer, Boolean> {
         // Configure callback to track if the file was created as part of opening
         var fileCreated = false
         val callback = DataInitializationCallback {
             fileCreated = true
         }
         realm_config_set_data_initialization_function(config, callback)
 
-        // create a custom Scheduler for JVM if a Coroutine Dispatcher is provided other wise
-        // pass null to use the generic one
-        val realmPtr = LongPointerWrapper<LiveRealmT>(
-            realmc.open_realm_with_scheduler(
-                (config as LongPointerWrapper).ptr,
-                if (dispatcher != null) JVMScheduler(dispatcher) else null
-            )
-        )
+        realmc.realm_config_set_scheduler(config.cptr(), scheduler.cptr())
+        val realmPtr = LongPointerWrapper<LiveRealmT>(realmc.realm_open(config.cptr()))
+
         // Ensure that we can read version information, etc.
         realm_begin_read(realmPtr)
         return Pair(realmPtr, fileCreated)
@@ -2093,13 +2096,8 @@ private class JVMScheduler(dispatcher: CoroutineDispatcher) {
     val scope: CoroutineScope = CoroutineScope(dispatcher)
 
     fun notifyCore(schedulerPointer: Long) {
-        val function: suspend CoroutineScope.() -> Unit = {
+        scope.launch {
             realmc.invoke_core_notify_callback(schedulerPointer)
         }
-        scope.launch(
-            scope.coroutineContext,
-            CoroutineStart.DEFAULT,
-            function
-        )
     }
 }

packages/cinterop/src/nativeDarwin/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -19,7 +19,6 @@
 package io.realm.kotlin.internal.interop
 
 import io.realm.kotlin.internal.interop.Constants.ENCRYPTION_KEY_LENGTH
-import io.realm.kotlin.internal.interop.RealmInterop.safeKString
 import io.realm.kotlin.internal.interop.sync.ApiKeyWrapper
 import io.realm.kotlin.internal.interop.sync.AppError
 import io.realm.kotlin.internal.interop.sync.AuthProvider
@@ -80,7 +79,6 @@ import kotlinx.cinterop.usePinned
 import kotlinx.cinterop.value
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.launch
 import org.mongodb.kbson.BsonObjectId
 import org.mongodb.kbson.ObjectId
@@ -494,11 +492,10 @@ actual object RealmInterop {
         )
     }
 
-    actual fun realm_open(config: RealmConfigurationPointer, dispatcher: CoroutineDispatcher?): Pair<LiveRealmPointer, Boolean> {
+    actual fun realm_open(config: RealmConfigurationPointer, scheduler: RealmSchedulerPointer): Pair<LiveRealmPointer, Boolean> {
         val fileCreated = atomic(false)
         val callback = DataInitializationCallback {
             fileCreated.value = true
-            true
         }
         realm_wrapper.realm_config_set_data_initialization_function(
             config.cptr(),
@@ -516,21 +513,75 @@ actual object RealmInterop {
         //      val dispatcher = runBlocking { coroutineContext[CoroutineDispatcher.Key] }
         //  but requires opting in for @ExperimentalStdlibApi, and have really gotten it to play
         //  for default cases.
-        if (dispatcher != null) {
-            val scheduler = checkedPointerResult(createSingleThreadDispatcherScheduler(dispatcher))
-            realm_wrapper.realm_config_set_scheduler(config.cptr(), scheduler)
-        } else {
-            // If there is no notification dispatcher use the default scheduler.
-            // Re-verify if this is actually needed when notification scheduler is fully in place.
-            val scheduler = checkedPointerResult(realm_wrapper.realm_scheduler_make_default())
-            realm_wrapper.realm_config_set_scheduler(config.cptr(), scheduler)
-        }
+        realm_wrapper.realm_config_set_scheduler(config.cptr(), scheduler.cptr())
+
         val realmPtr = CPointerWrapper<LiveRealmT>(realm_wrapper.realm_open(config.cptr()))
         // Ensure that we can read version information, etc.
         realm_begin_read(realmPtr)
         return Pair(realmPtr, fileCreated.value)
     }
 
+    actual fun realm_create_scheduler(): RealmSchedulerPointer {
+        // If there is no notification dispatcher use the default scheduler.
+        // Re-verify if this is actually needed when notification scheduler is fully in place.
+        val scheduler = checkedPointerResult(realm_wrapper.realm_scheduler_make_default())
+        return CPointerWrapper<RealmSchedulerT>(scheduler)
+    }
+
+    actual fun realm_create_scheduler(dispatcher: CoroutineDispatcher): RealmSchedulerPointer {
+        printlntid("createSingleThreadDispatcherScheduler")
+        val scheduler = SingleThreadDispatcherScheduler(tid(), dispatcher)
+
+        val capi_scheduler: CPointer<realm_scheduler_t> = checkedPointerResult(
+            realm_wrapper.realm_scheduler_new(
+                // userdata: kotlinx.cinterop.CValuesRef<*>?,
+                scheduler.ref,
+
+                // free: realm_wrapper.realm_free_userdata_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Unit>>? */,
+                staticCFunction<COpaquePointer?, Unit> { userdata ->
+                    printlntid("free")
+                    userdata?.asStableRef<SingleThreadDispatcherScheduler>()?.dispose()
+                },
+
+                // notify: realm_wrapper.realm_scheduler_notify_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Unit>>? */,
+                staticCFunction<COpaquePointer?, Unit> { userdata ->
+                    // Must be thread safe
+                    val scheduler =
+                        userdata!!.asStableRef<SingleThreadDispatcherScheduler>().get()
+                    printlntid("$scheduler notify")
+                    try {
+                        scheduler.notify()
+                    } catch (e: Exception) {
+                        // Should never happen, but is included for development to get some indicators
+                        // on errors instead of silent crashes.
+                        e.printStackTrace()
+                    }
+                },
+
+                // is_on_thread: realm_wrapper.realm_scheduler_is_on_thread_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Boolean>>? */,
+                staticCFunction<COpaquePointer?, Boolean> { userdata ->
+                    // Must be thread safe
+                    val scheduler =
+                        userdata!!.asStableRef<SingleThreadDispatcherScheduler>().get()
+                    printlntid("is_on_thread[$scheduler] ${scheduler.threadId} " + tid())
+                    scheduler.threadId == tid()
+                },
+
+                // is_same_as: realm_wrapper.realm_scheduler_is_same_as_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */, kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Boolean>>? */,
+                staticCFunction<COpaquePointer?, COpaquePointer?, Boolean> { userdata, other ->
+                    userdata == other
+                },
+
+                // can_deliver_notifications: realm_wrapper.realm_scheduler_can_deliver_notifications_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Boolean>>? */,
+                staticCFunction<COpaquePointer?, Boolean> { _ -> true },
+            )
+        ) ?: error("Couldn't create scheduler")
+
+        scheduler.set_scheduler(capi_scheduler)
+
+        return CPointerWrapper<RealmSchedulerT>(capi_scheduler)
+    }
+
     actual fun realm_open_synchronized(config: RealmConfigurationPointer): RealmAsyncOpenTaskPointer {
         return CPointerWrapper(realm_wrapper.realm_open_synchronized(config.cptr()))
     }
@@ -3241,61 +3292,6 @@ actual object RealmInterop {
             ?: throw NullPointerException(identifier?.let { "'$identifier' shouldn't be null." })
     }
 
-    private fun createSingleThreadDispatcherScheduler(
-        dispatcher: CoroutineDispatcher
-    ): CPointer<realm_scheduler_t> {
-        printlntid("createSingleThreadDispatcherScheduler")
-        val scheduler = SingleThreadDispatcherScheduler(tid(), dispatcher)
-
-        val capi_scheduler: CPointer<realm_scheduler_t> = checkedPointerResult(
-            realm_wrapper.realm_scheduler_new(
-                // userdata: kotlinx.cinterop.CValuesRef<*>?,
-                scheduler.ref,
-
-                // free: realm_wrapper.realm_free_userdata_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Unit>>? */,
-                staticCFunction<COpaquePointer?, Unit> { userdata ->
-                    printlntid("free")
-                    userdata?.asStableRef<SingleThreadDispatcherScheduler>()?.dispose()
-                },
-
-                // notify: realm_wrapper.realm_scheduler_notify_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Unit>>? */,
-                staticCFunction<COpaquePointer?, Unit> { userdata ->
-                    // Must be thread safe
-                    val scheduler =
-                        userdata!!.asStableRef<SingleThreadDispatcherScheduler>().get()
-                    printlntid("$scheduler notify")
-                    try {
-                        scheduler.notify()
-                    } catch (e: Exception) {
-                        // Should never happen, but is included for development to get some indicators
-                        // on errors instead of silent crashes.
-                        e.printStackTrace()
-                    }
-                },
-
-                // is_on_thread: realm_wrapper.realm_scheduler_is_on_thread_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Boolean>>? */,
-                staticCFunction<COpaquePointer?, Boolean> { userdata ->
-                    // Must be thread safe
-                    val scheduler =
-                        userdata!!.asStableRef<SingleThreadDispatcherScheduler>().get()
-                    printlntid("is_on_thread[$scheduler] ${scheduler.threadId} " + tid())
-                    scheduler.threadId == tid()
-                },
-
-                // is_same_as: realm_wrapper.realm_scheduler_is_same_as_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */, kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Boolean>>? */,
-                staticCFunction<COpaquePointer?, COpaquePointer?, Boolean> { userdata, other ->
-                    userdata == other
-                },
-
-                // can_deliver_notifications: realm_wrapper.realm_scheduler_can_deliver_notifications_func_t? /* = kotlinx.cinterop.CPointer<kotlinx.cinterop.CFunction<(kotlinx.cinterop.COpaquePointer? /* = kotlinx.cinterop.CPointer<out kotlinx.cinterop.CPointed>? */) -> kotlin.Boolean>>? */,
-                staticCFunction<COpaquePointer?, Boolean> { userdata -> true },
-            )
-        ) ?: error("Couldn't create scheduler")
-        scheduler.set_scheduler(capi_scheduler)
-        scheduler
-        return capi_scheduler
-    }
-
     private fun <R> handleAppCallback(
         userData: COpaquePointer?,
         error: CPointer<realm_app_error_t>?,
@@ -3401,7 +3397,7 @@ actual object RealmInterop {
         }
 
         override fun notify() {
-            val function: suspend CoroutineScope.() -> Unit = {
+            scope.launch {
                 try {
                     printlntid("on dispatcher")
                     realm_wrapper.realm_scheduler_perform_work(scheduler)
@@ -3411,11 +3407,6 @@ actual object RealmInterop {
                     e.printStackTrace()
                 }
             }
-            scope.launch(
-                scope.coroutineContext,
-                CoroutineStart.DEFAULT,
-                function
-            )
         }
     }
 }

packages/cinterop/src/nativeDarwinTest/kotlin/io/realm/kotlin/test/CinteropTest.kt

@@ -28,6 +28,7 @@ import io.realm.kotlin.internal.interop.SchemaMode
 import io.realm.kotlin.internal.interop.SchemaValidationMode
 import io.realm.kotlin.internal.interop.set
 import io.realm.kotlin.internal.interop.toKotlinString
+import io.realm.kotlin.internal.interop.use
 import kotlinx.cinterop.BooleanVar
 import kotlinx.cinterop.CPointer
 import kotlinx.cinterop.CPointerVarOf
@@ -178,9 +179,12 @@ class CinteropTest {
                 SchemaMode.RLM_SCHEMA_MODE_AUTOMATIC
             )
             RealmInterop.realm_config_set_schema_version(nativeConfig, 1)
-
-            val (realm, fileCreated) = RealmInterop.realm_open(nativeConfig)
-            assertEquals(1L, RealmInterop.realm_get_num_classes(realm))
+            RealmInterop.realm_create_scheduler()
+                .use { scheduler ->
+                    val (realm, fileCreated) = RealmInterop.realm_open(nativeConfig, scheduler)
+                    assertEquals(1L, RealmInterop.realm_get_num_classes(realm))
+                    RealmInterop.realm_close(realm)
+                }
         }
     }
 

packages/jni-swig-stub/realm.i

@@ -301,7 +301,7 @@ return $jnicall;
                realm_flx_sync_mutable_subscription_set_t*, realm_flx_sync_subscription_desc_t*,
                realm_set_t*, realm_async_open_task_t*, realm_dictionary_t*,
                realm_sync_session_connection_state_notification_token_t*,
-               realm_dictionary_changes_t* };
+               realm_dictionary_changes_t*, realm_scheduler_t* };
 
 // For all functions returning a pointer or bool, check for null/false and throw an error if
 // realm_get_last_error returns true.

packages/jni-swig-stub/src/main/jni/realm_api_helpers.cpp

@@ -313,8 +313,8 @@ void invoke_core_notify_callback(int64_t scheduler) {
     realm_scheduler_perform_work(reinterpret_cast<realm_scheduler_t *>(scheduler));
 }
 
-realm_t *open_realm_with_scheduler(int64_t config_ptr, jobject dispatchScheduler) {
-    auto config = reinterpret_cast<realm_config_t *>(config_ptr);
+realm_scheduler_t*
+realm_create_scheduler(jobject dispatchScheduler) {
     if (dispatchScheduler) {
         auto jvmScheduler = new CustomJVMScheduler(dispatchScheduler);
         auto scheduler = realm_scheduler_new(
@@ -326,13 +326,9 @@ realm_t *open_realm_with_scheduler(int64_t config_ptr, jobject dispatchScheduler
                 [](void *userdata) { return static_cast<CustomJVMScheduler *>(userdata)->can_invoke(); }
         );
         jvmScheduler->set_scheduler(scheduler);
-        realm_config_set_scheduler(config, scheduler);
-    } else {
-        // TODO refactor to use public C-API https://github.com/realm/realm-kotlin/issues/496
-        auto scheduler =  new realm_scheduler_t{realm::util::Scheduler::make_generic()};
-        realm_config_set_scheduler(config, scheduler);
+        return scheduler;
     }
-    return realm_open(config);
+    throw std::runtime_error("Null dispatchScheduler");
 }
 
 jobject convert_to_jvm_app_error(JNIEnv* env, const realm_app_error_t* error) {

packages/jni-swig-stub/src/main/jni/realm_api_helpers.h

@@ -47,8 +47,8 @@ realm_network_transport_new(jobject network_transport);
 void
 set_log_callback(jint log_level, jobject log_callback);
 
-realm_t*
-open_realm_with_scheduler(int64_t config_ptr, jobject dispatchScheduler);
+realm_scheduler_t*
+realm_create_scheduler(jobject dispatchScheduler);
 
 bool
 realm_should_compact_callback(void* userdata, uint64_t total_bytes, uint64_t used_bytes);