Pivot to specifying all of MALLOC_CONF in flag

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

misc/python/materialize/mzcompose/__init__.py

@@ -77,6 +77,7 @@ def get_minimal_system_parameters(
         # -----
         # Others (ordered by name)
         "allow_real_time_recency": "true",
+        "clusterd_malloc_conf": "thp:always",
         "constraint_based_timestamp_selection": "verify",
         "enable_compute_peek_response_stash": "true",
         "enable_0dt_deployment": "true" if zero_downtime else "false",
@@ -113,7 +114,6 @@ def get_minimal_system_parameters(
         "enable_sql_server_source": "true",
         "enable_statement_lifecycle_logging": "true",
         "enable_compute_temporal_bucketing": "true",
-        "enabel_transparent_hugepages": "true",
         "enable_variadic_left_join_lowering": "true",
         "enable_worker_core_affinity": "true",
         "grpc_client_http2_keep_alive_timeout": "5s",
@@ -565,7 +565,7 @@ def get_default_system_parameters(
     "enable_ctp_cluster_protocols",
     "enable_paused_cluster_readhold_downgrade",
     "force_swap_for_cc_sizes",
-    "enable_transparent_hugepages",
+    "clusterd_malloc_conf",
 ]
 
 

misc/python/materialize/parallel_workload/action.py

@@ -1322,7 +1322,7 @@ def __init__(
             "enable_mz_join_core_v2",
             "force_swap_for_cc_sizes",
             "enable_with_ordinality_legacy_fallback",
-            "enable_transparent_hugepages",
+            "clusterd_malloc_conf",
         ]
 
     def run(self, exe: Executor) -> bool:

src/controller-types/src/dyncfgs.rs

@@ -80,10 +80,10 @@ pub const ENABLE_PAUSED_CLUSTER_READHOLD_DOWNGRADE: Config<bool> = Config::new(
     "Aggressively downgrade input read holds for indexes on zero-replica clusters.",
 );
 
-pub const ENABLE_TRANSPARENT_HUGEPAGES: Config<bool> = Config::new(
-    "enable_transparent_hugepages",
-    false,
-    "Enable transparent hugepages in clusterd with jemalloc.",
+pub const CLUSTERD_MALLOC_CONF: Config<&str> = Config::new(
+    "clusterd_malloc_conf",
+    "",
+    "MALLOC_CONF to pass to clusterd. If empty, no MALLOC_CONF is set.",
 );
 
 /// Adds the full set of all controller `Config`s.
@@ -96,7 +96,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&WALLCLOCK_LAG_HISTOGRAM_PERIOD_INTERVAL)
         .add(&ENABLE_TIMELY_ZERO_COPY)
         .add(&ENABLE_TIMELY_ZERO_COPY_LGALLOC)
-        .add(&ENABLE_TRANSPARENT_HUGEPAGES)
+        .add(&CLUSTERD_MALLOC_CONF)
         .add(&TIMELY_ZERO_COPY_LIMIT)
         .add(&ARRANGEMENT_EXERT_PROPORTIONALITY)
         .add(&ENABLE_CTP_CLUSTER_PROTOCOLS)

src/controller/src/clusters.rs

@@ -26,9 +26,9 @@ use mz_compute_client::logging::LogVariant;
 use mz_compute_client::service::{ComputeClient, ComputeGrpcClient};
 use mz_compute_types::config::{ComputeReplicaConfig, ComputeReplicaLogging};
 use mz_controller_types::dyncfgs::{
-    ARRANGEMENT_EXERT_PROPORTIONALITY, CONTROLLER_PAST_GENERATION_REPLICA_CLEANUP_RETRY_INTERVAL,
-    ENABLE_CTP_CLUSTER_PROTOCOLS, ENABLE_TIMELY_ZERO_COPY, ENABLE_TIMELY_ZERO_COPY_LGALLOC,
-    ENABLE_TRANSPARENT_HUGEPAGES, TIMELY_ZERO_COPY_LIMIT,
+    ARRANGEMENT_EXERT_PROPORTIONALITY, CLUSTERD_MALLOC_CONF,
+    CONTROLLER_PAST_GENERATION_REPLICA_CLEANUP_RETRY_INTERVAL, ENABLE_CTP_CLUSTER_PROTOCOLS,
+    ENABLE_TIMELY_ZERO_COPY, ENABLE_TIMELY_ZERO_COPY_LGALLOC, TIMELY_ZERO_COPY_LIMIT,
 };
 use mz_controller_types::{ClusterId, ReplicaId};
 use mz_orchestrator::NamespacedOrchestrator;
@@ -665,6 +665,10 @@ where
             });
         }
 
+        let clusterd_malloc_conf = CLUSTERD_MALLOC_CONF.get(&self.dyncfg);
+        let clusterd_malloc_conf =
+            (!clusterd_malloc_conf.is_empty()).then_some(clusterd_malloc_conf);
+
         let service = self.orchestrator.ensure_service(
             &service_name,
             ServiceConfig {
@@ -819,7 +823,7 @@ where
                 }],
                 disk_limit,
                 node_selector: location.allocation.selectors,
-                enable_transparent_hugepages: ENABLE_TRANSPARENT_HUGEPAGES.get(&self.dyncfg),
+                clusterd_malloc_conf,
             },
         )?;
 

src/orchestrator-kubernetes/src/lib.rs

@@ -568,7 +568,7 @@ impl NamespacedOrchestrator for NamespacedKubernetesOrchestrator {
             replicas_selector,
             disk_limit,
             node_selector,
-            enable_transparent_hugepages,
+            clusterd_malloc_conf,
         }: ServiceConfig,
     ) -> Result<Box<dyn Service>, anyhow::Error> {
         // This is extremely cheap to clone, so just look into the lock once.
@@ -972,30 +972,27 @@ impl NamespacedOrchestrator for NamespacedKubernetesOrchestrator {
             }]
         });
 
-        let mut env = if self.config.coverage {
-            Some(vec![EnvVar {
+        let mut env = Vec::new();
+
+        if self.config.coverage {
+            env.push(EnvVar {
                 name: "LLVM_PROFILE_FILE".to_string(),
                 value: Some(format!("/coverage/{}-%p-%9m%c.profraw", self.namespace)),
                 ..Default::default()
-            }])
-        } else {
-            None
-        };
+            });
+        }
 
-        // Configure jemalloc to use transparent hugepages.
-        if enable_transparent_hugepages {
-            let var = EnvVar {
+        // Configure jemalloc's MALLOC_CONF environment symbol.
+        if let Some(malloc_conf) = clusterd_malloc_conf {
+            env.push(EnvVar {
                 name: "MALLOC_CONF".to_string(),
-                value: Some("thp:always".to_string()),
+                value: Some(malloc_conf),
                 ..Default::default()
-            };
-            if let Some(env) = &mut env {
-                env.push(var);
-            } else {
-                env = Some(vec![var]);
-            }
+            });
         }
 
+        let env = (!env.is_empty()).then_some(env);
+
         let mut volume_mounts = vec![];
 
         if self.config.coverage {

src/orchestrator-process/src/lib.rs

@@ -345,7 +345,7 @@ impl NamespacedOrchestrator for NamespacedProcessOrchestrator {
             cpu_limit: config.cpu_limit,
             scale: config.scale,
             labels: config.labels,
-            enable_transparent_hugepages: config.enable_transparent_hugepages,
+            clusterd_malloc_conf: config.clusterd_malloc_conf,
             disk,
         };
 
@@ -462,8 +462,8 @@ struct EnsureServiceConfig {
     pub labels: BTreeMap<String, String>,
     /// Whether scratch disk space should be allocated for the service.
     pub disk: bool,
-    /// Whether to enable transparent hugepages.
-    pub enable_transparent_hugepages: bool,
+    /// Optional jemalloc configuration.
+    pub clusterd_malloc_conf: Option<String>,
 }
 
 /// A task executing blocking work for a [`NamespacedProcessOrchestrator`] in the background.
@@ -575,7 +575,7 @@ impl OrchestratorWorker {
             scale,
             labels,
             disk,
-            enable_transparent_hugepages,
+            clusterd_malloc_conf,
         }: EnsureServiceConfig,
     ) -> Result<(), anyhow::Error> {
         let full_id = self.config.full_id(&id);
@@ -690,7 +690,7 @@ impl OrchestratorWorker {
                         memory_limit,
                         cpu_limit,
                         launch_spec: self.config.launch_spec,
-                        enable_transparent_hugepages,
+                        clusterd_malloc_conf: clusterd_malloc_conf.clone(),
                     }),
                 );
 
@@ -796,7 +796,7 @@ impl OrchestratorWorker {
             memory_limit,
             cpu_limit,
             launch_spec,
-            enable_transparent_hugepages,
+            clusterd_malloc_conf,
         }: ServiceProcessConfig,
     ) -> impl Future<Output = ()> + use<> {
         let suppress_output = self.config.suppress_output;
@@ -844,8 +844,8 @@ impl OrchestratorWorker {
                     memory_limit.as_ref(),
                     cpu_limit.as_ref(),
                 );
-                if enable_transparent_hugepages {
-                    cmd.env("MALLOC_CONF", "thp:always");
+                if let Some(malloc_conf) = &clusterd_malloc_conf {
+                    cmd.env("MALLOC_CONF", malloc_conf);
                 }
                 info!(
                     "launching {full_id}-{i} via {} {}...",
@@ -942,7 +942,7 @@ struct ServiceProcessConfig {
     memory_limit: Option<MemoryLimit>,
     cpu_limit: Option<CpuLimit>,
     launch_spec: LaunchSpec,
-    enable_transparent_hugepages: bool,
+    clusterd_malloc_conf: Option<String>,
 }
 
 struct ServiceProcessPort {

src/orchestrator/src/lib.rs

@@ -243,8 +243,8 @@ pub struct ServiceConfig {
     pub disk_limit: Option<DiskLimit>,
     /// Node selector for this service.
     pub node_selector: BTreeMap<String, String>,
-    /// Whether to enable transparent hugepages.
-    pub enable_transparent_hugepages: bool,
+    /// Optional jemalloc configuration.
+    pub clusterd_malloc_conf: Option<String>,
 }
 
 /// A named port associated with a service.