Implemented that master process restarts only as many times as configured

ktoso · ktoso · commit 658ffe4dd38e · 2019-08-29T23:34:34.000+09:00
diff --git a/IntegrationTests/tests_02_process_isolated/it_ProcessIsolated_escalatingWorkers/main.swift b/IntegrationTests/tests_02_process_isolated/it_ProcessIsolated_escalatingWorkers/main.swift
@@ -33,8 +33,8 @@ pprint("Started process: \(getpid()) with roles: \(isolated.roles)")
 struct OnPurposeBoom: Error {}
 
 isolated.run(on: .master) {
-    isolated.spawnServantProcess(supervision: .restart(atMost: 1, within: nil), args: ["fatalError"])
-    isolated.spawnServantProcess(supervision: .restart(atMost: 1, within: nil), args: ["escalateError"])
+    isolated.spawnServantProcess(supervision: .replace(atMost: 1, within: nil), args: ["fatalError"])
+    isolated.spawnServantProcess(supervision: .replace(atMost: 1, within: nil), args: ["escalateError"])
 }
 
 try isolated.run(on: .servant) {
@@ -58,8 +58,8 @@ try isolated.run(on: .servant) {
                     // since we .escalate and are a top-level actor, this will cause the process to die as well
                     throw OnPurposeBoom()
                 } else {
-                    context.log.error("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly.")
-                    fatalError("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly.")
+                    context.log.error("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly. \(CommandLine.arguments)")
+                    fatalError("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly. \(CommandLine.arguments)")
                 }
             }
         })
diff --git a/IntegrationTests/tests_02_process_isolated/it_ProcessIsolated_noLeaking/main.swift b/IntegrationTests/tests_02_process_isolated/it_ProcessIsolated_noLeaking/main.swift
@@ -43,7 +43,7 @@ try isolated.run(on: .master) {
 
     /// spawn a servant
 
-    isolated.spawnServantProcess(supervision: .restart(atMost: 100, within: .seconds(1)), args: ["ALPHA"])
+    isolated.spawnServantProcess(supervision: .replace(atMost: 100, within: .seconds(1)), args: ["ALPHA"])
 }
 
 // finally, once prepared, you have to invoke the following:
diff --git a/IntegrationTests/tests_02_process_isolated/it_ProcessIsolated_respawnsServants/main.swift b/IntegrationTests/tests_02_process_isolated/it_ProcessIsolated_respawnsServants/main.swift
@@ -49,7 +49,7 @@ try isolated.run(on: .master) {
     })
 
     // should we allow anyone to issue this, or only on master? we could `runOnMaster { control` etc
-    isolated.spawnServantProcess(supervision: .restart(atMost: 100, within: .seconds(1)), args: ["ALPHA"])
+    isolated.spawnServantProcess(supervision: .replace(atMost: 100, within: .seconds(1)), args: ["ALPHA"])
 }
 
 // Notice that master has no workers, just the pool...
diff --git a/IntegrationTests/tests_02_process_isolated/shared.sh b/IntegrationTests/tests_02_process_isolated/shared.sh
@@ -13,6 +13,10 @@
 ##
 ##===----------------------------------------------------------------------===##
 
+function echoerr() {
+    echo "$@" 1>&2;
+}
+
 function _killall() {
     set +e
     local killall_app_name="$1"
diff --git a/IntegrationTests/tests_02_process_isolated/test_02_kill_servant_master_restarts_it.sh b/IntegrationTests/tests_02_process_isolated/test_02_kill_servant_master_restarts_it.sh
@@ -58,7 +58,8 @@ await_n_processes "$app_name" 2
 
 if [[ $(ps aux | awk '{print $2}' | grep ${pid_servant}  | grep -v 'grep' | wc -l) -ne 0 ]]; then
     echo "ERROR: Seems the servant was not killed!!!"
-    exit -2
+    _killall ${app_name}
+    exit -1
 fi
 
 await_n_processes "$app_name" 2
diff --git a/IntegrationTests/tests_02_process_isolated/test_03_servant_spawning_not_leak_fds.sh b/IntegrationTests/tests_02_process_isolated/test_03_servant_spawning_not_leak_fds.sh
@@ -52,7 +52,9 @@ for pid_servant in $pid_servants; do
     if [[ $(lsof -p $pid_servant | wc -l) -gt 100 ]]; then
         lsof -p $pid_servant
         printf "${RED}ERROR: Seems the servant [${pid_servant}] has too many FDs open, did the masters FD leak?${RST}\n"
-        exit -2
+
+        _killall ${app_name}
+        exit -1
     fi
 done
 
diff --git a/IntegrationTests/tests_02_process_isolated/test_04_failing_servants_to_cause_servant_restart.sh b/IntegrationTests/tests_02_process_isolated/test_04_failing_servants_to_cause_servant_restart.sh
@@ -37,32 +37,51 @@ _killall ${app_name}
 
 swift build # synchronously ensure built
 
-swift run ${app_name} &
+declare -r log_file="/tmp/${app_name}.log"
+rm -f ${log_file}
+swift run ${app_name} > ${log_file} &
+
+declare -r supervision_replace_grep_txt='supervision: REPLACE'
+declare -r supervision_stop_grep_txt='supervision: STOP'
+
+# we want to wait until 2 STOPs are found in the logs; then we can check if the other conditions are as we expect
+echo "Waiting for servants to REPLACE and STOP..."
+spin=1 # spin counter
+max_spins=20
+while [[ $(cat ${log_file} | grep "${supervision_stop_grep_txt}" | wc -l) -ne 2 ]]; do
+    sleep 1
+    spin=$((spin+1))
+    if [[ ${spin} -eq ${max_spins} ]]; then
+        echoerr "Never saw enough '${supervision_stop_grep_txt}' in logs."
+        cat ${log_file}
+        exit -1
+    fi
+done
 
-await_n_processes "$app_name" 3
-
-pid_master=$(ps aux | grep ${app_name} | grep -v grep | grep -v servant | awk '{ print $2 }')
-pid_servant=$(ps aux | grep ${app_name} | grep -v grep | grep servant | head -n1 | awk '{ print $2 }')
-
-echo "> PID Master: ${pid_master}"
-echo "> PID Servant: ${pid_servant}"
+echo '~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
+cat ${log_file} | grep "${supervision_replace_grep_txt}"
+echo '~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
 
-echo '~~~~~~~~~~~~BEFORE KILL~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
-ps aux | grep ${app_name}
 echo '~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
+cat ${log_file} | grep "${supervision_stop_grep_txt}"
+echo '~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
+
+if [[ $(cat ${log_file} | grep "${supervision_replace_grep_txt}" | wc -l) -ne 2 ]]; then
+    echoerr "ERROR: We expected 2 servants to only restart once, yet more restarts were detected!"
+    cat ${log_file}
 
-sleep 3 # TODO rather, sleep until another proc replaces the servant automatically
+    _killall ${app_name}
+    exit -1
+fi
 
-# the 1 servant should die, but be restarted so we'll be back at two processes
-await_n_processes "$app_name" 3
+if [[ $(cat ${log_file} | grep "${supervision_stop_grep_txt}" | wc -l) -ne 2 ]]; then
+    echoerr "ERROR: Expected the servants to STOP after they are replaced once!"
+    cat ${log_file}
 
-if [[ $(ps aux | awk '{print $2}' | grep ${pid_servant}  | grep -v 'grep' | wc -l) -ne 0 ]]; then
-    echo "ERROR: Seems the servant was not killed!!!"
+    _killall ${app_name}
     exit -2
 fi
 
-await_n_processes "$app_name" 2
-
 # === cleanup ----------------------------------------------------------------------------------------------------------
 
 _killall ${app_name}
diff --git a/Sources/DistributedActors/Cluster/ClusterShell.swift b/Sources/DistributedActors/Cluster/ClusterShell.swift
@@ -678,27 +678,3 @@ extension ClusterShell {
         }
     }
 }
-
-// ==== ----------------------------------------------------------------------------------------------------------------
-// MARK: ActorSystem extensions
-
-extension ActorSystem {
-    internal var clusterShell: ActorRef<ClusterShell.Message> {
-        return self._cluster?.ref ?? self.deadLetters.adapt(from: ClusterShell.Message.self)
-    }
-
-    // TODO: not sure how to best expose, but for now this is better than having to make all internal messages public.
-    public func join(node: Node) {
-        self.clusterShell.tell(.command(.join(node)))
-    }
-
-    // TODO: not sure how to best expose, but for now this is better than having to make all internal messages public.
-    public func _dumpAssociations() {
-        let ref: ActorRef<Set<UniqueNode>> = try! self.spawn(.anonymous, .receive { context, nodes in
-            let stringlyNodes = nodes.map { String(reflecting: $0) }.joined(separator: "\n     ")
-            context.log.info("~~~~ ASSOCIATED NODES ~~~~~\n     \(stringlyNodes)")
-            return .stop
-        })
-        self.clusterShell.tell(.query(.associatedNodes(ref)))
-    }
-}
diff --git a/Sources/DistributedActors/Cluster/NodeDeathWatcher.swift b/Sources/DistributedActors/Cluster/NodeDeathWatcher.swift
@@ -135,13 +135,12 @@ enum NodeDeathWatcherShell {
             let instance = NodeDeathWatcherInstance(selfNode: context.system.settings.cluster.uniqueBindNode)
 
             context.system.cluster.events.subscribe(context.subReceive(ClusterEvent.self) { event in
-                context.log.info("EVENT::::: \(event)")
                 switch event {
                 case .membership(.memberDown(let member)):
                     let change = MembershipChange(node: member.node, fromStatus: .none, toStatus: .down)
                     instance.handleAddressDown(change)
                 default:
-                    () // ignore for now...
+                    () // ignore other changes, we only need to react on nodes becoming DOWN
                 }
             })
 
diff --git a/Sources/DistributedActors/Cluster/SWIM/SWIMSettings.swift b/Sources/DistributedActors/Cluster/SWIM/SWIMSettings.swift
@@ -36,11 +36,11 @@ public struct SWIMSettings {
     /// These logs will contain SWIM.Instance metadata, as offered by `SWIM.Instance.metadata`.
     /// All logs will be prefixed using `[tracelog:SWIM]`, for easier grepping and inspecting only logs related to the SWIM instance.
     // TODO: how to make this nicely dynamically changeable during runtime
-//    #if SACT_TRACELOG_SWIM
+    #if SACT_TRACELOG_SWIM
     var traceLogLevel: Logger.Level? = .warning
-//    #else
-//    var traceLogLevel: Logger.Level?
-//    #endif
+    #else
+    var traceLogLevel: Logger.Level?
+    #endif
 }
 
 extension SWIM {
diff --git a/Sources/DistributedActors/ProcessIsolated/ProcessCommander.swift b/Sources/DistributedActors/ProcessIsolated/ProcessCommander.swift
@@ -14,21 +14,25 @@
 
 /// EXPERIMENTAL.
 // Master (Process) and Commander (Actor): The Far Side of the World
-public struct ProcessCommander {
-    public static let naming: ActorNaming = "processCommander"
+internal struct ProcessCommander {
     public static let name: String = "processCommander"
+    public static let naming: ActorNaming = .unique(name)
 
-    public enum Command {
-        case requestSpawnServant(ServantProcessSupervisionStrategy, args: [String], delay: TimeAmount?)
+    internal enum Command {
+        case requestSpawnServant(ServantProcessSupervisionStrategy, args: [String])
+        case requestReplaceServant(ServantProcess, delay: TimeAmount?)
     }
 
-    private let funRemoveServantPid: (Int) -> Void
+    private let funRemoveServantByPID: (Int) -> Void
     private let funSpawnServantProcess: (ServantProcessSupervisionStrategy, [String]) -> Void
+    private let funReplaceServantProcess: (ServantProcess) -> Void
 
-    public init(funSpawnServantProcess: @escaping (ServantProcessSupervisionStrategy, [String]) -> Void,
-                funKillServantProcess: @escaping (Int) -> Void) {
+    init(funSpawnServantProcess: @escaping (ServantProcessSupervisionStrategy, [String]) -> Void,
+         funReplaceServantProcess: @escaping (ServantProcess) -> Void,
+         funKillServantProcess: @escaping (Int) -> Void) {
         self.funSpawnServantProcess = funSpawnServantProcess
-        self.funRemoveServantPid = funKillServantProcess
+        self.funReplaceServantProcess = funReplaceServantProcess
+        self.funRemoveServantByPID = funKillServantProcess
     }
 
     private var _servants: [Int: ServantProcess] = [:]
@@ -54,13 +58,17 @@ public struct ProcessCommander {
 
             return .receiveMessage { message in
                 switch message {
-                case .requestSpawnServant(let supervision, let args, .none):
+                case .requestSpawnServant(let supervision, let args):
                     context.log.info("Spawning new servant process; Supervision \(supervision), arguments: \(args)")
                     self.funSpawnServantProcess(supervision, args)
 
-                case .requestSpawnServant(let supervision, let args, .some(let delay)):
-                    context.log.info("Scheduling spawning of new servant process in [\(delay.prettyDescription)]; Supervision \(supervision), arguments: \(args)")
-                    context.timers.startSingle(key: nextSpawnServantTimerKey(), message: .requestSpawnServant(supervision, args: args, delay: nil), delay: delay)
+                case .requestReplaceServant(let servant, .some(let delay)):
+                    context.log.info("Scheduling spawning of new servant process in [\(delay.prettyDescription)]; Servant to be replaced: \(servant), in \(delay.prettyDescription)")
+                    context.timers.startSingle(key: nextSpawnServantTimerKey(), message: .requestReplaceServant(servant, delay: nil), delay: delay)
+                case .requestReplaceServant(let servant, .none):
+                    // restart immediately
+                    context.log.info("Spawning replacement servant process; Supervision \(servant.supervisionStrategy), arguments: \(servant.args)")
+                    self.funReplaceServantProcess(servant)
                 }
                 return .same
             }
diff --git a/Sources/DistributedActors/ProcessIsolated/ProcessIsolated+Supervision.swift b/Sources/DistributedActors/ProcessIsolated/ProcessIsolated+Supervision.swift
@@ -0,0 +1,114 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Distributed Actors open source project
+//
+// Copyright (c) 2018-2019 Apple Inc. and the Swift Distributed Actors project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.md for the list of Swift Distributed Actors project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+#if os(OSX)
+import Darwin.C
+#else
+import Glibc
+#endif
+
+import DistributedActorsConcurrencyHelpers
+
+// ==== ----------------------------------------------------------------------------------------------------------------
+// MARK: Servant Supervision
+
+/// Configures supervision for a specific su
+///
+/// Similar to `SupervisionStrategy` (which is for actors), however in effect for servant processes.
+///
+/// - SeeAlso: `SupervisionStrategy` for detailed documentation on supervision and timing semantics.
+public struct ServantProcessSupervisionStrategy {
+    internal let underlying: SupervisionStrategy
+
+    /// Stopping supervision strategy, meaning that terminated servant processes will not get automatically spawned replacements.
+    ///
+    /// It is useful if you want to manually manage replacements and servant processes, however note that without restarting
+    /// servants, the system may end up in a state with no servants, and only the master running, so you should plan to take
+    /// action in case this happens (e.g. by terminating the master itself, and relying on a higher level orchestrator to restart
+    /// the entire system).
+    public static var stop: ServantProcessSupervisionStrategy {
+        return .init(underlying: .stop)
+    }
+
+    /// Supervision strategy binding the lifecycle of the master process with the given servant process,
+    /// i.e. if a servant process supervised using this strategy terminates (exits, fails, for whatever reason),
+    /// the master parent will also terminate (with an error exit code).
+    public static var escalate: ServantProcessSupervisionStrategy {
+        return .init(underlying: .escalate)
+    }
+
+    /// The replacing strategy allows the supervised servant process to be restarted `atMost` times `within` a time period.
+    /// In addition, each subsequent restart _may_ be performed after a certain backoff.
+    ///
+    /// ### Servant Replacement vs. Actor Restart semantics
+    /// While servant `.replace` supervision may, on the surface, seem identical to `restart` supervision of actors,
+    /// it differs in one crucial aspect: supervising actors with `.restart` allows them to retain the existing mailbox
+    /// and create a new instance of the initial behavior to continue serving the same mailbox, i.e. only a single message is lost upon restart.
+    /// Servant process replacement sadly cannot guarantee this, and all mailboxes and state in the servant process is lost, including all mailboxes,
+    /// thus explaining the slightly different naming and semantics implications of this supervision strategy.
+    ///
+    /// - SeeAlso: The actor `SupervisionStrategy` documentation, which explains the exact semantics of this supervision mechanism in-depth.
+    ///
+    /// - parameter atMost: number of attempts allowed restarts within a single failure period (defined by the `within` parameter. MUST be > 0).
+    /// - parameter within: amount of time within which the `atMost` failures are allowed to happen. This defines the so called "failure period",
+    ///                     which runs from the first failure encountered for `within` time, and if more than `atMost` failures happen in this time amount then
+    ///                     no restart is performed and the failure is escalated (and the actor terminates in the process).
+    /// - parameter backoff: strategy to be used for suspending the failed actor for a given (backoff) amount of time before completing the restart.
+    public static func replace(atMost: Int, within: TimeAmount?, backoff: BackoffStrategy? = nil) -> ServantProcessSupervisionStrategy {
+        return .init(underlying: .restart(atMost: atMost, within: within, backoff: backoff))
+    }
+}
+
+extension ProcessIsolated {
+    func monitorServants() {
+        let res = POSIXProcessUtils.nonBlockingWaitPID(pid: 0)
+        if res.pid > 0 {
+            let maybeServant = self.removeServant(pid: res.pid)
+
+            guard var servant = maybeServant else {
+                self.system.log.warning("Unknown PID died, ignoring... PID was: \(res.pid)")
+                return
+            }
+
+            // always DOWN the node that we know has terminated
+            self.system.cluster.down(node: servant.node)
+            // TODO: we could aggressively tell other nodes about the down rather rely on the gossip...?
+
+            // if we have a restart supervision logic, we should apply it.
+            guard var decision = servant.recordFailure() else {
+                self.system.log.info("Servant \(servant.node) (pid:\(res.pid)) has no supervision / restart strategy defined, NO replacement servant will be spawned in its place.")
+                return
+            }
+
+            let messagePrefix = "Servant [\(servant.node) @ pid:\(res.pid)] supervision"
+            switch decision {
+            case .stop:
+                self.system.log.info("\(messagePrefix): STOP, as decided by: \(servant.restartLogic); No replacement process will be spawned for servant.")
+
+            case .escalate:
+                self.system.log.info("\(messagePrefix): ESCALATE, as decided by: \(servant.restartLogic)")
+                self.system.cluster.down(node: self.system.cluster.node)
+                // TODO: ensure we exit the master process as well
+
+            case .restartImmediately:
+                self.system.log.info("\(messagePrefix): REPLACE, as decided by: \(servant.restartLogic)")
+                self.control.requestServantRestart(servant, delay: nil)
+
+            case .restartBackoff(let delay):
+                self.system.log.info("\(messagePrefix): REPLACE BACKOFF, as decided by: \(servant.restartLogic)")
+                self.control.requestServantRestart(servant, delay: delay)
+            }
+        }
+    }
+}
diff --git a/Sources/DistributedActors/ProcessIsolated/ProcessIsolated.swift b/Sources/DistributedActors/ProcessIsolated/ProcessIsolated.swift
diff --git a/Sources/DistributedActors/Supervision.swift b/Sources/DistributedActors/Supervision.swift

Original file line number	Diff line number	Diff line change
`@@ -33,8 +33,8 @@ pprint("Started process: \(getpid()) with roles: \(isolated.roles)")`
`33`	`33`	`struct OnPurposeBoom: Error {}`
`34`	`34`
`35`	`35`	`isolated.run(on: .master) {`
`36`		`- isolated.spawnServantProcess(supervision: .restart(atMost: 1, within: nil), args: ["fatalError"])`
`37`		`- isolated.spawnServantProcess(supervision: .restart(atMost: 1, within: nil), args: ["escalateError"])`
	`36`	`+ isolated.spawnServantProcess(supervision: .replace(atMost: 1, within: nil), args: ["fatalError"])`
	`37`	`+ isolated.spawnServantProcess(supervision: .replace(atMost: 1, within: nil), args: ["escalateError"])`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`try isolated.run(on: .servant) {`
`@@ -58,8 +58,8 @@ try isolated.run(on: .servant) {`
`58`	`58`	`// since we .escalate and are a top-level actor, this will cause the process to die as well`
`59`	`59`	`throw OnPurposeBoom()`
`60`	`60`	`} else {`
`61`		`- context.log.error("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly.")`
`62`		`- fatalError("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly.")`
	`61`	`+ context.log.error("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly. \(CommandLine.arguments)")`
	`62`	`+ fatalError("MISSING FAILURE MODE ARGUMENT!!! Test is constructed not properly, or arguments were not passed properly. \(CommandLine.arguments)")`
`63`	`63`	`}`
`64`	`64`	`}`
`65`	`65`	`})`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ try isolated.run(on: .master) {`
`43`	`43`
`44`	`44`	`/// spawn a servant`
`45`	`45`
`46`		`- isolated.spawnServantProcess(supervision: .restart(atMost: 100, within: .seconds(1)), args: ["ALPHA"])`
	`46`	`+ isolated.spawnServantProcess(supervision: .replace(atMost: 100, within: .seconds(1)), args: ["ALPHA"])`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`// finally, once prepared, you have to invoke the following:`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ try isolated.run(on: .master) {`
`49`	`49`	`})`
`50`	`50`
`51`	`51`	// should we allow anyone to issue this, or only on master? we could `runOnMaster { control` etc
`52`		`- isolated.spawnServantProcess(supervision: .restart(atMost: 100, within: .seconds(1)), args: ["ALPHA"])`
	`52`	`+ isolated.spawnServantProcess(supervision: .replace(atMost: 100, within: .seconds(1)), args: ["ALPHA"])`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`// Notice that master has no workers, just the pool...`