1use std::cell::{Cell, OnceCell, RefCell};
6use std::collections::VecDeque;
7use std::io::{self, ErrorKind, Read};
8use std::rc::Rc;
9
10use rustc_target::spec::Os;
11
12use crate::concurrency::VClock;
13use crate::shims::files::{
14 EvalContextExt as _, FileDescription, FileDescriptionRef, WeakFileDescriptionRef,
15};
16use crate::shims::readiness::DelayedReadinessUpdates;
17use crate::shims::sig::Varargs;
18use crate::shims::unix::UnixFileDescription;
19use crate::shims::unix::socket::UnixSocketFileDescription;
20use crate::*;
21
22const MAX_SOCKETPAIR_BUFFER_CAPACITY: usize = 0x34000;
26
27#[derive(Debug, PartialEq)]
28enum VirtualSocketType {
29 Socketpair,
31 PipeRead,
33 PipeWrite,
35}
36
37#[derive(Debug)]
39struct VirtualSocket {
40 readbuf: Option<RefCell<Buffer>>,
43 peer_fd: OnceCell<WeakFileDescriptionRef<VirtualSocket>>,
47 peer_lost_data: Cell<bool>,
51 blocked_read_tid: RefCell<Vec<ThreadId>>,
54 blocked_write_tid: RefCell<Vec<ThreadId>>,
57 is_nonblock: Cell<bool>,
59 fd_type: VirtualSocketType,
61 delayed_readiness_updates: Rc<DelayedReadinessUpdates>,
64 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 if let Some(peer_fd) = self.peer_fd().upgrade() {
89 if let Some(readbuf) = &self.readbuf {
92 if !readbuf.borrow().buf.is_empty() {
93 peer_fd.peer_lost_data.set(true);
94 }
95 }
96 self.delayed_readiness_updates.add(peer_fd);
98 }
99 }
100}
101
102impl FileDescription for VirtualSocket {
103 fn name(&self) -> &'static str {
104 match self.fd_type {
105 VirtualSocketType::Socketpair => "socketpair",
106 VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => "pipe",
107 }
108 }
109
110 fn metadata<'tcx>(
111 &self,
112 ) -> InterpResult<'tcx, Either<io::Result<std::fs::Metadata>, &'static str>> {
113 let mode_name = match self.fd_type {
114 VirtualSocketType::Socketpair => "S_IFSOCK",
115 VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => "S_IFIFO",
116 };
117 interp_ok(Either::Right(mode_name))
118 }
119
120 fn read<'tcx>(
121 self: FileDescriptionRef<Self>,
122 _communicate_allowed: bool,
123 ptr: Pointer,
124 len: usize,
125 ecx: &mut MiriInterpCx<'tcx>,
126 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
127 ) -> InterpResult<'tcx> {
128 ecx.virtual_socket_read(self, ptr, len, false, finish)
129 }
130
131 fn write<'tcx>(
132 self: FileDescriptionRef<Self>,
133 _communicate_allowed: bool,
134 ptr: Pointer,
135 len: usize,
136 ecx: &mut MiriInterpCx<'tcx>,
137 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
138 ) -> InterpResult<'tcx> {
139 ecx.virtual_socket_write(self, ptr, len, false, finish)
140 }
141
142 fn short_fd_operations(&self) -> bool {
143 false
148 }
149
150 fn as_unix(self: FileDescriptionRef<Self>) -> FileDescriptionRef<dyn UnixFileDescription> {
151 self
152 }
153
154 fn readiness_watched(&self) -> Option<&ReadinessWatched> {
155 Some(&self.watched)
156 }
157
158 fn readiness(&self) -> Readiness {
159 let mut readiness = Readiness::EMPTY;
163
164 if let Some(readbuf) = &self.readbuf {
166 if !readbuf.borrow().buf.is_empty() {
167 readiness.readable = true;
168 }
169 } else {
170 readiness.readable = true;
172 }
173
174 if let Some(peer_fd) = self.peer_fd().upgrade() {
176 if let Some(writebuf) = &peer_fd.readbuf {
177 let data_size = writebuf.borrow().buf.len();
178 let available_space = MAX_SOCKETPAIR_BUFFER_CAPACITY.strict_sub(data_size);
179 if available_space != 0 {
180 readiness.writable = true;
181 }
182 } else {
183 readiness.writable = true;
185 }
186 } else {
187 readiness.read_closed = true;
190 readiness.write_closed = true;
191 readiness.readable = true;
195 readiness.writable = true;
196 if self.peer_lost_data.get() {
198 readiness.error = true;
199 }
200 }
201 readiness
202 }
203}
204
205impl UnixFileDescription for VirtualSocket {
206 fn get_flags<'tcx>(&self, ecx: &mut MiriInterpCx<'tcx>) -> InterpResult<'tcx, Scalar> {
207 let mut flags = 0;
208
209 match self.fd_type {
214 VirtualSocketType::Socketpair => {
215 flags |= ecx.eval_libc_i32("O_RDWR");
216 }
217 VirtualSocketType::PipeRead => {
218 flags |= ecx.eval_libc_i32("O_RDONLY");
219 }
220 VirtualSocketType::PipeWrite => {
221 flags |= ecx.eval_libc_i32("O_WRONLY");
222 }
223 }
224
225 if self.is_nonblock.get() {
227 flags |= ecx.eval_libc_i32("O_NONBLOCK");
228 }
229
230 interp_ok(Scalar::from_i32(flags))
231 }
232
233 fn set_flags<'tcx>(
234 &self,
235 mut flag: i32,
236 ecx: &mut MiriInterpCx<'tcx>,
237 ) -> InterpResult<'tcx, Scalar> {
238 let o_nonblock = ecx.eval_libc_i32("O_NONBLOCK");
239
240 if flag & o_nonblock == o_nonblock {
242 self.is_nonblock.set(true);
243 flag &= !o_nonblock;
244 } else {
245 self.is_nonblock.set(false);
246 }
247
248 if flag != 0 {
250 throw_unsup_format!(
251 "fcntl: only O_NONBLOCK is supported for F_SETFL on socketpairs and pipes"
252 )
253 }
254
255 interp_ok(Scalar::from_i32(0))
256 }
257
258 fn ioctl<'tcx>(
259 &self,
260 op: Scalar,
261 args: Varargs<'tcx, '_>,
262 ecx: &mut MiriInterpCx<'tcx>,
263 ) -> InterpResult<'tcx, i32> {
264 match self.fd_type {
265 VirtualSocketType::Socketpair => { }
266 VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => {
267 throw_unsup_format!("cannot use ioctl on pipe");
271 }
272 }
273
274 let fionbio = ecx.eval_libc("FIONBIO");
275
276 if op == fionbio {
277 if !matches!(ecx.tcx.sess.target.os, Os::Linux | Os::Android | Os::MacOs | Os::FreeBsd)
280 {
281 throw_unsup_format!(
286 "ioctl: setting FIONBIO on sockets is unsupported on target {}",
287 ecx.tcx.sess.target.os
288 );
289 }
290
291 let ([value_ptr], _) = ecx.check_varargs(shim_varargs![*i32], args, "ioctl")?;
292 let value = ecx.deref_pointer_as(value_ptr, ecx.machine.layouts.i32)?;
293 let non_block = ecx.read_scalar(&value)?.to_i32()? != 0;
294 self.is_nonblock.set(non_block);
295 return interp_ok(0);
296 }
297
298 throw_unsup_format!("ioctl: unsupported operation {op:#x} on socket");
299 }
300
301 fn as_socket(
302 self: FileDescriptionRef<Self>,
303 ) -> Option<FileDescriptionRef<dyn UnixSocketFileDescription>> {
304 match self.fd_type {
305 VirtualSocketType::Socketpair => Some(self),
306 VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => None,
307 }
308 }
309}
310
311impl UnixSocketFileDescription for VirtualSocket {
312 fn send<'tcx>(
313 self: FileDescriptionRef<Self>,
314 _communicate_allowed: bool,
315 ptr: Pointer,
316 len: usize,
317 is_non_block: bool,
318 ecx: &mut MiriInterpCx<'tcx>,
319 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
320 ) -> InterpResult<'tcx> {
321 ecx.virtual_socket_write(self, ptr, len, is_non_block, finish)
322 }
323
324 fn recv<'tcx>(
325 self: FileDescriptionRef<Self>,
326 _communicate_allowed: bool,
327 ptr: Pointer,
328 len: usize,
329 is_peek: bool,
330 is_non_block: bool,
331 ecx: &mut MiriInterpCx<'tcx>,
332 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
333 ) -> InterpResult<'tcx> {
334 if is_peek {
335 throw_unsup_format!("socketpair: virtual sockets don't support peeking")
336 }
337
338 ecx.virtual_socket_read(self, ptr, len, is_non_block, finish)
339 }
340}
341
342impl<'tcx> EvalContextPrivExt<'tcx> for crate::MiriInterpCx<'tcx> {}
343trait EvalContextPrivExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
344 fn virtual_socket_write(
350 &mut self,
351 socket: FileDescriptionRef<VirtualSocket>,
352 ptr: Pointer,
353 len: usize,
354 is_non_block: bool,
355 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
356 ) -> InterpResult<'tcx> {
357 let this = self.eval_context_mut();
358
359 if len == 0 {
362 return finish.call(this, Ok(0));
363 }
364
365 let Some(peer_fd) = socket.peer_fd().upgrade() else {
367 return finish.call(this, Err(ErrorKind::BrokenPipe.into()));
370 };
371
372 let Some(writebuf) = &peer_fd.readbuf else {
373 return finish.call(this, Err(IoError::LibcError("EBADF")));
375 };
376
377 let available_space =
379 MAX_SOCKETPAIR_BUFFER_CAPACITY.strict_sub(writebuf.borrow().buf.len());
380 if available_space == 0 {
381 if socket.is_nonblock.get() || is_non_block {
382 return finish.call(this, Err(ErrorKind::WouldBlock.into()));
384 } else {
385 socket.blocked_write_tid.borrow_mut().push(this.active_thread());
386 this.block_thread(
387 BlockReason::VirtualSocket,
388 None,
389 callback!(
390 @capture<'tcx> {
391 socket: FileDescriptionRef<VirtualSocket>,
392 ptr: Pointer,
393 len: usize,
394 is_non_block: bool,
395 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
396 }
397 |this, unblock: UnblockKind| {
398 assert_eq!(unblock, UnblockKind::Ready);
399 this.virtual_socket_write(socket, ptr, len, is_non_block, finish)
400 }
401 ),
402 );
403 }
404 } else {
405 let mut writebuf = writebuf.borrow_mut();
407 this.release_clock(|clock| {
409 writebuf.clock.join(clock);
410 })?;
411 let write_size = len.min(available_space);
413 let actual_write_size =
414 this.write_to_host(&mut writebuf.buf, write_size, ptr)?.unwrap();
415 assert_eq!(actual_write_size, write_size);
416
417 drop(writebuf);
419
420 let waiting_threads = std::mem::take(&mut *peer_fd.blocked_read_tid.borrow_mut());
422 for thread_id in waiting_threads {
424 this.unblock_thread(thread_id, BlockReason::VirtualSocket)?;
425 }
426 this.update_fd_readiness(socket, ReadinessUpdateFlags::DEFAULT)?;
430 this.update_fd_readiness(peer_fd, ReadinessUpdateFlags::FORCE_EDGE)?;
431
432 return finish.call(this, Ok(write_size));
433 }
434 interp_ok(())
435 }
436
437 fn virtual_socket_read(
443 &mut self,
444 socket: FileDescriptionRef<VirtualSocket>,
445 ptr: Pointer,
446 len: usize,
447 is_non_block: bool,
448 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
449 ) -> InterpResult<'tcx> {
450 let this = self.eval_context_mut();
451
452 if len == 0 {
454 return finish.call(this, Ok(0));
455 }
456
457 let Some(readbuf) = &socket.readbuf else {
458 throw_unsup_format!("reading from the write end of a pipe")
461 };
462
463 if readbuf.borrow_mut().buf.is_empty() {
464 if socket.peer_fd().upgrade().is_none() {
465 return finish.call(this, Ok(0));
468 } else if socket.is_nonblock.get() || is_non_block {
469 return finish.call(this, Err(ErrorKind::WouldBlock.into()));
475 } else {
476 socket.blocked_read_tid.borrow_mut().push(this.active_thread());
477 this.block_thread(
478 BlockReason::VirtualSocket,
479 None,
480 callback!(
481 @capture<'tcx> {
482 socket: FileDescriptionRef<VirtualSocket>,
483 ptr: Pointer,
484 len: usize,
485 is_non_block: bool,
486 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
487 }
488 |this, unblock: UnblockKind| {
489 assert_eq!(unblock, UnblockKind::Ready);
490 this.virtual_socket_read(socket, ptr, len, is_non_block, finish)
491 }
492 ),
493 );
494 }
495 } else {
496 let mut readbuf = readbuf.borrow_mut();
498 this.acquire_clock(&readbuf.clock)?;
502
503 let read_size = this.read_from_host(|buf| readbuf.buf.read(buf), len, ptr)?.unwrap();
506 let readbuf_now_empty = readbuf.buf.is_empty();
507
508 drop(readbuf);
510
511 if let Some(peer_fd) = socket.peer_fd().upgrade() {
519 let waiting_threads = std::mem::take(&mut *peer_fd.blocked_write_tid.borrow_mut());
521 for thread_id in waiting_threads {
523 this.unblock_thread(thread_id, BlockReason::VirtualSocket)?;
524 }
525 this.update_fd_readiness(
530 peer_fd,
531 if readbuf_now_empty {
532 ReadinessUpdateFlags::FORCE_EDGE
533 } else {
534 ReadinessUpdateFlags::DEFAULT
535 },
536 )?;
537 };
538 this.update_fd_readiness(socket, ReadinessUpdateFlags::DEFAULT)?;
540
541 return finish.call(this, Ok(read_size));
542 }
543 interp_ok(())
544 }
545}
546
547impl<'tcx> EvalContextExt<'tcx> for crate::MiriInterpCx<'tcx> {}
548pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
549 fn socketpair(
552 &mut self,
553 domain: &OpTy<'tcx>,
554 type_: &OpTy<'tcx>,
555 protocol: &OpTy<'tcx>,
556 sv: &OpTy<'tcx>,
557 ) -> InterpResult<'tcx, Scalar> {
558 let this = self.eval_context_mut();
559
560 let domain = this.read_scalar(domain)?.to_i32()?;
561 let mut flags = this.read_scalar(type_)?.to_i32()?;
562 let protocol = this.read_scalar(protocol)?.to_i32()?;
563 let sv = this.deref_pointer_as(sv, this.machine.layouts.i32)?;
565
566 let mut is_sock_nonblock = false;
567
568 if matches!(
571 this.tcx.sess.target.os,
572 Os::Linux | Os::Android | Os::FreeBsd | Os::Solaris | Os::Illumos
573 ) {
574 let sock_nonblock = this.eval_libc_i32("SOCK_NONBLOCK");
577 let sock_cloexec = this.eval_libc_i32("SOCK_CLOEXEC");
578 if flags & sock_nonblock == sock_nonblock {
579 is_sock_nonblock = true;
580 flags &= !sock_nonblock;
581 }
582 if flags & sock_cloexec == sock_cloexec {
583 flags &= !sock_cloexec;
584 }
585 }
586
587 if domain != this.eval_libc_i32("AF_UNIX") && domain != this.eval_libc_i32("AF_LOCAL") {
591 throw_unsup_format!(
592 "socketpair: domain {:#x} is unsupported, only AF_UNIX \
593 and AF_LOCAL are allowed",
594 domain
595 );
596 } else if flags != this.eval_libc_i32("SOCK_STREAM") {
597 throw_unsup_format!(
598 "socketpair: type {:#x} is unsupported, only SOCK_STREAM, \
599 SOCK_CLOEXEC and SOCK_NONBLOCK are allowed",
600 flags
601 );
602 } else if protocol != 0 {
603 throw_unsup_format!(
604 "socketpair: socket protocol {protocol} is unsupported, \
605 only 0 is allowed",
606 );
607 }
608
609 let fds = &mut this.machine.fds;
611 let fd0 = fds.new_ref(VirtualSocket {
612 readbuf: Some(RefCell::new(Buffer::new())),
613 peer_fd: OnceCell::new(),
614 peer_lost_data: Cell::new(false),
615 blocked_read_tid: RefCell::new(Vec::new()),
616 blocked_write_tid: RefCell::new(Vec::new()),
617 is_nonblock: Cell::new(is_sock_nonblock),
618 fd_type: VirtualSocketType::Socketpair,
619 delayed_readiness_updates: Rc::clone(&this.machine.delayed_readiness_updates),
620 watched: ReadinessWatched::default(),
621 });
622 let fd1 = fds.new_ref(VirtualSocket {
623 readbuf: Some(RefCell::new(Buffer::new())),
624 peer_fd: OnceCell::new(),
625 peer_lost_data: Cell::new(false),
626 blocked_read_tid: RefCell::new(Vec::new()),
627 blocked_write_tid: RefCell::new(Vec::new()),
628 is_nonblock: Cell::new(is_sock_nonblock),
629 fd_type: VirtualSocketType::Socketpair,
630 delayed_readiness_updates: Rc::clone(&this.machine.delayed_readiness_updates),
631 watched: ReadinessWatched::default(),
632 });
633
634 fd0.peer_fd.set(FileDescriptionRef::downgrade(&fd1)).unwrap();
636 fd1.peer_fd.set(FileDescriptionRef::downgrade(&fd0)).unwrap();
637
638 let sv0 = fds.insert(fd0);
640 let sv1 = fds.insert(fd1);
641
642 let sv0 = Scalar::from_int(sv0, sv.layout.size);
644 let sv1 = Scalar::from_int(sv1, sv.layout.size);
645 this.write_scalar(sv0, &sv)?;
646 this.write_scalar(sv1, &sv.offset(sv.layout.size, sv.layout, this)?)?;
647
648 interp_ok(Scalar::from_i32(0))
649 }
650
651 fn pipe2(
652 &mut self,
653 pipefd: &OpTy<'tcx>,
654 flags: Option<&OpTy<'tcx>>,
655 ) -> InterpResult<'tcx, Scalar> {
656 let this = self.eval_context_mut();
657
658 let pipefd = this.deref_pointer_as(pipefd, this.machine.layouts.i32)?;
659 let mut flags = match flags {
660 Some(flags) => this.read_scalar(flags)?.to_i32()?,
661 None => 0,
662 };
663
664 let cloexec = this.eval_libc_i32("O_CLOEXEC");
665 let o_nonblock = this.eval_libc_i32("O_NONBLOCK");
666
667 let mut is_nonblock = false;
670 if flags & o_nonblock == o_nonblock {
671 is_nonblock = true;
672 flags &= !o_nonblock;
673 }
674 if flags & cloexec == cloexec {
676 flags &= !cloexec;
677 }
678 if flags != 0 {
679 throw_unsup_format!("unsupported flags in `pipe2`");
680 }
681
682 let fds = &mut this.machine.fds;
685 let fd0 = fds.new_ref(VirtualSocket {
686 readbuf: Some(RefCell::new(Buffer::new())),
687 peer_fd: OnceCell::new(),
688 peer_lost_data: Cell::new(false),
689 blocked_read_tid: RefCell::new(Vec::new()),
690 blocked_write_tid: RefCell::new(Vec::new()),
691 is_nonblock: Cell::new(is_nonblock),
692 fd_type: VirtualSocketType::PipeRead,
693 delayed_readiness_updates: Rc::clone(&this.machine.delayed_readiness_updates),
694 watched: ReadinessWatched::default(),
695 });
696 let fd1 = fds.new_ref(VirtualSocket {
697 readbuf: None,
698 peer_fd: OnceCell::new(),
699 peer_lost_data: Cell::new(false),
700 blocked_read_tid: RefCell::new(Vec::new()),
701 blocked_write_tid: RefCell::new(Vec::new()),
702 is_nonblock: Cell::new(is_nonblock),
703 fd_type: VirtualSocketType::PipeWrite,
704 delayed_readiness_updates: Rc::clone(&this.machine.delayed_readiness_updates),
705 watched: ReadinessWatched::default(),
706 });
707
708 fd0.peer_fd.set(FileDescriptionRef::downgrade(&fd1)).unwrap();
710 fd1.peer_fd.set(FileDescriptionRef::downgrade(&fd0)).unwrap();
711
712 let pipefd0 = fds.insert(fd0);
714 let pipefd1 = fds.insert(fd1);
715
716 let pipefd0 = Scalar::from_int(pipefd0, pipefd.layout.size);
718 let pipefd1 = Scalar::from_int(pipefd1, pipefd.layout.size);
719 this.write_scalar(pipefd0, &pipefd)?;
720 this.write_scalar(pipefd1, &pipefd.offset(pipefd.layout.size, pipefd.layout, this)?)?;
721
722 interp_ok(Scalar::from_i32(0))
723 }
724}