From 69552822b4118ebbde774e1c2c824720f1b19060 Mon Sep 17 00:00:00 2001 From: Alexey Kondratov Date: Mon, 3 Apr 2023 20:37:54 +0200 Subject: [PATCH] Use Condvar and make configuration API blocking --- compute_tools/src/bin/compute_ctl.rs | 114 ++++++++++++++------------- compute_tools/src/compute.rs | 34 +++++--- compute_tools/src/configurator.rs | 65 ++++++++------- compute_tools/src/http/api.rs | 48 ++++++----- compute_tools/src/monitor.rs | 7 +- 5 files changed, 141 insertions(+), 127 deletions(-) diff --git a/compute_tools/src/bin/compute_ctl.rs b/compute_tools/src/bin/compute_ctl.rs index bb73d8e086..15a1679c56 100644 --- a/compute_tools/src/bin/compute_ctl.rs +++ b/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::(); - - 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); diff --git a/compute_tools/src/compute.rs b/compute_tools/src/compute.rs index 4dd162821c..1402b95cc2 100644 --- a/compute_tools/src/compute.rs +++ b/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, + /// 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, Condvar), } fn rfc3339_serialize(x: &DateTime, s: S) -> Result @@ -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 { - 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, diff --git a/compute_tools/src/configurator.rs b/compute_tools/src/configurator.rs index 1029bd725c..229f673914 100644 --- a/compute_tools/src/configurator.rs +++ b/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, mut rx: UnboundedReceiver) { +#[instrument(skip(compute))] +fn configurator_main_loop(compute: &Arc) { 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, - rx: UnboundedReceiver, -) -> Result> { +pub fn launch_configurator(compute: &Arc) -> Result> { 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"); })?) } diff --git a/compute_tools/src/http/api.rs b/compute_tools/src/http/api.rs index ca6e18ffde..06b0098b2a 100644 --- a/compute_tools/src/http/api.rs +++ b/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, - compute: &Arc, - tx: &UnboundedSender, -) -> Response { +async fn routes(req: Request, compute: &Arc) -> Response { // // 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::(&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, tx: UnboundedSender) { +async fn serve(state: Arc) { 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| { 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, tx: UnboundedSender) { // 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, tx: UnboundedSender) { } /// Launch a separate Hyper HTTP API server thread and return its `JoinHandle`. -pub fn launch_http_server( - state: &Arc, - tx: UnboundedSender, -) -> Result> { +pub fn launch_http_server(state: &Arc) -> Result> { let state = Arc::clone(state); Ok(thread::Builder::new() .name("http-endpoint".into()) - .spawn(move || serve(state, tx))?) + .spawn(move || serve(state))?) } diff --git a/compute_tools/src/monitor.rs b/compute_tools/src/monitor.rs index 7c9878ffcf..8b11f9aca2 100644 --- a/compute_tools/src/monitor.rs +++ b/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> = 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);