Use the Dart isolate ownership API on the root isolate (#163703)

The isolate ownership API was [introduced
recently](https://dart-review.googlesource.com/c/sdk/+/407700) to solve
[some deadlock bugs](https://github.com/dart-lang/native/issues/1908) in
native callbacks.

A native callback is a call from native code into a Dart function.
Currently all such callbacks must run that Dart function in the isolate
that created the callback (called the target isolate). The only native
callback primitives at the moment are `NativeCallable.isolateLocal`
(blocking, but must be invoked from the same thread as the target
isolate, and the target isolate must be currently entered on that
thread) and `NativeCallable.listener` (non-blocking, can be invoked from
any thread).

To build blocking callbacks that can be called from any thread, we can
use a `NativeCallable.listener`, and use a synchronization object like a
mutex or a condition variable to block until the callback is complete.
However, if we try to do this on the thread that is currently entered in
the target isolate, we will deadlock: we invoke the listener, a message
is sent to the target isolate, and we block waiting for the message to
be handled, so we never pass control flow back to the isolate to handle
the message, and never stop waiting.

To fix this deadlock, Ffigen and Jnigen both have a mechanism that
checks if we're on the target isolate's thread first:
- If the native caller is already on the same thread as the target
isolate, and the target isolate is entered:
- Call the Dart function directly using `NativeCallable.isolateLocal` or
similar
- Otherwise, if the native caller is coming from a different thread:
- Call the Dart function asynchronously using `NativeCallable.listener`
or similar
  - Block until the callback finishes

However, this neglects the case where we're on the target isolate's
thread, but not entered into the isolate. This case happens in Flutter
when the callback is invoked from the UI thread (or the platform thread
when thread merging is enabled), and the target isolate is the root
isolate. When the native callback is invoked, the root isolate is not
entered, so we hit the second case: we send a message to the root
isolate, and block to wait for a response. Since the root isolate is
exclusively run on the UI thread, and we're blocking the UI thread, the
message will never be handled, and we deadlock.

The isolate ownership API fixes this by allowing the embedder to inform
the VM that it will run a particular isolate exclusively on a particular
thread, using `Dart_SetCurrentThreadOwnsIsolate`. Other native code can
then query that ownership using `Dart_GetCurrentThreadOwnsIsolate`. This
lets us add a third case to our conditional:

- If the native caller is on the thread that is currently entered in the
target isolate:
- Call the Dart function directly using `NativeCallable.isolateLocal` or
similar
- Otherwise, if the native caller is on the thread that owns the target
isolate
  - Enter the target isolate
- Call the Dart function directly using `NativeCallable.isolateLocal `or
similar
  - Exit the target isolate
- Otherwise, the native caller is coming from an unrelated thread:
- Call the Dart function asynchronously using `NativeCallable.listener`
or similar
  - Block until the callback finishes

**Note:** We don't need to set the ownership of VM managed threads,
because they run in a thread pool exclusively used by the VM, so there's
no way for native code to be executed on the thread (except by FFI, in
which case we're entered into the isolate anyway). We only need this for
Flutter's root isolate because work can be sent to the UI
thread/platform thread using OS specific APIs like Android's
`Looper.getMainLooper()`.

engine/src/flutter/runtime/dart_isolate.cc

@@ -153,11 +153,15 @@ std::weak_ptr<DartIsolate> DartIsolate::CreateRunningRootIsolate(
     return {};
   }
 
-  if (settings.root_isolate_create_callback) {
-    // Isolate callbacks always occur in isolate scope and before user code has
-    // had a chance to run.
+  {
     tonic::DartState::Scope scope(isolate.get());
-    settings.root_isolate_create_callback(*isolate.get());
+    Dart_SetCurrentThreadOwnsIsolate();
+
+    if (settings.root_isolate_create_callback) {
+      // Isolate callbacks always occur in isolate scope and before user code
+      // has had a chance to run.
+      settings.root_isolate_create_callback(*isolate.get());
+    }
   }
 
   if (root_isolate_create_callback) {

engine/src/flutter/runtime/dart_isolate_unittests.cc

@@ -1113,6 +1113,61 @@ TEST_F(DartIsolateTest, PlatformIsolateMainThrowsError) {
   // root isolate will be auto-shutdown
 }
 
+TEST_F(DartIsolateTest, RootIsolateIsOwnedByMainThread) {
+  ASSERT_FALSE(DartVMRef::IsInstanceRunning());
+  auto settings = CreateSettingsForFixture();
+  auto vm_ref = DartVMRef::Create(settings);
+  ASSERT_TRUE(vm_ref);
+  auto vm_data = vm_ref.GetVMData();
+  ASSERT_TRUE(vm_data);
+  TaskRunners task_runners(GetCurrentTestName(),    //
+                           GetCurrentTaskRunner(),  //
+                           GetCurrentTaskRunner(),  //
+                           GetCurrentTaskRunner(),  //
+                           GetCurrentTaskRunner()   //
+  );
+
+  auto isolate_configuration =
+      IsolateConfiguration::InferFromSettings(settings);
+
+  UIDartState::Context context(task_runners);
+  context.advisory_script_uri = "main.dart";
+  context.advisory_script_entrypoint = "main";
+  auto weak_isolate = DartIsolate::CreateRunningRootIsolate(
+      vm_data->GetSettings(),              // settings
+      vm_data->GetIsolateSnapshot(),       // isolate snapshot
+      nullptr,                             // platform configuration
+      DartIsolate::Flags{},                // flags
+      nullptr,                             // root_isolate_create_callback
+      settings.isolate_create_callback,    // isolate create callback
+      settings.isolate_shutdown_callback,  // isolate shutdown callback
+      "main",                              // dart entrypoint
+      std::nullopt,                        // dart entrypoint library
+      {},                                  // dart entrypoint arguments
+      std::move(isolate_configuration),    // isolate configuration
+      context                              // engine context
+  );
+  auto root_isolate = weak_isolate.lock();
+
+  Dart_Port main_port;
+  {
+    tonic::DartState::Scope scope(root_isolate.get());
+    main_port = Dart_GetMainPortId();
+
+    ASSERT_TRUE(Dart_GetCurrentThreadOwnsIsolate(main_port));
+  }
+
+  ASSERT_TRUE(Dart_GetCurrentThreadOwnsIsolate(main_port));
+
+  std::thread([main_port]() {
+    ASSERT_FALSE(Dart_GetCurrentThreadOwnsIsolate(main_port));
+  }).join();
+
+  ASSERT_TRUE(root_isolate->Shutdown());
+
+  ASSERT_FALSE(Dart_GetCurrentThreadOwnsIsolate(main_port));
+}
+
 }  // namespace testing
 }  // namespace flutter