wasmtime_wasi/p3/filesystem/
host.rs

1use crate::filesystem::sys;
2use crate::filesystem::{Descriptor, Dir, File, WasiFilesystem, WasiFilesystemCtxView};
3use crate::p3::bindings::clocks::system_clock;
4use crate::p3::bindings::filesystem::types::{
5    self, Advice, DescriptorFlags, DescriptorStat, DescriptorType, DirectoryEntry, ErrorCode,
6    Filesize, MetadataHashValue, NewTimestamp, OpenFlags, PathFlags,
7};
8use crate::p3::filesystem::{FilesystemError, FilesystemResult, preopens};
9use crate::p3::{DEFAULT_BUFFER_CAPACITY, FallibleIteratorProducer};
10use bytes::BytesMut;
11use core::pin::Pin;
12use core::task::{Context, Poll, ready};
13use core::{iter, mem};
14use std::io;
15use std::sync::Arc;
16use std::time::SystemTime;
17use tokio::sync::{mpsc, oneshot};
18use tokio::task::{JoinHandle, spawn_blocking};
19use wasmtime::StoreContextMut;
20use wasmtime::component::{
21    Access, Accessor, Destination, FutureReader, Resource, ResourceTable, Source, StreamConsumer,
22    StreamProducer, StreamReader, StreamResult,
23};
24use wasmtime::error::Context as _;
25
26fn get_descriptor<'a>(
27    table: &'a ResourceTable,
28    fd: &'a Resource<Descriptor>,
29) -> FilesystemResult<&'a Descriptor> {
30    table
31        .get(fd)
32        .context("failed to get descriptor resource from table")
33        .map_err(FilesystemError::trap)
34}
35
36fn get_file<'a>(
37    table: &'a ResourceTable,
38    fd: &'a Resource<Descriptor>,
39) -> FilesystemResult<&'a File> {
40    let file = get_descriptor(table, fd).map(Descriptor::file)??;
41    Ok(file)
42}
43
44fn get_dir<'a>(
45    table: &'a ResourceTable,
46    fd: &'a Resource<Descriptor>,
47) -> FilesystemResult<&'a Dir> {
48    let dir = get_descriptor(table, fd).map(Descriptor::dir)??;
49    Ok(dir)
50}
51
52trait AccessorExt {
53    fn get_descriptor(&self, fd: &Resource<Descriptor>) -> FilesystemResult<Descriptor>;
54    fn get_file(&self, fd: &Resource<Descriptor>) -> FilesystemResult<File>;
55    fn get_dir(&self, fd: &Resource<Descriptor>) -> FilesystemResult<Dir>;
56    fn get_dir_pair(
57        &self,
58        a: &Resource<Descriptor>,
59        b: &Resource<Descriptor>,
60    ) -> FilesystemResult<(Dir, Dir)>;
61}
62
63impl<T> AccessorExt for Accessor<T, WasiFilesystem> {
64    fn get_descriptor(&self, fd: &Resource<Descriptor>) -> FilesystemResult<Descriptor> {
65        self.with(|mut store| {
66            let fd = get_descriptor(store.get().table, fd)?;
67            Ok(fd.clone())
68        })
69    }
70
71    fn get_file(&self, fd: &Resource<Descriptor>) -> FilesystemResult<File> {
72        self.with(|mut store| {
73            let file = get_file(store.get().table, fd)?;
74            Ok(file.clone())
75        })
76    }
77
78    fn get_dir(&self, fd: &Resource<Descriptor>) -> FilesystemResult<Dir> {
79        self.with(|mut store| {
80            let dir = get_dir(store.get().table, fd)?;
81            Ok(dir.clone())
82        })
83    }
84
85    fn get_dir_pair(
86        &self,
87        a: &Resource<Descriptor>,
88        b: &Resource<Descriptor>,
89    ) -> FilesystemResult<(Dir, Dir)> {
90        self.with(|mut store| {
91            let table = store.get().table;
92            let a = get_dir(table, a)?;
93            let b = get_dir(table, b)?;
94            Ok((a.clone(), b.clone()))
95        })
96    }
97}
98
99fn systemtime_from(t: system_clock::Instant) -> Result<std::time::SystemTime, ErrorCode> {
100    if let Ok(seconds) = t.seconds.try_into() {
101        std::time::SystemTime::UNIX_EPOCH
102            .checked_add(core::time::Duration::new(seconds, t.nanoseconds))
103            .ok_or(ErrorCode::Overflow)
104    } else {
105        std::time::SystemTime::UNIX_EPOCH
106            .checked_sub(core::time::Duration::new(
107                t.seconds.unsigned_abs(),
108                t.nanoseconds,
109            ))
110            .ok_or(ErrorCode::Overflow)
111    }
112}
113
114fn systemtimespec_from(t: NewTimestamp) -> Result<Option<SystemTime>, ErrorCode> {
115    match t {
116        NewTimestamp::NoChange => Ok(None),
117        NewTimestamp::Now => Ok(Some(SystemTime::now())),
118        NewTimestamp::Timestamp(st) => Ok(Some(systemtime_from(st)?)),
119    }
120}
121
122struct ReadStreamProducer {
123    file: File,
124    offset: u64,
125    result: Option<oneshot::Sender<Result<(), ErrorCode>>>,
126    task: Option<JoinHandle<std::io::Result<BytesMut>>>,
127}
128
129impl Drop for ReadStreamProducer {
130    fn drop(&mut self) {
131        self.close(Ok(()))
132    }
133}
134
135impl ReadStreamProducer {
136    fn close(&mut self, res: Result<(), ErrorCode>) {
137        if let Some(tx) = self.result.take() {
138            _ = tx.send(res);
139        }
140    }
141
142    /// Update the internal `offset` field after reading `amt` bytes from the file.
143    fn complete_read(&mut self, amt: usize) -> StreamResult {
144        let Ok(amt) = amt.try_into() else {
145            self.close(Err(ErrorCode::Overflow));
146            return StreamResult::Dropped;
147        };
148        let Some(amt) = self.offset.checked_add(amt) else {
149            self.close(Err(ErrorCode::Overflow));
150            return StreamResult::Dropped;
151        };
152        self.offset = amt;
153        StreamResult::Completed
154    }
155}
156
157impl<D> StreamProducer<D> for ReadStreamProducer {
158    type Item = u8;
159    type Buffer = BytesMut;
160
161    fn poll_produce<'a>(
162        mut self: Pin<&mut Self>,
163        cx: &mut Context<'_>,
164        store: StoreContextMut<'a, D>,
165        mut dst: Destination<'a, Self::Item, Self::Buffer>,
166        finish: bool,
167    ) -> Poll<wasmtime::Result<StreamResult>> {
168        if let Some(file) = self.file.as_blocking_file() {
169            // Once a blocking file, always a blocking file, so assert as such.
170            assert!(self.task.is_none());
171            let mut dst = dst.as_direct(store, DEFAULT_BUFFER_CAPACITY);
172            let buf = dst.remaining();
173            if buf.is_empty() {
174                return Poll::Ready(Ok(StreamResult::Completed));
175            }
176            return match sys::read_at_cursor_unspecified(file, buf, self.offset) {
177                Ok(0) => {
178                    self.close(Ok(()));
179                    Poll::Ready(Ok(StreamResult::Dropped))
180                }
181                Ok(n) => {
182                    dst.mark_written(n);
183                    Poll::Ready(Ok(self.complete_read(n)))
184                }
185                Err(err) => {
186                    self.close(Err(err.into()));
187                    Poll::Ready(Ok(StreamResult::Dropped))
188                }
189            };
190        }
191
192        // Lazily spawn a read task if one hasn't already been spawned yet.
193        let me = &mut *self;
194        let task = me.task.get_or_insert_with(|| {
195            let mut buf = dst.take_buffer();
196            buf.resize(DEFAULT_BUFFER_CAPACITY, 0);
197            let file = Arc::clone(me.file.as_file());
198            let offset = me.offset;
199            spawn_blocking(move || {
200                sys::read_at_cursor_unspecified(&file, &mut buf, offset).map(|n| {
201                    buf.truncate(n);
202                    buf
203                })
204            })
205        });
206
207        // Await the completion of the read task. Note that this is not a
208        // cancellable await point because we can't cancel the other task, so
209        // the `finish` parameter is ignored.
210        let result = match Pin::new(&mut *task).poll(cx) {
211            // If cancellation is requested, then flag that to Tokio. Note that
212            // this still waits for the actual completion of the spawned task,
213            // which won't actually happen if it's already executing.
214            Poll::Pending if finish => {
215                task.abort();
216                ready!(Pin::new(task).poll(cx))
217            }
218            other => ready!(other),
219        };
220        self.task = None;
221        match result {
222            Ok(Ok(buf)) if buf.is_empty() => {
223                self.close(Ok(()));
224                Poll::Ready(Ok(StreamResult::Dropped))
225            }
226            Ok(Ok(buf)) => {
227                let n = buf.len();
228                dst.set_buffer(buf);
229                Poll::Ready(Ok(self.complete_read(n)))
230            }
231            Ok(Err(err)) => {
232                self.close(Err(err.into()));
233                Poll::Ready(Ok(StreamResult::Dropped))
234            }
235            Err(err) => {
236                if err.is_cancelled() {
237                    return Poll::Ready(Ok(StreamResult::Cancelled));
238                }
239                panic!("I/O task should not panic: {err}")
240            }
241        }
242    }
243}
244
245fn map_dir_entry(
246    entry: std::io::Result<crate::filesystem::primitives::DirEntry>,
247) -> Result<Option<DirectoryEntry>, ErrorCode> {
248    match entry {
249        Ok(entry) => {
250            let meta = entry.metadata()?;
251            let Ok(name) = entry.file_name().into_string() else {
252                return Err(ErrorCode::IllegalByteSequence);
253            };
254            Ok(Some(DirectoryEntry {
255                type_: meta.file_type().into(),
256                name,
257            }))
258        }
259        Err(err) => {
260            // On windows, filter out files like `C:\DumpStack.log.tmp` which we
261            // can't get full metadata for.
262            #[cfg(windows)]
263            {
264                use windows_sys::Win32::Foundation::{
265                    ERROR_ACCESS_DENIED, ERROR_SHARING_VIOLATION,
266                };
267                if err.raw_os_error() == Some(ERROR_SHARING_VIOLATION as i32)
268                    || err.raw_os_error() == Some(ERROR_ACCESS_DENIED as i32)
269                {
270                    return Ok(None);
271                }
272            }
273            Err(err.into())
274        }
275    }
276}
277
278struct ReadDirStream {
279    rx: mpsc::Receiver<DirectoryEntry>,
280    task: JoinHandle<Result<(), ErrorCode>>,
281    result: Option<oneshot::Sender<Result<(), ErrorCode>>>,
282}
283
284impl ReadDirStream {
285    fn new(
286        dir: Arc<std::fs::File>,
287        result: oneshot::Sender<Result<(), ErrorCode>>,
288    ) -> ReadDirStream {
289        let (tx, rx) = mpsc::channel(1);
290        ReadDirStream {
291            task: spawn_blocking(move || {
292                let entries = crate::filesystem::primitives::read_base_dir(&dir)?;
293                for entry in entries {
294                    if let Some(entry) = map_dir_entry(entry)? {
295                        if let Err(_) = tx.blocking_send(entry) {
296                            break;
297                        }
298                    }
299                }
300                Ok(())
301            }),
302            rx,
303            result: Some(result),
304        }
305    }
306
307    fn close(&mut self, res: Result<(), ErrorCode>) {
308        self.rx.close();
309        self.task.abort();
310        let _ = self.result.take().unwrap().send(res);
311    }
312}
313
314impl<D> StreamProducer<D> for ReadDirStream {
315    type Item = DirectoryEntry;
316    type Buffer = Option<DirectoryEntry>;
317
318    fn poll_produce<'a>(
319        mut self: Pin<&mut Self>,
320        cx: &mut Context<'_>,
321        mut store: StoreContextMut<'a, D>,
322        mut dst: Destination<'a, Self::Item, Self::Buffer>,
323        finish: bool,
324    ) -> Poll<wasmtime::Result<StreamResult>> {
325        // If this is a 0-length read then `mpsc::Receiver` does not expose an
326        // API to wait for an item to be available without taking it out of the
327        // channel. In lieu of that just say that we're complete and ready for a
328        // read.
329        if dst.remaining(&mut store) == Some(0) {
330            return Poll::Ready(Ok(StreamResult::Completed));
331        }
332
333        match self.rx.poll_recv(cx) {
334            // If an item is on the channel then send that along and say that
335            // the read is now complete with one item being yielded.
336            Poll::Ready(Some(item)) => {
337                dst.set_buffer(Some(item));
338                Poll::Ready(Ok(StreamResult::Completed))
339            }
340
341            // If there's nothing left on the channel then that means that an
342            // error occurred or the iterator is done. In both cases an
343            // un-cancellable wait for the spawned task is entered and we await
344            // its completion. Upon completion there our own stream is closed
345            // with the result (sending an error code on our oneshot) and then
346            // the stream is reported as dropped.
347            Poll::Ready(None) => {
348                let result = ready!(Pin::new(&mut self.task).poll(cx))
349                    .expect("spawned task should not panic");
350                self.close(result);
351                Poll::Ready(Ok(StreamResult::Dropped))
352            }
353
354            // If an item isn't ready yet then cancel this outstanding request
355            // if `finish` is set, otherwise propagate the `Pending` status.
356            Poll::Pending if finish => Poll::Ready(Ok(StreamResult::Cancelled)),
357            Poll::Pending => Poll::Pending,
358        }
359    }
360}
361
362impl Drop for ReadDirStream {
363    fn drop(&mut self) {
364        if self.result.is_some() {
365            self.close(Ok(()));
366        }
367    }
368}
369
370struct WriteStreamConsumer {
371    file: File,
372    location: WriteLocation,
373    result: Option<oneshot::Sender<Result<(), ErrorCode>>>,
374    buffer: BytesMut,
375    task: Option<JoinHandle<std::io::Result<(BytesMut, usize)>>>,
376}
377
378#[derive(Copy, Clone)]
379enum WriteLocation {
380    End,
381    Offset(u64),
382}
383
384impl WriteStreamConsumer {
385    fn new_at(file: File, offset: u64, result: oneshot::Sender<Result<(), ErrorCode>>) -> Self {
386        Self {
387            file,
388            location: WriteLocation::Offset(offset),
389            result: Some(result),
390            buffer: BytesMut::default(),
391            task: None,
392        }
393    }
394
395    fn new_append(file: File, result: oneshot::Sender<Result<(), ErrorCode>>) -> Self {
396        Self {
397            file,
398            location: WriteLocation::End,
399            result: Some(result),
400            buffer: BytesMut::default(),
401            task: None,
402        }
403    }
404
405    fn close(&mut self, res: Result<(), ErrorCode>) {
406        _ = self.result.take().unwrap().send(res);
407    }
408
409    /// Update the internal `offset` field after writing `amt` bytes from the file.
410    fn complete_write(&mut self, amt: usize) -> StreamResult {
411        match &mut self.location {
412            WriteLocation::End => StreamResult::Completed,
413            WriteLocation::Offset(offset) => {
414                let Ok(amt) = amt.try_into() else {
415                    self.close(Err(ErrorCode::Overflow));
416                    return StreamResult::Dropped;
417                };
418                let Some(amt) = offset.checked_add(amt) else {
419                    self.close(Err(ErrorCode::Overflow));
420                    return StreamResult::Dropped;
421                };
422                *offset = amt;
423                StreamResult::Completed
424            }
425        }
426    }
427}
428
429impl WriteLocation {
430    fn write(&self, file: &std::fs::File, bytes: &[u8]) -> io::Result<usize> {
431        match *self {
432            WriteLocation::End => sys::append_cursor_unspecified(file, bytes),
433            WriteLocation::Offset(at) => sys::write_at_cursor_unspecified(file, bytes, at),
434        }
435    }
436}
437
438impl<D> StreamConsumer<D> for WriteStreamConsumer {
439    type Item = u8;
440
441    fn poll_consume(
442        mut self: Pin<&mut Self>,
443        cx: &mut Context<'_>,
444        store: StoreContextMut<D>,
445        src: Source<Self::Item>,
446        finish: bool,
447    ) -> Poll<wasmtime::Result<StreamResult>> {
448        let mut src = src.as_direct(store);
449        if let Some(file) = self.file.as_blocking_file() {
450            // Once a blocking file, always a blocking file, so assert as such.
451            assert!(self.task.is_none());
452            return match self.location.write(file, src.remaining()) {
453                Ok(n) => {
454                    src.mark_read(n);
455                    Poll::Ready(Ok(self.complete_write(n)))
456                }
457                Err(err) => {
458                    self.close(Err(err.into()));
459                    Poll::Ready(Ok(StreamResult::Dropped))
460                }
461            };
462        }
463        let me = &mut *self;
464        let task = me.task.get_or_insert_with(|| {
465            debug_assert!(me.buffer.is_empty());
466            let remaining = src.remaining();
467            let n = remaining.len().min(DEFAULT_BUFFER_CAPACITY);
468            me.buffer.extend_from_slice(&remaining[..n]);
469            let buf = mem::take(&mut me.buffer);
470            let file = Arc::clone(me.file.as_file());
471            let location = me.location;
472            spawn_blocking(move || location.write(&file, &buf).map(|n| (buf, n)))
473        });
474        let result = match Pin::new(&mut *task).poll(cx) {
475            // If cancellation is requested, then flag that to Tokio. Note that
476            // this still waits for the actual completion of the spawned task,
477            // which won't actually happen if it's already executing.
478            Poll::Pending if finish => {
479                task.abort();
480                ready!(Pin::new(task).poll(cx))
481            }
482            other => ready!(other),
483        };
484        self.task = None;
485        match result {
486            Ok(Ok((buf, n))) => {
487                src.mark_read(n);
488                self.buffer = buf;
489                self.buffer.clear();
490                Poll::Ready(Ok(self.complete_write(n)))
491            }
492            Ok(Err(err)) => {
493                self.close(Err(err.into()));
494                Poll::Ready(Ok(StreamResult::Dropped))
495            }
496            Err(err) => {
497                if err.is_cancelled() {
498                    return Poll::Ready(Ok(StreamResult::Cancelled));
499                }
500                panic!("I/O task should not panic: {err}")
501            }
502        }
503    }
504}
505
506impl Drop for WriteStreamConsumer {
507    fn drop(&mut self) {
508        if self.result.is_some() {
509            self.close(Ok(()))
510        }
511    }
512}
513
514impl types::Host for WasiFilesystemCtxView<'_> {
515    fn convert_error_code(&mut self, error: FilesystemError) -> wasmtime::Result<ErrorCode> {
516        error.downcast()
517    }
518}
519
520impl<U> types::HostDescriptorWithStore<U> for WasiFilesystem {
521    fn read_via_stream(
522        mut store: Access<U, Self>,
523        fd: Resource<Descriptor>,
524        offset: Filesize,
525    ) -> wasmtime::Result<(StreamReader<u8>, FutureReader<Result<(), ErrorCode>>)> {
526        let file = match get_descriptor(store.get().table, &fd)? {
527            Descriptor::File(file) => file.clone(),
528            Descriptor::Dir(_) => {
529                return Ok((
530                    StreamReader::new(&mut store, iter::empty())?,
531                    FutureReader::new(&mut store, async move {
532                        wasmtime::error::Ok(Err(ErrorCode::IsDirectory))
533                    })?,
534                ));
535            }
536        };
537        let (result_tx, result_rx) = oneshot::channel();
538        Ok((
539            StreamReader::new(
540                &mut store,
541                ReadStreamProducer {
542                    file,
543                    offset,
544                    result: Some(result_tx),
545                    task: None,
546                },
547            )?,
548            FutureReader::new(&mut store, result_rx)?,
549        ))
550    }
551
552    fn write_via_stream(
553        mut store: Access<'_, U, Self>,
554        fd: Resource<Descriptor>,
555        mut data: StreamReader<u8>,
556        offset: Filesize,
557    ) -> wasmtime::Result<FutureReader<Result<(), ErrorCode>>> {
558        let (result_tx, result_rx) = oneshot::channel();
559        match get_file(store.get().table, &fd).and_then(|file| {
560            if file.perms.write_not_permitted() {
561                Err(ErrorCode::NotPermitted.into())
562            } else {
563                Ok(file.clone())
564            }
565        }) {
566            Ok(file) => {
567                data.pipe(
568                    &mut store,
569                    WriteStreamConsumer::new_at(file, offset, result_tx),
570                )?;
571            }
572            Err(err) => {
573                data.close(&mut store)?;
574                let _ = result_tx.send(Err(err.downcast().unwrap_or(ErrorCode::Io)));
575            }
576        }
577        FutureReader::new(&mut store, result_rx)
578    }
579
580    fn append_via_stream(
581        mut store: Access<'_, U, Self>,
582        fd: Resource<Descriptor>,
583        mut data: StreamReader<u8>,
584    ) -> wasmtime::Result<FutureReader<Result<(), ErrorCode>>> {
585        let (result_tx, result_rx) = oneshot::channel();
586        match get_file(store.get().table, &fd).and_then(|file| {
587            if file.perms.write_not_permitted() {
588                Err(ErrorCode::NotPermitted.into())
589            } else {
590                Ok(file.clone())
591            }
592        }) {
593            Ok(file) => {
594                data.pipe(&mut store, WriteStreamConsumer::new_append(file, result_tx))?;
595            }
596            Err(err) => {
597                data.close(&mut store)?;
598                let _ = result_tx.send(Err(err.downcast().unwrap_or(ErrorCode::Io)));
599            }
600        }
601        FutureReader::new(&mut store, result_rx)
602    }
603
604    async fn advise(
605        store: &Accessor<U, Self>,
606        fd: Resource<Descriptor>,
607        offset: Filesize,
608        length: Filesize,
609        advice: Advice,
610    ) -> FilesystemResult<()> {
611        let file = store.get_file(&fd)?;
612        file.advise(offset, length, advice.into()).await?;
613        Ok(())
614    }
615
616    async fn sync_data(
617        store: &Accessor<U, Self>,
618        fd: Resource<Descriptor>,
619    ) -> FilesystemResult<()> {
620        let fd = store.get_descriptor(&fd)?;
621        fd.sync_data().await?;
622        Ok(())
623    }
624
625    async fn get_flags(
626        store: &Accessor<U, Self>,
627        fd: Resource<Descriptor>,
628    ) -> FilesystemResult<DescriptorFlags> {
629        let fd = store.get_descriptor(&fd)?;
630        let flags = fd.get_flags().await?;
631        Ok(flags.into())
632    }
633
634    async fn get_type(
635        store: &Accessor<U, Self>,
636        fd: Resource<Descriptor>,
637    ) -> FilesystemResult<DescriptorType> {
638        let fd = store.get_descriptor(&fd)?;
639        let ty = fd.get_type().await?;
640        Ok(ty.into())
641    }
642
643    async fn set_size(
644        store: &Accessor<U, Self>,
645        fd: Resource<Descriptor>,
646        size: Filesize,
647    ) -> FilesystemResult<()> {
648        let file = store.get_file(&fd)?;
649        file.set_size(size).await?;
650        Ok(())
651    }
652
653    async fn set_times(
654        store: &Accessor<U, Self>,
655        fd: Resource<Descriptor>,
656        data_access_timestamp: NewTimestamp,
657        data_modification_timestamp: NewTimestamp,
658    ) -> FilesystemResult<()> {
659        let fd = store.get_descriptor(&fd)?;
660        let atim = systemtimespec_from(data_access_timestamp)?;
661        let mtim = systemtimespec_from(data_modification_timestamp)?;
662        fd.set_times(atim, mtim).await?;
663        Ok(())
664    }
665
666    fn read_directory(
667        mut store: Access<'_, U, Self>,
668        fd: Resource<Descriptor>,
669    ) -> wasmtime::Result<(
670        StreamReader<DirectoryEntry>,
671        FutureReader<Result<(), ErrorCode>>,
672    )> {
673        let (result_tx, result_rx) = oneshot::channel();
674        let stream = match get_dir(store.get().table, &fd) {
675            Ok(dir) => {
676                let allow_blocking_current_thread = dir.allow_blocking_current_thread;
677                let dir = Arc::clone(dir.as_dir());
678                if allow_blocking_current_thread {
679                    match crate::filesystem::primitives::read_base_dir(&dir) {
680                        Ok(readdir) => StreamReader::new(
681                            &mut store,
682                            FallibleIteratorProducer::new(
683                                readdir.filter_map(|e| map_dir_entry(e).transpose()),
684                                result_tx,
685                            ),
686                        )?,
687                        Err(e) => {
688                            let _ = result_tx.send(Err(e.into()));
689                            StreamReader::new(&mut store, iter::empty())?
690                        }
691                    }
692                } else {
693                    StreamReader::new(&mut store, ReadDirStream::new(dir, result_tx))?
694                }
695            }
696            Err(err) => {
697                let _ = result_tx.send(Err(err.downcast().unwrap_or(ErrorCode::Io)));
698                StreamReader::new(&mut store, iter::empty())?
699            }
700        };
701        Ok((stream, FutureReader::new(&mut store, result_rx)?))
702    }
703
704    async fn sync(store: &Accessor<U, Self>, fd: Resource<Descriptor>) -> FilesystemResult<()> {
705        let fd = store.get_descriptor(&fd)?;
706        fd.sync().await?;
707        Ok(())
708    }
709
710    async fn create_directory_at(
711        store: &Accessor<U, Self>,
712        fd: Resource<Descriptor>,
713        path: String,
714    ) -> FilesystemResult<()> {
715        let dir = store.get_dir(&fd)?;
716        dir.create_directory_at(path).await?;
717        Ok(())
718    }
719
720    async fn stat(
721        store: &Accessor<U, Self>,
722        fd: Resource<Descriptor>,
723    ) -> FilesystemResult<DescriptorStat> {
724        let fd = store.get_descriptor(&fd)?;
725        let stat = fd.stat().await?;
726        Ok(stat.into())
727    }
728
729    async fn stat_at(
730        store: &Accessor<U, Self>,
731        fd: Resource<Descriptor>,
732        path_flags: PathFlags,
733        path: String,
734    ) -> FilesystemResult<DescriptorStat> {
735        let dir = store.get_dir(&fd)?;
736        let stat = dir.stat_at(path_flags.into(), path).await?;
737        Ok(stat.into())
738    }
739
740    async fn set_times_at(
741        store: &Accessor<U, Self>,
742        fd: Resource<Descriptor>,
743        path_flags: PathFlags,
744        path: String,
745        data_access_timestamp: NewTimestamp,
746        data_modification_timestamp: NewTimestamp,
747    ) -> FilesystemResult<()> {
748        let dir = store.get_dir(&fd)?;
749        let atim = systemtimespec_from(data_access_timestamp)?;
750        let mtim = systemtimespec_from(data_modification_timestamp)?;
751        dir.set_times_at(path_flags.into(), path, atim, mtim)
752            .await?;
753        Ok(())
754    }
755
756    async fn link_at(
757        store: &Accessor<U, Self>,
758        fd: Resource<Descriptor>,
759        old_path_flags: PathFlags,
760        old_path: String,
761        new_fd: Resource<Descriptor>,
762        new_path: String,
763    ) -> FilesystemResult<()> {
764        let (old_dir, new_dir) = store.get_dir_pair(&fd, &new_fd)?;
765        old_dir
766            .link_at(old_path_flags.into(), old_path, &new_dir, new_path)
767            .await?;
768        Ok(())
769    }
770
771    async fn open_at(
772        store: &Accessor<U, Self>,
773        fd: Resource<Descriptor>,
774        path_flags: PathFlags,
775        path: String,
776        open_flags: OpenFlags,
777        flags: DescriptorFlags,
778    ) -> FilesystemResult<Resource<Descriptor>> {
779        let (allow_blocking_current_thread, dir) = store.with(|mut store| {
780            let store = store.get();
781            let dir = get_dir(&store.table, &fd)?;
782            FilesystemResult::Ok((store.ctx.allow_blocking_current_thread, dir.clone()))
783        })?;
784        let fd = dir
785            .open_at(
786                path_flags.into(),
787                path,
788                open_flags.into(),
789                flags.into(),
790                allow_blocking_current_thread,
791            )
792            .await?;
793        let fd = store.with(|mut store| store.get().table.push(fd))?;
794        Ok(fd)
795    }
796
797    async fn readlink_at(
798        store: &Accessor<U, Self>,
799        fd: Resource<Descriptor>,
800        path: String,
801    ) -> FilesystemResult<String> {
802        let dir = store.get_dir(&fd)?;
803        let path = dir.readlink_at(path).await?;
804        Ok(path)
805    }
806
807    async fn remove_directory_at(
808        store: &Accessor<U, Self>,
809        fd: Resource<Descriptor>,
810        path: String,
811    ) -> FilesystemResult<()> {
812        let dir = store.get_dir(&fd)?;
813        dir.remove_directory_at(path).await?;
814        Ok(())
815    }
816
817    async fn rename_at(
818        store: &Accessor<U, Self>,
819        fd: Resource<Descriptor>,
820        old_path: String,
821        new_fd: Resource<Descriptor>,
822        new_path: String,
823    ) -> FilesystemResult<()> {
824        let (old_dir, new_dir) = store.get_dir_pair(&fd, &new_fd)?;
825        old_dir.rename_at(old_path, &new_dir, new_path).await?;
826        Ok(())
827    }
828
829    async fn symlink_at(
830        store: &Accessor<U, Self>,
831        fd: Resource<Descriptor>,
832        old_path: String,
833        new_path: String,
834    ) -> FilesystemResult<()> {
835        let dir = store.get_dir(&fd)?;
836        dir.symlink_at(old_path, new_path).await?;
837        Ok(())
838    }
839
840    async fn unlink_file_at(
841        store: &Accessor<U, Self>,
842        fd: Resource<Descriptor>,
843        path: String,
844    ) -> FilesystemResult<()> {
845        let dir = store.get_dir(&fd)?;
846        dir.unlink_file_at(path).await?;
847        Ok(())
848    }
849
850    async fn is_same_object(
851        store: &Accessor<U, Self>,
852        fd: Resource<Descriptor>,
853        other: Resource<Descriptor>,
854    ) -> wasmtime::Result<bool> {
855        let (fd, other) = store.with(|mut store| {
856            let table = store.get().table;
857            let fd = get_descriptor(table, &fd)?.clone();
858            let other = get_descriptor(table, &other)?.clone();
859            wasmtime::error::Ok((fd, other))
860        })?;
861        fd.is_same_object(&other).await
862    }
863
864    async fn metadata_hash(
865        store: &Accessor<U, Self>,
866        fd: Resource<Descriptor>,
867    ) -> FilesystemResult<MetadataHashValue> {
868        let fd = store.get_descriptor(&fd)?;
869        let meta = fd.metadata_hash().await?;
870        Ok(meta.into())
871    }
872
873    async fn metadata_hash_at(
874        store: &Accessor<U, Self>,
875        fd: Resource<Descriptor>,
876        path_flags: PathFlags,
877        path: String,
878    ) -> FilesystemResult<MetadataHashValue> {
879        let dir = store.get_dir(&fd)?;
880        let meta = dir.metadata_hash_at(path_flags.into(), path).await?;
881        Ok(meta.into())
882    }
883}
884
885impl types::HostDescriptor for WasiFilesystemCtxView<'_> {
886    fn drop(&mut self, fd: Resource<Descriptor>) -> wasmtime::Result<()> {
887        self.table
888            .delete(fd)
889            .context("failed to delete descriptor resource from table")?;
890        Ok(())
891    }
892}
893
894impl preopens::Host for WasiFilesystemCtxView<'_> {
895    fn get_directories(&mut self) -> wasmtime::Result<Vec<(Resource<Descriptor>, String)>> {
896        self.get_directories()
897    }
898}
wasmtime_wasi/p3/filesystem/host.rs

wasmtime_wasi/p3/filesystem/
host.rs
