diff --git a/src/server.rs b/src/server.rs index 0ea4db10b..6180e7610 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1095,48 +1095,39 @@ where let me = Arc::new(self); + // The reader loop below feeds requests into `reqs_tx`; this task drains + // them, calls the service, and writes responses back to the sink. + // + // `_handle` must stay bound for the lifetime of `bind` (not `let _ =`, + // which would drop it immediately): it is an `AbortOnDropHandle`, so + // dropping it aborts this task and its in-flight compile subtasks. That + // is the cancellation we want once the client disconnects and the reader + // loop below exits. let _handle = util::spawn(async move { while let Some(req) = reqs_rx.recv().await { - match req { - Ok(req) => { - let res = match util::spawn(me.clone().call(req)).await? { - Ok(res) => res, - Err(err) => { - return Err(err); - } - }; - match res { - Message::WithoutBody(message) => { - sink.send(Frame::Message { message }).await?; - } - Message::WithBody(message, body) => { - sink.send(Frame::Message { message }).await?; - sink.send(Frame::Body { - chunk: Some(util::spawn(body).await??), - }) - .await?; - sink.send(Frame::Body { chunk: None }).await?; - } - } + match util::spawn(me.clone().call(req)).await?? { + Message::WithoutBody(message) => { + sink.send(Frame::Message { message }).await?; } - Err(err) => { - return Err(err); + Message::WithBody(message, body) => { + sink.send(Frame::Message { message }).await?; + sink.send(Frame::Body { + chunk: Some(util::spawn(body).await??), + }) + .await?; + sink.send(Frame::Body { chunk: None }).await?; } } } - Ok(()) + Ok::<_, Error>(()) }); + // Read requests until the client disconnects (stream ends) or the + // stream errors. A send error means the handler task above has ended, + // so we stop reading in that case too. while let Some(req) = stream.next().await { - match req { - Ok(req) => { - reqs_tx.send(Ok(req))?; - } - Err(err) => { - return Err(err); - } - } + reqs_tx.send(req?)?; } Ok(()) @@ -1469,218 +1460,210 @@ where let me = self.clone(); - // This redundant async block exists to reduce whitespace-only - // changes when comparing this diff with upstream/main. - // TODO: remove this before merging - #[allow(clippy::redundant_async_block)] util::spawn_on(&self.rt, async move { - async move { - let result = match me.dist_client.get_client().await { - Ok(client) => std::panic::AssertUnwindSafe(hasher.get_cached_or_compile( - &me, - client, - me.creator.clone(), - me.storage.clone(), - arguments, - cwd, - env_vars, - cache_control, - me.rt.clone(), - )) - .catch_unwind() - .await - .map_err(|e| { - let panic = e - .downcast_ref::<&str>() - .map(|s| &**s) - .or_else(|| e.downcast_ref::().map(|s| &**s)) - .unwrap_or("An unknown panic was caught."); - let thread = std::thread::current(); - let thread_name = thread.name().unwrap_or("unnamed"); - if let Some((file, line, column)) = PANIC_LOCATION.with(|l| l.take()) { - anyhow!( - "thread '{thread_name}' panicked at {file}:{line}:{column}: {panic}" - ) - } else { - anyhow!("thread '{thread_name}' panicked: {panic}") - } - }) - .and_then(std::convert::identity), - Err(e) => Err(e), - }; + let result = match me.dist_client.get_client().await { + Ok(client) => std::panic::AssertUnwindSafe(hasher.get_cached_or_compile( + &me, + client, + me.creator.clone(), + me.storage.clone(), + arguments, + cwd, + env_vars, + cache_control, + me.rt.clone(), + )) + .catch_unwind() + .await + .map_err(|e| { + let panic = e + .downcast_ref::<&str>() + .map(|s| &**s) + .or_else(|| e.downcast_ref::().map(|s| &**s)) + .unwrap_or("An unknown panic was caught."); + let thread = std::thread::current(); + let thread_name = thread.name().unwrap_or("unnamed"); + if let Some((file, line, column)) = PANIC_LOCATION.with(|l| l.take()) { + anyhow!( + "thread '{thread_name}' panicked at {file}:{line}:{column}: {panic}" + ) + } else { + anyhow!("thread '{thread_name}' panicked: {panic}") + } + }) + .and_then(std::convert::identity), + Err(e) => Err(e), + }; - let mut cache_write = None; - let mut res = CompileFinished { - color_mode, - ..Default::default() - }; + let mut cache_write = None; + let mut res = CompileFinished { + color_mode, + ..Default::default() + }; - let mut stats = me.stats.lock().await; + let mut stats = me.stats.lock().await; - match result { - Ok((compiled, out)) => { - let mut dist_type = DistType::NoDist; + match result { + Ok((compiled, out)) => { + let mut dist_type = DistType::NoDist; - match compiled { - CompileResult::Error => { - debug!("[{}]: compile result: cache error", out_pretty); + match compiled { + CompileResult::Error => { + debug!("[{}]: compile result: cache error", out_pretty); - stats.cache_errors.increment(&kind, &lang); - } - CompileResult::CacheHit(duration) => { - debug!("[{}]: compile result: cache hit", out_pretty); + stats.cache_errors.increment(&kind, &lang); + } + CompileResult::CacheHit(duration) => { + debug!("[{}]: compile result: cache hit", out_pretty); - stats.cache_hits.increment(&kind, &lang); - stats.cache_read_hit_duration += duration; - } - CompileResult::CacheMiss(miss_type, dt, duration, future) => { - debug!("[{}]: compile result: cache miss", out_pretty); - dist_type = dt; - - match miss_type { - MissType::Normal => {} - MissType::ForcedNoCache => {} - MissType::ForcedRecache => { - stats.forced_recaches += 1; - } - MissType::TimedOut => { - stats.cache_timeouts += 1; - } - MissType::CacheReadError => { - stats.cache_errors.increment(&kind, &lang); - } + stats.cache_hits.increment(&kind, &lang); + stats.cache_read_hit_duration += duration; + } + CompileResult::CacheMiss(miss_type, dt, duration, future) => { + debug!("[{}]: compile result: cache miss", out_pretty); + dist_type = dt; + + match miss_type { + MissType::Normal => {} + MissType::ForcedNoCache => {} + MissType::ForcedRecache => { + stats.forced_recaches += 1; + } + MissType::TimedOut => { + stats.cache_timeouts += 1; + } + MissType::CacheReadError => { + stats.cache_errors.increment(&kind, &lang); } - stats.compilations += 1; - stats.cache_misses.increment(&kind, &lang); - stats.compiler_write_duration += duration; - debug!("stats after compile result: {stats:?}"); - cache_write = Some(future); - } - CompileResult::NotCached(dt, duration) => { - debug!("[{}]: compile result: not cached", out_pretty); - dist_type = dt; - stats.compilations += 1; - stats.compiler_write_duration += duration; - } - CompileResult::NotCacheable(dt, duration) => { - debug!("[{}]: compile result: not cacheable", out_pretty); - dist_type = dt; - stats.compilations += 1; - stats.compiler_write_duration += duration; - stats.non_cacheable_compilations += 1; - } - CompileResult::CompileFailed(dt, duration) => { - debug!("[{}]: compile result: compile failed", out_pretty); - dist_type = dt; - stats.compilations += 1; - stats.compiler_write_duration += duration; - stats.compile_fails += 1; } + stats.compilations += 1; + stats.cache_misses.increment(&kind, &lang); + stats.compiler_write_duration += duration; + debug!("stats after compile result: {stats:?}"); + cache_write = Some(future); } - - match dist_type { - DistType::NoDist => {} - DistType::Ok(id) => { - let server = id.addr().to_string(); - let server_count = stats.dist_compiles.entry(server).or_insert(0); - *server_count += 1; - } - DistType::Error => stats.dist_errors += 1, + CompileResult::NotCached(dt, duration) => { + debug!("[{}]: compile result: not cached", out_pretty); + dist_type = dt; + stats.compilations += 1; + stats.compiler_write_duration += duration; + } + CompileResult::NotCacheable(dt, duration) => { + debug!("[{}]: compile result: not cacheable", out_pretty); + dist_type = dt; + stats.compilations += 1; + stats.compiler_write_duration += duration; + stats.non_cacheable_compilations += 1; } + CompileResult::CompileFailed(dt, duration) => { + debug!("[{}]: compile result: compile failed", out_pretty); + dist_type = dt; + stats.compilations += 1; + stats.compiler_write_duration += duration; + stats.compile_fails += 1; + } + } - // Make sure the write guard has been dropped ASAP. - drop(stats); + match dist_type { + DistType::NoDist => {} + DistType::Ok(id) => { + let server = id.addr().to_string(); + let server_count = stats.dist_compiles.entry(server).or_insert(0); + *server_count += 1; + } + DistType::Error => stats.dist_errors += 1, + } - let Output { - status, - stdout, - stderr, - } = out; + // Make sure the write guard has been dropped ASAP. + drop(stats); - trace!("CompileFinished retcode: {}", status); + let Output { + status, + stdout, + stderr, + } = out; - match status.code() { - Some(code) => res.retcode = Some(code), - None => res.signal = Some(get_signal(status)), - } + trace!("CompileFinished retcode: {}", status); - res.stdout = stdout; - res.stderr = stderr; + match status.code() { + Some(code) => res.retcode = Some(code), + None => res.signal = Some(get_signal(status)), } - Err(err) => { - match err.downcast::() { - Ok(ProcessError(output)) => { - debug!("Compilation failed: {:?}", output); - stats.compile_fails += 1; + + res.stdout = stdout; + res.stderr = stderr; + } + Err(err) => { + match err.downcast::() { + Ok(ProcessError(output)) => { + debug!("Compilation failed: {:?}", output); + stats.compile_fails += 1; + // Make sure the write guard has been dropped ASAP. + drop(stats); + + match output.status.code() { + Some(code) => res.retcode = Some(code), + None => res.signal = Some(get_signal(output.status)), + } + res.stdout = output.stdout; + res.stderr = output.stderr; + } + Err(err) => match err.downcast::() { + Ok(HttpClientError(msg)) => { + // Make sure the write guard has been dropped ASAP. + drop(stats); + me.dist_client.reset_state().await; + let errmsg = + format!("[{:?}] http error status: {}", out_pretty, msg); + error!("{}", errmsg); + res.retcode = Some(1); + res.stderr = errmsg.as_bytes().to_vec(); + } + Err(err) => { + stats.cache_errors.increment(&kind, &lang); // Make sure the write guard has been dropped ASAP. drop(stats); - match output.status.code() { - Some(code) => res.retcode = Some(code), - None => res.signal = Some(get_signal(output.status)), + use std::fmt::Write; + + error!("[{:?}] fatal error: {}", out_pretty, err); + + let mut error = "sccache: encountered fatal error\n".to_string(); + let _ = writeln!(error, "sccache: error: {}", err); + for e in err.chain() { + error!("[{:?}] \t{}", out_pretty, e); + let _ = writeln!(error, "sccache: caused by: {}", e); } - res.stdout = output.stdout; - res.stderr = output.stderr; + //TODO: figure out a better way to communicate this? + res.retcode = Some(-2); + res.stderr = error.into_bytes(); } - Err(err) => match err.downcast::() { - Ok(HttpClientError(msg)) => { - // Make sure the write guard has been dropped ASAP. - drop(stats); - me.dist_client.reset_state().await; - let errmsg = - format!("[{:?}] http error status: {}", out_pretty, msg); - error!("{}", errmsg); - res.retcode = Some(1); - res.stderr = errmsg.as_bytes().to_vec(); - } - Err(err) => { - stats.cache_errors.increment(&kind, &lang); - // Make sure the write guard has been dropped ASAP. - drop(stats); - - use std::fmt::Write; - - error!("[{:?}] fatal error: {}", out_pretty, err); - - let mut error = - "sccache: encountered fatal error\n".to_string(); - let _ = writeln!(error, "sccache: error: {}", err); - for e in err.chain() { - error!("[{:?}] \t{}", out_pretty, e); - let _ = writeln!(error, "sccache: caused by: {}", e); - } - //TODO: figure out a better way to communicate this? - res.retcode = Some(-2); - res.stderr = error.into_bytes(); - } - }, - } + }, } } + } - if let Some(cache_write) = cache_write { - match cache_write.await { - Err(e) => { - debug!("Error executing cache write: {}", e); - me.stats.lock().await.cache_write_errors += 1; - } - //TODO: save cache stats! - Ok(info) => { - debug!( - "[{}]: Cache write finished in {}", - info.object_file_pretty, - util::fmt_duration_as_secs(&info.duration) - ); - let mut stats = me.stats.lock().await; - stats.cache_writes += 1; - stats.cache_write_duration += info.duration; - } + if let Some(cache_write) = cache_write { + match cache_write.await { + Err(e) => { + debug!("Error executing cache write: {}", e); + me.stats.lock().await.cache_write_errors += 1; + } + //TODO: save cache stats! + Ok(info) => { + debug!( + "[{}]: Cache write finished in {}", + info.object_file_pretty, + util::fmt_duration_as_secs(&info.duration) + ); + let mut stats = me.stats.lock().await; + stats.cache_writes += 1; + stats.cache_write_duration += info.duration; } } - - Ok(res) } - .await + + Ok(res) }) .map_err(anyhow::Error::new) .await?