Deno/deno
1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-11-22 04:51:22 +00:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

fix(cli): synchronize async stdio/file reads and writes (#15092)

Browse Source 
Fixes a regression where async writes and reads could get out of order.
This commit is contained in:
David Sherret 2022-07-13 11:16:42 -04:00 committed by GitHub
parent 7470b2d2a7
commit 667812a297
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 457 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
cli/tests/integration/run_tests.rs
View File

@ -8,8 +8,14 @@ use test_util::TempDir;
use util::assert_contains;
itest!(stdout_write_all {
args: "run --quiet stdout_write_all.ts",
output: "stdout_write_all.out",
args: "run --quiet run/stdout_write_all.ts",
output: "run/stdout_write_all.out",
});
itest!(stdin_read_all {
args: "run --quiet run/stdin_read_all.ts",
output: "run/stdin_read_all.out",
input: Some("01234567890123456789012345678901234567890123456789"),
});
itest!(_001_hello {

1
cli/tests/testdata/run/stdin_read_all.out vendored Normal file
View File

@ -0,0 +1 @@
01234567890123456789012345678901234567890123456789

17
cli/tests/testdata/run/stdin_read_all.ts vendored Normal file
View File

@ -0,0 +1,17 @@
const encoder = new TextEncoder();
const pending = [];
// do this a bunch of times to ensure it doesn't race
// and everything happens in order
for (let i = 0; i < 50; i++) {
const buf = new Uint8Array(1);
pending.push(
Deno.stdin.read(buf).then(() => {
return Deno.stdout.write(buf);
}),
);
}
await Promise.all(pending);
await Deno.stdout.write(encoder.encode("\n"));

100
cli/tests/testdata/run/stdout_write_all.out vendored Normal file
View File

@ -0,0 +1,100 @@
Hello, world! 0
Hello, world! 1
Hello, world! 2
Hello, world! 3
Hello, world! 4
Hello, world! 5
Hello, world! 6
Hello, world! 7
Hello, world! 8
Hello, world! 9
Hello, world! 10
Hello, world! 11
Hello, world! 12
Hello, world! 13
Hello, world! 14
Hello, world! 15
Hello, world! 16
Hello, world! 17
Hello, world! 18
Hello, world! 19
Hello, world! 20
Hello, world! 21
Hello, world! 22
Hello, world! 23
Hello, world! 24
Hello, world! 25
Hello, world! 26
Hello, world! 27
Hello, world! 28
Hello, world! 29
Hello, world! 30
Hello, world! 31
Hello, world! 32
Hello, world! 33
Hello, world! 34
Hello, world! 35
Hello, world! 36
Hello, world! 37
Hello, world! 38
Hello, world! 39
Hello, world! 40
Hello, world! 41
Hello, world! 42
Hello, world! 43
Hello, world! 44
Hello, world! 45
Hello, world! 46
Hello, world! 47
Hello, world! 48
Hello, world! 49
Hello, world! 50
Hello, world! 51
Hello, world! 52
Hello, world! 53
Hello, world! 54
Hello, world! 55
Hello, world! 56
Hello, world! 57
Hello, world! 58
Hello, world! 59
Hello, world! 60
Hello, world! 61
Hello, world! 62
Hello, world! 63
Hello, world! 64
Hello, world! 65
Hello, world! 66
Hello, world! 67
Hello, world! 68
Hello, world! 69
Hello, world! 70
Hello, world! 71
Hello, world! 72
Hello, world! 73
Hello, world! 74
Hello, world! 75
Hello, world! 76
Hello, world! 77
Hello, world! 78
Hello, world! 79
Hello, world! 80
Hello, world! 81
Hello, world! 82
Hello, world! 83
Hello, world! 84
Hello, world! 85
Hello, world! 86
Hello, world! 87
Hello, world! 88
Hello, world! 89
Hello, world! 90
Hello, world! 91
Hello, world! 92
Hello, world! 93
Hello, world! 94
Hello, world! 95
Hello, world! 96
Hello, world! 97
Hello, world! 98
Hello, world! 99

13
cli/tests/testdata/run/stdout_write_all.ts vendored Normal file
View File

@ -0,0 +1,13 @@
const encoder = new TextEncoder();
const pending = [];
// do this a bunch of times to ensure it doesn't race
// and everything happens in order
for (let i = 0; i < 100; i++) {
pending.push(Deno.stdout.write(encoder.encode("Hello, ")));
pending.push(Deno.stdout.write(encoder.encode(`world! ${i}`)));
pending.push(Deno.stdout.write(encoder.encode("\n")));
}
await Promise.all(pending);

1
cli/tests/testdata/stdout_write_all.out vendored
View File

@ -1 +0,0 @@
Hello, world!

8
cli/tests/testdata/stdout_write_all.ts vendored
View File

@ -1,8 +0,0 @@
const encoder = new TextEncoder();
const pending = [
Deno.stdout.write(encoder.encode("Hello, ")),
Deno.stdout.write(encoder.encode("world!")),
];
await Promise.all(pending);
await Deno.stdout.write(encoder.encode("\n"));

32
cli/tests/unit/files_test.ts
View File

@ -263,6 +263,38 @@ Deno.test(
},
);
Deno.test(
{ permissions: { write: true } },
async function writeSyncWhileAsyncFails() {
const tempDir = await Deno.makeTempDir();
try {
const filePath = tempDir + "/file.txt";
const file = await Deno.open(filePath, { create: true, write: true });
const rid = file.rid;
try {
// set a file lock so the async write will be held up
await Deno.flock(rid, true);
let p: Promise<number> | undefined;
try {
p = Deno.write(rid, new TextEncoder().encode("test"));
assertThrows(
() => Deno.writeSync(rid, new TextEncoder().encode("test")),
Error,
"Resource is unavailable because it is in use by a promise",
);
} finally {
await Deno.funlock(rid);
}
await p;
} finally {
file.close();
}
} finally {
Deno.removeSync(tempDir, { recursive: true });
}
},
);
Deno.test(async function openOptions() {
const filename = "cli/tests/testdata/fixture.json";
await assertRejects(

126
runtime/ops/fs.rs
View File

@ -322,10 +322,9 @@ fn seek_helper(args: SeekArgs) -> Result<(u32, SeekFrom), AnyError> {
#[op]
fn op_seek_sync(state: &mut OpState, args: SeekArgs) -> Result<u64, AnyError> {
let (rid, seek_from) = seek_helper(args)?;
let pos = StdFileResource::with_file(state, rid, |std_file| {
StdFileResource::with_file(state, rid, |std_file| {
std_file.seek(seek_from).map_err(AnyError::from)
})?;
Ok(pos)
})
}
#[op]
@ -335,19 +334,10 @@ async fn op_seek_async(
) -> Result<u64, AnyError> {
let (rid, seek_from) = seek_helper(args)?;
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
tokio::task::spawn_blocking(move || {
let mut std_file = std_file.lock();
std_file.seek(seek_from)
StdFileResource::with_file_blocking_task(state, rid, move |std_file| {
std_file.seek(seek_from).map_err(AnyError::from)
})
.await?
.map_err(AnyError::from)
.await
}
#[op]
@ -357,8 +347,7 @@ fn op_fdatasync_sync(
) -> Result<(), AnyError> {
StdFileResource::with_file(state, rid, |std_file| {
std_file.sync_data().map_err(AnyError::from)
})?;
Ok(())
})
}
#[op]
@ -366,27 +355,17 @@ async fn op_fdatasync_async(
state: Rc<RefCell<OpState>>,
rid: ResourceId,
) -> Result<(), AnyError> {
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
tokio::task::spawn_blocking(move || {
let std_file = std_file.lock();
std_file.sync_data()
StdFileResource::with_file_blocking_task(state, rid, move |std_file| {
std_file.sync_data().map_err(AnyError::from)
})
.await?
.map_err(AnyError::from)
.await
}
#[op]
fn op_fsync_sync(state: &mut OpState, rid: ResourceId) -> Result<(), AnyError> {
StdFileResource::with_file(state, rid, |std_file| {
std_file.sync_all().map_err(AnyError::from)
})?;
Ok(())
})
}
#[op]
@ -394,19 +373,10 @@ async fn op_fsync_async(
state: Rc<RefCell<OpState>>,
rid: ResourceId,
) -> Result<(), AnyError> {
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
tokio::task::spawn_blocking(move || {
let std_file = std_file.lock();
std_file.sync_all()
StdFileResource::with_file_blocking_task(state, rid, move |std_file| {
std_file.sync_all().map_err(AnyError::from)
})
.await?
.map_err(AnyError::from)
.await
}
#[op]
@ -425,19 +395,11 @@ async fn op_fstat_async(
state: Rc<RefCell<OpState>>,
rid: ResourceId,
) -> Result<FsStat, AnyError> {
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
let metadata = tokio::task::spawn_blocking(move || {
let std_file = std_file.lock();
std_file.metadata()
})
.await?
.map_err(AnyError::from)?;
let metadata =
StdFileResource::with_file_blocking_task(state, rid, move |std_file| {
std_file.metadata().map_err(AnyError::from)
})
.await?;
Ok(get_stat(metadata))
}
@ -469,15 +431,7 @@ async fn op_flock_async(
use fs3::FileExt;
super::check_unstable2(&state, "Deno.flock");
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
tokio::task::spawn_blocking(move || -> Result<(), AnyError> {
let std_file = std_file.lock();
StdFileResource::with_file_blocking_task(state, rid, move |std_file| {
if exclusive {
std_file.lock_exclusive()?;
} else {
@ -485,7 +439,7 @@ async fn op_flock_async(
}
Ok(())
})
.await?
.await
}
#[op]
@ -510,19 +464,11 @@ async fn op_funlock_async(
use fs3::FileExt;
super::check_unstable2(&state, "Deno.funlock");
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
tokio::task::spawn_blocking(move || -> Result<(), AnyError> {
let std_file = std_file.lock();
StdFileResource::with_file_blocking_task(state, rid, move |std_file| {
std_file.unlock()?;
Ok(())
})
.await?
.await
}
#[op]
@ -1592,19 +1538,11 @@ async fn op_ftruncate_async(
let rid = args.rid;
let len = args.len as u64;
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
tokio::task::spawn_blocking(move || {
let std_file = std_file.lock();
std_file.set_len(len)
StdFileResource::with_file_blocking_task(state, rid, move |std_file| {
std_file.set_len(len)?;
Ok(())
})
.await?
.map_err(AnyError::from)
.await
}
#[derive(Deserialize)]
@ -1884,19 +1822,11 @@ async fn op_futime_async(
let atime = filetime::FileTime::from_unix_time(args.atime.0, args.atime.1);
let mtime = filetime::FileTime::from_unix_time(args.mtime.0, args.mtime.1);
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
tokio::task::spawn_blocking(move || {
let std_file = std_file.lock();
filetime::set_file_handle_times(&std_file, Some(atime), Some(mtime))?;
StdFileResource::with_file_blocking_task(state, rid, move |std_file| {
filetime::set_file_handle_times(std_file, Some(atime), Some(mtime))?;
Ok(())
})
.await
.unwrap()
}
#[derive(Deserialize)]

308
runtime/ops/io.rs
View File

@ -1,8 +1,8 @@
// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
use deno_core::error::resource_unavailable;
use deno_core::error::AnyError;
use deno_core::op;
use deno_core::parking_lot::Mutex;
use deno_core::AsyncMutFuture;
use deno_core::AsyncRefCell;
use deno_core::AsyncResult;
@ -22,7 +22,6 @@ use std::io::ErrorKind;
use std::io::Read;
use std::io::Write;
use std::rc::Rc;
use std::sync::Arc;
use tokio::io::AsyncRead;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWrite;
@ -125,23 +124,27 @@ pub fn init_stdio(stdio: Stdio) -> Extension {
let t = &mut state.resource_table;
t.add(StdFileResource::stdio(
match stdio.stdin {
StdioPipe::Inherit => StdFileResourceInner::Stdin(Arc::new(
Mutex::new(STDIN_HANDLE.try_clone().unwrap()),
)),
StdioPipe::Inherit => {
StdFileResourceInner::Stdin(STDIN_HANDLE.try_clone().unwrap())
}
StdioPipe::File(pipe) => StdFileResourceInner::file(pipe),
},
"stdin",
));
t.add(StdFileResource::stdio(
match stdio.stdout {
StdioPipe::Inherit => StdFileResourceInner::Stdout,
StdioPipe::Inherit => {
StdFileResourceInner::Stdout(STDOUT_HANDLE.try_clone().unwrap())
}
StdioPipe::File(pipe) => StdFileResourceInner::file(pipe),
},
"stdout",
));
t.add(StdFileResource::stdio(
match stdio.stderr {
StdioPipe::Inherit => StdFileResourceInner::Stderr,
StdioPipe::Inherit => {
StdFileResourceInner::Stderr(STDERR_HANDLE.try_clone().unwrap())
}
StdioPipe::File(pipe) => StdFileResourceInner::file(pipe),
},
"stderr",
@ -305,31 +308,28 @@ impl Resource for ChildStderrResource {
}
}
#[derive(Clone)]
enum StdFileResourceInner {
File(Arc<Mutex<StdFile>>),
Stdin(Arc<Mutex<StdFile>>),
// Ideally we would store stdio as an StdFile, but we get some Windows
// specific functionality for free by using Rust std's wrappers. So we
// take a bit of a complexity hit here in order to not have to duplicate
// the functionality in Rust's std/src/sys/windows/stdio.rs
Stdout,
Stderr,
File(StdFile),
Stdin(StdFile),
// For stdout and stderr, we sometimes instead use std::io::stdout() directly,
// because we get some Windows specific functionality for free by using Rust
// std's wrappers. So we take a bit of a complexity hit in order to not
// have to duplicate the functionality in Rust's std/src/sys/windows/stdio.rs
Stdout(StdFile),
Stderr(StdFile),
}
impl StdFileResourceInner {
pub fn file(fs_file: StdFile) -> Self {
StdFileResourceInner::File(Arc::new(Mutex::new(fs_file)))
StdFileResourceInner::File(fs_file)
}
pub fn with_file<R>(&self, mut f: impl FnMut(&mut StdFile) -> R) -> R {
pub fn with_file<R>(&mut self, f: impl FnOnce(&mut StdFile) -> R) -> R {
match self {
Self::File(file) | Self::Stdin(file) => {
let mut file = file.lock();
f(&mut file)
}
Self::Stdout => f(&mut STDOUT_HANDLE.try_clone().unwrap()),
Self::Stderr => f(&mut STDERR_HANDLE.try_clone().unwrap()),
Self::File(file)
| Self::Stdin(file)
| Self::Stdout(file)
| Self::Stderr(file) => f(file),
}
}
@ -337,120 +337,163 @@ impl StdFileResourceInner {
&mut self,
buf: &[u8],
) -> Result<usize, AnyError> {
let nwritten = self.write(buf)?;
if !matches!(self, StdFileResourceInner::File(_)) {
// Rust will line buffer and we don't want that behavior
// (see https://github.com/denoland/deno/issues/948), so flush.
// Although an alternative solution could be to bypass Rust's std by
// using the raw fds/handles, it will cause encoding issues on Windows
// that we get solved for free by using Rust's stdio wrappers (see
// std/src/sys/windows/stdio.rs in Rust's source code).
self.flush()?;
// Rust will line buffer and we don't want that behavior
// (see https://github.com/denoland/deno/issues/948), so flush stdout and stderr.
// Although an alternative solution could be to bypass Rust's std by
// using the raw fds/handles, it will cause encoding issues on Windows
// that we get solved for free by using Rust's stdio wrappers (see
// std/src/sys/windows/stdio.rs in Rust's source code).
match self {
Self::File(file) => Ok(file.write(buf)?),
Self::Stdin(_) => {
Err(Into::<std::io::Error>::into(ErrorKind::Unsupported).into())
}
Self::Stdout(_) => {
// bypass the file and use std::io::stdout()
let mut stdout = std::io::stdout().lock();
let nwritten = stdout.write(buf)?;
stdout.flush()?;
Ok(nwritten)
}
Self::Stderr(_) => {
// bypass the file and use std::io::stderr()
let mut stderr = std::io::stderr().lock();
let nwritten = stderr.write(buf)?;
stderr.flush()?;
Ok(nwritten)
}
}
}
pub fn write_all_and_maybe_flush(
&mut self,
buf: &[u8],
) -> Result<(), AnyError> {
// this method exists instead of using a `Write` implementation
// so that we can acquire the locks once and do both actions
match self {
Self::File(file) => Ok(file.write_all(buf)?),
Self::Stdin(_) => {
Err(Into::<std::io::Error>::into(ErrorKind::Unsupported).into())
}
Self::Stdout(_) => {
// bypass the file and use std::io::stdout()
let mut stdout = std::io::stdout().lock();
stdout.write_all(buf)?;
stdout.flush()?;
Ok(())
}
Self::Stderr(_) => {
// bypass the file and use std::io::stderr()
let mut stderr = std::io::stderr().lock();
stderr.write_all(buf)?;
stderr.flush()?;
Ok(())
}
}
Ok(nwritten)
}
}
impl Read for StdFileResourceInner {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
match self {
Self::File(file) | Self::Stdin(file) => file.lock().read(buf),
Self::Stdout => Err(ErrorKind::Unsupported.into()),
Self::Stderr => Err(ErrorKind::Unsupported.into()),
}
}
}
impl Write for StdFileResourceInner {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
match self {
Self::File(file) => file.lock().write(buf),
Self::Stdin(_) => Err(ErrorKind::Unsupported.into()),
Self::Stdout => std::io::stdout().write(buf),
Self::Stderr => std::io::stderr().write(buf),
}
}
fn flush(&mut self) -> std::io::Result<()> {
match self {
Self::File(file) => file.lock().flush(),
Self::Stdin(_) => Err(ErrorKind::Unsupported.into()),
Self::Stdout => std::io::stdout().flush(),
Self::Stderr => std::io::stderr().flush(),
Self::File(file) | Self::Stdin(file) => file.read(buf),
Self::Stdout(_) | Self::Stderr(_) => Err(ErrorKind::Unsupported.into()),
}
}
}
pub struct StdFileResource {
inner: StdFileResourceInner,
metadata: RefCell<FileMetadata>,
name: String,
cell: AsyncRefCell<Option<(StdFileResourceInner, FileMetadata)>>,
}
impl StdFileResource {
fn stdio(inner: StdFileResourceInner, name: &str) -> Self {
Self {
inner,
metadata: Default::default(),
cell: AsyncRefCell::new(Some((inner, Default::default()))),
name: name.to_string(),
}
}
pub fn fs_file(fs_file: StdFile) -> Self {
Self {
inner: StdFileResourceInner::file(fs_file),
metadata: Default::default(),
cell: AsyncRefCell::new(Some((
StdFileResourceInner::file(fs_file),
Default::default(),
))),
name: "fsFile".to_string(),
}
}
pub fn std_file(&self) -> Arc<Mutex<StdFile>> {
match &self.inner {
StdFileResourceInner::File(fs_file)
| StdFileResourceInner::Stdin(fs_file) => fs_file.clone(),
StdFileResourceInner::Stdout => {
Arc::new(Mutex::new(STDOUT_HANDLE.try_clone().unwrap()))
}
StdFileResourceInner::Stderr => {
Arc::new(Mutex::new(STDERR_HANDLE.try_clone().unwrap()))
fn with_inner_and_metadata<TResult>(
self: Rc<Self>,
action: impl FnOnce(
&mut StdFileResourceInner,
&mut FileMetadata,
) -> Result<TResult, AnyError>,
) -> Result<TResult, AnyError> {
match RcRef::map(&self, |r| &r.cell).try_borrow_mut() {
Some(mut cell) => {
let mut file = cell.take().unwrap();
let result = action(&mut file.0, &mut file.1);
cell.replace(file);
result
}
None => Err(resource_unavailable()),
}
}
pub fn metadata_mut(&self) -> std::cell::RefMut<FileMetadata> {
self.metadata.borrow_mut()
async fn with_inner_blocking_task<F, R: Send + 'static>(
self: Rc<Self>,
action: F,
) -> R
where
F: FnOnce(&mut StdFileResourceInner) -> R + Send + 'static,
{
// we take the value out of the cell, use it on a blocking task,
// then put it back into the cell when we're done
let mut cell = RcRef::map(&self, |r| &r.cell).borrow_mut().await;
let mut file = cell.take().unwrap();
let (file, result) = tokio::task::spawn_blocking(move || {
let result = action(&mut file.0);
(file, result)
})
.await
.unwrap();
cell.replace(file);
result
}
async fn read(
self: Rc<Self>,
mut buf: ZeroCopyBuf,
) -> Result<(usize, ZeroCopyBuf), AnyError> {
let mut inner = self.inner.clone();
tokio::task::spawn_blocking(
move || -> Result<(usize, ZeroCopyBuf), AnyError> {
Ok((inner.read(&mut buf)?, buf))
},
)
.await?
self
.with_inner_blocking_task(
move |inner| -> Result<(usize, ZeroCopyBuf), AnyError> {
Ok((inner.read(&mut buf)?, buf))
},
)
.await
}
async fn write(self: Rc<Self>, buf: ZeroCopyBuf) -> Result<usize, AnyError> {
let mut inner = self.inner.clone();
tokio::task::spawn_blocking(move || inner.write_and_maybe_flush(&buf))
.await?
.map_err(AnyError::from)
self
.with_inner_blocking_task(move |inner| inner.write_and_maybe_flush(&buf))
.await
}
fn with_inner<F, R>(
fn with_resource<F, R>(
state: &mut OpState,
rid: ResourceId,
mut f: F,
f: F,
) -> Result<R, AnyError>
where
F: FnMut(StdFileResourceInner) -> Result<R, AnyError>,
F: FnOnce(Rc<StdFileResource>) -> Result<R, AnyError>,
{
let resource = state.resource_table.get::<StdFileResource>(rid)?;
f(resource.inner.clone())
f(resource)
}
pub fn with_file<F, R>(
@ -459,10 +502,44 @@ impl StdFileResource {
f: F,
) -> Result<R, AnyError>
where
F: FnMut(&mut StdFile) -> Result<R, AnyError>,
F: FnOnce(&mut StdFile) -> Result<R, AnyError>,
{
let resource = state.resource_table.get::<StdFileResource>(rid)?;
resource.inner.with_file(f)
Self::with_resource(state, rid, move |resource| {
resource.with_inner_and_metadata(move |inner, _| inner.with_file(f))
})
}
pub fn with_file_and_metadata<F, R>(
state: &mut OpState,
rid: ResourceId,
f: F,
) -> Result<R, AnyError>
where
F: FnOnce(&mut StdFile, &mut FileMetadata) -> Result<R, AnyError>,
{
Self::with_resource(state, rid, move |resource| {
resource.with_inner_and_metadata(move |inner, metadata| {
inner.with_file(move |file| f(file, metadata))
})
})
}
pub async fn with_file_blocking_task<F, R: Send + 'static>(
state: Rc<RefCell<OpState>>,
rid: ResourceId,
f: F,
) -> Result<R, AnyError>
where
F: (FnOnce(&mut StdFile) -> Result<R, AnyError>) + Send + 'static,
{
let resource = state
.borrow_mut()
.resource_table
.get::<StdFileResource>(rid)?;
resource
.with_inner_blocking_task(move |inner| inner.with_file(f))
.await
}
pub fn clone_file(
@ -478,12 +555,14 @@ impl StdFileResource {
state: &mut OpState,
rid: u32,
) -> Result<std::process::Stdio, AnyError> {
Self::with_inner(state, rid, |inner| match inner {
StdFileResourceInner::File(file) => {
let file = file.lock().try_clone()?;
Ok(file.into())
}
_ => Ok(std::process::Stdio::inherit()),
Self::with_resource(state, rid, |resource| {
resource.with_inner_and_metadata(|inner, _| match inner {
StdFileResourceInner::File(file) => {
let file = file.try_clone()?;
Ok(file.into())
}
_ => Ok(std::process::Stdio::inherit()),
})
})
}
}
@ -513,10 +592,11 @@ pub fn op_print(
is_err: bool,
) -> Result<(), AnyError> {
let rid = if is_err { 2 } else { 1 };
StdFileResource::with_inner(state, rid, move |mut inner| {
inner.write_all(msg.as_bytes())?;
inner.flush().unwrap();
Ok(())
StdFileResource::with_resource(state, rid, move |resource| {
resource.with_inner_and_metadata(|inner, _| {
inner.write_all_and_maybe_flush(msg.as_bytes())?;
Ok(())
})
})
}
@ -526,11 +606,13 @@ fn op_read_sync(
rid: ResourceId,
mut buf: ZeroCopyBuf,
) -> Result<u32, AnyError> {
StdFileResource::with_inner(state, rid, move |mut inner| {
inner
.read(&mut buf)
.map(|n: usize| n as u32)
.map_err(AnyError::from)
StdFileResource::with_resource(state, rid, move |resource| {
resource.with_inner_and_metadata(|inner, _| {
inner
.read(&mut buf)
.map(|n: usize| n as u32)
.map_err(AnyError::from)
})
})
}
@ -540,10 +622,12 @@ fn op_write_sync(
rid: ResourceId,
buf: ZeroCopyBuf,
) -> Result<u32, AnyError> {
StdFileResource::with_inner(state, rid, move |mut inner| {
inner
.write_and_maybe_flush(&buf)
.map(|nwritten: usize| nwritten as u32)
.map_err(AnyError::from)
StdFileResource::with_resource(state, rid, move |resource| {
resource.with_inner_and_metadata(|inner, _| {
inner
.write_and_maybe_flush(&buf)
.map(|nwritten: usize| nwritten as u32)
.map_err(AnyError::from)
})
})
}

121
runtime/ops/tty.rs
View File

@ -82,82 +82,85 @@ fn op_set_raw(state: &mut OpState, args: SetRawArgs) -> Result<(), AnyError> {
use winapi::shared::minwindef::FALSE;
use winapi::um::{consoleapi, handleapi};
let resource = state.resource_table.get::<StdFileResource>(rid)?;
if cbreak {
return Err(deno_core::error::not_supported());
}
let std_file = resource.std_file();
let std_file = std_file.lock(); // hold the lock
let handle = std_file.as_raw_handle();
StdFileResource::with_file(state, rid, move |std_file| {
let handle = std_file.as_raw_handle();
if handle == handleapi::INVALID_HANDLE_VALUE {
return Err(Error::last_os_error().into());
} else if handle.is_null() {
return Err(custom_error("ReferenceError", "null handle"));
}
let mut original_mode: DWORD = 0;
if unsafe { consoleapi::GetConsoleMode(handle, &mut original_mode) }
== FALSE
{
return Err(Error::last_os_error().into());
}
let new_mode = if is_raw {
original_mode & !RAW_MODE_MASK
} else {
original_mode | RAW_MODE_MASK
};
if unsafe { consoleapi::SetConsoleMode(handle, new_mode) } == FALSE {
return Err(Error::last_os_error().into());
}
if handle == handleapi::INVALID_HANDLE_VALUE {
return Err(Error::last_os_error().into());
} else if handle.is_null() {
return Err(custom_error("ReferenceError", "null handle"));
}
let mut original_mode: DWORD = 0;
if unsafe { consoleapi::GetConsoleMode(handle, &mut original_mode) }
== FALSE
{
return Err(Error::last_os_error().into());
}
let new_mode = if is_raw {
original_mode & !RAW_MODE_MASK
} else {
original_mode | RAW_MODE_MASK
};
if unsafe { consoleapi::SetConsoleMode(handle, new_mode) } == FALSE {
return Err(Error::last_os_error().into());
}
Ok(())
Ok(())
})
}
#[cfg(unix)]
{
use std::os::unix::io::AsRawFd;
let resource = state.resource_table.get::<StdFileResource>(rid)?;
let std_file = resource.std_file();
let raw_fd = std_file.lock().as_raw_fd();
let mut meta_data = resource.metadata_mut();
let maybe_tty_mode = &mut meta_data.tty.mode;
StdFileResource::with_file_and_metadata(
state,
rid,
move |std_file, meta_data| {
let raw_fd = std_file.as_raw_fd();
let maybe_tty_mode = &mut meta_data.tty.mode;
if is_raw {
if maybe_tty_mode.is_none() {
// Save original mode.
let original_mode = termios::tcgetattr(raw_fd)?;
maybe_tty_mode.replace(original_mode);
}
if is_raw {
if maybe_tty_mode.is_none() {
// Save original mode.
let original_mode = termios::tcgetattr(raw_fd)?;
maybe_tty_mode.replace(original_mode);
}
let mut raw = maybe_tty_mode.clone().unwrap();
let mut raw = maybe_tty_mode.clone().unwrap();
raw.input_flags &= !(termios::InputFlags::BRKINT
| termios::InputFlags::ICRNL
| termios::InputFlags::INPCK
| termios::InputFlags::ISTRIP
| termios::InputFlags::IXON);
raw.input_flags &= !(termios::InputFlags::BRKINT
| termios::InputFlags::ICRNL
| termios::InputFlags::INPCK
| termios::InputFlags::ISTRIP
| termios::InputFlags::IXON);
raw.control_flags |= termios::ControlFlags::CS8;
raw.control_flags |= termios::ControlFlags::CS8;
raw.local_flags &= !(termios::LocalFlags::ECHO
| termios::LocalFlags::ICANON
| termios::LocalFlags::IEXTEN);
if !cbreak {
raw.local_flags &= !(termios::LocalFlags::ISIG);
}
raw.control_chars[termios::SpecialCharacterIndices::VMIN as usize] = 1;
raw.control_chars[termios::SpecialCharacterIndices::VTIME as usize] = 0;
termios::tcsetattr(raw_fd, termios::SetArg::TCSADRAIN, &raw)?;
} else {
// Try restore saved mode.
if let Some(mode) = maybe_tty_mode.take() {
termios::tcsetattr(raw_fd, termios::SetArg::TCSADRAIN, &mode)?;
}
}
raw.local_flags &= !(termios::LocalFlags::ECHO
| termios::LocalFlags::ICANON
| termios::LocalFlags::IEXTEN);
if !cbreak {
raw.local_flags &= !(termios::LocalFlags::ISIG);
}
raw.control_chars[termios::SpecialCharacterIndices::VMIN as usize] =
1;
raw.control_chars[termios::SpecialCharacterIndices::VTIME as usize] =
0;
termios::tcsetattr(raw_fd, termios::SetArg::TCSADRAIN, &raw)?;
} else {
// Try restore saved mode.
if let Some(mode) = maybe_tty_mode.take() {
termios::tcsetattr(raw_fd, termios::SetArg::TCSADRAIN, &mode)?;
}
}
Ok(())
Ok(())
},
)
}
}