feat: Introduce MultiIsolateRelicServer (#216)

- Implements multi-isolate support for `RelicServer` to enable
concurrent request handling across multiple CPU cores.
- Refactors `RelicServer` into a sealed interface with the existing
implementation moved to `_RelicServer`.
- Adds `IsolatedObject<T>`, a generic wrapper that spawns an object in a
separate isolate and communicates via message passing.
- Introduces `_IsolatedRelicServer` and `_MultiIsolateRelicServer`
implementations of `RelicServer`.
- Adds `noOfIsolates` optional named parameter to `RelicServer`
constructor (which is now a factory, delegating to either _RelicServer
or _MultiIsolateRelicServer), and `RelicApp.run`.

Author: nielsenko

doc/history.md

@@ -32,7 +32,7 @@ typedef Handler = FutureOr<HandledContext> Function(NewContext ctx);
 ```
 This seems more complex, why the added complexity? The problem is that not everything follows the request response model. Today we have web-sockets, server-side events, etc. that hijacks the underlying socket of a request, and hands lifetime handling to the developer.
 
-Shelf does support hijacking, but it does so by using exceptions for control flow. It will throw an exception `HijackException` that is only supposed to be caught and handled by a Shelf adaptor.
+Shelf does support hijacking, but it does so by using exceptions for control flow. It will throw an exception `HijackException` that is only supposed to be caught and handled by a Shelf adapter.
 
 As we consider using exceptions for control flow an anti-pattern, we have opted for introducing a `RequestContext` and an associated state machine instead.
 

example/basic/as_handle.dart

@@ -8,15 +8,15 @@ Future<void> main() async {
   final router = RelicRouter()
     ..use('/', logRequests()) // log all request from / and down
     ..any(
-        '/**',
-        respondWith(
-          (final request) => Response.ok(
-            body: Body.fromString('Hello, Relic!'),
-          ),
-        ));
+      '/**',
+      respondWith(
+        (final request) => Response.ok(body: Body.fromString('Hello, Relic!')),
+      ),
+    );
 
   // Start a server that forward request to the handler
-  final adapter = await IOAdapter.bind(InternetAddress.anyIPv4, port: 8080);
-  final server = RelicServer(adapter);
+  final server = RelicServer(
+    () => IOAdapter.bind(InternetAddress.anyIPv4, port: 8080),
+  );
   await server.mountAndStart(router.asHandler); // use asHandler to
 }

example/routing/pipeline.dart

@@ -9,14 +9,16 @@ Future<void> main() async {
   final handler = const Pipeline()
       // Pipelines allows middleware to run before routing
       .addMiddleware(logRequests())
-      .addHandler(respondWith(
-        (final request) => Response.ok(
-          body: Body.fromString('Hello, Relic!'),
+      .addHandler(
+        respondWith(
+          (final request) =>
+              Response.ok(body: Body.fromString('Hello, Relic!')),
         ),
-      )); // handles any verb, and any path
+      ); // handles any verb, and any path
 
   // Start a server that forward request to the handler
-  final adapter = await IOAdapter.bind(InternetAddress.anyIPv4, port: 8080);
-  final server = RelicServer(adapter);
+  final server = RelicServer(
+    () => IOAdapter.bind(InternetAddress.anyIPv4, port: 8080),
+  );
   await server.mountAndStart(handler);
 }

lib/src/adapter/adapter.dart

@@ -38,6 +38,9 @@ abstract class AdapterRequest {
 /// [AdapterRequest] format, and then handing them off to the Relic core.
 /// They also handle sending back responses or managing hijacked connections.
 abstract class Adapter {
+  /// The port this adapter is bound to.
+  int get port;
+
   /// A stream of incoming requests from the underlying source.
   ///
   /// Each event in the stream is an [AdapterRequest] representing a new

lib/src/adapter/io/io_adapter.dart

@@ -60,7 +60,7 @@ class IOAdapter extends Adapter {
   /// The [io.InternetAddress] the underlying server is listening on.
   io.InternetAddress get address => _server.address;
 
-  /// The port number the underlying server is listening on.
+  @override
   int get port => _server.port;
 
   @override

lib/src/isolated_object.dart

@@ -0,0 +1,123 @@
+import 'dart:async';
+import 'dart:isolate';
+
+import 'util/util.dart';
+
+typedef Factory<T> = FutureOr<T> Function();
+
+typedef _Action<T> = ({int id, dynamic Function(T) function});
+typedef _Response = ({int id, dynamic result});
+typedef _Inflight = Map<int, Completer>;
+typedef _Setup = (SendPort, ReceivePort, _Inflight);
+
+class IsolatedObject<T> {
+  final Future<_Setup> _connected;
+
+  IsolatedObject(final Factory<T> create) : _connected = _connect(create);
+
+  static Future<_Setup> _connect<T>(
+    final Factory<T> create,
+  ) async {
+    final parentPort = RawReceivePort();
+    final setupDone = Completer<_Setup>();
+
+    parentPort.handler = (final dynamic message) async {
+      if (message case final RemoteError e) {
+        setupDone.completeError(e, e.stackTrace);
+      } else {
+        final toChild = message as SendPort;
+        final fromChild = ReceivePort.fromRawReceivePort(parentPort);
+        final inflight = _Inflight();
+        setupDone.complete((toChild, fromChild, inflight));
+      }
+    };
+
+    try {
+      await _spawn(create, parentPort.sendPort);
+    } catch (_) {
+      parentPort.close();
+      rethrow;
+    }
+
+    final result = await setupDone.future;
+    final (toChild, fromChild, inflight) = result;
+
+    fromChild.asyncListen((final message) async {
+      if (message case final _Response response) {
+        final completer = inflight.remove(response.id);
+        assert(completer != null, 'PROTOCOL BUG. No such ID ${response.id}');
+        if (completer == null) return; // coverage: ignore-line
+        switch (response.result) {
+          case final RemoteError e:
+            completer.completeError(e, e.stackTrace);
+          default:
+            completer.complete(await response.result);
+        }
+      }
+    }, onDone: () {
+      // ReceivePort closed. Fail any pending requests to avoid hangs.
+      for (final c in inflight.values) {
+        if (!c.isCompleted) {
+          c.completeError(StateError('IsolatedObject<$T> channel closed'));
+        }
+      }
+      inflight.clear();
+    });
+
+    return result;
+  }
+
+  static Future<Isolate> _spawn<T>(
+    final Factory<T> create,
+    final SendPort toParent,
+  ) {
+    return Isolate.spawn(
+      (final toParent) async {
+        final childPort = ReceivePort();
+        final T isolatedObject;
+        try {
+          isolatedObject = await create();
+          toParent.send(childPort.sendPort); // complete handshake
+        } catch (e, st) {
+          toParent.send(RemoteError('$e', '$st'));
+          return;
+        }
+
+        // process inbound actions
+        await for (final message in childPort) {
+          if (message == null) {
+            // shutdown signal received
+            childPort.close();
+            break;
+          } else if (message case final _Action<T> action) {
+            try {
+              final result = await action.function(isolatedObject);
+              toParent.send((id: action.id, result: result)); // return result
+            } catch (e, st) {
+              toParent.send((id: action.id, result: RemoteError('$e', '$st')));
+            }
+          }
+        }
+      },
+      toParent,
+    );
+  }
+
+  int _nextId = 0;
+  Future<U> evaluate<U>(final FutureOr<U> Function(T) function) async {
+    final (toChild, _, inflight) = await _connected;
+
+    final id = _nextId++;
+    final completer = Completer<U>();
+    inflight[id] = completer;
+
+    toChild.send((id: id, function: function));
+    return await completer.future;
+  }
+
+  Future<void> close() async {
+    final (toChild, fromChild, _) = await _connected;
+    toChild.send(null); // shutdown signal
+    fromChild.close();
+  }
+}

lib/src/relic_server.dart

@@ -6,45 +6,80 @@ import 'body/body.dart';
 import 'handler/handler.dart';
 import 'headers/exception/header_exception.dart';
 import 'headers/standard_headers_extensions.dart';
+import 'isolated_object.dart';
 import 'logger/logger.dart';
 import 'message/request.dart';
 import 'message/response.dart';
 import 'util/util.dart';
 
-/// A server that uses a [Adapter] to handle HTTP requests.
-class RelicServer {
-  /// The underlying adapter.
-  final Adapter adapter;
+sealed class RelicServer {
+  /// Mounts a [handler] to the server and starts listening for requests.
+  ///
+  /// Only one [handler] can be mounted at a time, but it will be replaced
+  /// on each call.
+  Future<void> mountAndStart(final Handler handler);
 
-  /// Whether [mountAndStart] has been called.
-  Handler? _handler;
+  /// Close the server
+  Future<void> close();
 
+  /// The port this server is bound to.
+  ///
+  /// This will throw a [LateInitializationError], if called before [mountAndStart].
+  int get port;
+
+  factory RelicServer(
+    final Factory<Adapter> adapterFactory, {
+    final int noOfIsolates = 1,
+  }) {
+    return switch (noOfIsolates) {
+      < 1 => throw RangeError.value(
+          noOfIsolates,
+          'noOfIsolates',
+          'Must be larger than 0',
+        ),
+      == 1 => _RelicServer(adapterFactory),
+      _ => _MultiIsolateRelicServer(adapterFactory, noOfIsolates),
+    };
+  }
+}
+
+/// A server that uses a [Adapter] to handle HTTP requests.
+final class _RelicServer implements RelicServer {
+  final FutureOr<Adapter> _adapter;
+  Handler? _handler;
   StreamSubscription<AdapterRequest>? _subscription;
 
   /// Creates a server with the given parameters.
-  RelicServer(this.adapter);
+  _RelicServer(final Factory<Adapter> adapterFactory)
+      : _adapter = adapterFactory();
 
   /// Mounts a handler to the server and starts listening for requests.
   ///
   /// Only one handler can be mounted at a time.
+  @override
   Future<void> mountAndStart(final Handler handler) async {
+    port = (await _adapter).port;
     _handler = _wrapHandlerWithMiddleware(handler);
     if (_subscription == null) await _startListening();
   }
 
-  /// Close the server
+  @override
   Future<void> close() async {
     await _stopListening();
-    await adapter.close();
+    await (await _adapter).close();
   }
 
+  @override
+  late final int port;
+
   Future<void> _stopListening() async {
     await _subscription?.cancel();
     _handler = null;
   }
 
   /// Starts listening for requests.
   Future<void> _startListening() async {
+    final adapter = await _adapter;
     catchTopLevelErrors(() {
       _subscription = adapter.requests.listen(_handleRequest);
     }, (final error, final stackTrace) {
@@ -60,6 +95,8 @@ class RelicServer {
     final handler = _handler;
     if (handler == null) return; // if close has been called
 
+    final adapter = await _adapter;
+
     // Wrap the handler with our middleware
     late Request request;
     try {
@@ -143,3 +180,47 @@ void _logError(
     type: LoggerType.error,
   );
 }
+
+final class _IsolatedRelicServer extends IsolatedObject<RelicServer>
+    implements RelicServer {
+  _IsolatedRelicServer(final Factory<Adapter> adapterFactory)
+      : super(() => RelicServer(adapterFactory));
+
+  @override
+  Future<void> close() async {
+    await evaluate((final r) => r.close());
+    await super.close();
+  }
+
+  @override
+  Future<void> mountAndStart(final Handler handler) async {
+    await evaluate((final r) => r.mountAndStart(handler));
+    port = await evaluate((final r) => r.port);
+  }
+
+  @override
+  late final int port;
+}
+
+final class _MultiIsolateRelicServer implements RelicServer {
+  final List<RelicServer> _children;
+
+  _MultiIsolateRelicServer(
+    final Factory<Adapter> adapterFactory,
+    final int noOfIsolates,
+  ) : _children = List.generate(
+            noOfIsolates, (final _) => _IsolatedRelicServer(adapterFactory));
+
+  @override
+  Future<void> close() async {
+    await _children.map((final c) => c.close()).wait;
+  }
+
+  @override
+  Future<void> mountAndStart(final Handler handler) async {
+    await _children.map((final c) => c.mountAndStart(handler)).wait;
+  }
+
+  @override
+  int get port => _children.first.port;
+}

lib/src/router/relic_app.dart

@@ -17,6 +17,11 @@ final class RelicApp implements RelicRouter {
   /// The adapter factory pattern allows for deferred initialization of server
   /// resources.
   ///
+  /// If [noOfIsolates] equals 1, the server is started on the current isolate.
+  /// Otherwise [noOfIsolates] isolates are spun up. When using multiple isolates make
+  /// sure that any handler and middleware configured are sendable.
+  /// (see https://api.dart.dev/stable/dart-isolate/SendPort/send.html).
+  ///
   /// Returns a [Future] that completes with the running [RelicServer] instance.
   ///
   /// [RelicApp] instances supports hot-reload. They will re-configure the internal
@@ -28,7 +33,7 @@ final class RelicApp implements RelicRouter {
   /// final app = RelicApp()
   ///   ..get('/', (ctx) => ctx.ok('Hello!'));
   ///
-  /// final adaptorFactory = () => IOAdapter.bind(InternetAddress.loopbackIPv4, port: 8080);
+  /// final adapterFactory = () => IOAdapter.bind(InternetAddress.loopbackIPv4, port: 8080);
   /// final server = await app.run(adapterFactory);
   ///
   /// // later .. when done
@@ -39,10 +44,11 @@ final class RelicApp implements RelicRouter {
   /// Check out [RelicAppIOServeEx.serve] if you are using `dart:io` to avoid
   /// specifying [adapterFactory] explicitly.
   Future<RelicServer> run(
-    final FutureOr<Adapter> Function() adapterFactory,
-  ) async {
+    final FutureOr<Adapter> Function() adapterFactory, {
+    final int noOfIsolates = 1,
+  }) async {
     if (_server != null) throw StateError('Cannot call run twice');
-    _server = RelicServer(await adapterFactory());
+    _server = RelicServer(adapterFactory, noOfIsolates: noOfIsolates);
     await _init();
     _reloadSubscription = await _hotReloader.register(this);
     return _server!;
@@ -136,9 +142,7 @@ class _HotReloader {
   Future<StreamSubscription?> register(final RelicApp app) async {
     final reloadStream = await _reloadStream;
     if (reloadStream != null) {
-      return reloadStream
-          .asyncMap((final _) => app._reload())
-          .listen((final _) {});
+      return reloadStream.asyncListen((final _) => app._reload());
     }
     return null;
   }