extern/nushell
1
0
Fork 1
mirror of https://github.com/nushell/nushell.git synced 2025-08-09 12:15:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add and use new Signals struct (#13314)

Browse Source 
# Description
This PR introduces a new `Signals` struct to replace our adhoc passing
around of `ctrlc: Option<Arc<AtomicBool>>`. Doing so has a few benefits:
- We can better enforce when/where resetting or triggering an interrupt
is allowed.
- Consolidates `nu_utils::ctrl_c::was_pressed` and other ad-hoc
re-implementations into a single place: `Signals::check`.
- This allows us to add other types of signals later if we want. E.g.,
exiting or suspension.
- Similarly, we can more easily change the underlying implementation if
we need to in the future.
- Places that used to have a `ctrlc` of `None` now use
`Signals::empty()`, so we can double check these usages for correctness
in the future.
This commit is contained in:
Ian Manske
2024-07-07 22:29:01 +00:00
committed by GitHub
parent c6b6b1b7a8
commit 399a7c8836
246 changed files with 1332 additions and 1234 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
crates/nu-protocol/src/engine/engine_state.rs
View File

@ -8,7 +8,7 @@ use crate::{
},
eval_const::create_nu_constant,
BlockId, Category, Config, DeclId, FileId, GetSpan, HistoryConfig, Module, ModuleId, OverlayId,
ShellError, Signature, Span, SpanId, Type, Value, VarId, VirtualPathId,
ShellError, Signals, Signature, Span, SpanId, Type, Value, VarId, VirtualPathId,
};
use fancy_regex::Regex;
use lru::LruCache;
@ -84,7 +84,7 @@ pub struct EngineState {
pub spans: Vec<Span>,
usage: Usage,
pub scope: ScopeFrame,
pub ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
pub env_vars: Arc<EnvVars>,
pub previous_env_vars: Arc<HashMap<String, Value>>,
pub config: Arc<Config>,
@ -144,7 +144,7 @@ impl EngineState {
0,
false,
),
ctrlc: None,
signals: Signals::empty(),
env_vars: Arc::new(
[(DEFAULT_OVERLAY_NAME.to_string(), HashMap::new())]
.into_iter()
@ -177,6 +177,18 @@ impl EngineState {
}
}
pub fn signals(&self) -> &Signals {
&self.signals
}
pub fn reset_signals(&mut self) {
self.signals.reset()
}
pub fn set_signals(&mut self, signals: Signals) {
self.signals = signals;
}
/// Merges a `StateDelta` onto the current state. These deltas come from a system, like the parser, that
/// creates a new set of definitions and visible symbols in the current scope. We make this transactional
/// as there are times when we want to run the parser and immediately throw away the results (namely:

7
crates/nu-protocol/src/errors/shell_error.rs
View File

@ -1187,6 +1187,13 @@ pub enum ShellError {
span: Option<Span>,
},
/// Operation interrupted
#[error("Operation interrupted")]
Interrupted {
#[label("This operation was interrupted")]
span: Span,
},
/// Operation interrupted by user
#[error("Operation interrupted by user")]
InterruptedByUser {

178
crates/nu-protocol/src/pipeline/byte_stream.rs
View File

@ -1,9 +1,8 @@
use serde::{Deserialize, Serialize};
use crate::{
process::{ChildPipe, ChildProcess, ExitStatus},
ErrSpan, IntoSpanned, OutDest, PipelineData, ShellError, Span, Type, Value,
ErrSpan, IntoSpanned, OutDest, PipelineData, ShellError, Signals, Span, Type, Value,
};
use serde::{Deserialize, Serialize};
#[cfg(unix)]
use std::os::fd::OwnedFd;
#[cfg(windows)]
@ -13,10 +12,6 @@ use std::{
fs::File,
io::{self, BufRead, BufReader, Cursor, ErrorKind, Read, Write},
process::Stdio,
sync::{
atomic::{AtomicBool, Ordering},
Arc,
},
thread,
};
@ -182,7 +177,7 @@ impl From<ByteStreamType> for Type {
pub struct ByteStream {
stream: ByteStreamSource,
span: Span,
ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
type_: ByteStreamType,
known_size: Option<u64>,
}
@ -192,13 +187,13 @@ impl ByteStream {
pub fn new(
stream: ByteStreamSource,
span: Span,
interrupt: Option<Arc<AtomicBool>>,
signals: Signals,
type_: ByteStreamType,
) -> Self {
Self {
stream,
span,
ctrlc: interrupt,
signals,
type_,
known_size: None,
}
@ -208,33 +203,33 @@ impl ByteStream {
pub fn read(
reader: impl Read + Send + 'static,
span: Span,
interrupt: Option<Arc<AtomicBool>>,
signals: Signals,
type_: ByteStreamType,
) -> Self {
Self::new(
ByteStreamSource::Read(Box::new(reader)),
span,
interrupt,
signals,
type_,
)
}
/// Create a [`ByteStream`] from a string. The type of the stream is always `String`.
pub fn read_string(string: String, span: Span, interrupt: Option<Arc<AtomicBool>>) -> Self {
pub fn read_string(string: String, span: Span, signals: Signals) -> Self {
let len = string.len();
ByteStream::read(
Cursor::new(string.into_bytes()),
span,
interrupt,
signals,
ByteStreamType::String,
)
.with_known_size(Some(len as u64))
}
/// Create a [`ByteStream`] from a byte vector. The type of the stream is always `Binary`.
pub fn read_binary(bytes: Vec<u8>, span: Span, interrupt: Option<Arc<AtomicBool>>) -> Self {
pub fn read_binary(bytes: Vec<u8>, span: Span, signals: Signals) -> Self {
let len = bytes.len();
ByteStream::read(Cursor::new(bytes), span, interrupt, ByteStreamType::Binary)
ByteStream::read(Cursor::new(bytes), span, signals, ByteStreamType::Binary)
.with_known_size(Some(len as u64))
}
@ -242,11 +237,11 @@ impl ByteStream {
///
/// The type is implicitly `Unknown`, as it's not typically known whether files will
/// return text or binary.
pub fn file(file: File, span: Span, interrupt: Option<Arc<AtomicBool>>) -> Self {
pub fn file(file: File, span: Span, signals: Signals) -> Self {
Self::new(
ByteStreamSource::File(file),
span,
interrupt,
signals,
ByteStreamType::Unknown,
)
}
@ -259,7 +254,7 @@ impl ByteStream {
Self::new(
ByteStreamSource::Child(Box::new(child)),
span,
None,
Signals::empty(),
ByteStreamType::Unknown,
)
}
@ -271,14 +266,19 @@ impl ByteStream {
pub fn stdin(span: Span) -> Result<Self, ShellError> {
let stdin = os_pipe::dup_stdin().err_span(span)?;
let source = ByteStreamSource::File(convert_file(stdin));
Ok(Self::new(source, span, None, ByteStreamType::Unknown))
Ok(Self::new(
source,
span,
Signals::empty(),
ByteStreamType::Unknown,
))
}
/// Create a [`ByteStream`] from a generator function that writes data to the given buffer
/// when called, and returns `Ok(false)` on end of stream.
pub fn from_fn(
span: Span,
interrupt: Option<Arc<AtomicBool>>,
signals: Signals,
type_: ByteStreamType,
generator: impl FnMut(&mut Vec<u8>) -> Result<bool, ShellError> + Send + 'static,
) -> Self {
@ -288,7 +288,7 @@ impl ByteStream {
generator,
},
span,
interrupt,
signals,
type_,
)
}
@ -301,12 +301,7 @@ impl ByteStream {
/// Create a new [`ByteStream`] from an [`Iterator`] of bytes slices.
///
/// The returned [`ByteStream`] will have a [`ByteStreamSource`] of `Read`.
pub fn from_iter<I>(
iter: I,
span: Span,
interrupt: Option<Arc<AtomicBool>>,
type_: ByteStreamType,
) -> Self
pub fn from_iter<I>(iter: I, span: Span, signals: Signals, type_: ByteStreamType) -> Self
where
I: IntoIterator,
I::IntoIter: Send + 'static,
@ -314,7 +309,7 @@ impl ByteStream {
{
let iter = iter.into_iter();
let cursor = Some(Cursor::new(I::Item::default()));
Self::read(ReadIterator { iter, cursor }, span, interrupt, type_)
Self::read(ReadIterator { iter, cursor }, span, signals, type_)
}
/// Create a new [`ByteStream`] from an [`Iterator`] of [`Result`] bytes slices.
@ -323,7 +318,7 @@ impl ByteStream {
pub fn from_result_iter<I, T>(
iter: I,
span: Span,
interrupt: Option<Arc<AtomicBool>>,
signals: Signals,
type_: ByteStreamType,
) -> Self
where
@ -333,7 +328,7 @@ impl ByteStream {
{
let iter = iter.into_iter();
let cursor = Some(Cursor::new(T::default()));
Self::read(ReadResultIterator { iter, cursor }, span, interrupt, type_)
Self::read(ReadResultIterator { iter, cursor }, span, signals, type_)
}
/// Set the known size, in number of bytes, of the [`ByteStream`].
@ -378,7 +373,7 @@ impl ByteStream {
Some(Reader {
reader: BufReader::new(reader),
span: self.span,
ctrlc: self.ctrlc,
signals: self.signals,
})
}
@ -394,7 +389,7 @@ impl ByteStream {
Some(Lines {
reader: BufReader::new(reader),
span: self.span,
ctrlc: self.ctrlc,
signals: self.signals,
})
}
@ -415,7 +410,7 @@ impl ByteStream {
/// then the stream is considered empty and `None` will be returned.
pub fn chunks(self) -> Option<Chunks> {
let reader = self.stream.reader()?;
Some(Chunks::new(reader, self.span, self.ctrlc, self.type_))
Some(Chunks::new(reader, self.span, self.signals, self.type_))
}
/// Convert the [`ByteStream`] into its inner [`ByteStreamSource`].
@ -552,7 +547,7 @@ impl ByteStream {
pub fn drain(self) -> Result<Option<ExitStatus>, ShellError> {
match self.stream {
ByteStreamSource::Read(read) => {
copy_with_interrupt(read, io::sink(), self.span, self.ctrlc.as_deref())?;
copy_with_signals(read, io::sink(), self.span, &self.signals)?;
Ok(None)
}
ByteStreamSource::File(_) => Ok(None),
@ -578,14 +573,14 @@ impl ByteStream {
/// then the [`ExitStatus`] of the [`ChildProcess`] is returned.
pub fn write_to(self, dest: impl Write) -> Result<Option<ExitStatus>, ShellError> {
let span = self.span;
let ctrlc = self.ctrlc.as_deref();
let signals = &self.signals;
match self.stream {
ByteStreamSource::Read(read) => {
copy_with_interrupt(read, dest, span, ctrlc)?;
copy_with_signals(read, dest, span, signals)?;
Ok(None)
}
ByteStreamSource::File(file) => {
copy_with_interrupt(file, dest, span, ctrlc)?;
copy_with_signals(file, dest, span, signals)?;
Ok(None)
}
ByteStreamSource::Child(mut child) => {
@ -597,10 +592,10 @@ impl ByteStream {
if let Some(stdout) = child.stdout.take() {
match stdout {
ChildPipe::Pipe(pipe) => {
copy_with_interrupt(pipe, dest, span, ctrlc)?;
copy_with_signals(pipe, dest, span, signals)?;
}
ChildPipe::Tee(tee) => {
copy_with_interrupt(tee, dest, span, ctrlc)?;
copy_with_signals(tee, dest, span, signals)?;
}
}
}
@ -615,21 +610,21 @@ impl ByteStream {
stderr: &OutDest,
) -> Result<Option<ExitStatus>, ShellError> {
let span = self.span;
let ctrlc = self.ctrlc.as_deref();
let signals = &self.signals;
match self.stream {
ByteStreamSource::Read(read) => {
write_to_out_dest(read, stdout, true, span, ctrlc)?;
write_to_out_dest(read, stdout, true, span, signals)?;
Ok(None)
}
ByteStreamSource::File(file) => {
match stdout {
OutDest::Pipe | OutDest::Capture | OutDest::Null => {}
OutDest::Inherit => {
copy_with_interrupt(file, io::stdout(), span, ctrlc)?;
copy_with_signals(file, io::stdout(), span, signals)?;
}
OutDest::File(f) => {
copy_with_interrupt(file, f.as_ref(), span, ctrlc)?;
copy_with_signals(file, f.as_ref(), span, signals)?;
}
}
Ok(None)
@ -643,20 +638,20 @@ impl ByteStream {
.name("stderr writer".into())
.spawn_scoped(s, || match err {
ChildPipe::Pipe(pipe) => {
write_to_out_dest(pipe, stderr, false, span, ctrlc)
write_to_out_dest(pipe, stderr, false, span, signals)
}
ChildPipe::Tee(tee) => {
write_to_out_dest(tee, stderr, false, span, ctrlc)
write_to_out_dest(tee, stderr, false, span, signals)
}
})
.err_span(span);
match out {
ChildPipe::Pipe(pipe) => {
write_to_out_dest(pipe, stdout, true, span, ctrlc)
write_to_out_dest(pipe, stdout, true, span, signals)
}
ChildPipe::Tee(tee) => {
write_to_out_dest(tee, stdout, true, span, ctrlc)
write_to_out_dest(tee, stdout, true, span, signals)
}
}?;
@ -672,11 +667,11 @@ impl ByteStream {
}
(Some(out), None) => {
// single output stream, we can consume directly
write_to_out_dest(out, stdout, true, span, ctrlc)?;
write_to_out_dest(out, stdout, true, span, signals)?;
}
(None, Some(err)) => {
// single output stream, we can consume directly
write_to_out_dest(err, stderr, false, span, ctrlc)?;
write_to_out_dest(err, stderr, false, span, signals)?;
}
(None, None) => {}
}
@ -749,7 +744,7 @@ where
pub struct Reader {
reader: BufReader<SourceReader>,
span: Span,
ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
}
impl Reader {
@ -760,14 +755,8 @@ impl Reader {
impl Read for Reader {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
if nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
Err(ShellError::InterruptedByUser {
span: Some(self.span),
}
.into())
} else {
self.reader.read(buf)
}
self.signals.check(self.span)?;
self.reader.read(buf)
}
}
@ -784,7 +773,7 @@ impl BufRead for Reader {
pub struct Lines {
reader: BufReader<SourceReader>,
span: Span,
ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
}
impl Lines {
@ -797,7 +786,7 @@ impl Iterator for Lines {
type Item = Result<String, ShellError>;
fn next(&mut self) -> Option<Self::Item> {
if nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
if self.signals.interrupted() {
None
} else {
let mut buf = Vec::new();
@ -826,23 +815,18 @@ pub struct Chunks {
pos: u64,
error: bool,
span: Span,
ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
type_: ByteStreamType,
}
impl Chunks {
fn new(
reader: SourceReader,
span: Span,
ctrlc: Option<Arc<AtomicBool>>,
type_: ByteStreamType,
) -> Self {
fn new(reader: SourceReader, span: Span, signals: Signals, type_: ByteStreamType) -> Self {
Self {
reader: BufReader::new(reader),
pos: 0,
error: false,
span,
ctrlc,
signals,
type_,
}
}
@ -922,7 +906,7 @@ impl Iterator for Chunks {
type Item = Result<Value, ShellError>;
fn next(&mut self) -> Option<Self::Item> {
if self.error || nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
if self.error || self.signals.interrupted() {
None
} else {
match self.type_ {
@ -988,14 +972,14 @@ fn write_to_out_dest(
stream: &OutDest,
stdout: bool,
span: Span,
ctrlc: Option<&AtomicBool>,
signals: &Signals,
) -> Result<(), ShellError> {
match stream {
OutDest::Pipe | OutDest::Capture => return Ok(()),
OutDest::Null => copy_with_interrupt(read, io::sink(), span, ctrlc),
OutDest::Inherit if stdout => copy_with_interrupt(read, io::stdout(), span, ctrlc),
OutDest::Inherit => copy_with_interrupt(read, io::stderr(), span, ctrlc),
OutDest::File(file) => copy_with_interrupt(read, file.as_ref(), span, ctrlc),
OutDest::Null => copy_with_signals(read, io::sink(), span, signals),
OutDest::Inherit if stdout => copy_with_signals(read, io::stdout(), span, signals),
OutDest::Inherit => copy_with_signals(read, io::stderr(), span, signals),
OutDest::File(file) => copy_with_signals(read, file.as_ref(), span, signals),
}?;
Ok(())
}
@ -1012,28 +996,13 @@ pub(crate) fn convert_file<T: From<OwnedHandle>>(file: impl Into<OwnedHandle>) -
const DEFAULT_BUF_SIZE: usize = 8192;
pub fn copy_with_interrupt(
pub fn copy_with_signals(
mut reader: impl Read,
mut writer: impl Write,
span: Span,
interrupt: Option<&AtomicBool>,
signals: &Signals,
) -> Result<u64, ShellError> {
if let Some(interrupt) = interrupt {
// #[cfg(any(target_os = "linux", target_os = "android"))]
// {
// return crate::sys::kernel_copy::copy_spec(reader, writer);
// }
match generic_copy(&mut reader, &mut writer, span, interrupt) {
Ok(len) => {
writer.flush().err_span(span)?;
Ok(len)
}
Err(err) => {
let _ = writer.flush();
Err(err)
}
}
} else {
if signals.is_empty() {
match io::copy(&mut reader, &mut writer) {
Ok(n) => {
writer.flush().err_span(span)?;
@ -1044,6 +1013,21 @@ pub fn copy_with_interrupt(
Err(err.into_spanned(span).into())
}
}
} else {
// #[cfg(any(target_os = "linux", target_os = "android"))]
// {
// return crate::sys::kernel_copy::copy_spec(reader, writer);
// }
match generic_copy(&mut reader, &mut writer, span, signals) {
Ok(len) => {
writer.flush().err_span(span)?;
Ok(len)
}
Err(err) => {
let _ = writer.flush();
Err(err)
}
}
}
}
@ -1052,14 +1036,12 @@ fn generic_copy(
mut reader: impl Read,
mut writer: impl Write,
span: Span,
interrupt: &AtomicBool,
signals: &Signals,
) -> Result<u64, ShellError> {
let buf = &mut [0; DEFAULT_BUF_SIZE];
let mut len = 0;
loop {
if interrupt.load(Ordering::Relaxed) {
return Err(ShellError::InterruptedByUser { span: Some(span) });
}
signals.check(span)?;
let n = match reader.read(buf) {
Ok(0) => break,
Ok(n) => n,
@ -1134,7 +1116,7 @@ mod tests {
Chunks::new(
SourceReader::Read(Box::new(reader)),
Span::test_data(),
None,
Signals::empty(),
type_,
)
}

29
crates/nu-protocol/src/pipeline/list_stream.rs
View File

@ -1,8 +1,5 @@
use crate::{Config, PipelineData, ShellError, Span, Value};
use std::{
fmt::Debug,
sync::{atomic::AtomicBool, Arc},
};
use crate::{Config, PipelineData, ShellError, Signals, Span, Value};
use std::fmt::Debug;
pub type ValueIterator = Box<dyn Iterator<Item = Value> + Send + 'static>;
@ -21,10 +18,10 @@ impl ListStream {
pub fn new(
iter: impl Iterator<Item = Value> + Send + 'static,
span: Span,
interrupt: Option<Arc<AtomicBool>>,
signals: Signals,
) -> Self {
Self {
stream: Box::new(Interrupt::new(iter, interrupt)),
stream: Box::new(InterruptIter::new(iter, signals)),
span,
}
}
@ -69,10 +66,10 @@ impl ListStream {
/// E.g., `take`, `filter`, `step_by`, and more.
///
/// ```
/// use nu_protocol::{ListStream, Span, Value};
/// use nu_protocol::{ListStream, Signals, Span, Value};
///
/// let span = Span::unknown();
/// let stream = ListStream::new(std::iter::repeat(Value::int(0, span)), span, None);
/// let stream = ListStream::new(std::iter::repeat(Value::int(0, span)), span, Signals::empty());
/// let new_stream = stream.modify(|iter| iter.take(100));
/// ```
pub fn modify<I>(self, f: impl FnOnce(ValueIterator) -> I) -> Self
@ -128,22 +125,22 @@ impl Iterator for IntoIter {
}
}
struct Interrupt<I: Iterator> {
struct InterruptIter<I: Iterator> {
iter: I,
interrupt: Option<Arc<AtomicBool>>,
signals: Signals,
}
impl<I: Iterator> Interrupt<I> {
fn new(iter: I, interrupt: Option<Arc<AtomicBool>>) -> Self {
Self { iter, interrupt }
impl<I: Iterator> InterruptIter<I> {
fn new(iter: I, signals: Signals) -> Self {
Self { iter, signals }
}
}
impl<I: Iterator> Iterator for Interrupt<I> {
impl<I: Iterator> Iterator for InterruptIter<I> {
type Item = <I as Iterator>::Item;
fn next(&mut self) -> Option<Self::Item> {
if nu_utils::ctrl_c::was_pressed(&self.interrupt) {
if self.signals.interrupted() {
None
} else {
self.iter.next()

2
crates/nu-protocol/src/pipeline/mod.rs
View File

@ -3,9 +3,11 @@ pub mod list_stream;
mod metadata;
mod out_dest;
mod pipeline_data;
mod signals;
pub use byte_stream::*;
pub use list_stream::*;
pub use metadata::*;
pub use out_dest::*;
pub use pipeline_data::*;
pub use signals::*;

112
crates/nu-protocol/src/pipeline/pipeline_data.rs
View File

@ -3,13 +3,10 @@ use crate::{
engine::{EngineState, Stack},
process::{ChildPipe, ChildProcess, ExitStatus},
ByteStream, ByteStreamType, Config, ErrSpan, ListStream, OutDest, PipelineMetadata, Range,
ShellError, Span, Type, Value,
ShellError, Signals, Span, Type, Value,
};
use nu_utils::{stderr_write_all_and_flush, stdout_write_all_and_flush};
use std::{
io::{Cursor, Read, Write},
sync::{atomic::AtomicBool, Arc},
};
use std::io::{Cursor, Read, Write};
const LINE_ENDING_PATTERN: &[char] = &['\r', '\n'];
@ -196,19 +193,23 @@ impl PipelineData {
let val_span = value.span();
match value {
Value::List { vals, .. } => PipelineIteratorInner::ListStream(
ListStream::new(vals.into_iter(), val_span, None).into_iter(),
ListStream::new(vals.into_iter(), val_span, Signals::empty()).into_iter(),
),
Value::Binary { val, .. } => PipelineIteratorInner::ListStream(
ListStream::new(
val.into_iter().map(move |x| Value::int(x as i64, val_span)),
val_span,
None,
Signals::empty(),
)
.into_iter(),
),
Value::Range { val, .. } => PipelineIteratorInner::ListStream(
ListStream::new(val.into_range_iter(val_span, None), val_span, None)
.into_iter(),
ListStream::new(
val.into_range_iter(val_span, Signals::empty()),
val_span,
Signals::empty(),
)
.into_iter(),
),
// Propagate errors by explicitly matching them before the final case.
Value::Error { error, .. } => return Err(*error),
@ -301,11 +302,7 @@ impl PipelineData {
}
/// Simplified mapper to help with simple values also. For full iterator support use `.into_iter()` instead
pub fn map<F>(
self,
mut f: F,
ctrlc: Option<Arc<AtomicBool>>,
) -> Result<PipelineData, ShellError>
pub fn map<F>(self, mut f: F, signals: &Signals) -> Result<PipelineData, ShellError>
where
Self: Sized,
F: FnMut(Value) -> Value + 'static + Send,
@ -314,13 +311,14 @@ impl PipelineData {
PipelineData::Value(value, metadata) => {
let span = value.span();
let pipeline = match value {
Value::List { vals, .. } => {
vals.into_iter().map(f).into_pipeline_data(span, ctrlc)
}
Value::Range { val, .. } => val
.into_range_iter(span, ctrlc.clone())
Value::List { vals, .. } => vals
.into_iter()
.map(f)
.into_pipeline_data(span, ctrlc),
.into_pipeline_data(span, signals.clone()),
Value::Range { val, .. } => val
.into_range_iter(span, Signals::empty())
.map(f)
.into_pipeline_data(span, signals.clone()),
value => match f(value) {
Value::Error { error, .. } => return Err(*error),
v => v.into_pipeline_data(),
@ -339,11 +337,7 @@ impl PipelineData {
}
/// Simplified flatmapper. For full iterator support use `.into_iter()` instead
pub fn flat_map<U, F>(
self,
mut f: F,
ctrlc: Option<Arc<AtomicBool>>,
) -> Result<PipelineData, ShellError>
pub fn flat_map<U, F>(self, mut f: F, signals: &Signals) -> Result<PipelineData, ShellError>
where
Self: Sized,
U: IntoIterator<Item = Value> + 'static,
@ -355,14 +349,17 @@ impl PipelineData {
PipelineData::Value(value, metadata) => {
let span = value.span();
let pipeline = match value {
Value::List { vals, .. } => {
vals.into_iter().flat_map(f).into_pipeline_data(span, ctrlc)
}
Value::Range { val, .. } => val
.into_range_iter(span, ctrlc.clone())
Value::List { vals, .. } => vals
.into_iter()
.flat_map(f)
.into_pipeline_data(span, ctrlc),
value => f(value).into_iter().into_pipeline_data(span, ctrlc),
.into_pipeline_data(span, signals.clone()),
Value::Range { val, .. } => val
.into_range_iter(span, Signals::empty())
.flat_map(f)
.into_pipeline_data(span, signals.clone()),
value => f(value)
.into_iter()
.into_pipeline_data(span, signals.clone()),
};
Ok(pipeline.set_metadata(metadata))
}
@ -380,18 +377,16 @@ impl PipelineData {
}
Err(err) => f(Value::binary(err.into_bytes(), span)),
};
Ok(iter
.into_iter()
.into_pipeline_data_with_metadata(span, ctrlc, metadata))
Ok(iter.into_iter().into_pipeline_data_with_metadata(
span,
signals.clone(),
metadata,
))
}
}
}
pub fn filter<F>(
self,
mut f: F,
ctrlc: Option<Arc<AtomicBool>>,
) -> Result<PipelineData, ShellError>
pub fn filter<F>(self, mut f: F, signals: &Signals) -> Result<PipelineData, ShellError>
where
Self: Sized,
F: FnMut(&Value) -> bool + 'static + Send,
@ -401,13 +396,14 @@ impl PipelineData {
PipelineData::Value(value, metadata) => {
let span = value.span();
let pipeline = match value {
Value::List { vals, .. } => {
vals.into_iter().filter(f).into_pipeline_data(span, ctrlc)
}
Value::Range { val, .. } => val
.into_range_iter(span, ctrlc.clone())
Value::List { vals, .. } => vals
.into_iter()
.filter(f)
.into_pipeline_data(span, ctrlc),
.into_pipeline_data(span, signals.clone()),
Value::Range { val, .. } => val
.into_range_iter(span, Signals::empty())
.filter(f)
.into_pipeline_data(span, signals.clone()),
value => {
if f(&value) {
value.into_pipeline_data()
@ -538,7 +534,8 @@ impl PipelineData {
}
}
}
let range_values: Vec<Value> = val.into_range_iter(span, None).collect();
let range_values: Vec<Value> =
val.into_range_iter(span, Signals::empty()).collect();
Ok(PipelineData::Value(Value::list(range_values, span), None))
}
x => Ok(PipelineData::Value(x, metadata)),
@ -638,10 +635,15 @@ impl IntoIterator for PipelineData {
let span = value.span();
match value {
Value::List { vals, .. } => PipelineIteratorInner::ListStream(
ListStream::new(vals.into_iter(), span, None).into_iter(),
ListStream::new(vals.into_iter(), span, Signals::empty()).into_iter(),
),
Value::Range { val, .. } => PipelineIteratorInner::ListStream(
ListStream::new(val.into_range_iter(span, None), span, None).into_iter(),
ListStream::new(
val.into_range_iter(span, Signals::empty()),
span,
Signals::empty(),
)
.into_iter(),
),
x => PipelineIteratorInner::Value(x),
}
@ -703,11 +705,11 @@ where
}
pub trait IntoInterruptiblePipelineData {
fn into_pipeline_data(self, span: Span, ctrlc: Option<Arc<AtomicBool>>) -> PipelineData;
fn into_pipeline_data(self, span: Span, signals: Signals) -> PipelineData;
fn into_pipeline_data_with_metadata(
self,
span: Span,
ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
metadata: impl Into<Option<PipelineMetadata>>,
) -> PipelineData;
}
@ -718,18 +720,18 @@ where
I::IntoIter: Send + 'static,
<I::IntoIter as Iterator>::Item: Into<Value>,
{
fn into_pipeline_data(self, span: Span, ctrlc: Option<Arc<AtomicBool>>) -> PipelineData {
ListStream::new(self.into_iter().map(Into::into), span, ctrlc).into()
fn into_pipeline_data(self, span: Span, signals: Signals) -> PipelineData {
ListStream::new(self.into_iter().map(Into::into), span, signals).into()
}
fn into_pipeline_data_with_metadata(
self,
span: Span,
ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
metadata: impl Into<Option<PipelineMetadata>>,
) -> PipelineData {
PipelineData::ListStream(
ListStream::new(self.into_iter().map(Into::into), span, ctrlc),
ListStream::new(self.into_iter().map(Into::into), span, signals),
metadata.into(),
)
}

76
crates/nu-protocol/src/pipeline/signals.rs Normal file
View File

@ -0,0 +1,76 @@
use crate::{ShellError, Span};
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
/// Used to check for signals to suspend or terminate the execution of Nushell code.
///
/// For now, this struct only supports interruption (ctrl+c or SIGINT).
#[derive(Debug, Clone)]
pub struct Signals {
signals: Option<Arc<AtomicBool>>,
}
impl Signals {
/// A [`Signals`] that is not hooked up to any event/signals source.
///
/// So, this [`Signals`] will never be interrupted.
pub const EMPTY: Self = Signals { signals: None };
/// Create a new [`Signals`] with `ctrlc` as the interrupt source.
///
/// Once `ctrlc` is set to `true`, [`check`](Self::check) will error
/// and [`interrupted`](Self::interrupted) will return `true`.
pub fn new(ctrlc: Arc<AtomicBool>) -> Self {
Self {
signals: Some(ctrlc),
}
}
/// Create a [`Signals`] that is not hooked up to any event/signals source.
///
/// So, the returned [`Signals`] will never be interrupted.
///
/// This should only be used in test code, or if the stream/iterator being created
/// already has an underlying [`Signals`].
pub const fn empty() -> Self {
Self::EMPTY
}
/// Returns an `Err` if an interrupt has been triggered.
///
/// Otherwise, returns `Ok`.
#[inline]
pub fn check(&self, span: Span) -> Result<(), ShellError> {
#[inline]
#[cold]
fn interrupt_error(span: Span) -> Result<(), ShellError> {
Err(ShellError::Interrupted { span })
}
if self.interrupted() {
interrupt_error(span)
} else {
Ok(())
}
}
/// Returns whether an interrupt has been triggered.
#[inline]
pub fn interrupted(&self) -> bool {
self.signals
.as_deref()
.is_some_and(|b| b.load(Ordering::Relaxed))
}
pub(crate) fn is_empty(&self) -> bool {
self.signals.is_none()
}
pub(crate) fn reset(&self) {
if let Some(signals) = &self.signals {
signals.store(false, Ordering::Relaxed);
}
}
}

7
crates/nu-protocol/src/value/mod.rs
View File

@ -23,7 +23,7 @@ use crate::{
ast::{Bits, Boolean, CellPath, Comparison, Math, Operator, PathMember},
did_you_mean,
engine::{Closure, EngineState},
Config, ShellError, Span, Type,
Config, ShellError, Signals, Span, Type,
};
use chrono::{DateTime, Datelike, FixedOffset, Locale, TimeZone};
use chrono_humanize::HumanTime;
@ -1017,8 +1017,9 @@ impl Value {
}
}
Value::Range { ref val, .. } => {
if let Some(item) =
val.into_range_iter(current.span(), None).nth(*count)
if let Some(item) = val
.into_range_iter(current.span(), Signals::empty())
.nth(*count)
{
current = item;
} else if *optional {

53
crates/nu-protocol/src/value/range.rs
View File

@ -1,25 +1,13 @@
//! A Range is an iterator over integers or floats.
use crate::{ast::RangeInclusion, ShellError, Signals, Span, Value};
use serde::{Deserialize, Serialize};
use std::{
cmp::Ordering,
fmt::Display,
sync::{atomic::AtomicBool, Arc},
};
use crate::{ast::RangeInclusion, ShellError, Span, Value};
use std::{cmp::Ordering, fmt::Display};
mod int_range {
use std::{
cmp::Ordering,
fmt::Display,
ops::Bound,
sync::{atomic::AtomicBool, Arc},
};
use crate::{ast::RangeInclusion, ShellError, Signals, Span, Value};
use serde::{Deserialize, Serialize};
use crate::{ast::RangeInclusion, ShellError, Span, Value};
use std::{cmp::Ordering, fmt::Display, ops::Bound};
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct IntRange {
@ -123,12 +111,12 @@ mod int_range {
}
}
pub fn into_range_iter(self, ctrlc: Option<Arc<AtomicBool>>) -> Iter {
pub fn into_range_iter(self, signals: Signals) -> Iter {
Iter {
current: Some(self.start),
step: self.step,
end: self.end,
ctrlc,
signals,
}
}
}
@ -202,7 +190,7 @@ mod int_range {
current: Option<i64>,
step: i64,
end: Bound<i64>,
ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
}
impl Iterator for Iter {
@ -218,7 +206,7 @@ mod int_range {
(_, Bound::Unbounded) => true, // will stop once integer overflows
};
if not_end && !nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
if not_end && !self.signals.interrupted() {
self.current = current.checked_add(self.step);
Some(current)
} else {
@ -233,16 +221,9 @@ mod int_range {
}
mod float_range {
use std::{
cmp::Ordering,
fmt::Display,
ops::Bound,
sync::{atomic::AtomicBool, Arc},
};
use crate::{ast::RangeInclusion, IntRange, Range, ShellError, Signals, Span, Value};
use serde::{Deserialize, Serialize};
use crate::{ast::RangeInclusion, IntRange, Range, ShellError, Span, Value};
use std::{cmp::Ordering, fmt::Display, ops::Bound};
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct FloatRange {
@ -365,13 +346,13 @@ mod float_range {
}
}
pub fn into_range_iter(self, ctrlc: Option<Arc<AtomicBool>>) -> Iter {
pub fn into_range_iter(self, signals: Signals) -> Iter {
Iter {
start: self.start,
step: self.step,
end: self.end,
iter: Some(0),
ctrlc,
signals,
}
}
}
@ -477,7 +458,7 @@ mod float_range {
step: f64,
end: Bound<f64>,
iter: Option<u64>,
ctrlc: Option<Arc<AtomicBool>>,
signals: Signals,
}
impl Iterator for Iter {
@ -495,7 +476,7 @@ mod float_range {
(_, Bound::Unbounded) => current.is_finite(),
};
if not_end && !nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
if not_end && !self.signals.interrupted() {
self.iter = iter.checked_add(1);
Some(current)
} else {
@ -549,10 +530,10 @@ impl Range {
}
}
pub fn into_range_iter(self, span: Span, ctrlc: Option<Arc<AtomicBool>>) -> Iter {
pub fn into_range_iter(self, span: Span, signals: Signals) -> Iter {
match self {
Range::IntRange(range) => Iter::IntIter(range.into_range_iter(ctrlc), span),
Range::FloatRange(range) => Iter::FloatIter(range.into_range_iter(ctrlc), span),
Range::IntRange(range) => Iter::IntIter(range.into_range_iter(signals), span),
Range::FloatRange(range) => Iter::FloatIter(range.into_range_iter(signals), span),
}
}
}