Illegal argument in isolate message: (object is a ReceivePort)

[gmb119943]

I'm trying to write code that runs a task in a separate isolate. If the task is not completed within some maximum time, then an error is generated.

The code uses the closePorts local function, which closes multiple ports. This results in an "Illegal argument in isolate message: (object is a ReceivePort)" error. The same error will occur even if this function is not used anywhere at all.

What am I doing wrong?

import 'dart:async';

import 'dart:isolate';

class Timeout implements Exception {
  const Timeout();
}

class RunTask {
  late final Isolate _isolate;
  final _initCompleter = Completer();
  final _taskCompleter = Completer();

  RunTask(Function func) {
    _run(func);
  }

  get future => _taskCompleter.future;

  Future<dynamic> _run(func) async {
    var outputPort = ReceivePort("output");
    var errorPort = ReceivePort("error");
    var exitPort = ReceivePort("exit");

    _isolate = await Isolate.spawn(
      (sendPort) async {
        var result = await func();
        Isolate.exit(sendPort, result);
      },
      outputPort.sendPort,
      onError: errorPort.sendPort,
      onExit: exitPort.sendPort,
    );
    _initCompleter.complete();

    void closePorts() {
      // It is assumed that calling close on a closed port will
      // not result in an error. If the port is not closed, then the application
      // will not terminate.
      outputPort.close();
      exitPort.close();
      errorPort.close();
    }

    outputPort.first.then((result) {
      closePorts();
      _taskCompleter.complete(result);
    });
    errorPort.first.then((error) {
      closePorts();
      _taskCompleter.completeError(error);
    });
    exitPort.first.then((_) {
      print(_);
      closePorts();
    });

    return _taskCompleter.future;
  }

  Future<void> kill() async {
    await _initCompleter.future;
    _isolate.kill();
  }
}

Future<dynamic> timeout(Duration timeout, Function func) async {
  var internalCompleter = Completer();

  var timer = Timer(timeout, () {
    assert(!internalCompleter.isCompleted);
    internalCompleter.completeError(const Timeout());
  });

  var task = RunTask(func);
  task.future.then((result) {
    if (!internalCompleter.isCompleted) {
      timer.cancel();
      internalCompleter.complete(result);
    }
  });

  return internalCompleter.future.catchError((e) async {
    await task.kill();
    throw e;
  });
}

void main() {
  timeout(Duration(milliseconds: 100), () async {
    for (;;) {}
  });
}

[rmacnak-google]

      (sendPort) async {
        var result = await func();
        Isolate.exit(sendPort, result);
      },

captures func and shares a parent context with

   void closePorts() {
      // It is assumed that calling close on a closed port will
      // not result in an error. If the port is not closed, then the application
      // will not terminate.
      outputPort.close();
      exitPort.close();
      errorPort.close();
    }

that captures outputPort, exitPort and errorPort, so the closure passed to Isolate.spawn indirectly references those receive ports. See #36983.

I recommend factoring out the entry closure so it only has what is necessary in scope:

// At top level.
forwardResult(func) => (sendPort) => Isolate.exit(sendPort, await func());

...
Isolate.spawn(forwardResult(func), ...)

[gmb119943]

Thank you, it works. But now there are new questions.

Why if the hello function is removed from forwardResult, then there will be no errors? Data not being captured?

Why if only globalport is specified in the hello function, then there are no errors? And if localport is specified, will there be an error?

When I read the documentation on dart.dev, I did not find any coverage of this issue.

The new code looks like this.

import 'dart:async';

import 'dart:isolate';

class Timeout implements Exception {
  const Timeout();
}

var globalPort = ReceivePort();
dynamic forwardResult(func) {
  var localPort = ReceivePort();

  void hello() {
    localPort;
    globalPort;
  }

  return (sendPort) async {
    return Isolate.exit(sendPort, await func());
  };
}

class RunTask {
  late final Isolate _isolate;
  final _initCompleter = Completer();
  final _taskCompleter = Completer();

  RunTask(Function func) {
    _run(func);
  }

  get future => _taskCompleter.future;

  Future<dynamic> _run(func) async {
    var outputPort = ReceivePort("output");
    var errorPort = ReceivePort("error");
    var exitPort = ReceivePort("exit");

    _isolate = await Isolate.spawn(
      forwardResult(func),
      outputPort.sendPort,
      onError: errorPort.sendPort,
      onExit: exitPort.sendPort,
    );
    _initCompleter.complete();

    void closePorts() {
      // It is assumed that calling close on a closed port will
      // not result in an error. If the port is not closed, then the application
      // will not terminate.
      outputPort.close();
      exitPort.close();
      errorPort.close();
    }

    outputPort.first.then((result) {
      closePorts();
      _taskCompleter.complete(result);
    });
    errorPort.first.then((error) {
      closePorts();
      _taskCompleter.completeError(error);
    });
    exitPort.first.then((_) {
      print(_);
      closePorts();
    });

    return _taskCompleter.future;
  }

  Future<void> kill() async {
    await _initCompleter.future;
    _isolate.kill();
  }
}

Future<dynamic> timeout(Duration timeout, Function func) async {
  var internalCompleter = Completer();

  var timer = Timer(timeout, () {
    assert(!internalCompleter.isCompleted);
    internalCompleter.completeError(const Timeout());
  });

  var task = RunTask(func);
  task.future.then((result) {
    if (!internalCompleter.isCompleted) {
      timer.cancel();
      internalCompleter.complete(result);
    }
  });

  return internalCompleter.future.catchError((e) async {
    await task.kill();
    throw e;
  });
}

void main() {
  timeout(Duration(milliseconds: 100), () async {
    for (;;) {}
  });
}

[gmb119943]

The global variable in the new isolate will be a copy, right? And in the case of a local variable, will there be an attempt to transfer it to the context of the isolate?

[gmb119943]

I think this should be explained in the documentation.

[brianquinlan]

I think that we can document this. How about we present an example in the Isolate.run documentation like:

void serializeAndWriteIfEmpty(File f, Object o) async {
  final stream = await f.open(mode: FileMode.write);
  writeNew() async {
    final encoded = await Isolate.run(() => jsonEncode(o));
    await stream.writeString(encoded);
    await stream.flush();
    await stream.close();
  }

  if (await stream.position() == 0) {
    await writeNew();
  }
}

And suggest that, if the code fails with Illegal argument in isolate message, then the developer mechanically creates a new top-level function that makes the required arguments explicit e.g.

Future<String> doEncode(Object o) async {
  return await Isolate.run(() => jsonEncode(o));
}

void serializeAndWrite(File f, Object o) async {
  final stream = await f.open(mode: FileMode.write);
  writeNew() async {
    final encoded = await doEncode(o);
    await stream.writeString(encoded);
    await stream.flush();
    await stream.close();
  }

  if (await stream.position() == 0) {
    await writeNew();
  }
}

If the Illegal argument in isolate message failure persists then the developer will know that they are actually trying to pass an illegal argument to the Isolate.

[brianquinlan]

I have a proposed change here:
https://dart-review.googlesource.com/c/sdk/+/288140/1/sdk/lib/isolate/isolate.dart#221

I'm not super happy with the writing or example so feedback would be appreciated!

[mkustermann]

@gmb119943 : Side-note: @aam recently improved our error reporting here, the error on master should now tell you the exact path of how the closure references an object that cannot be sent (aka copied) across isolates.

[mkustermann]

@aam: Maybe it would make sense that our retaining path printer will print some extra information for closure contexts?

Instead of printing

Invalid argument(s): Illegal argument in isolate message: (object is a ReceivePort)
 <- Instance of '_ReceivePortImpl' (from dart:isolate)
 <- Context num_variables: 2
 <- Closure: () => void (from dart:core)

we could print

Invalid argument(s): While trying to send a message we encountered an object that cannot be sent: a ReceivePort
 <- ReceivePort (from dart:isolate)
 <- Context (This may unintentionally capture more than needed. Rewriting your code may help. See http://dartbug.com/36983)
 <- Closure: () => void (from file:///.../test.dart)

in JIT mode we could even detect the name of the captured variable name and print:

...
 <- ReceivePort (from dart:isolate)
 <- Context.<variable> (If this variable is not actually needed by the closure, rewriting your code may fix this problem. See http://dartbug.com/36983)
 <- Closure: () => void (from file:///.../test.dart)

[lrhn]

Consider also if it would be possible to just not over-capture.

A closure holding on to objects that it doesn't reference will potentially keep those values a live past their expected lifetime.
It's also exposing VM internals in a very unpredictable way.

Consider this example;

import "dart:isolate";
import "dart:convert";

Future asArgs(ReceivePort r, Object o) async {
  try {
    await Isolate.run<void>(() => o);
    print("Test 0 success");
  } catch (e) {
    print("Test 0 failure: $e");
  }

  try {
    await Isolate.run<void>(() => jsonEncode(o));
    print("Test 0 encode success");
  } catch (e) {
    print("Test 0 encode failure: $e");
  }
}

void asArgsSync(ReceivePort r, Object o) {
  try {
    Isolate.run<void>(() => o).catchError((e) {
      print("Test 0 sync delayed failure: $e");
    });
    print("Test 0 sync success");
  } catch (e) {
    print("Test 0 sync failure: $e");
  }

  try {
    Isolate.run<void>(() => jsonEncode(o)).catchError((e) {
      print("Test 0 sync encode delayed failure: $e");
    });
    print("Test 0 sync encode success");
  } catch (e) {
    print("Test 0 sync encode failure: $e");
  }
}

void main() async {
  var r = ReceivePort()..forEach(print);
  var o = "null";

  asArgsSync(r, o);
  await asArgs(r, o);

  try {
    Isolate.run<void>(() => o).catchError((e) {
      print("Test 1 sync delayed failure: $e");
    }); // Captures r?
    print("Test 1 sync success");
  } catch (e) {
    print("Test 1 sync failure: $e");
  }

  try {
    await Isolate.run<void>(() => o); // Captures r?
    print("Test 1 async success");
  } catch (e) {
    print("Test 1 async failure: $e");
  }

  try {
    Isolate.run<void>(() => jsonEncode(o)).catchError((e) {
      print("Test 1 sync encode delayed failure: $e");
    }); // Captures r?
    print("Test 1 sync encode success");
  } catch (e) {
    print("Test 1 sync encode failure: $e");
  }

  try {
    await Isolate.run<void>(() => jsonEncode(o)); // Captures r?
    print("Test 1 async encode success");
  } catch (e) {
    print("Test 1 async encode failure: $e");
  }

  try {
    var o2 = o;
    await Isolate.run<void>(() => jsonEncode(o2));  // Captures r?
    print("Test 2 encode success");
  } catch (e) {
    print("Test 2 encode failure: $e");
  }

  try {
    await () async {
      var o3 = o;
      await Isolate.run<void>(() => o3);  // Captures r?
    }();
    print("Test 3 success");
  } catch (e) {
    print("Test 3 failure: $e");
  }

  await Future.delayed(Duration(seconds: 1), () {});

  r.close();
}  

From the description, and the text in the CL, this seems like it should fail a lot.

It runs without any issues or errors, even though r is clearly in the scope.

Compare that to;

import "dart:isolate";
import "dart:convert";

Future asArgs(RawReceivePort r, Object o) async {
  try {
    await Isolate.run<void>(() => o);
    print("Test 0 success");
  } catch (e) {
    print("Test 0 failure: $e");
  }

  try {
    await Isolate.run<void>(() => jsonEncode(o));
    print("Test 0 encode success");
  } catch (e) {
    print("Test 0 encode failure: $e");
  }
}

void asArgsSync(RawReceivePort r, Object o) {
  try {
    Isolate.run<void>(() => o).catchError((e) {
      print("Test 0 sync delayed failure: $e");
    });
    print("Test 0 sync success");
  } catch (e) {
    print("Test 0 sync failure: $e");
  }

  try {
    Isolate.run<void>(() => jsonEncode(o)).catchError((e) {
      print("Test 0 sync encode delayed failure: $e");
    });;
    print("Test 0 sync encode success");
  } catch (e) {
    print("Test 0 sync encode failure: $e");
  }
}

void main() async {
  var r = RawReceivePort();
  r.handler = (m) {
    print(m);
    if (m == "Done") r.close();
  };
  r.sendPort.send("Port is used");
  var o = "null";

  asArgsSync(r, o);
  await asArgs(r, o);

  try {
    Isolate.run<void>(() => o).catchError((e) {
      print("Test 1 sync delayed failure: $e");
    });; // Captures r?
    print("Test 1 sync success");
  } catch (e) {
    print("Test 1 sync failure: $e");
  }

  try {
    await Isolate.run<void>(() => o); // Captures r?
    print("Test 1 async success");
  } catch (e) {
    print("Test 1 async failure: $e");
  }

  try {
    Isolate.run<void>(() => jsonEncode(o)).catchError((e) {
      print("Test 1 sync encode delayed failure: $e");
    }); // Captures r?
    print("Test 1 sync encode success");
  } catch (e) {
    print("Test 1 sync encode failure: $e");
  }

  try {
    await Isolate.run<void>(() => jsonEncode(o)); // Captures r?
    print("Test 1 async encode success");
  } catch (e) {
    print("Test 1 async encode failure: $e");
  }

  try {
    var o2 = o;
    await Isolate.run<void>(() => jsonEncode(o2));  // Captures r?
    print("Test 2 encode success");
  } catch (e) {
    print("Test 2 encode failure: $e");
  }

  try {
    await () async {
      var o3 = o;
      await Isolate.run<void>(() => o3);  // Captures r?
    }();
    print("Test 3 success");
  } catch (e) {
    print("Test 3 failure: $e");
  }

  r.sendPort.send("Done");
}  

which has several of the sends fail because it captures r. The code is almost the same, the only difference is that I actually send port messages.

Or rather, I'm guessing the difference is that r is captured in a closure, even if it is a different closure.

That's unpredictable effect-at-a-distance and an internal VM behavior that I don't think we should be exposing users to.
(And it's wasteful.)

I'm slightly surprised that test 2 doesn't work, because I had expected that the o2 variable would be in a separate scope. I guess something like

  try {
    await (o4) async {
      await Isolate.run<void>(() => o4);  // Captures r?
    }(o);
    print("Test 4 success");
  } catch (e) {
    print("Test 4 failure: $e");
  }

I'd also have expected that this would work, because o5 should be in a new scope, since there is one scope per iteration.

  for (var o5 in [o, o]) {
    try {
      await Isolate.run<void>(() => o5);  // Captures r?
      print("Test 5 success");
    } catch (e) {
      print("Test 5 failure: $e");
    }
  }

It doesn't, so you are capturing the surrounding context object, even though it's not used inside this scope.

TL;DR: I consider over-capturing a VM issue, not only for isolate communication, although it gets very visible there instead of just retaining arbitrary objects past their lifetime. It's not a library issue, and not something that should be pushed onto users to fix.

[mkustermann]

Consider also if it would be possible to just not over-capture.

We do have a tracking bug for it: #36983

It'd need some resources to do it, but it's clear how it can be implemented. What's less clear is whether we'd regress

• performance of existing code (e.g. due to more allocations & initializations of those)
• developer experience (e.g. due to seeing less information when debugging)

Once we have the resources, I think we should definitely try

[aam]

I think it's worth to consider something else, that showed up as 'Illegal argument in isolate message' when attempt to use Futures in in dart apps that use package:test.

The following

import 'dart:async';
import 'dart:isolate';
import 'package:test/test.dart';

main() async {
  test('my test', () async {
    final f = Future.value(123);
    await Isolate.run(() async {
      print(f);
    });
  });
}

will fail with

 Invalid argument(s): Illegal argument in isolate message: (object is a SuspendState)

The root cause is that Future f is created and references CustomZone(created by test infra) with zone values of Futures with suspended listeners. Basically similar to the following non-package-test reproduction.

import 'dart:async';
import 'dart:isolate';

greeting(String name) {
  final value = Future.value(123);
  return Isolate.run(() {
    // this is consequential, but in unexpected way.
    // Capturing futures is not a problem in itself, however here it causes
    // problems because unbeknownst to the developer the zone(created below)
    // that is referenced by the Future makes that Future non-passable.
    // *** Comment the line below to make the code print "Hello, world"
    value;
    return 'Hello, $name';
  });
}

Stream<int> yield1() async* {
  yield 1;
}

main() async {
  runZoned(() async {
    print('${await greeting("world")}');
  }, zoneValues: {
    // These zone values are part of CustomZone.
    // Futures created in such CustomZone can't go through isolate boundary
    // because the closure function has SuspendState object(never passable).
    // *** Comment the line below to make the code print "Hello, world"
    0: yield1()
  });
}

The point I guess is that Illegal argument in isolate message can be tricky to get to the bottom of.

[lrhn]

Sending a Future is tricky. Here, the Future.value(123) seems like it should have completed, but at the current time where Isolate.run is called, it's probably waiting for a microtask to do the actual completion. That microtask won't exist in the new isolate, so the future will never complete. Whoops. (Well, here you just print it, so that doesn't matter, but that's not why you'd send a future.)

Also, as you noticed, futures hang onto zones, which I'd love to have themstop doing after completion, but the current implementation schedules microtasks in the future's own zone to report results to later listeners, not in the listener's zone. And changing that breaks a lot of zone-fragile (mostly test) code that assumes every microtasks runs exactly where they did when the test was written. (Bad test! Bad!)

So sending futures, and their zones, is dangerous, and should generally be avoided. (But zones are just normal Dart objects, so nothing really prevents a zone from being sent, it's just that a lot of other stuff hangs onto zones and stores weird objects in zones. And therefore copying a zone might not do what one expects. For example, a zone which implements its own microtask queue, if copied might make the same microtask happen twice, in separate isolates. Or it might never work, because the original scheduleMicrotask it needs to get it running wasn't copied.)
Copying a zone is a crap-shot at best. Same for futures and streams.

[mkustermann]

We could decide to provide an annotation that will prevent an object of a class to be sent across isolates. In the VM we could store this as a bit on the Class object (just like we already track the number of native fields on the class). Sending can then be a simple bit check in the class object.

Classes like Future could then be marked and will never be sendable (compared to now, where they may or may not be sendable, depending on whether they have a pointer to custom zone, that has pointers to closures with suspend state, ...)

[brianquinlan]

Yeah, what about this - we introduce a new lint that makes it an error to use a non top-level function as an argument to Isolate.spawn and Isolate.run. Tools like VSCode could offer to do the transformation.

[mkustermann]

we introduce a new lint ... that makes it an error to use a non top-level function as an argument to ... Isolate.run

Isolate.run()s API was designed under assumption people use a closure. Otherwise there'd be no way to pass data to the function being invoked (compare that with much older compute() function from flutter, which takes function and argument separately).

We now print retaining paths to problematic objects. We could slightly improve that more. We also have concrete suggestions how to avoid the problem. So that leaves only the concern about over-capturing. But notice that this problem exists even without using isolates: Memory cannot reclaimed even though it cannot be accessed anymore. So I consider this a separate issue.

[brianquinlan]

The idea would be that the lint warns you that the function calling Isolate.run has a statement other than the Isolate.run call in it.

void main() async {
  final x = 5;
  final y = 6;
  final z = [1,2,3];
  Isolate.run(() => x + y);
  ^^^^^^^^^^^^^^^^^. Isolate may capture more state than necessary [Fix]
}

And "Fix" would do this transformation:

int runner(int x, int y) async {
  return Isolate.run(() => x + y);
}

void main() async {
  final x = 5;
  final y = 6;
  runner(x, y);
}

I'm not sure if this lint would be expressible. I don't have strong feelings about this though.

[brianquinlan]

I've updated the docs for Isolate.spawn and Isolate.run - do you think that we can close this issue?

[mkustermann]

I've updated the docs for Isolate.spawn and Isolate.run - do you think that we can close this issue?

Let's close it. @aam still has a CL for review, but that's separately tracked at #51722