+system allow parking() when clustering, to be refactored when cluster (#459)

becomes a Transport()

Author: ktoso

IntegrationTests/tests_03_xpc_actorable/it_XPCActorable_echo_service/main.swift

@@ -36,5 +36,5 @@ try! _file.append("service booted...\n")
 let service = try XPCActorableService(system, XPCEchoService.init)
 
 service.park()
-//system.park()
+system.park() // TODO system park should invoke the service park, we only need to park once for XPC to kickoff dispatch_main
 // unreachable, park never exits

Samples/Sources/SampleCluster/main.swift

@@ -93,5 +93,4 @@ if system.cluster.node.port == 7337 { // <2>
 }
 // end::cluster-sample-actors-discover-and-chat[]
 
-system.park()
-Thread.sleep(.seconds(60)) // TODO: make park halt execution here
+system.park(atMost: .seconds(60))

Samples/Sources/SampleDiningPhilosophers/DiningPhilosophers.swift

@@ -32,6 +32,6 @@ struct DiningPhilosophers {
         let _: Philosopher.Ref = try system.spawn("Cory", Philosopher(left: fork3, right: fork4).behavior)
         let _: Philosopher.Ref = try system.spawn("Norman", Philosopher(left: fork4, right: fork5).behavior)
 
-        Thread.sleep(time)
+        system.park(atMost: time)
     }
 }

Samples/Sources/SampleDiningPhilosophers/DistributedDiningPhilosophers.swift

@@ -73,6 +73,6 @@ struct DistributedDiningPhilosophers {
         _ = try systemC.spawn("Cory", Philosopher(left: fork3, right: fork4).behavior)
         _ = try systemC.spawn("Norman", Philosopher(left: fork4, right: fork5).behavior)
 
-        Thread.sleep(time)
+        systemA.park(atMost: time)
     }
 }

Sources/DistributedActors/ActorSystem.swift

@@ -103,8 +103,8 @@ public final class ActorSystem {
 
     // ==== ----------------------------------------------------------------------------------------------------------------
     // MARK: Shutdown
-    private var shutdownReceptacle: BlockingReceptacle<Void>?
-    private let shutdownLock: Lock = Lock()
+    private var shutdownReceptacle = BlockingReceptacle<Void>()
+    private let shutdownLock = Lock()
 
     /// Greater than 0 shutdown has been initiated / is in progress.
     private let shutdownFlag = Atomic(value: 0)
@@ -277,14 +277,26 @@ public final class ActorSystem {
         self.init("ActorSystem")
     }
 
-    /// Some transports require being able to take over the applications thread.
-    /// E.g. this may mean invoking `dispatchMain()` or TODO: `ProcessIsolated.park()`
-    public func park() {
-        self.log.info("Parking actor system...")
+    /// Parks the current thread (usually "main thread") until the system is terminated,
+    /// of the optional timeout is exceeded.
+    ///
+    /// This call is also offered to underlying transports which may have to perform the blocking wait themselves
+    /// (most notably, `ProcessIsolated` does so). Please refer to your configured transports documentation,
+    /// to learn about exact semantics of parking a system while using them.
+    public func park(atMost parkTimeout: TimeAmount? = nil) {
+        let howLongParkingMsg = parkTimeout == nil ? "indefinitely" : "for \(parkTimeout!.prettyDescription)"
+        self.log.info("Parking actor system \(howLongParkingMsg)...")
+
         for transport in self.settings.transports {
             self.log.info("Offering transport [\(transport.protocolName)] chance to park the thread...")
             transport.onActorSystemPark()
         }
+
+        if let maxParkingTime = parkTimeout {
+            self.shutdownReceptacle.wait(atMost: maxParkingTime)
+        } else {
+            self.shutdownReceptacle.wait()
+        }
     }
 
     #if SACT_TESTS_LEAKS
@@ -317,24 +329,11 @@ public final class ActorSystem {
     /// - Returns: A `Shutdown` value that can be waited upon until the system has completed the shutdown.
     @discardableResult
     public func shutdown() -> Shutdown {
-        let (shutdownAlreadyRunning, receptacle) = self.shutdownLock.withLock { () -> (Bool, BlockingReceptacle<Void>) in
-            if let receptacle = self.shutdownReceptacle {
-                return (true, receptacle)
-            } else {
-                let receptacle = BlockingReceptacle<Void>()
-                self.shutdownReceptacle = receptacle
-                return (false, receptacle)
-            }
+        guard self.shutdownFlag.add(1) == 0 else {
+            // shutdown already kicked off by someone else
+            return Shutdown(receptacle: self.shutdownReceptacle)
         }
 
-        // if the shutdown has previously been initiated, we just return the
-        // existing receptacle
-        if shutdownAlreadyRunning {
-            return Shutdown(receptacle: receptacle)
-        }
-
-        _ = self.shutdownFlag.add(1)
-
         self.serialization = nil
         self._cluster = nil
 
@@ -352,10 +351,10 @@ public final class ActorSystem {
             self.dispatcher.shutdown()
             try! self._eventLoopGroup.syncShutdownGracefully()
             self._receptionist = self.deadLetters.adapted()
-            receptacle.offerOnce(())
+            self.shutdownReceptacle.offerOnce(())
         }
 
-        return Shutdown(receptacle: receptacle)
+        return Shutdown(receptacle: self.shutdownReceptacle)
     }
 
     public var cluster: ClusterControl {