Finishing clusterd

Package.swift

@@ -121,7 +121,14 @@ var targets: [PackageDescription.Target] = [
             .product(name: "ArgumentParser", package: "swift-argument-parser"),
         ]
     ),
-
+    .executableTarget(
+        name: "swift-clusterd",
+        dependencies: [
+            "DistributedCluster",
+            .product(name: "ArgumentParser", package: "swift-argument-parser"),
+        ],
+        path: "Sources/Clusterd"
+    ),
     // ==== ------------------------------------------------------------------------------------------------------------
     // MARK: Multi Node Tests
 

Sources/Clusterd/boot+ClusterD.swift

@@ -0,0 +1,47 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Distributed Actors open source project
+//
+// Copyright (c) 2020-2024 Apple Inc. and the Swift Distributed Actors project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of Swift Distributed Actors project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import ArgumentParser
+import DistributedCluster
+import Logging
+
+@main
+struct ClusterDBoot: AsyncParsableCommand {
+    @Option(name: .shortAndLong, help: "The port to bind the cluster daemon on.")
+    var port: Int = ClusterDaemon.defaultEndpoint.port
+
+    @Option(help: "The host address to bid the cluster daemon on.")
+    var host: String = ClusterDaemon.defaultEndpoint.host
+
+    func run() async throws {
+        let daemon = await ClusterSystem.startClusterDaemon(configuredWith: self.configure)
+
+        #if DEBUG
+        daemon.system.log.warning("RUNNING ClusterD DEBUG binary, operation is likely to be negatively affected. Please build/run the ClusterD process using '-c release' configuration!")
+        #endif
+
+        try daemon.system.park()
+    }
+
+    func configure(_ settings: inout ClusterSystemSettings) {
+        // other nodes will be discovering us, not the opposite
+        settings.discovery = .init(static: [])
+
+        settings.endpoint = Cluster.Endpoint(
+            systemName: "clusterd",
+            host: self.host,
+            port: self.port
+        )
+    }
+}

Sources/DistributedCluster/Cluster/DiscoveryShell.swift

@@ -34,6 +34,12 @@ final class DiscoveryShell {
 
     var behavior: _Behavior<Message> {
         .setup { context in
+            // FIXME: should have a behavior to bridge the async world...
+            context.log.info("Initializing discovery: \(self.settings.implementation)")
+            // Try to initialise clusterd if needed
+            self.settings.initializeClusterd(context.system)
+            context.log.info("Initializing discovery, done.")
+
             self.subscription = self.settings.subscribe(
                 onNext: { result in
                     switch result {

Sources/DistributedCluster/ClusterSystem+Clusterd.swift

@@ -0,0 +1,88 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Distributed Actors open source project
+//
+// Copyright (c) 2018-2022 Apple Inc. and the Swift Distributed Actors project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of Swift Distributed Actors project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Atomics
+import Backtrace
+import CDistributedActorsMailbox
+import Dispatch
+import Distributed
+import DistributedActorsConcurrencyHelpers
+import Foundation  // for UUID
+import Logging
+import NIO
+
+extension ClusterSystem {
+    public static func startClusterDaemon(configuredWith configureSettings: (inout ClusterSystemSettings) -> Void = { _ in () }) async -> ClusterDaemon {
+        let system = await ClusterSystem("clusterd") { settings in
+            settings.endpoint = ClusterDaemon.defaultEndpoint
+            configureSettings(&settings)
+        }
+
+        return ClusterDaemon(system: system)
+    }
+}
+
+public struct ClusterDaemon {
+    public let system: ClusterSystem
+    public var settings: ClusterSystemSettings {
+        self.system.settings
+    }
+
+    public init(system: ClusterSystem) {
+        self.system = system
+    }
+}
+
+extension ClusterDaemon {
+    /// Suspends until the ``ClusterSystem`` is terminated by a call to ``shutdown()``.
+    public var terminated: Void {
+        get async throws {
+            try await self.system.terminated
+        }
+    }
+
+    /// Returns `true` if the system was already successfully terminated (i.e. awaiting ``terminated`` would resume immediately).
+    public var isTerminated: Bool {
+        self.system.isTerminated
+    }
+
+    /// Forcefully stops this actor system and all actors that live within it.
+    /// This is an asynchronous operation and will be executed on a separate thread.
+    ///
+    /// You can use `shutdown().wait()` to synchronously await on the system's termination,
+    /// or provide a callback to be executed after the system has completed it's shutdown.
+    ///
+    /// - Returns: A `Shutdown` value that can be waited upon until the system has completed the shutdown.
+    @discardableResult
+    public func shutdown() throws -> ClusterSystem.Shutdown {
+        try self.system.shutdown()
+    }
+}
+
+extension ClusterDaemon {
+    /// The default endpoint
+    public static let defaultEndpoint = Cluster.Endpoint(host: "127.0.0.1", port: 3137)
+}
+
+internal distributed actor ClusterDaemonServant {
+    typealias ActorSystem = ClusterSystem
+
+    @ActorID.Metadata(\.wellKnown)
+    public var wellKnownName: String
+
+    init(system: ClusterSystem) async {
+        self.actorSystem = system
+        self.wellKnownName = "$cluster-daemon-servant"
+    }
+}

Sources/DistributedCluster/ClusterSystemSettings.swift

@@ -405,18 +405,19 @@ protocol ClusterSystemInstrumentationProvider {
 /// all the nodes of an existing cluster.
 public struct ServiceDiscoverySettings {
     let implementation: ServiceDiscoveryImplementation
-    private let _subscribe: (@escaping (Result<[Cluster.Endpoint], Error>) -> Void, @escaping (CompletionReason) -> Void) -> CancellationToken?
 
     public init<Discovery, S>(_ implementation: Discovery, service: S)
     where
         Discovery: ServiceDiscovery,
         Discovery.Instance == Cluster.Endpoint,
         S == Discovery.Service
     {
-        self.implementation = .dynamic(AnyServiceDiscovery(implementation))
-        self._subscribe = { onNext, onComplete in
-            implementation.subscribe(to: service, onNext: onNext, onComplete: onComplete)
-        }
+        self.implementation = .dynamic(
+            serviceDiscovery: AnyServiceDiscovery(implementation),
+            subscribe: { onNext, onComplete in
+                implementation.subscribe(to: service, onNext: onNext, onComplete: onComplete)
+            }
+        )
     }
 
     public init<Discovery, S>(_ implementation: Discovery, service: S, mapInstanceToNode transformer: @escaping (Discovery.Instance) throws -> Cluster.Endpoint)
@@ -425,33 +426,78 @@ public struct ServiceDiscoverySettings {
         S == Discovery.Service
     {
         let mappedDiscovery: MapInstanceServiceDiscovery<Discovery, Cluster.Endpoint> = implementation.mapInstance(transformer)
-        self.implementation = .dynamic(AnyServiceDiscovery(mappedDiscovery))
-        self._subscribe = { onNext, onComplete in
-            mappedDiscovery.subscribe(to: service, onNext: onNext, onComplete: onComplete)
+        self.implementation = .dynamic(
+            serviceDiscovery: AnyServiceDiscovery(mappedDiscovery),
+            subscribe: { onNext, onComplete in
+                mappedDiscovery.subscribe(to: service, onNext: onNext, onComplete: onComplete)
+            }
+        )
+    }
+
+    init(clusterdEndpoint: Cluster.Endpoint) {
+        self.implementation = .clusterDaemon(
+            endpoint: clusterdEndpoint,
+            initialize: { system in
+                system.log.info("Joining [clusterd] at \(clusterdEndpoint)")
+                system.cluster.join(endpoint: clusterdEndpoint)
+            }
+        )
+    }
+
+    /// Locate the default `ClusterD` process and use it for discovering cluster nodes.
+    public static var clusterd: Self {
+        get {
+            Self.clusterd(endpoint: nil)
         }
     }
 
+    public static func clusterd(endpoint: Cluster.Endpoint?) -> Self {
+        ServiceDiscoverySettings(clusterdEndpoint: endpoint ?? ClusterDaemon.defaultEndpoint)
+    }
+
+    public static func seed(nodes: Set<Cluster.Endpoint>) -> Self {
+        .init(static: nodes)
+    }
+
     public init(static nodes: Set<Cluster.Endpoint>) {
-        self.implementation = .static(nodes)
-        self._subscribe = { onNext, _ in
-            // Call onNext once and never again since the list of nodes doesn't change
-            onNext(.success(Array(nodes)))
-            // Ignore onComplete because static service discovery never terminates
-
-            // No cancellation token
-            return nil
-        }
+        self.implementation = .static(
+            endpoints: nodes,
+            subscribe: { onNext, _ in
+                // Call onNext once and never again since the list of nodes doesn't change
+                onNext(.success(Array(nodes)))
+                // Ignore onComplete because static service discovery never terminates
+
+                // No cancellation token
+                return nil
+            }
+        )
     }
 
     /// Similar to `ServiceDiscovery.subscribe` however it allows the handling of the listings to be generic and handled by the cluster system.
     /// This function is only intended for internal use by the `DiscoveryShell`.
     func subscribe(onNext nextResultHandler: @escaping (Result<[Cluster.Endpoint], Error>) -> Void, onComplete completionHandler: @escaping (CompletionReason) -> Void) -> CancellationToken? {
-        self._subscribe(nextResultHandler, completionHandler)
+        switch self.implementation {
+        case .static(_, let subscribe),
+            .dynamic(_, let subscribe):
+            subscribe(nextResultHandler, completionHandler)
+        case .clusterDaemon:
+            .none
+        }
+    }
+
+    func initializeClusterd(_ system: ClusterSystem) {
+        switch self.implementation {
+        case .clusterDaemon(_, let initialize):
+            initialize(system)
+        default:
+            break
+        }
     }
 
     enum ServiceDiscoveryImplementation {
-        case `static`(Set<Cluster.Endpoint>)
-        case dynamic(AnyServiceDiscovery)
+        case `static`(endpoints: Set<Cluster.Endpoint>, subscribe: (@escaping (Result<[Cluster.Endpoint], Error>) -> Void, @escaping (CompletionReason) -> Void) -> CancellationToken?)
+        case dynamic(serviceDiscovery: AnyServiceDiscovery, subscribe: (@escaping (Result<[Cluster.Endpoint], Error>) -> Void, @escaping (CompletionReason) -> Void) -> CancellationToken?)
+        case clusterDaemon(endpoint: Cluster.Endpoint, initialize: (ClusterSystem) -> Void)
     }
 }
 

Sources/MultiNodeTestKit/MultiNodeTestKit.swift

@@ -36,6 +36,7 @@ public struct MultiNodeTest {
     public let crashRegex: String?
     public let runTest: (any MultiNodeTestControlProtocol) async throws -> Void
     public let configureActorSystem: (inout ClusterSystemSettings) -> Void
+    public let startNode: (ClusterSystemSettings) async throws -> ClusterSystem
     public let configureMultiNodeTest: (inout MultiNodeTestSettings) -> Void
     public let makeControl: (String) -> any MultiNodeTestControlProtocol
 
@@ -51,6 +52,7 @@ public struct MultiNodeTest {
         }
 
         self.configureActorSystem = TestSuite.configureActorSystem
+        self.startNode = TestSuite.startNode
         self.configureMultiNodeTest = TestSuite.configureMultiNodeTest
 
         self.makeControl = { nodeName -> Control<TestSuite.Nodes> in
@@ -80,6 +82,7 @@ public protocol MultiNodeTestSuite {
     init()
     associatedtype Nodes: MultiNodeNodes
     static func configureActorSystem(settings: inout ClusterSystemSettings)
+    static func startNode(settings: ClusterSystemSettings) async throws -> ClusterSystem
     static func configureMultiNodeTest(settings: inout MultiNodeTestSettings)
 }
 
@@ -88,8 +91,8 @@ extension MultiNodeTestSuite {
         "\(Self.self)".split(separator: ".").last.map(String.init) ?? ""
     }
 
-    public func configureActorSystem(settings: inout ClusterSystemSettings) {
-        // do nothing by default
+    public static func startNode(settings: ClusterSystemSettings) async throws -> ClusterSystem {
+        await ClusterSystem(settings: settings)
     }
 
     var nodeNames: [String] {

Sources/MultiNodeTestKitRunner/boot+MultiNodeTestKitRunner+Exec.swift

@@ -50,27 +50,34 @@ extension MultiNodeTestKitRunnerBoot {
             )
         }
 
-        let actorSystem = await ClusterSystem(nodeName) { settings in
-            settings.bindHost = myNode.host
-            settings.bindPort = myNode.port
-
-            /// Configure a nicer logger, that pretty prints metadata and also includes source location of logs
-            if multiNodeSettings.installPrettyLogger {
-                settings.logging.baseLogger = Logger(
-                    label: nodeName,
-                    factory: { label in
-                        PrettyMultiNodeLogHandler(nodeName: label, settings: multiNodeSettings.logCapture)
-                    }
-                )
-            }
+        var settings = ClusterSystemSettings(name: nodeName)
+        settings.bindHost = myNode.host
+        settings.bindPort = myNode.port
+
+        /// Configure a nicer logger, that pretty prints metadata and also includes source location of logs
+        if multiNodeSettings.installPrettyLogger {
+            settings.logging.baseLogger = Logger(
+                label: nodeName,
+                factory: { label in
+                    PrettyMultiNodeLogHandler(nodeName: label, settings: multiNodeSettings.logCapture)
+                }
+            )
+        }
 
-            // we use the singleton to implement a simple Coordinator
-            // TODO: if the node hosting the coordinator dies we'd potentially have some races at hand
-            //       there's a few ways to solve this... but for now this is good enough.
-            settings += ClusterSingletonPlugin()
+        // we use the singleton to implement a simple Coordinator
+        // TODO: if the node hosting the coordinator dies we'd potentially have some races at hand
+        //       there's a few ways to solve this... but for now this is good enough.
+        settings += ClusterSingletonPlugin()
 
-            multiNodeTest.configureActorSystem(&settings)
-        }
+        multiNodeTest.configureActorSystem(&settings)
+
+        // we use the singleton to implement a simple Coordinator
+        // TODO: if the node hosting the coordinator dies we'd potentially have some races at hand
+        //       there's a few ways to solve this... but for now this is good enough.
+        settings += ClusterSingletonPlugin()
+        multiNodeTest.configureActorSystem(&settings)
+
+        let actorSystem = try await multiNodeTest.startNode(settings)
         control._actorSystem = actorSystem
 
         let signalQueue = DispatchQueue(label: "multi.node.\(multiNodeTest.testSuiteName).\(multiNodeTest.testName).\(nodeName).SignalHandlerQueue")