diff --git a/crates/kumo-api-types/src/egress_path.rs b/crates/kumo-api-types/src/egress_path.rs index 530f4eda..ac9a7f5c 100644 --- a/crates/kumo-api-types/src/egress_path.rs +++ b/crates/kumo-api-types/src/egress_path.rs @@ -126,6 +126,15 @@ pub fn find_rustls_cipher_suite(name: &str) -> Option { None } +#[derive(Deserialize, Serialize, Debug, Clone, Default, Copy, PartialEq, Eq)] +#[cfg_attr(feature = "lua", derive(FromLua))] +pub enum MemoryReductionPolicy { + #[default] + ShrinkDataAndMeta, + ShrinkData, + NoShrink, +} + #[derive(Deserialize, Serialize, Debug, Clone, Default, Copy, PartialEq, Eq)] #[cfg_attr(feature = "lua", derive(FromLua))] pub enum ConfigRefreshStrategy { @@ -285,6 +294,14 @@ pub struct EgressPathConfig { /// Which thread pool to use for processing the ready queue #[serde(default)] pub readyq_pool_name: Option, + + /// What to do to newly inserted messages when memory is low + #[serde(default)] + pub low_memory_reduction_policy: MemoryReductionPolicy, + + /// What to do to newly inserted messages when memory is over the soft limit + #[serde(default)] + pub no_memory_reduction_policy: MemoryReductionPolicy, } #[cfg(feature = "lua")] @@ -333,6 +350,8 @@ impl Default for EgressPathConfig { use_lmtp: false, reconnect_strategy: ReconnectStrategy::default(), readyq_pool_name: None, + low_memory_reduction_policy: MemoryReductionPolicy::default(), + no_memory_reduction_policy: MemoryReductionPolicy::default(), } } } diff --git a/crates/kumo-api-types/src/shaping.rs b/crates/kumo-api-types/src/shaping.rs index 1b666c39..7c65aa9b 100644 --- a/crates/kumo-api-types/src/shaping.rs +++ b/crates/kumo-api-types/src/shaping.rs @@ -1913,6 +1913,8 @@ MergedEntry { use_lmtp: false, reconnect_strategy: ConnectNextHost, readyq_pool_name: None, + low_memory_reduction_policy: ShrinkDataAndMeta, + no_memory_reduction_policy: ShrinkDataAndMeta, }, sources: {}, automation: [ @@ -2053,6 +2055,8 @@ MergedEntry { use_lmtp: false, reconnect_strategy: ConnectNextHost, readyq_pool_name: None, + low_memory_reduction_policy: ShrinkDataAndMeta, + no_memory_reduction_policy: ShrinkDataAndMeta, }, sources: { "my source name": EgressPathConfig { @@ -2106,6 +2110,8 @@ MergedEntry { use_lmtp: false, reconnect_strategy: ConnectNextHost, readyq_pool_name: None, + low_memory_reduction_policy: ShrinkDataAndMeta, + no_memory_reduction_policy: ShrinkDataAndMeta, }, }, automation: [ @@ -2252,6 +2258,8 @@ MergedEntry { use_lmtp: false, reconnect_strategy: ConnectNextHost, readyq_pool_name: None, + low_memory_reduction_policy: ShrinkDataAndMeta, + no_memory_reduction_policy: ShrinkDataAndMeta, }, sources: {}, automation: [ diff --git a/crates/kumo-server-memory/src/lib.rs b/crates/kumo-server-memory/src/lib.rs index a709ed4a..054cf09b 100644 --- a/crates/kumo-server-memory/src/lib.rs +++ b/crates/kumo-server-memory/src/lib.rs @@ -486,6 +486,24 @@ pub fn low_memory() -> bool { LOW_MEM.load(Ordering::SeqCst) } +/// Indicates the overall memory status +pub fn memory_status() -> MemoryStatus { + if get_headroom() == 0 { + MemoryStatus::NoMemory + } else if low_memory() { + MemoryStatus::LowMemory + } else { + MemoryStatus::Ok + } +} + +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub enum MemoryStatus { + Ok, + LowMemory, + NoMemory, +} + /// Returns a receiver that will notify when memory status /// changes from OK -> !OK or vice versa. pub fn subscribe_to_memory_status_changes() -> Option> { diff --git a/crates/kumod/src/ready_queue.rs b/crates/kumod/src/ready_queue.rs index 983066d2..5a6b0016 100644 --- a/crates/kumod/src/ready_queue.rs +++ b/crates/kumod/src/ready_queue.rs @@ -21,10 +21,12 @@ use async_trait::async_trait; use config::epoch::ConfigEpoch; use config::{load_config, CallbackSignature}; use dns_resolver::MailExchanger; -use kumo_api_types::egress_path::{ConfigRefreshStrategy, EgressPathConfig}; +use kumo_api_types::egress_path::{ConfigRefreshStrategy, EgressPathConfig, MemoryReductionPolicy}; use kumo_server_common::config_handle::ConfigHandle; use kumo_server_lifecycle::{is_shutting_down, Activity, ShutdownSubcription}; -use kumo_server_memory::{get_headroom, low_memory, subscribe_to_memory_status_changes_async}; +use kumo_server_memory::{ + get_headroom, memory_status, subscribe_to_memory_status_changes_async, MemoryStatus, +}; use kumo_server_runtime::{get_named_runtime, spawn, Runtime}; use message::message::{MessageList, QueueNameComponents}; use message::Message; @@ -614,8 +616,19 @@ impl ReadyQueue { pub async fn insert(&self, msg: Message) -> Result<(), Message> { let _timer = INSERT_LATENCY.start_timer(); - if low_memory() { - msg.save_and_shrink().await.ok(); + let action = match memory_status() { + MemoryStatus::LowMemory => self.path_config.borrow().low_memory_reduction_policy, + MemoryStatus::NoMemory => self.path_config.borrow().no_memory_reduction_policy, + MemoryStatus::Ok => MemoryReductionPolicy::NoShrink, + }; + match action { + MemoryReductionPolicy::NoShrink => {} + MemoryReductionPolicy::ShrinkDataAndMeta => { + msg.save_and_shrink().await.ok(); + } + MemoryReductionPolicy::ShrinkData => { + msg.save_and_shrink_data().await.ok(); + } } match self.ready.push(msg) { Ok(()) => { diff --git a/docs/changelog/main.md b/docs/changelog/main.md index de0fc53d..5004c030 100644 --- a/docs/changelog/main.md +++ b/docs/changelog/main.md @@ -22,6 +22,10 @@ for a list of the additional functions. * New [kumo.string.eval_template](../reference/string/eval_template.md) function for expanding minijinja template strings. +* New [low_memory_reduction_policy](../reference/kumo/make_egress_path/low_memory_reduction_policy.md) and + [no_memory_reduction_policy](../reference/kumo/make_egress_path/no_memory_reduction_policy.md) + options give advanced control over memory vs. spool IO trade-offs when + available is memory low. ## Fixes diff --git a/docs/reference/kumo/make_egress_path/low_memory_reduction_policy.md b/docs/reference/kumo/make_egress_path/low_memory_reduction_policy.md new file mode 100644 index 00000000..9252512e --- /dev/null +++ b/docs/reference/kumo/make_egress_path/low_memory_reduction_policy.md @@ -0,0 +1,36 @@ +--- +tags: + - memory +--- + +# low_memory_reduction_policy + +{{since('dev')}} + +Specifies what action should be taken when a message is added to the ready +queue corresponding to this egress path when the memory usage is above the *low +memory threshold*. + +Possible values for this option are: + +* `"ShrinkDataAndMeta"` - this is the default (and is the implicit action for + older versions of KumoMTA). Both the message data and metadata will be saved + (if modified since the prior save, or if the message has not yet been saved + to spool), then both will be released, freeing up the corresponding memory. +* `"ShrinkData"` - Both the message data and metadata will be saved + (if modified since the prior save, or if the message has not yet been saved + to spool), then just the message data will be released, freeing up that memory. + The metadata will be preserved. +* `"NoShrink"` - do not save or free up any message memory. + +This setting allows you more control in the trade-off of memory usage against +spool IO pressure. The default is relatively conservative, aiming to avoid OOM +killing at the cost of throughput (increased spool IO). Setting this option to +`"ShrinkData"` or `"NoShrink"` will allow the system to use more memory and +reduce pressure on the spool, but you increase the risk of memory usage +exceeding limits and being targeted by the OOM killer if there is a burst +in your workload. + +See also: +* [no_memory_reduction_policy](no_memory_reduction_policy.md) +* [Memory Management](../../memory.md) diff --git a/docs/reference/kumo/make_egress_path/no_memory_reduction_policy.md b/docs/reference/kumo/make_egress_path/no_memory_reduction_policy.md new file mode 100644 index 00000000..053e37ad --- /dev/null +++ b/docs/reference/kumo/make_egress_path/no_memory_reduction_policy.md @@ -0,0 +1,38 @@ +--- +tags: + - memory +--- + +# no_memory_reduction_policy + +{{since('dev')}} + +Specifies what action should be taken when a message is added to the ready +queue corresponding to this egress path when the memory usage is above the *soft +memory limit*. When the system is in this state, additional active measures +will also be applied to reduce overall memory consumption. + +Possible values for this option are: + +* `"ShrinkDataAndMeta"` - this is the default (and is the implicit action for + older versions of KumoMTA). Both the message data and metadata will be saved + (if modified since the prior save, or if the message has not yet been saved + to spool), then both will be released, freeing up the corresponding memory. +* `"ShrinkData"` - Both the message data and metadata will be saved + (if modified since the prior save, or if the message has not yet been saved + to spool), then just the message data will be released, freeing up that memory. + The metadata will be preserved. +* `"NoShrink"` - do not save or free up any message memory. + +This setting allows you more control in the trade-off of memory usage against +spool IO pressure. The default is relatively conservative, aiming to avoid OOM +killing at the cost of throughput (increased spool IO). Setting this option to +`"ShrinkData"` or `"NoShrink"` will allow the system to use more memory and +reduce pressure on the spool, but you increase the risk of memory usage +exceeding limits and being targeted by the OOM killer if there is a burst +in your workload. + +See also: +* [low_memory_reduction_policy](low_memory_reduction_policy.md) +* [Memory Management](../../memory.md) +