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<'tcx>(
151 self: FileDescriptionRef<Self>,
152 _ecx: &MiriInterpCx<'tcx>,
153 ) -> FileDescriptionRef<dyn UnixFileDescription> {
154 self
155 }
156
157 fn get_flags<'tcx>(&self, ecx: &mut MiriInterpCx<'tcx>) -> InterpResult<'tcx, Scalar> {
158 let mut flags = 0;
159
160 match self.fd_type {
165 VirtualSocketType::Socketpair => {
166 flags |= ecx.eval_libc_i32("O_RDWR");
167 }
168 VirtualSocketType::PipeRead => {
169 flags |= ecx.eval_libc_i32("O_RDONLY");
170 }
171 VirtualSocketType::PipeWrite => {
172 flags |= ecx.eval_libc_i32("O_WRONLY");
173 }
174 }
175
176 if self.is_nonblock.get() {
178 flags |= ecx.eval_libc_i32("O_NONBLOCK");
179 }
180
181 interp_ok(Scalar::from_i32(flags))
182 }
183
184 fn set_flags<'tcx>(
185 &self,
186 mut flag: i32,
187 ecx: &mut MiriInterpCx<'tcx>,
188 ) -> InterpResult<'tcx, Scalar> {
189 let o_nonblock = ecx.eval_libc_i32("O_NONBLOCK");
190
191 if flag & o_nonblock == o_nonblock {
193 self.is_nonblock.set(true);
194 flag &= !o_nonblock;
195 } else {
196 self.is_nonblock.set(false);
197 }
198
199 if flag != 0 {
201 throw_unsup_format!(
202 "fcntl: only O_NONBLOCK is supported for F_SETFL on socketpairs and pipes"
203 )
204 }
205
206 interp_ok(Scalar::from_i32(0))
207 }
208
209 fn readiness_watched(&self) -> Option<&ReadinessWatched> {
210 Some(&self.watched)
211 }
212
213 fn readiness(&self) -> Readiness {
214 let mut readiness = Readiness::EMPTY;
218
219 if let Some(readbuf) = &self.readbuf {
221 if !readbuf.borrow().buf.is_empty() {
222 readiness.readable = true;
223 }
224 } else {
225 readiness.readable = true;
227 }
228
229 if let Some(peer_fd) = self.peer_fd().upgrade() {
231 if let Some(writebuf) = &peer_fd.readbuf {
232 let data_size = writebuf.borrow().buf.len();
233 let available_space = MAX_SOCKETPAIR_BUFFER_CAPACITY.strict_sub(data_size);
234 if available_space != 0 {
235 readiness.writable = true;
236 }
237 } else {
238 readiness.writable = true;
240 }
241 } else {
242 readiness.read_closed = true;
245 readiness.write_closed = true;
246 readiness.readable = true;
250 readiness.writable = true;
251 if self.peer_lost_data.get() {
253 readiness.error = true;
254 }
255 }
256 readiness
257 }
258}
259
260impl UnixFileDescription for VirtualSocket {
261 fn ioctl<'tcx>(
262 &self,
263 op: Scalar,
264 args: Varargs<'tcx, '_>,
265 ecx: &mut MiriInterpCx<'tcx>,
266 ) -> InterpResult<'tcx, i32> {
267 match self.fd_type {
268 VirtualSocketType::Socketpair => { }
269 VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => {
270 throw_unsup_format!("cannot use ioctl on pipe");
274 }
275 }
276
277 let fionbio = ecx.eval_libc("FIONBIO");
278
279 if op == fionbio {
280 if !matches!(ecx.tcx.sess.target.os, Os::Linux | Os::Android | Os::MacOs | Os::FreeBsd)
283 {
284 throw_unsup_format!(
289 "ioctl: setting FIONBIO on sockets is unsupported on target {}",
290 ecx.tcx.sess.target.os
291 );
292 }
293
294 let ([value_ptr], _) = ecx.check_varargs(shim_varargs![*_], args, "ioctl")?;
295 let value = ecx.deref_pointer_as(value_ptr, ecx.machine.layouts.i32)?;
296 let non_block = ecx.read_scalar(&value)?.to_i32()? != 0;
297 self.is_nonblock.set(non_block);
298 return interp_ok(0);
299 }
300
301 throw_unsup_format!("ioctl: unsupported operation {op:#x} on socket");
302 }
303
304 fn as_socket<'tcx>(
305 self: FileDescriptionRef<Self>,
306 _ecx: &MiriInterpCx<'tcx>,
307 ) -> Option<FileDescriptionRef<dyn UnixSocketFileDescription>> {
308 match self.fd_type {
309 VirtualSocketType::Socketpair => Some(self),
310 VirtualSocketType::PipeRead | VirtualSocketType::PipeWrite => None,
311 }
312 }
313}
314
315impl UnixSocketFileDescription for VirtualSocket {
316 fn send<'tcx>(
317 self: FileDescriptionRef<Self>,
318 _communicate_allowed: bool,
319 ptr: Pointer,
320 len: usize,
321 is_non_block: bool,
322 ecx: &mut MiriInterpCx<'tcx>,
323 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
324 ) -> InterpResult<'tcx> {
325 ecx.virtual_socket_write(self, ptr, len, is_non_block, finish)
326 }
327
328 fn recv<'tcx>(
329 self: FileDescriptionRef<Self>,
330 _communicate_allowed: bool,
331 ptr: Pointer,
332 len: usize,
333 is_peek: bool,
334 is_non_block: bool,
335 ecx: &mut MiriInterpCx<'tcx>,
336 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
337 ) -> InterpResult<'tcx> {
338 if is_peek {
339 throw_unsup_format!("socketpair: virtual sockets don't support peeking")
340 }
341
342 ecx.virtual_socket_read(self, ptr, len, is_non_block, finish)
343 }
344}
345
346impl<'tcx> EvalContextPrivExt<'tcx> for crate::MiriInterpCx<'tcx> {}
347trait EvalContextPrivExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
348 fn virtual_socket_write(
354 &mut self,
355 socket: FileDescriptionRef<VirtualSocket>,
356 ptr: Pointer,
357 len: usize,
358 is_non_block: bool,
359 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
360 ) -> InterpResult<'tcx> {
361 let this = self.eval_context_mut();
362
363 if len == 0 {
366 return finish.call(this, Ok(0));
367 }
368
369 let Some(peer_fd) = socket.peer_fd().upgrade() else {
371 return finish.call(this, Err(ErrorKind::BrokenPipe.into()));
374 };
375
376 let Some(writebuf) = &peer_fd.readbuf else {
377 return finish.call(this, Err(IoError::LibcError("EBADF")));
379 };
380
381 let available_space =
383 MAX_SOCKETPAIR_BUFFER_CAPACITY.strict_sub(writebuf.borrow().buf.len());
384 if available_space == 0 {
385 if socket.is_nonblock.get() || is_non_block {
386 return finish.call(this, Err(ErrorKind::WouldBlock.into()));
388 } else {
389 socket.blocked_write_tid.borrow_mut().push(this.active_thread());
390 this.block_thread(
391 BlockReason::VirtualSocket,
392 None,
393 callback!(
394 @capture<'tcx> {
395 socket: FileDescriptionRef<VirtualSocket>,
396 ptr: Pointer,
397 len: usize,
398 is_non_block: bool,
399 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
400 }
401 |this, unblock: UnblockKind| {
402 assert_eq!(unblock, UnblockKind::Ready);
403 this.virtual_socket_write(socket, ptr, len, is_non_block, finish)
404 }
405 ),
406 );
407 }
408 } else {
409 let mut writebuf = writebuf.borrow_mut();
411 this.release_clock(|clock| {
413 writebuf.clock.join(clock);
414 })?;
415 let write_size = len.min(available_space);
417 let actual_write_size =
418 this.write_to_host(&mut writebuf.buf, write_size, ptr)?.unwrap();
419 assert_eq!(actual_write_size, write_size);
420
421 drop(writebuf);
423
424 let waiting_threads = std::mem::take(&mut *peer_fd.blocked_read_tid.borrow_mut());
426 for thread_id in waiting_threads {
428 this.unblock_thread(thread_id, BlockReason::VirtualSocket)?;
429 }
430 this.update_fd_readiness(socket, ReadinessUpdateFlags::DEFAULT)?;
434 this.update_fd_readiness(peer_fd, ReadinessUpdateFlags::FORCE_EDGE)?;
435
436 return finish.call(this, Ok(write_size));
437 }
438 interp_ok(())
439 }
440
441 fn virtual_socket_read(
447 &mut self,
448 socket: FileDescriptionRef<VirtualSocket>,
449 ptr: Pointer,
450 len: usize,
451 is_non_block: bool,
452 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
453 ) -> InterpResult<'tcx> {
454 let this = self.eval_context_mut();
455
456 if len == 0 {
458 return finish.call(this, Ok(0));
459 }
460
461 let Some(readbuf) = &socket.readbuf else {
462 throw_unsup_format!("reading from the write end of a pipe")
465 };
466
467 if readbuf.borrow_mut().buf.is_empty() {
468 if socket.peer_fd().upgrade().is_none() {
469 return finish.call(this, Ok(0));
472 } else if socket.is_nonblock.get() || is_non_block {
473 return finish.call(this, Err(ErrorKind::WouldBlock.into()));
479 } else {
480 socket.blocked_read_tid.borrow_mut().push(this.active_thread());
481 this.block_thread(
482 BlockReason::VirtualSocket,
483 None,
484 callback!(
485 @capture<'tcx> {
486 socket: FileDescriptionRef<VirtualSocket>,
487 ptr: Pointer,
488 len: usize,
489 is_non_block: bool,
490 finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
491 }
492 |this, unblock: UnblockKind| {
493 assert_eq!(unblock, UnblockKind::Ready);
494 this.virtual_socket_read(socket, ptr, len, is_non_block, finish)
495 }
496 ),
497 );
498 }
499 } else {
500 let mut readbuf = readbuf.borrow_mut();
502 this.acquire_clock(&readbuf.clock)?;
506
507 let read_size = this.read_from_host(|buf| readbuf.buf.read(buf), len, ptr)?.unwrap();
510 let readbuf_now_empty = readbuf.buf.is_empty();
511
512 drop(readbuf);
514
515 if let Some(peer_fd) = socket.peer_fd().upgrade() {
523 let waiting_threads = std::mem::take(&mut *peer_fd.blocked_write_tid.borrow_mut());
525 for thread_id in waiting_threads {
527 this.unblock_thread(thread_id, BlockReason::VirtualSocket)?;
528 }
529 this.update_fd_readiness(
534 peer_fd,
535 if readbuf_now_empty {
536 ReadinessUpdateFlags::FORCE_EDGE
537 } else {
538 ReadinessUpdateFlags::DEFAULT
539 },
540 )?;
541 };
542 this.update_fd_readiness(socket, ReadinessUpdateFlags::DEFAULT)?;
544
545 return finish.call(this, Ok(read_size));
546 }
547 interp_ok(())
548 }
549}
550
551impl<'tcx> EvalContextExt<'tcx> for crate::MiriInterpCx<'tcx> {}
552pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
553 fn socketpair(
556 &mut self,
557 domain: &OpTy<'tcx>,
558 type_: &OpTy<'tcx>,
559 protocol: &OpTy<'tcx>,
560 sv: &OpTy<'tcx>,
561 ) -> InterpResult<'tcx, Scalar> {
562 let this = self.eval_context_mut();
563
564 let domain = this.read_scalar(domain)?.to_i32()?;
565 let mut flags = this.read_scalar(type_)?.to_i32()?;
566 let protocol = this.read_scalar(protocol)?.to_i32()?;
567 let sv = this.deref_pointer_as(sv, this.machine.layouts.i32)?;
569
570 let mut is_sock_nonblock = false;
571
572 if matches!(
575 this.tcx.sess.target.os,
576 Os::Linux | Os::Android | Os::FreeBsd | Os::Solaris | Os::Illumos
577 ) {
578 let sock_nonblock = this.eval_libc_i32("SOCK_NONBLOCK");
581 let sock_cloexec = this.eval_libc_i32("SOCK_CLOEXEC");
582 if flags & sock_nonblock == sock_nonblock {
583 is_sock_nonblock = true;
584 flags &= !sock_nonblock;
585 }
586 if flags & sock_cloexec == sock_cloexec {
587 flags &= !sock_cloexec;
588 }
589 }
590
591 if domain != this.eval_libc_i32("AF_UNIX") && domain != this.eval_libc_i32("AF_LOCAL") {
595 throw_unsup_format!(
596 "socketpair: domain {:#x} is unsupported, only AF_UNIX \
597 and AF_LOCAL are allowed",
598 domain
599 );
600 } else if flags != this.eval_libc_i32("SOCK_STREAM") {
601 throw_unsup_format!(
602 "socketpair: type {:#x} is unsupported, only SOCK_STREAM, \
603 SOCK_CLOEXEC and SOCK_NONBLOCK are allowed",
604 flags
605 );
606 } else if protocol != 0 {
607 throw_unsup_format!(
608 "socketpair: socket protocol {protocol} is unsupported, \
609 only 0 is allowed",
610 );
611 }
612
613 let fds = &mut this.machine.fds;
615 let fd0 = fds.new_ref(VirtualSocket {
616 readbuf: Some(RefCell::new(Buffer::new())),
617 peer_fd: OnceCell::new(),
618 peer_lost_data: Cell::new(false),
619 blocked_read_tid: RefCell::new(Vec::new()),
620 blocked_write_tid: RefCell::new(Vec::new()),
621 is_nonblock: Cell::new(is_sock_nonblock),
622 fd_type: VirtualSocketType::Socketpair,
623 delayed_readiness_updates: Rc::clone(&this.machine.delayed_readiness_updates),
624 watched: ReadinessWatched::default(),
625 });
626 let fd1 = fds.new_ref(VirtualSocket {
627 readbuf: Some(RefCell::new(Buffer::new())),
628 peer_fd: OnceCell::new(),
629 peer_lost_data: Cell::new(false),
630 blocked_read_tid: RefCell::new(Vec::new()),
631 blocked_write_tid: RefCell::new(Vec::new()),
632 is_nonblock: Cell::new(is_sock_nonblock),
633 fd_type: VirtualSocketType::Socketpair,
634 delayed_readiness_updates: Rc::clone(&this.machine.delayed_readiness_updates),
635 watched: ReadinessWatched::default(),
636 });
637
638 fd0.peer_fd.set(FileDescriptionRef::downgrade(&fd1)).unwrap();
640 fd1.peer_fd.set(FileDescriptionRef::downgrade(&fd0)).unwrap();
641
642 let sv0 = fds.insert(fd0);
644 let sv1 = fds.insert(fd1);
645
646 let sv0 = Scalar::from_int(sv0, sv.layout.size);
648 let sv1 = Scalar::from_int(sv1, sv.layout.size);
649 this.write_scalar(sv0, &sv)?;
650 this.write_scalar(sv1, &sv.offset(sv.layout.size, sv.layout, this)?)?;
651
652 interp_ok(Scalar::from_i32(0))
653 }
654
655 fn pipe2(
656 &mut self,
657 pipefd: &OpTy<'tcx>,
658 flags: Option<&OpTy<'tcx>>,
659 ) -> InterpResult<'tcx, Scalar> {
660 let this = self.eval_context_mut();
661
662 let pipefd = this.deref_pointer_as(pipefd, this.machine.layouts.i32)?;
663 let mut flags = match flags {
664 Some(flags) => this.read_scalar(flags)?.to_i32()?,
665 None => 0,
666 };
667
668 let cloexec = this.eval_libc_i32("O_CLOEXEC");
669 let o_nonblock = this.eval_libc_i32("O_NONBLOCK");
670
671 let mut is_nonblock = false;
674 if flags & o_nonblock == o_nonblock {
675 is_nonblock = true;
676 flags &= !o_nonblock;
677 }
678 if flags & cloexec == cloexec {
680 flags &= !cloexec;
681 }
682 if flags != 0 {
683 throw_unsup_format!("unsupported flags in `pipe2`");
684 }
685
686 let fds = &mut this.machine.fds;
689 let fd0 = fds.new_ref(VirtualSocket {
690 readbuf: Some(RefCell::new(Buffer::new())),
691 peer_fd: OnceCell::new(),
692 peer_lost_data: Cell::new(false),
693 blocked_read_tid: RefCell::new(Vec::new()),
694 blocked_write_tid: RefCell::new(Vec::new()),
695 is_nonblock: Cell::new(is_nonblock),
696 fd_type: VirtualSocketType::PipeRead,
697 delayed_readiness_updates: Rc::clone(&this.machine.delayed_readiness_updates),
698 watched: ReadinessWatched::default(),
699 });
700 let fd1 = fds.new_ref(VirtualSocket {
701 readbuf: None,
702 peer_fd: OnceCell::new(),
703 peer_lost_data: Cell::new(false),
704 blocked_read_tid: RefCell::new(Vec::new()),
705 blocked_write_tid: RefCell::new(Vec::new()),
706 is_nonblock: Cell::new(is_nonblock),
707 fd_type: VirtualSocketType::PipeWrite,
708 delayed_readiness_updates: Rc::clone(&this.machine.delayed_readiness_updates),
709 watched: ReadinessWatched::default(),
710 });
711
712 fd0.peer_fd.set(FileDescriptionRef::downgrade(&fd1)).unwrap();
714 fd1.peer_fd.set(FileDescriptionRef::downgrade(&fd0)).unwrap();
715
716 let pipefd0 = fds.insert(fd0);
718 let pipefd1 = fds.insert(fd1);
719
720 let pipefd0 = Scalar::from_int(pipefd0, pipefd.layout.size);
722 let pipefd1 = Scalar::from_int(pipefd1, pipefd.layout.size);
723 this.write_scalar(pipefd0, &pipefd)?;
724 this.write_scalar(pipefd1, &pipefd.offset(pipefd.layout.size, pipefd.layout, this)?)?;
725
726 interp_ok(Scalar::from_i32(0))
727 }
728}