Use Condvar and make configuration API blocking

compute_tools/src/bin/compute_ctl.rs

@@ -34,13 +34,12 @@ use std::fs::File;
 use std::panic;
 use std::path::Path;
 use std::process::exit;
-use std::sync::{Arc, RwLock};
+use std::sync::{Arc, Condvar, Mutex};
 use std::{thread, time::Duration};
 
 use anyhow::{Context, Result};
 use chrono::Utc;
 use clap::Arg;
-use tokio::sync::mpsc;
 use tracing::{error, info};
 use url::Url;
 
@@ -108,46 +107,75 @@ fn main() -> Result<()> {
         }
     };
 
-    // We have one configurator thread and async http server, so generally we
-    // have single consumer - multiple producers pattern here. That's why we use
-    // `mpsc` channel, not `tokio::sync::watch`. Actually, concurrency of
-    // producers is limited to one due to code logic, but we still need to
-    // pass `Sender` to several threads.
-    //
-    // Next, we use async `hyper` + `tokio` http server, but all the other code
-    // is completely synchronous. So we need to send data from async to sync,
-    // that's why we use `mpsc::unbounded_channel` here, not `mpsc::channel`.
-    // It doesn't make much sense to rewrite all code to async now, but we can
-    // consider doing this in the future. Unbound should not sound scary here,
-    // as again, only one unprocessed spec could be sent to channel at any
-    // given moment.
-    let (spec_tx, mut spec_rx) = mpsc::unbounded_channel::<ComputeSpec>();
-
-    let compute_state = ComputeNode {
+    let mut new_state = ComputeState::new();
+    if spec_set {
+        new_state.spec = spec;
+    }
+    // Volatile compute state under mutex and condition variable to notify everyone
+    // who is interested in the state changes.
+    let compute_state = (Mutex::new(new_state), Condvar::new());
+    let compute_node = ComputeNode {
         start_time: Utc::now(),
         connstr: Url::parse(connstr).context("cannot parse connstr as a URL")?,
         pgdata: pgdata.to_string(),
         pgbin: pgbin.to_string(),
         live_config_allowed,
         metrics: ComputeMetrics::default(),
-        state: RwLock::new(ComputeState::new()),
+        state: compute_state,
     };
-    let compute = Arc::new(compute_state);
+    let compute = Arc::new(compute_node);
 
     // Launch http service first, so we were able to serve control-plane
     // requests, while configuration is still in progress.
-    let _http_handle =
-        launch_http_server(&compute, spec_tx).expect("cannot launch http endpoint thread");
+    let _http_handle = launch_http_server(&compute).expect("cannot launch http endpoint thread");
 
     if !spec_set {
+        // No spec provided, hang waiting for it.
         info!("no compute spec provided, waiting");
-        if let Some(s) = spec_rx.blocking_recv() {
-            spec = s;
-        } else {
-            panic!("no compute spec received");
+        let (state, state_changed) = &compute.state;
+        let mut state = state.lock().unwrap();
+        while state.status != ComputeStatus::ConfigurationPending {
+            state = state_changed.wait(state).unwrap();
+
+            if state.status == ComputeStatus::ConfigurationPending {
+                info!("got spec, continue configuration");
+                // Spec is already set by the http server handler.
+                break;
+            }
         }
     }
 
+    // We got all we need, fill in the state.
+    let (state, _) = &compute.state;
+    let mut state = state.lock().unwrap();
+    let pageserver_connstr = state
+        .spec
+        .cluster
+        .settings
+        .find("neon.pageserver_connstring")
+        .expect("pageserver connstr should be provided");
+    let storage_auth_token = state.spec.storage_auth_token.clone();
+    let tenant = state
+        .spec
+        .cluster
+        .settings
+        .find("neon.tenant_id")
+        .expect("tenant id should be provided");
+    let timeline = state
+        .spec
+        .cluster
+        .settings
+        .find("neon.timeline_id")
+        .expect("tenant id should be provided");
+    let startup_tracing_context = state.spec.startup_tracing_context.clone();
+
+    state.pageserver_connstr = pageserver_connstr;
+    state.storage_auth_token = storage_auth_token;
+    state.tenant = tenant;
+    state.timeline = timeline;
+    state.status = ComputeStatus::Init;
+    drop(state);
+
     // Extract OpenTelemetry context for the startup actions from the spec, and
     // attach it to the current tracing context.
     //
@@ -163,7 +191,7 @@ fn main() -> Result<()> {
     // postgres is configured and up-and-running, we exit this span. Any other
     // actions that are performed on incoming HTTP requests, for example, are
     // performed in separate spans.
-    let startup_context_guard = if let Some(ref carrier) = spec.startup_tracing_context {
+    let startup_context_guard = if let Some(ref carrier) = startup_tracing_context {
         use opentelemetry::propagation::TextMapPropagator;
         use opentelemetry::sdk::propagation::TraceContextPropagator;
         Some(TraceContextPropagator::new().extract(carrier).attach())
@@ -171,37 +199,10 @@ fn main() -> Result<()> {
         None
     };
 
-    let pageserver_connstr = spec
-        .cluster
-        .settings
-        .find("neon.pageserver_connstring")
-        .expect("pageserver connstr should be provided");
-    let storage_auth_token = spec.storage_auth_token.clone();
-    let tenant = spec
-        .cluster
-        .settings
-        .find("neon.tenant_id")
-        .expect("tenant id should be provided");
-    let timeline = spec
-        .cluster
-        .settings
-        .find("neon.timeline_id")
-        .expect("tenant id should be provided");
-
-    // We got all we need, fill in the state.
-    let mut state = compute.state.write().unwrap();
-    state.spec = spec;
-    state.pageserver_connstr = pageserver_connstr;
-    state.storage_auth_token = storage_auth_token;
-    state.tenant = tenant;
-    state.timeline = timeline;
-    state.status = ComputeStatus::Init;
-    drop(state);
-
     // Launch remaining service threads
     let _monitor_handle = launch_monitor(&compute).expect("cannot launch compute monitor thread");
     let _configurator_handle =
-        launch_configurator(&compute, spec_rx).expect("cannot launch configurator thread");
+        launch_configurator(&compute).expect("cannot launch configurator thread");
 
     // Start Postgres
     let mut delay_exit = false;
@@ -210,7 +211,8 @@ fn main() -> Result<()> {
         Ok(pg) => Some(pg),
         Err(err) => {
             error!("could not start the compute node: {:?}", err);
-            let mut state = compute.state.write().unwrap();
+            let (state, _) = &compute.state;
+            let mut state = state.lock().unwrap();
             state.error = Some(format!("{:?}", err));
             state.status = ComputeStatus::Failed;
             drop(state);

compute_tools/src/compute.rs

@@ -20,7 +20,8 @@ use std::path::Path;
 use std::process::{Command, Stdio};
 use std::str::FromStr;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::RwLock;
+// use std::sync::RwLock;
+use std::sync::{Condvar, Mutex};
 
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
@@ -50,10 +51,13 @@ pub struct ComputeNode {
     // - but then something happens and compute pod / VM is destroyed,
     //   so k8s controller starts it again with the **old** spec
     pub live_config_allowed: bool,
-    /// Volatile part of the `ComputeNode` so should be used under `RwLock`
-    /// to allow HTTP API server to serve status requests, while configuration
-    /// is in progress.
-    pub state: RwLock<ComputeState>,
+    /// Volatile part of the `ComputeNode`, which should be used under `Mutex`.
+    /// Coupled with `Condvar` to allow notifying HTTP API and configurator
+    /// thread about state changes. To allow HTTP API server to serving status
+    /// requests, while configuration is in progress, lock should be held only
+    /// for short periods of time to do read/write, not the whole configuration
+    /// process.
+    pub state: (Mutex<ComputeState>, Condvar),
 }
 
 fn rfc3339_serialize<S>(x: &DateTime<Utc>, s: S) -> Result<S::Ok, S::Error>
@@ -132,11 +136,16 @@ pub struct ComputeMetrics {
 
 impl ComputeNode {
     pub fn set_status(&self, status: ComputeStatus) {
-        self.state.write().unwrap().status = status;
+        let (state, state_changed) = &self.state;
+        let mut state = state.lock().unwrap();
+        state.status = status;
+        state_changed.notify_all();
     }
 
     pub fn get_status(&self) -> ComputeStatus {
-        self.state.read().unwrap().status
+        // self.state.read().unwrap().status
+        let (state, _) = &self.state;
+        state.lock().unwrap().status
     }
 
     // Remove `pgdata` directory and create it again with right permissions.
@@ -370,8 +379,11 @@ impl ComputeNode {
 
     /// Similar to `apply_config()`, but does a bit different sequence of operations,
     /// as it's used to reconfigure a previously started and configured Postgres node.
-    #[instrument(skip(self, spec))]
-    pub fn reconfigure(&self, spec: ComputeSpec) -> Result<()> {
+    #[instrument(skip(self))]
+    pub fn reconfigure(&self) -> Result<()> {
+        let (state, _) = &self.state;
+        let spec = state.lock().unwrap().spec.clone();
+
         // Write new config
         let pgdata_path = Path::new(&self.pgdata);
         config::write_postgres_conf(&pgdata_path.join("postgresql.conf"), &spec)?;
@@ -401,7 +413,9 @@ impl ComputeNode {
 
     #[instrument(skip(self))]
     pub fn start_compute(&self) -> Result<std::process::Child> {
-        let compute_state = self.state.read().unwrap().clone();
+        // let compute_state = self.state.read().unwrap().clone();
+        let (state, _) = &self.state;
+        let compute_state = state.lock().unwrap().clone();
         info!(
             "starting compute for project {}, operation {}, tenant {}, timeline {}",
             compute_state.spec.cluster.cluster_id,

compute_tools/src/configurator.rs

@@ -2,54 +2,53 @@ use std::sync::Arc;
 use std::thread;
 
 use anyhow::Result;
-use tokio::sync::mpsc::UnboundedReceiver;
 use tracing::{error, info, instrument};
 
 use crate::compute::{ComputeNode, ComputeStatus};
-use crate::spec::ComputeSpec;
 
-#[instrument(skip(compute, rx))]
-fn configurator_main_loop(compute: &Arc<ComputeNode>, mut rx: UnboundedReceiver<ComputeSpec>) {
+#[instrument(skip(compute))]
+fn configurator_main_loop(compute: &Arc<ComputeNode>) {
     info!("waiting for reconfiguration requests");
-    while let Some(spec) = rx.blocking_recv() {
-        info!("got spec = {:?}", &spec);
+    let (state, state_changed) = &compute.state;
+    loop {
+        let state = state.lock().unwrap();
+        let mut state = state_changed.wait(state).unwrap();
 
-        let status = compute.get_status();
-        // Sanity check, should never happen.
-        if status != ComputeStatus::ConfigurationPending {
-            error!(
-                "unexpected compute status: {:?}, expected {:?}",
-                status,
-                ComputeStatus::ConfigurationPending
-            );
-            compute.set_status(ComputeStatus::Failed);
-            continue;
+        if state.status == ComputeStatus::ConfigurationPending {
+            info!("got reconfiguration request");
+            state.status = ComputeStatus::Reconfiguration;
+            state_changed.notify_all();
+            drop(state);
+
+            let mut new_status = ComputeStatus::Failed;
+            if let Err(e) = compute.reconfigure() {
+                error!("could not reconfigure compute node: {}", e);
+            } else {
+                new_status = ComputeStatus::Running;
+                info!("compute node reconfigured");
+            }
+
+            // XXX: used to test that API is blocking
+            // TODO: remove before merge
+            std::thread::sleep(std::time::Duration::from_millis(2000));
+
+            compute.set_status(new_status);
+        } else if state.status == ComputeStatus::Failed {
+            info!("compute node is now in Failed state, exiting");
+            break;
         } else {
-            compute.set_status(ComputeStatus::Reconfiguration);
+            info!("woken up for compute status: {:?}, sleeping", state.status);
         }
-
-        let mut new_status = ComputeStatus::Failed;
-        if let Err(e) = compute.reconfigure(spec) {
-            error!("could not reconfigure compute node: {}", e);
-        } else {
-            new_status = ComputeStatus::Running;
-            info!("compute node reconfigured");
-        }
-
-        compute.set_status(new_status);
     }
-    info!("configurator thread is exiting");
 }
 
-pub fn launch_configurator(
-    compute: &Arc<ComputeNode>,
-    rx: UnboundedReceiver<ComputeSpec>,
-) -> Result<thread::JoinHandle<()>> {
+pub fn launch_configurator(compute: &Arc<ComputeNode>) -> Result<thread::JoinHandle<()>> {
     let compute = Arc::clone(compute);
 
     Ok(thread::Builder::new()
         .name("compute-configurator".into())
         .spawn(move || {
-            configurator_main_loop(&compute, rx);
+            configurator_main_loop(&compute);
+            info!("configurator thread is exited");
         })?)
 }

compute_tools/src/http/api.rs

@@ -11,16 +11,11 @@ use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Method, Request, Response, Server, StatusCode};
 use num_cpus;
 use serde_json;
-use tokio::sync::mpsc::UnboundedSender;
 use tracing::{error, info};
 use tracing_utils::http::OtelName;
 
 // Service function to handle all available routes.
-async fn routes(
-    req: Request<Body>,
-    compute: &Arc<ComputeNode>,
-    tx: &UnboundedSender<ComputeSpec>,
-) -> Response<Body> {
+async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body> {
     //
     // NOTE: The URI path is currently included in traces. That's OK because
     // it doesn't contain any variable parts or sensitive information. But
@@ -30,7 +25,9 @@ async fn routes(
         // Serialized compute state.
         (&Method::GET, "/status") => {
             info!("serving /status GET request");
-            let state = compute.state.read().unwrap();
+            // let state = compute.state.read().unwrap();
+            let (state, _) = &compute.state;
+            let state = state.lock().unwrap();
             Response::new(Body::from(serde_json::to_string(&*state).unwrap()))
         }
 
@@ -105,7 +102,8 @@ async fn routes(
             let body_bytes = hyper::body::to_bytes(req.into_body()).await.unwrap();
             let spec_raw = String::from_utf8(body_bytes.to_vec()).unwrap();
             if let Ok(spec) = serde_json::from_str::<ComputeSpec>(&spec_raw) {
-                let mut state = compute.state.write().unwrap();
+                let (state, state_changed) = &compute.state;
+                let mut state = state.lock().unwrap();
                 if !(state.status == ComputeStatus::WaitingSpec
                     || state.status == ComputeStatus::Running)
                 {
@@ -116,19 +114,22 @@ async fn routes(
                     error!(msg);
                     return Response::new(Body::from(msg));
                 }
+                state.spec = spec;
                 state.status = ComputeStatus::ConfigurationPending;
+                state_changed.notify_all();
                 drop(state);
+                info!("set new spec and notified configurator");
 
-                if let Err(e) = tx.send(spec) {
-                    error!("failed to send spec request to configurator thread: {}", e);
-                    Response::new(Body::from(format!(
-                        "could not request reconfiguration: {}",
-                        e
-                    )))
-                } else {
-                    info!("sent spec request to configurator");
-                    Response::new(Body::from("ok"))
+                let (state, state_changed) = &compute.state;
+                let mut state = state.lock().unwrap();
+                while state.status != ComputeStatus::Running {
+                    state = state_changed.wait(state).unwrap();
+                    info!(
+                        "waiting for compute to become Running, current status: {:?}",
+                        state.status
+                    );
                 }
+                Response::new(Body::from("ok"))
             } else {
                 let msg = "invalid spec";
                 error!(msg);
@@ -147,16 +148,14 @@ async fn routes(
 
 // Main Hyper HTTP server function that runs it and blocks waiting on it forever.
 #[tokio::main]
-async fn serve(state: Arc<ComputeNode>, tx: UnboundedSender<ComputeSpec>) {
+async fn serve(state: Arc<ComputeNode>) {
     let addr = SocketAddr::from(([0, 0, 0, 0], 3080));
 
     let make_service = make_service_fn(move |_conn| {
         let state = state.clone();
-        let tx = tx.clone();
         async move {
             Ok::<_, Infallible>(service_fn(move |req: Request<Body>| {
                 let state = state.clone();
-                let tx = tx.clone();
                 async move {
                     Ok::<_, Infallible>(
                         // NOTE: We include the URI path in the string. It
@@ -164,7 +163,7 @@ async fn serve(state: Arc<ComputeNode>, tx: UnboundedSender<ComputeSpec>) {
                         // information in this API.
                         tracing_utils::http::tracing_handler(
                             req,
-                            |req| routes(req, &state, &tx),
+                            |req| routes(req, &state),
                             OtelName::UriPath,
                         )
                         .await,
@@ -185,12 +184,9 @@ async fn serve(state: Arc<ComputeNode>, tx: UnboundedSender<ComputeSpec>) {
 }
 
 /// Launch a separate Hyper HTTP API server thread and return its `JoinHandle`.
-pub fn launch_http_server(
-    state: &Arc<ComputeNode>,
-    tx: UnboundedSender<ComputeSpec>,
-) -> Result<thread::JoinHandle<()>> {
+pub fn launch_http_server(state: &Arc<ComputeNode>) -> Result<thread::JoinHandle<()>> {
     let state = Arc::clone(state);
     Ok(thread::Builder::new()
         .name("http-endpoint".into())
-        .spawn(move || serve(state, tx))?)
+        .spawn(move || serve(state))?)
 }

compute_tools/src/monitor.rs

@@ -46,7 +46,9 @@ fn watch_compute_activity(compute: &ComputeNode) {
                             AND usename != 'cloud_admin';", // XXX: find a better way to filter other monitors?
                         &[],
                     );
-                let mut last_active = compute.state.read().unwrap().last_active;
+                // let mut last_active = compute.state.read().unwrap().last_active;
+                let (state, _) = &compute.state;
+                let mut last_active = state.lock().unwrap().last_active;
 
                 if let Ok(backs) = backends {
                     let mut idle_backs: Vec<DateTime<Utc>> = vec![];
@@ -87,7 +89,8 @@ fn watch_compute_activity(compute: &ComputeNode) {
                 }
 
                 // Update the last activity in the shared state if we got a more recent one.
-                let mut state = compute.state.write().unwrap();
+                let (state, _) = &compute.state;
+                let mut state = state.lock().unwrap();
                 if last_active > state.last_active {
                     state.last_active = last_active;
                     debug!("set the last compute activity time to: {}", last_active);