mvarjon/plan9 - Change FPX4JMCJTVSWMV76OY5NIZ6IOZ4GGNLGAX5J43EBBHR3XRXTLMWAC

Implement UNIX signal receiver for controlled shutdown.

Created by mvarjon on August 5, 2022

FPX4JMCJTVSWMV76OY5NIZ6IOZ4GGNLGAX5J43EBBHR3XRXTLMWAC

Dependencies

In channels

main

Change contents

Replacement in src/signal_guard.rs at line 1 [2.40]

B:BD[2.40] → [2.41:239]

use tokio::signal;
pub struct SignalGuard {}
impl SignalGuard {
    pub async fn wait_for_stop() -> Result<(), std::io::Error> {
        let guard = signal::ctrl_c().await;
        guard
    }
}

[2.40]

use futures::prelude::*;
use futures::stream::FuturesUnordered;
use log::info;
use std::collections::HashMap;
use tokio::signal::unix::{signal, SignalKind};
use tokio::sync::broadcast::Sender;
pub struct SignalGuard {}
impl SignalGuard {
    pub async fn wait_for_stop(signal_forwarder: &Sender<i32>) -> Option<()> {
        // Default behavior in UNIX is to terminate process when any of the
        // following signals are received. In each of these case we initiate
        // the controlled shutdown.
        let mut interrupt_stream = signal(SignalKind::interrupt()).ok()?;
        let mut terminate_stream = signal(SignalKind::terminate()).ok()?;
        let mut alarm_stream = signal(SignalKind::alarm()).ok()?;
        let mut hangup_stream = signal(SignalKind::hangup()).ok()?;
        let mut quit_stream = signal(SignalKind::quit()).ok()?;
        let mut user_defined1_stream = signal(SignalKind::user_defined1()).ok()?;
        let mut user_defined2_stream = signal(SignalKind::user_defined2()).ok()?;
        let interrupt_future = interrupt_stream.recv();
        let terminate_future = terminate_stream.recv();
        let alarm_future = alarm_stream.recv();
        let hangup_future = hangup_stream.recv();
        let quit_future = quit_stream.recv();
        let user_defined1_future = user_defined1_stream.recv();
        let user_defined2_future = user_defined2_stream.recv();
        let all_futures = vec![
            interrupt_future,
            terminate_future,
            alarm_future,
            hangup_future,
            quit_future,
            user_defined1_future,
            user_defined2_future,
        ];
        let mut signal_map = HashMap::with_capacity(all_futures.len());
        signal_map.insert(0, SignalKind::interrupt().as_raw_value());
        signal_map.insert(1, SignalKind::terminate().as_raw_value());
        signal_map.insert(2, SignalKind::alarm().as_raw_value());
        signal_map.insert(3, SignalKind::hangup().as_raw_value());
        signal_map.insert(4, SignalKind::quit().as_raw_value());
        signal_map.insert(5, SignalKind::user_defined1().as_raw_value());
        signal_map.insert(6, SignalKind::user_defined2().as_raw_value());
        let all_futures = all_futures
            .into_iter()
            .enumerate()
            .map(|(i, fut)| fut.map(move |res| (i, res)))
            .collect::<FuturesUnordered<_>>();
        // use collection as a stream, await only first 1 future to complete
        let first_one = all_futures.take(1).collect::<Vec<_>>().await;
        let signal_index = first_one[0].0;
        // Currently we mask all signals to SIGTERM.
        let received = signal_map[&signal_index];
        info!("SIGNAL_GUARD: RECEIVED CODE {} FROM OS", received);
        signal_forwarder.send(received).unwrap();
        Option::Some(())
    }
}

Replacement in src/shutdown.rs at line 1 [3.36]

B:BD[3.36] → [2.240:1058]

// Adapted from: https://github.com/tokio-rs/mini-redis/blob/069a8e5ee0de445ad3e642548d196ef6b4fa8bc1/src/shutdown.rs
use tokio::sync::broadcast;
pub struct Shutdown {
    /// The receive half of the channel used to listen for shutdown.
    notify: broadcast::Receiver<i32>,
}
impl Shutdown {
    /// Create a new `Shutdown` backed by the given `broadcast::Receiver`.
    pub fn new(notify: broadcast::Receiver<i32>) -> Self {
        Self { notify }
    }
    #[allow(dead_code)]
    pub fn replicate(&self) -> Self {
        Self {
            notify: self.notify.resubscribe(),
        }
    }
    /// Receive the shutdown notice, waiting if necessary.
    pub async fn recv(&mut self) {
        // Cannot receive a "lag error" as only one value is ever sent.
        let _ = self.notify.recv().await;
    }
}

[3.36]

// Adapted from: https://github.com/tokio-rs/mini-redis/blob/069a8e5ee0de445ad3e642548d196ef6b4fa8bc1/src/shutdown.rs
use tokio::sync::broadcast;
pub struct Shutdown {
    /// The receive half of the channel used to listen for shutdown.
    notify: broadcast::Receiver<i32>,
}
impl Shutdown {
    /// Create a new `Shutdown` backed by the given `broadcast::Receiver`.
    pub fn new(notify: broadcast::Receiver<i32>) -> Self {
        Self { notify }
    }
    #[allow(dead_code)]
    pub fn replicate(&self) -> Self {
        Self {
            notify: self.notify.resubscribe(),
        }
    }
    /// Receive the shutdown notice, waiting if necessary.
    pub async fn recv(&mut self) {
        // Cannot receive a "lag error" as only one value is ever sent.
        let _ = self.notify.recv().await;
    }
}

Replacement in src/service.rs at line 1 [3.891]

B:BD[3.891] → [2.1059:1498]

use tokio::time::{sleep, Duration};
pub struct Service {}
pub struct ServiceError {}
impl Service {
    pub fn new() -> Self {
        Self {}
    }
    pub async fn wait(&self) -> Result<u64, ServiceError> {
        sleep(Duration::from_secs(7)).await;
        Result::Ok(117)
    }
    pub async fn serve(&self, _handle: u64) -> Result<(), ServiceError> {
        sleep(Duration::from_secs(15)).await;
        Result::Ok(())
    }
}

[3.891]

use tokio::time::{sleep, Duration};
pub struct Service {}
pub struct ServiceError {}
impl Service {
    pub fn new() -> Self {
        Self {}
    }
    pub async fn wait(&self) -> Result<u64, ServiceError> {
        sleep(Duration::from_secs(7)).await;
        Result::Ok(117)
    }
    pub async fn serve(&self, _handle: u64) -> Result<(), ServiceError> {
        sleep(Duration::from_secs(15)).await;
        Result::Ok(())
    }
}

Insertion in src/main.rs at line 4 [5.49]
[2.1537]
[4.2673]
```
use exitcode;
```

Replacement in src/main.rs at line 34 [5.49]

B:BD[3.3490] → [2.1557:1604]

∅:D[2.1604] → [3.3525:3733]

B:BD[3.3525] → [3.3525:3733]

B:BD[3.3733] → [2.1605:1662]

∅:D[2.1662] → [3.3782:3891]

B:BD[3.3782] → [3.3782:3891]

B:BD[3.3891] → [2.1663:1726]

∅:D[2.1726] → [3.3939:4160]

B:BD[3.3939] → [3.3939:4160]

    match SignalGuard::wait_for_stop().await {
        Ok(()) =>
        // Once we receive the signal here we need
        // to be sure that we exit the application
        // as e.g. CTRL-C is no longer processed if
        // pressed again.
        {
            info!("MAIN: STOP SIGNAL RECEIVED FROM OS");
            // SIGTERM is 15
            let received = 15;
            shutdown_tx.send(received).unwrap();
            info!("MAIN: SEND STOP CONTROL SIGNAL TO SERVER");
        }
        Err(err) => {
            error!("MAIN: Unable to listen for shutdown signal: {}", err);
            // we also shut down in case of error
            panic!("MAIN: Abnormal shutdown.");
        }
    }

[3.3490]

[6.2200]

    info!("MAIN: PARKED");
    SignalGuard::wait_for_stop(&shutdown_tx).await;

Insertion in src/main.rs at line 37 [5.49]

[6.2201]

[3.4161]

    info!("MAIN: STOP SIGNAL RECEIVED FROM OS");
    info!("MAIN: SEND STOP CONTROL SIGNAL TO SERVER");

Insertion in src/main.rs at line 40 [5.49]
[3.4198]
[6.2201]
Replacement in src/main.rs at line 46 [5.49]
B:BD[3.4259] → [3.4259:4280]
```
            exit(0);
```
[3.4259]
[3.4280]
```
            exit(exitcode::OK);
```
Replacement in src/main.rs at line 50 [5.49]
B:BD[2.1819] → [2.1819:1841]
```
            exit(-1);
```
[2.1819]
[2.1841]
```
            exit(exitcode::SOFTWARE);
```

Replacement in log4rs.config.yml at line 1 [4.2991]

B:BD[4.2991] → [2.1852:1965]

refresh_rate: 30 seconds
appenders:
  stdout:
    kind: console
root:
  level: trace
  appenders:
    - stdout

[4.2991]

refresh_rate: 30 seconds
appenders:
  stdout:
    kind: console
root:
  level: trace
  appenders:
    - stdout

Replacement in adr/Design.md at line 1 [7.1080]

B:BD[7.1080] → [2.1966:5479]

# Concepts
This chapter describes the core concepts around the implementation and the IRC specification. Some material may be restated from the source but it is gathered here for reference purposes.
## IRC
For the core IRC concepts see: [IRC concepts](https://modern.ircdocs.horse/#irc-concepts)
## Implementation
### Client
This is something that sends bytes to our server according to the IRC protocol (something that communicates with the server but is not a server). This is also something outside of this implementation and this is modelled as something from which we receive and send bytes to.
### Server
This is our implementation which communicates with the client and other servers in the same network. According to IRC specification network consists of servers and even though there is server to server protocols available in practise one network consists of same server implementations and they can use any protocol they wish to communicate state changes between them. We will utilize this in our implementation.
### User
This entity models a real user (either human or bot) which uses the client to communicate with our server.
### Channel
Channels represent the logical conversation groups in an IRC network. These are synced between servers in the network. This entity ties users to channels in which they can exchange messages.
### Capability
When the client connects to the server they first have an exchange of information which tells the client what capabilities the server provides. This allows the client to establish reasonable communication with the server. In the implementation different capabilities are implemented in a modular manner which are queried during runtime to provide the capability information to the client.
# Architecture
In general principle we follow modular monolith approach where we have a single executable for the application but that application is composed of modules which are independent and focus on specific features (capabilities) and the runtime then ties these together. In addition there are modules which handles the low-level work of communication.
## Components
### Server
This is the main high-level component that drives our server instance.
#### Capability
There are multiple capability components that each implement a specific functionality to the server.
### Pool
This is the component which tracks the running service instance in the network it is attached to. Network can consist of single server instance. State changes in this instance are then propagated to other servers in pool.
#### Server Communicator
This sub-component of the pool manages the communication between server instances.
### State Manager
This component maintains the state information (users, channels, etc.) for the current instance. It is linked to the pool component and the pool observes the changes in the state manager and propagates them forward and vice versa. That is pool may receive state update from a different server instance and pool then signals these changes to this instance's state manager.
### Protocol Stack
This component maintains the reference information for different communication and message components about specific IRC code constants and other similar information in the IRC protocol.
### Communicator
This component is responsible for sending and receiving bytes between client and server. It ensures the correct formatting of messages and their validation. This component is also linked to the state manager.

[7.1080]

# Concepts
This chapter describes the core concepts around the implementation and the IRC specification. Some material may be restated from the source but it is gathered here for reference purposes.
## IRC
For the core IRC concepts see: [IRC concepts](https://modern.ircdocs.horse/#irc-concepts)
## Implementation
### Client
This is something that sends bytes to our server according to the IRC protocol (something that communicates with the server but is not a server). This is also something outside of this implementation and this is modelled as something from which we receive and send bytes to.
### Server
This is our implementation which communicates with the client and other servers in the same network. According to IRC specification network consists of servers and even though there is server to server protocols available in practise one network consists of same server implementations and they can use any protocol they wish to communicate state changes between them. We will utilize this in our implementation.
### User
This entity models a real user (either human or bot) which uses the client to communicate with our server.
### Channel
Channels represent the logical conversation groups in an IRC network. These are synced between servers in the network. This entity ties users to channels in which they can exchange messages.
### Capability
When the client connects to the server they first have an exchange of information which tells the client what capabilities the server provides. This allows the client to establish reasonable communication with the server. In the implementation different capabilities are implemented in a modular manner which are queried during runtime to provide the capability information to the client.
# Architecture
In general principle we follow modular monolith approach where we have a single executable for the application but that application is composed of modules which are independent and focus on specific features (capabilities) and the runtime then ties these together. In addition there are modules which handles the low-level work of communication.
## Components
### Server
This is the main high-level component that drives our server instance.
#### Capability
There are multiple capability components that each implement a specific functionality to the server.
### Pool
This is the component which tracks the running service instance in the network it is attached to. Network can consist of single server instance. State changes in this instance are then propagated to other servers in pool.
#### Server Communicator
This sub-component of the pool manages the communication between server instances.
### State Manager
This component maintains the state information (users, channels, etc.) for the current instance. It is linked to the pool component and the pool observes the changes in the state manager and propagates them forward and vice versa. That is pool may receive state update from a different server instance and pool then signals these changes to this instance's state manager.
### Protocol Stack
This component maintains the reference information for different communication and message components about specific IRC code constants and other similar information in the IRC protocol.
### Communicator
This component is responsible for sending and receiving bytes between client and server. It ensures the correct formatting of messages and their validation. This component is also linked to the state manager.

Insertion in Cargo.toml at line 11 [5.123]
[5.395]
[4.3106]
```
exitcode = "1.1.2"
```
Insertion in Cargo.toml at line 13 [5.123]
[4.3121]
[4.3121]
```
futures = "0.3.21"
```

Insertion in Cargo.lock at line 66 [5.431]

[8.1955]

[4.4128]

name = "exitcode"
version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "de853764b47027c2e862a995c34978ffa63c1501f2e15f987ba11bd4f9bba193"
[[package]]

Insertion in Cargo.lock at line 76 [5.431]

[4.4302]


[[package]]
name = "futures"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f73fe65f54d1e12b726f517d3e2135ca3125a437b6d998caf1962961f7172d9e"
dependencies = [
 "futures-channel",
 "futures-core",
 "futures-executor",
 "futures-io",
 "futures-sink",
 "futures-task",
 "futures-util",
]
[[package]]
name = "futures-channel"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3083ce4b914124575708913bca19bfe887522d6e2e6d0952943f5eac4a74010"
dependencies = [
 "futures-core",
 "futures-sink",
]
[[package]]
name = "futures-core"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c09fd04b7e4073ac7156a9539b57a484a8ea920f79c7c675d05d289ab6110d3"
[[package]]
name = "futures-executor"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9420b90cfa29e327d0429f19be13e7ddb68fa1cccb09d65e5706b8c7a749b8a6"
dependencies = [
 "futures-core",
 "futures-task",
 "futures-util",
]
[[package]]
name = "futures-io"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc4045962a5a5e935ee2fdedaa4e08284547402885ab326734432bed5d12966b"
[[package]]
name = "futures-macro"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "33c1e13800337f4d4d7a316bf45a567dbcb6ffe087f16424852d97e97a91f512"
dependencies = [
 "proc-macro2",
 "quote",
 "syn",
]
[[package]]
name = "futures-sink"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "21163e139fa306126e6eedaf49ecdb4588f939600f0b1e770f4205ee4b7fa868"
[[package]]
name = "futures-task"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "57c66a976bf5909d801bbef33416c41372779507e7a6b3a5e25e4749c58f776a"
[[package]]
name = "futures-util"
version = "0.3.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d8b7abd5d659d9b90c8cba917f6ec750a74e2dc23902ef9cd4cc8c8b22e6036a"
dependencies = [
 "futures-channel",
 "futures-core",
 "futures-io",
 "futures-macro",
 "futures-sink",
 "futures-task",
 "memchr",
 "pin-project-lite",
 "pin-utils",
 "slab",
]

Insertion in Cargo.lock at line 271 [5.431]
[8.2829]
[4.5964]
```
 "exitcode",
 "futures",
```

Insertion in Cargo.lock at line 368 [5.431]

[8.4444]


[[package]]
name = "pin-utils"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"

Insertion in Cargo.lock at line 474 [5.431]

[8.5556]

]
[[package]]
name = "slab"
version = "0.4.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4614a76b2a8be0058caa9dbbaf66d988527d86d003c11a94fbd335d7661edcef"
dependencies = [
 "autocfg",

Implement UNIX signal receiver for controlled shutdown.

Dependencies

In channels

Change contents

Replacement in src/signal_guard.rs at line 1 [2.40]

Replacement in src/shutdown.rs at line 1 [3.36]

Replacement in src/service.rs at line 1 [3.891]

Insertion in src/main.rs at line 4 [5.49]

Replacement in src/main.rs at line 34 [5.49]

Insertion in src/main.rs at line 37 [5.49]

Insertion in src/main.rs at line 40 [5.49]

Replacement in src/main.rs at line 46 [5.49]

Replacement in src/main.rs at line 50 [5.49]

Replacement in log4rs.config.yml at line 1 [4.2991]

Replacement in adr/Design.md at line 1 [7.1080]

Insertion in Cargo.toml at line 11 [5.123]

Insertion in Cargo.toml at line 13 [5.123]

Insertion in Cargo.lock at line 66 [5.431]

Insertion in Cargo.lock at line 76 [5.431]

Insertion in Cargo.lock at line 271 [5.431]

Insertion in Cargo.lock at line 368 [5.431]

Insertion in Cargo.lock at line 474 [5.431]