miri/shims/unix/
virtual_socket.rs

1//! This implements "virtual" sockets, that do not correspond to anything on the host system and
2//! are entirely implemented inside Miri.
3//! This is used to implement `socketpair` and `pipe`.
4
5use std::cell::{Cell, OnceCell, RefCell};
6use std::collections::VecDeque;
7use std::io::{self, ErrorKind, Read, Write};
8use std::net::SocketAddr;
9use std::rc::Rc;
10
11use rustc_target::spec::Os;
12
13use super::fd::EvalContextExt as _;
14use crate::concurrency::VClock;
15use crate::concurrency::scheduler::DelayedCallbacks;
16use crate::shims::files::{
17    EvalContextExt as _, FileDescription, FileDescriptionRef, WeakFileDescriptionRef,
18};
19use crate::shims::unix::UnixFileDescription;
20use crate::shims::unix::socket::UnixSocketFileDescription;
21use crate::*;
22
23/// The maximum capacity of the socketpair buffer in bytes.
24/// This number is arbitrary as the value can always
25/// be configured in the real system.
26const MAX_SOCKETPAIR_BUFFER_CAPACITY: u64 = 0x34000;
27
28#[derive(Debug, PartialEq)]
29enum VirtualSocketType {
30    // Either end of the socketpair fd.
31    Socketpair,
32    // Read end of the pipe.
33    PipeRead,
34    // Write end of the pipe.
35    PipeWrite,
36}
37
38/// One end of a pair of connected virtual sockets.
39#[derive(Debug)]
40struct VirtualSocket {
41    /// The buffer we are reading from, or `None` if this is the writing end of a pipe.
42    /// (In that case, the peer FD will be the reading end of that pipe.)
43    readbuf: Option<RefCell<Buffer>>,
44    /// The `VirtualSocket` file descriptor that is our "peer", and that holds the buffer we are
45    /// writing to. This is a weak reference because the other side may be closed before us; all
46    /// future writes will then trigger EPIPE.
47    peer_fd: OnceCell<WeakFileDescriptionRef<VirtualSocket>>,
48    /// Indicates whether the peer has lost data when the file description is closed.
49    /// This flag is set to `true` if the peer's `readbuf` is non-empty at the time
50    /// of closure.
51    peer_lost_data: Cell<bool>,
52    /// A list of thread ids blocked because the buffer was empty.
53    /// Once another thread writes some bytes, these threads will be unblocked.
54    blocked_read_tid: RefCell<Vec<ThreadId>>,
55    /// A list of thread ids blocked because the buffer was full.
56    /// Once another thread reads some bytes, these threads will be unblocked.
57    blocked_write_tid: RefCell<Vec<ThreadId>>,
58    /// Whether this fd is non-blocking or not.
59    is_nonblock: Cell<bool>,
60    /// Differentiate between different virtual socket fd types.
61    fd_type: VirtualSocketType,
62    /// We need to unblock threads and update the peer_fd readiness when we get dropped.
63    delayed_callbacks: Rc<DelayedCallbacks>,
64    /// State for being watched by epoll.
65    watched: ReadinessWatched,
66}
67
68#[derive(Debug)]
69struct Buffer {
70    buf: VecDeque<u8>,
71    clock: VClock,
72}
73
74impl Buffer {
75    fn new() -> Self {
76        Buffer { buf: VecDeque::new(), clock: VClock::default() }
77    }
78}
79
80impl VirtualSocket {
81    fn peer_fd(&self) -> &WeakFileDescriptionRef<VirtualSocket> {
82        self.peer_fd.get().unwrap()
83    }
84}
85
86impl Drop for VirtualSocket {
87    fn drop(&mut self) {
88        // Blocked threads hold a reference, so when we get closed there cannot be blocked threads.
89        assert!(self.blocked_read_tid.get_mut().is_empty());
90        assert!(self.blocked_write_tid.get_mut().is_empty());
91
92        if let Some(peer_fd) = self.peer_fd().upgrade() {
93            // If the current readbuf is non-empty when the file description is closed,
94            // notify the peer that data loss has happened in current file description.
95            if let Some(readbuf) = &self.readbuf {
96                if !readbuf.borrow().buf.is_empty() {
97                    peer_fd.peer_lost_data.set(true);
98                }
99            }
100            // At the next opportunity, update our readiness and unblock the threads of our peer FD.
101            self.delayed_callbacks.add(Box::new(move |ecx| {
102                // Unblock all threads that are currently blocked on peer_fd (reads AND writes).
103                let waiting_readers = std::mem::take(&mut *peer_fd.blocked_read_tid.borrow_mut());
104                let waiting_writers = std::mem::take(&mut *peer_fd.blocked_write_tid.borrow_mut());
105                // FIXME: We can randomize the order of unblocking.
106                for thread_id in waiting_readers.into_iter().chain(waiting_writers) {
107                    ecx.unblock_thread(thread_id, BlockReason::VirtualSocket)?;
108                }
109
110                ecx.update_fd_readiness(peer_fd, ReadinessUpdateFlags::DEFAULT)?;
111
112                interp_ok(())
113            }));
114        }
115    }
116}
117
118impl FileDescription for VirtualSocket {
119    fn name(&self) -> &'static str {
120        match self.fd_type {
121            VirtualSocketType::Socketpair => "socketpair",
122            VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => "pipe",
123        }
124    }
125
126    fn metadata<'tcx>(
127        &self,
128        _communicate_allowed: bool,
129    ) -> InterpResult<'tcx, Either<io::Result<std::fs::Metadata>, &'static str>> {
130        let mode_name = match self.fd_type {
131            VirtualSocketType::Socketpair => "S_IFSOCK",
132            VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => "S_IFIFO",
133        };
134        interp_ok(Either::Right(mode_name))
135    }
136
137    fn read<'tcx>(
138        self: FileDescriptionRef<Self>,
139        _communicate_allowed: bool,
140        ptr: Pointer,
141        len: u64,
142        ecx: &mut MiriInterpCx<'tcx>,
143        finish: DynMachineCallback<'tcx, Result<u64, IoError>>,
144    ) -> InterpResult<'tcx> {
145        ecx.virtual_socket_read(self, ptr, len, /* is_non_block */ false, finish)
146    }
147
148    fn write<'tcx>(
149        self: FileDescriptionRef<Self>,
150        _communicate_allowed: bool,
151        ptr: Pointer,
152        len: u64,
153        ecx: &mut MiriInterpCx<'tcx>,
154        finish: DynMachineCallback<'tcx, Result<u64, IoError>>,
155    ) -> InterpResult<'tcx> {
156        ecx.virtual_socket_write(self, ptr, len, /* is_non_block */ false, finish)
157    }
158
159    fn short_fd_operations(&self) -> bool {
160        // Linux guarantees that when a read/write on a streaming socket comes back short,
161        // the kernel buffer is empty/full:
162        // See <https://man7.org/linux/man-pages/man7/epoll.7.html> in Q&A section.
163        // So we can't do short reads/writes here.
164        false
165    }
166
167    fn as_unix(self: FileDescriptionRef<Self>) -> FileDescriptionRef<dyn UnixFileDescription> {
168        self
169    }
170
171    fn readiness_watched(&self) -> Option<&ReadinessWatched> {
172        Some(&self.watched)
173    }
174
175    fn readiness(&self) -> Readiness {
176        // We only check the "readable", "writable", "read closed" and "write closed" readiness.
177        // If other event flags need to be supported in the future, the check should be added here.
178
179        let mut readiness = Readiness::EMPTY;
180
181        // Check if it is readable.
182        if let Some(readbuf) = &self.readbuf {
183            if !readbuf.borrow().buf.is_empty() {
184                readiness.readable = true;
185            }
186        } else {
187            // Without a read buffer, reading never blocks, so we are always ready.
188            readiness.readable = true;
189        }
190
191        // Check if is writable.
192        if let Some(peer_fd) = self.peer_fd().upgrade() {
193            if let Some(writebuf) = &peer_fd.readbuf {
194                let data_size = writebuf.borrow().buf.len().to_u64();
195                let available_space = MAX_SOCKETPAIR_BUFFER_CAPACITY.strict_sub(data_size);
196                if available_space != 0 {
197                    readiness.writable = true;
198                }
199            } else {
200                // Without a write buffer, writing never blocks.
201                readiness.writable = true;
202            }
203        } else {
204            // Peer FD has been closed. This always sets both the "read closed" and "write closed" flags
205            // as we do not support `shutdown` that could be used to partially close the stream.
206            readiness.read_closed = true;
207            readiness.write_closed = true;
208            // Since the peer is closed, even if no data is available reads will return EOF and
209            // writes will return EPIPE. In other words, they won't block, so we mark this as ready
210            // for read and write.
211            readiness.readable = true;
212            readiness.writable = true;
213            // If there is data lost in peer_fd, set error readiness.
214            if self.peer_lost_data.get() {
215                readiness.error = true;
216            }
217        }
218        readiness
219    }
220}
221
222impl UnixFileDescription for VirtualSocket {
223    fn get_flags<'tcx>(&self, ecx: &mut MiriInterpCx<'tcx>) -> InterpResult<'tcx, Scalar> {
224        let mut flags = 0;
225
226        // Get flag for file access mode.
227        // The flag for both socketpair and pipe will remain the same even when the peer
228        // fd is closed, so we need to look at the original type of this socket, not at whether
229        // the peer socket still exists.
230        match self.fd_type {
231            VirtualSocketType::Socketpair => {
232                flags |= ecx.eval_libc_i32("O_RDWR");
233            }
234            VirtualSocketType::PipeRead => {
235                flags |= ecx.eval_libc_i32("O_RDONLY");
236            }
237            VirtualSocketType::PipeWrite => {
238                flags |= ecx.eval_libc_i32("O_WRONLY");
239            }
240        }
241
242        // Get flag for blocking status.
243        if self.is_nonblock.get() {
244            flags |= ecx.eval_libc_i32("O_NONBLOCK");
245        }
246
247        interp_ok(Scalar::from_i32(flags))
248    }
249
250    fn set_flags<'tcx>(
251        &self,
252        mut flag: i32,
253        ecx: &mut MiriInterpCx<'tcx>,
254    ) -> InterpResult<'tcx, Scalar> {
255        let o_nonblock = ecx.eval_libc_i32("O_NONBLOCK");
256
257        // O_NONBLOCK flag can be set / unset by user.
258        if flag & o_nonblock == o_nonblock {
259            self.is_nonblock.set(true);
260            flag &= !o_nonblock;
261        } else {
262            self.is_nonblock.set(false);
263        }
264
265        // Throw error if there is any unsupported flag.
266        if flag != 0 {
267            throw_unsup_format!(
268                "fcntl: only O_NONBLOCK is supported for F_SETFL on socketpairs and pipes"
269            )
270        }
271
272        interp_ok(Scalar::from_i32(0))
273    }
274
275    fn as_socket(
276        self: FileDescriptionRef<Self>,
277    ) -> Option<FileDescriptionRef<dyn UnixSocketFileDescription>> {
278        match self.fd_type {
279            VirtualSocketType::Socketpair => Some(self),
280            VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => None,
281        }
282    }
283}
284
285impl UnixSocketFileDescription for VirtualSocket {
286    fn send<'tcx>(
287        self: FileDescriptionRef<Self>,
288        _communicate_allowed: bool,
289        ptr: Pointer,
290        len: u64,
291        is_non_block: bool,
292        // For stream-based sockets `send` and `sendto` are equivalent and
293        // we can thus ignore the destination address.
294        _dest_addr: Option<SocketAddr>,
295        ecx: &mut MiriInterpCx<'tcx>,
296        finish: DynMachineCallback<'tcx, Result<u64, IoError>>,
297    ) -> InterpResult<'tcx> {
298        ecx.virtual_socket_write(self, ptr, len, is_non_block, finish)
299    }
300
301    fn recv<'tcx>(
302        self: FileDescriptionRef<Self>,
303        _communicate_allowed: bool,
304        ptr: Pointer,
305        len: u64,
306        is_peek: bool,
307        is_non_block: bool,
308        ecx: &mut MiriInterpCx<'tcx>,
309        finish: DynMachineCallback<'tcx, Result<(u64, Option<SocketAddr>), IoError>>,
310    ) -> InterpResult<'tcx> {
311        if is_peek {
312            throw_unsup_format!("socketpair: virtual sockets don't support peeking")
313        }
314
315        let finish = callback!(
316            @capture<'tcx> {
317                finish: DynMachineCallback<'tcx, Result<(u64, Option<SocketAddr>), IoError>>,
318            } |this, result: Result<u64, IoError>| {
319                // The source address of the data isn't available for stream-based sockets.
320                finish.call(this, result.map(|bytes| (bytes, None)))
321            }
322        );
323        ecx.virtual_socket_read(self, ptr, len, is_non_block, finish)
324    }
325}
326
327impl<'tcx> EvalContextPrivExt<'tcx> for crate::MiriInterpCx<'tcx> {}
328trait EvalContextPrivExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
329    /// Attempt two write `len` bytes from the buffer pointed to by `ptr` into the
330    /// virtual socket `socket`.
331    /// `is_non_block` specifies whether the operation should be performed as if the
332    /// socket was non-blocking.
333    /// After a successful write, `finish` is called with the amount of bytes written.
334    fn virtual_socket_write(
335        &mut self,
336        socket: FileDescriptionRef<VirtualSocket>,
337        ptr: Pointer,
338        len: u64,
339        is_non_block: bool,
340        finish: DynMachineCallback<'tcx, Result<u64, IoError>>,
341    ) -> InterpResult<'tcx> {
342        let this = self.eval_context_mut();
343
344        // Always succeed on write size 0.
345        // ("If count is zero and fd refers to a file other than a regular file, the results are not specified.")
346        if len == 0 {
347            return finish.call(this, Ok(0));
348        }
349
350        // We are writing to our peer's readbuf.
351        let Some(peer_fd) = socket.peer_fd().upgrade() else {
352            // If the upgrade from Weak to Rc fails, it indicates that all read ends have been
353            // closed. It is an error to write even if there would be space.
354            return finish.call(this, Err(ErrorKind::BrokenPipe.into()));
355        };
356
357        let Some(writebuf) = &peer_fd.readbuf else {
358            // Writing to the read end of a pipe.
359            return finish.call(this, Err(IoError::LibcError("EBADF")));
360        };
361
362        // Let's see if we can write.
363        let available_space =
364            MAX_SOCKETPAIR_BUFFER_CAPACITY.strict_sub(writebuf.borrow().buf.len().to_u64());
365        if available_space == 0 {
366            if socket.is_nonblock.get() || is_non_block {
367                // Non-blocking socketpair with a full buffer.
368                return finish.call(this, Err(ErrorKind::WouldBlock.into()));
369            } else {
370                socket.blocked_write_tid.borrow_mut().push(this.active_thread());
371                this.block_thread(
372                    BlockReason::VirtualSocket,
373                    None,
374                    callback!(
375                        @capture<'tcx> {
376                            socket: FileDescriptionRef<VirtualSocket>,
377                            ptr: Pointer,
378                            len: u64,
379                            is_non_block: bool,
380                            finish: DynMachineCallback<'tcx, Result<u64, IoError>>,
381                        }
382                        |ecx, unblock: UnblockKind| {
383                            assert_eq!(unblock, UnblockKind::Ready);
384                            ecx.virtual_socket_write(socket, ptr, len, is_non_block, finish)
385                        }
386                    ),
387                );
388            }
389        } else {
390            // There is space to write!
391            let mut writebuf = writebuf.borrow_mut();
392            // Remember this clock so `read` can synchronize with us.
393            this.release_clock(|clock| {
394                writebuf.clock.join(clock);
395            })?;
396            // Do full write / partial write based on the space available.
397            let write_size = len.min(available_space);
398            let actual_write_size =
399                this.write_to_host(ptr, write_size, |buf| writebuf.buf.write(buf))?.unwrap();
400            assert_eq!(actual_write_size, write_size);
401
402            // Need to stop accessing peer_fd so that it can be notified.
403            drop(writebuf);
404
405            // Unblock all threads that are currently blocked on peer_fd's read.
406            let waiting_threads = std::mem::take(&mut *peer_fd.blocked_read_tid.borrow_mut());
407            // FIXME: We can randomize the order of unblocking.
408            for thread_id in waiting_threads {
409                this.unblock_thread(thread_id, BlockReason::VirtualSocket)?;
410            }
411            // Notify readiness watchers: we might be no longer writable, peer might now be readable.
412            // The notification to the peer seems to be always sent on Linux, even if the
413            // FD was readable before.
414            this.update_fd_readiness(socket, ReadinessUpdateFlags::DEFAULT)?;
415            this.update_fd_readiness(peer_fd, ReadinessUpdateFlags::FORCE_EDGE)?;
416
417            return finish.call(this, Ok(write_size));
418        }
419        interp_ok(())
420    }
421
422    /// Attempt to read `len` bytes from the virtual socket `socket` into the buffer
423    /// pointed to by `ptr`.
424    /// `is_non_block` specifies whether the operation should be performed as if the
425    /// socket was non-blocking.
426    /// After a successful read, `finish` is called with the amount of bytes read.
427    fn virtual_socket_read(
428        &mut self,
429        socket: FileDescriptionRef<VirtualSocket>,
430        ptr: Pointer,
431        len: u64,
432        is_non_block: bool,
433        finish: DynMachineCallback<'tcx, Result<u64, IoError>>,
434    ) -> InterpResult<'tcx> {
435        let this = self.eval_context_mut();
436
437        // Always succeed on read size 0.
438        if len == 0 {
439            return finish.call(this, Ok(0));
440        }
441
442        let Some(readbuf) = &socket.readbuf else {
443            // FIXME: This should return EBADF, but there's no nice way to do that as there's no
444            // corresponding ErrorKind variant.
445            throw_unsup_format!("reading from the write end of a pipe")
446        };
447
448        if readbuf.borrow_mut().buf.is_empty() {
449            if socket.peer_fd().upgrade().is_none() {
450                // Socketpair with no peer and empty buffer.
451                // 0 bytes successfully read indicates end-of-file.
452                return finish.call(this, Ok(0));
453            } else if socket.is_nonblock.get() || is_non_block {
454                // Non-blocking socketpair with writer and empty buffer.
455                // https://linux.die.net/man/2/read
456                // EAGAIN or EWOULDBLOCK can be returned for socket,
457                // POSIX.1-2001 allows either error to be returned for this case.
458                // Since there is no ErrorKind for EAGAIN, WouldBlock is used.
459                return finish.call(this, Err(ErrorKind::WouldBlock.into()));
460            } else {
461                socket.blocked_read_tid.borrow_mut().push(this.active_thread());
462                this.block_thread(
463                    BlockReason::VirtualSocket,
464                    None,
465                    callback!(
466                        @capture<'tcx> {
467                            socket: FileDescriptionRef<VirtualSocket>,
468                            ptr: Pointer,
469                            len: u64,
470                            is_non_block: bool,
471                            finish: DynMachineCallback<'tcx, Result<u64, IoError>>,
472                        }
473                        |ecx, unblock: UnblockKind| {
474                            assert_eq!(unblock, UnblockKind::Ready);
475                            ecx.virtual_socket_read(socket, ptr, len, is_non_block, finish)
476                        }
477                    ),
478                );
479            }
480        } else {
481            // There's data to be read!
482            let mut readbuf = readbuf.borrow_mut();
483            // Synchronize with all previous writes to this buffer.
484            // FIXME: this over-synchronizes; a more precise approach would be to
485            // only sync with the writes whose data we will read.
486            this.acquire_clock(&readbuf.clock)?;
487
488            // Do full read / partial read based on the space available.
489            // Conveniently, `read` exists on `VecDeque` and has exactly the desired behavior.
490            let read_size = this.read_from_host(ptr, len, |buf| readbuf.buf.read(buf))?.unwrap();
491            let readbuf_now_empty = readbuf.buf.is_empty();
492
493            // Need to drop before others can access the readbuf again.
494            drop(readbuf);
495
496            // A notification should be provided for the peer file description even when it can
497            // only write 1 byte. This implementation is not compliant with the actual Linux kernel
498            // implementation. For optimization reasons, the kernel will only mark the file description
499            // as "writable" when it can write more than a certain number of bytes. Since we
500            // don't know what that *certain number* is, we will provide a notification every time
501            // a read is successful. This might result in our readiness emulation providing more
502            // events than the real system.
503            if let Some(peer_fd) = socket.peer_fd().upgrade() {
504                // Unblock all threads that are currently blocked on peer_fd's write.
505                let waiting_threads = std::mem::take(&mut *peer_fd.blocked_write_tid.borrow_mut());
506                // FIXME: We can randomize the order of unblocking.
507                for thread_id in waiting_threads {
508                    this.unblock_thread(thread_id, BlockReason::VirtualSocket)?;
509                }
510                // Notify readiness watchers: peer is now writable.
511                // Linux seems to always notify the peer if the read buffer is now empty.
512                // (Linux also does that if this was a "big" read, but to avoid some arbitrary
513                // threshold, we do not match that.)
514                this.update_fd_readiness(
515                    peer_fd,
516                    if readbuf_now_empty {
517                        ReadinessUpdateFlags::FORCE_EDGE
518                    } else {
519                        ReadinessUpdateFlags::DEFAULT
520                    },
521                )?;
522            };
523            // Notify readiness watchers: we might be no longer readable.
524            this.update_fd_readiness(socket, ReadinessUpdateFlags::DEFAULT)?;
525
526            return finish.call(this, Ok(read_size));
527        }
528        interp_ok(())
529    }
530}
531
532impl<'tcx> EvalContextExt<'tcx> for crate::MiriInterpCx<'tcx> {}
533pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
534    /// For more information on the arguments see the socketpair manpage:
535    /// <https://linux.die.net/man/2/socketpair>
536    fn socketpair(
537        &mut self,
538        domain: &OpTy<'tcx>,
539        type_: &OpTy<'tcx>,
540        protocol: &OpTy<'tcx>,
541        sv: &OpTy<'tcx>,
542    ) -> InterpResult<'tcx, Scalar> {
543        let this = self.eval_context_mut();
544
545        let domain = this.read_scalar(domain)?.to_i32()?;
546        let mut flags = this.read_scalar(type_)?.to_i32()?;
547        let protocol = this.read_scalar(protocol)?.to_i32()?;
548        // This is really a pointer to `[i32; 2]` but we use a ptr-to-first-element representation.
549        let sv = this.deref_pointer(sv)?;
550
551        let mut is_sock_nonblock = false;
552
553        // Interpret the flag. Every flag we recognize is "subtracted" from `flags`, so
554        // if there is anything left at the end, that's an unsupported flag.
555        if matches!(
556            this.tcx.sess.target.os,
557            Os::Linux | Os::Android | Os::FreeBsd | Os::Solaris | Os::Illumos
558        ) {
559            // SOCK_NONBLOCK and SOCK_CLOEXEC only exist on Linux, Android, FreeBSD,
560            // Solaris, and Illumos targets.
561            let sock_nonblock = this.eval_libc_i32("SOCK_NONBLOCK");
562            let sock_cloexec = this.eval_libc_i32("SOCK_CLOEXEC");
563            if flags & sock_nonblock == sock_nonblock {
564                is_sock_nonblock = true;
565                flags &= !sock_nonblock;
566            }
567            if flags & sock_cloexec == sock_cloexec {
568                flags &= !sock_cloexec;
569            }
570        }
571
572        // Fail on unsupported input.
573        // AF_UNIX and AF_LOCAL are synonyms, so we accept both in case
574        // their values differ.
575        if domain != this.eval_libc_i32("AF_UNIX") && domain != this.eval_libc_i32("AF_LOCAL") {
576            throw_unsup_format!(
577                "socketpair: domain {:#x} is unsupported, only AF_UNIX \
578                                 and AF_LOCAL are allowed",
579                domain
580            );
581        } else if flags != this.eval_libc_i32("SOCK_STREAM") {
582            throw_unsup_format!(
583                "socketpair: type {:#x} is unsupported, only SOCK_STREAM, \
584                                 SOCK_CLOEXEC and SOCK_NONBLOCK are allowed",
585                flags
586            );
587        } else if protocol != 0 {
588            throw_unsup_format!(
589                "socketpair: socket protocol {protocol} is unsupported, \
590                                 only 0 is allowed",
591            );
592        }
593
594        // Generate file descriptions.
595        let fds = &mut this.machine.fds;
596        let fd0 = fds.new_ref(VirtualSocket {
597            readbuf: Some(RefCell::new(Buffer::new())),
598            peer_fd: OnceCell::new(),
599            peer_lost_data: Cell::new(false),
600            blocked_read_tid: RefCell::new(Vec::new()),
601            blocked_write_tid: RefCell::new(Vec::new()),
602            is_nonblock: Cell::new(is_sock_nonblock),
603            fd_type: VirtualSocketType::Socketpair,
604            delayed_callbacks: Rc::clone(&this.machine.delayed_callbacks),
605            watched: ReadinessWatched::default(),
606        });
607        let fd1 = fds.new_ref(VirtualSocket {
608            readbuf: Some(RefCell::new(Buffer::new())),
609            peer_fd: OnceCell::new(),
610            peer_lost_data: Cell::new(false),
611            blocked_read_tid: RefCell::new(Vec::new()),
612            blocked_write_tid: RefCell::new(Vec::new()),
613            is_nonblock: Cell::new(is_sock_nonblock),
614            fd_type: VirtualSocketType::Socketpair,
615            delayed_callbacks: Rc::clone(&this.machine.delayed_callbacks),
616            watched: ReadinessWatched::default(),
617        });
618
619        // Make the file descriptions point to each other.
620        fd0.peer_fd.set(FileDescriptionRef::downgrade(&fd1)).unwrap();
621        fd1.peer_fd.set(FileDescriptionRef::downgrade(&fd0)).unwrap();
622
623        // Insert the file description to the fd table, generating the file descriptors.
624        let Some(sv0) = fds.insert(fd0) else {
625            return this.set_errno_and_return_neg1_i32(LibcError("EMFILE"));
626        };
627        let Some(sv1) = fds.insert(fd1) else {
628            this.close(sv0)?; // don't leak the first FD
629            return this.set_errno_and_return_neg1_i32(LibcError("EMFILE"));
630        };
631
632        // Return socketpair file descriptors to the caller.
633        let sv0 = Scalar::from_int(sv0, sv.layout.size);
634        let sv1 = Scalar::from_int(sv1, sv.layout.size);
635        this.write_scalar(sv0, &sv)?;
636        this.write_scalar(sv1, &this.project_c_array_field(&sv, 1)?)?;
637
638        interp_ok(Scalar::from_i32(0))
639    }
640
641    fn pipe2(
642        &mut self,
643        pipefd: &OpTy<'tcx>,
644        flags: Option<&OpTy<'tcx>>,
645    ) -> InterpResult<'tcx, Scalar> {
646        let this = self.eval_context_mut();
647
648        let pipefd = this.deref_pointer(pipefd)?;
649        let mut flags = match flags {
650            Some(flags) => this.read_scalar(flags)?.to_i32()?,
651            None => 0,
652        };
653
654        let cloexec = this.eval_libc_i32("O_CLOEXEC");
655        let o_nonblock = this.eval_libc_i32("O_NONBLOCK");
656
657        // Interpret the flag. Every flag we recognize is "subtracted" from `flags`, so
658        // if there is anything left at the end, that's an unsupported flag.
659        let mut is_nonblock = false;
660        if flags & o_nonblock == o_nonblock {
661            is_nonblock = true;
662            flags &= !o_nonblock;
663        }
664        // As usual we ignore CLOEXEC.
665        if flags & cloexec == cloexec {
666            flags &= !cloexec;
667        }
668        if flags != 0 {
669            throw_unsup_format!("unsupported flags in `pipe2`");
670        }
671
672        // Generate file descriptions.
673        // pipefd[0] refers to the read end of the pipe.
674        let fds = &mut this.machine.fds;
675        let fd0 = fds.new_ref(VirtualSocket {
676            readbuf: Some(RefCell::new(Buffer::new())),
677            peer_fd: OnceCell::new(),
678            peer_lost_data: Cell::new(false),
679            blocked_read_tid: RefCell::new(Vec::new()),
680            blocked_write_tid: RefCell::new(Vec::new()),
681            is_nonblock: Cell::new(is_nonblock),
682            fd_type: VirtualSocketType::PipeRead,
683            delayed_callbacks: Rc::clone(&this.machine.delayed_callbacks),
684            watched: ReadinessWatched::default(),
685        });
686        let fd1 = fds.new_ref(VirtualSocket {
687            readbuf: None,
688            peer_fd: OnceCell::new(),
689            peer_lost_data: Cell::new(false),
690            blocked_read_tid: RefCell::new(Vec::new()),
691            blocked_write_tid: RefCell::new(Vec::new()),
692            is_nonblock: Cell::new(is_nonblock),
693            fd_type: VirtualSocketType::PipeWrite,
694            delayed_callbacks: Rc::clone(&this.machine.delayed_callbacks),
695            watched: ReadinessWatched::default(),
696        });
697
698        // Make the file descriptions point to each other.
699        fd0.peer_fd.set(FileDescriptionRef::downgrade(&fd1)).unwrap();
700        fd1.peer_fd.set(FileDescriptionRef::downgrade(&fd0)).unwrap();
701
702        // Insert the file description to the fd table, generating the file descriptors.
703        let Some(pipefd0) = fds.insert(fd0) else {
704            return this.set_errno_and_return_neg1_i32(LibcError("EMFILE"));
705        };
706        let Some(pipefd1) = fds.insert(fd1) else {
707            this.close(pipefd0)?; // don't leak the first FD
708            return this.set_errno_and_return_neg1_i32(LibcError("EMFILE"));
709        };
710
711        // Return file descriptors to the caller.
712        let pipefd0 = Scalar::from_int(pipefd0, pipefd.layout.size);
713        let pipefd1 = Scalar::from_int(pipefd1, pipefd.layout.size);
714        this.write_scalar(pipefd0, &pipefd)?;
715        this.write_scalar(pipefd1, &this.project_c_array_field(&pipefd, 1)?)?;
716
717        interp_ok(Scalar::from_i32(0))
718    }
719}
miri/shims/unix/virtual_socket.rs

miri/shims/unix/
virtual_socket.rs
