1pub const FORMATION_LIFELINE_ENV: &str = "MURMUR_FORMATION_LIFELINE";
41
42pub const SPAWNER_LIFELINE_ENV: &str = "MURMUR_SPAWNER_LIFELINE";
47
48#[cfg(unix)]
49pub use unix::{ChildLifeline, FormationLifeline, MemberLifeline, SpawnerLifeline};
50
51#[cfg(unix)]
52mod unix {
53 use std::ffi::OsString;
54 use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
55 use std::sync::atomic::{AtomicBool, Ordering};
56
57 use super::{FORMATION_LIFELINE_ENV, SPAWNER_LIFELINE_ENV};
58 use crate::errors::RuntimeError;
59
60 static FORMATION_TAKEN: AtomicBool = AtomicBool::new(false);
63
64 static SPAWNER_TAKEN: AtomicBool = AtomicBool::new(false);
68
69 const FIRST_LIFELINE_FD: RawFd = 3;
72
73 struct Kind {
76 thread: &'static str,
78 name: &'static str,
80 holder: &'static str,
82 ended: &'static str,
84 unreadable: fn(String) -> RuntimeError,
85 }
86
87 static FORMATION: Kind = Kind {
88 thread: "formation-lifeline",
89 name: "formation",
90 holder: "a member",
91 ended: "the formation",
92 unreadable: |reason| RuntimeError::FormationLifelineUnreadable { reason },
93 };
94
95 static SPAWNER: Kind = Kind {
96 thread: "spawner-lifeline",
97 name: "spawner",
98 holder: "a child",
99 ended: "the session that delegated to this one",
100 unreadable: |reason| RuntimeError::SpawnerLifelineUnreadable { reason },
101 };
102
103 impl Kind {
104 fn refuse(&self, reason: impl Into<String>) -> RuntimeError {
105 (self.unreadable)(reason.into())
106 }
107
108 #[allow(unsafe_code)]
115 fn read_end(
116 &self,
117 value: Option<OsString>,
118 taken: &AtomicBool,
119 ) -> Result<Option<OwnedFd>, RuntimeError> {
120 let Some(raw) = value else {
121 return Ok(None);
122 };
123 let raw = raw.to_string_lossy();
124 let value = raw.trim();
125 if value.is_empty() {
126 return Ok(None);
127 }
128 if taken.swap(true, Ordering::SeqCst) {
129 return Err(self.refuse(
130 "it was already read by this process, and a descriptor has one owner",
131 ));
132 }
133 let fd = self.parse_fd(value)?;
134 self.check_read_end(fd)?;
135 set_cloexec(fd).map_err(|error| {
136 self.refuse(format!(
137 "descriptor {fd} could not be made close-on-exec: {error}"
138 ))
139 })?;
140 Ok(Some(unsafe { OwnedFd::from_raw_fd(fd) }))
145 }
146
147 fn parse_fd(&self, value: &str) -> Result<RawFd, RuntimeError> {
148 if !value.bytes().all(|byte| byte.is_ascii_digit()) {
149 return Err(self.refuse("it is not a decimal descriptor number"));
150 }
151 let fd: RawFd = value
152 .parse()
153 .map_err(|_| self.refuse("it is not a descriptor number this platform has"))?;
154 if fd < FIRST_LIFELINE_FD {
155 return Err(self.refuse(format!(
156 "descriptor {fd} is a standard stream, and a lifeline is {FIRST_LIFELINE_FD} or above"
157 )));
158 }
159 Ok(fd)
160 }
161
162 #[allow(unsafe_code)]
164 fn check_read_end(&self, fd: RawFd) -> Result<(), RuntimeError> {
165 let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
168 if flags < 0 {
169 return Err(self.refuse(format!("descriptor {fd} is not open")));
170 }
171 let mode = unsafe {
173 let mut stat: libc::stat = std::mem::zeroed();
174 if libc::fstat(fd, &mut stat) != 0 {
175 return Err(self.refuse(format!(
176 "descriptor {fd} could not be examined: {}",
177 std::io::Error::last_os_error()
178 )));
179 }
180 stat.st_mode
181 };
182 if mode & libc::S_IFMT != libc::S_IFIFO {
183 let kind = match mode & libc::S_IFMT {
184 libc::S_IFCHR => "a character device",
185 libc::S_IFREG => "a regular file",
186 libc::S_IFDIR => "a directory",
187 libc::S_IFSOCK => "a socket",
188 _ => "not a pipe",
189 };
190 return Err(self.refuse(format!(
191 "descriptor {fd} is {kind}, and a lifeline is a pipe's read end"
192 )));
193 }
194 if flags & libc::O_ACCMODE != libc::O_RDONLY {
195 return Err(self.refuse(format!(
196 "descriptor {fd} is open for writing, and {} holds only a read end",
197 self.holder
198 )));
199 }
200 Ok(())
201 }
202
203 fn watch(
207 &'static self,
208 fd: OwnedFd,
209 on_closed: impl FnOnce() + Send + 'static,
210 ) -> std::io::Result<()> {
211 std::thread::Builder::new()
212 .name(self.thread.to_string())
213 .spawn(move || {
214 self.wait_for_eof(fd);
215 on_closed();
216 })
217 .map(|_| ())
218 }
219
220 fn wait_for_eof(&self, fd: OwnedFd) {
222 use std::io::Read;
223 let mut pipe = std::fs::File::from(fd);
224 let mut buffer = [0u8; 64];
225 loop {
226 match pipe.read(&mut buffer) {
227 Ok(0) => return,
228 Ok(_) => {}
229 Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
230 Err(error) => {
231 crate::runtime_err!(
232 "[capsule-runtime] the {} lifeline could not be read ({error}); \
233 treating {} as ended",
234 self.name,
235 self.ended
236 );
237 return;
238 }
239 }
240 }
241 }
242 }
243
244 #[derive(Debug)]
246 pub struct FormationLifeline {
247 fd: OwnedFd,
248 }
249
250 impl FormationLifeline {
251 pub fn from_env() -> Result<Option<Self>, RuntimeError> {
258 Self::from_env_once(&FORMATION_TAKEN)
259 }
260
261 pub(crate) fn from_env_once(taken: &AtomicBool) -> Result<Option<Self>, RuntimeError> {
262 let value = std::env::var_os(FORMATION_LIFELINE_ENV);
263 Ok(FORMATION.read_end(value, taken)?.map(|fd| Self { fd }))
264 }
265
266 pub fn watch(self, on_closed: impl FnOnce() + Send + 'static) -> std::io::Result<()> {
272 FORMATION.watch(self.fd, on_closed)
273 }
274 }
275
276 #[derive(Debug)]
279 pub struct SpawnerLifeline {
280 fd: OwnedFd,
281 }
282
283 impl SpawnerLifeline {
284 pub fn from_env() -> Result<Option<Self>, RuntimeError> {
289 Self::from_env_once(&SPAWNER_TAKEN)
290 }
291
292 pub(crate) fn from_env_once(taken: &AtomicBool) -> Result<Option<Self>, RuntimeError> {
293 let value = std::env::var_os(SPAWNER_LIFELINE_ENV);
294 Ok(SPAWNER.read_end(value, taken)?.map(|fd| Self { fd }))
295 }
296
297 pub fn watch(self, on_closed: impl FnOnce() + Send + 'static) -> std::io::Result<()> {
303 SPAWNER.watch(self.fd, on_closed)
304 }
305 }
306
307 #[allow(unsafe_code)]
308 fn set_cloexec(fd: RawFd) -> std::io::Result<()> {
309 let set = unsafe {
312 let flags = libc::fcntl(fd, libc::F_GETFD);
313 flags >= 0 && libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC) == 0
314 };
315 if set {
316 Ok(())
317 } else {
318 Err(std::io::Error::last_os_error())
319 }
320 }
321
322 #[derive(Debug)]
325 struct PipeEnds {
326 read: Option<OwnedFd>,
327 write: Option<OwnedFd>,
328 }
329
330 impl PipeEnds {
331 #[allow(unsafe_code)]
333 fn new() -> std::io::Result<Self> {
334 let [read, write] = cloexec_pipe()?;
335 let read = if read.as_raw_fd() < FIRST_LIFELINE_FD {
338 let duplicate = unsafe {
341 libc::fcntl(read.as_raw_fd(), libc::F_DUPFD_CLOEXEC, FIRST_LIFELINE_FD)
342 };
343 if duplicate < 0 {
344 return Err(std::io::Error::last_os_error());
345 }
346 unsafe { OwnedFd::from_raw_fd(duplicate) }
348 } else {
349 read
350 };
351 Ok(Self {
352 read: Some(read),
353 write: Some(write),
354 })
355 }
356
357 fn env_pair(&self, name: &'static str) -> Option<(&'static str, String)> {
359 self.read
360 .as_ref()
361 .map(|read| (name, read.as_raw_fd().to_string()))
362 }
363
364 #[allow(unsafe_code)]
368 fn hand_to(&self, name: &'static str, command: &mut std::process::Command) {
369 use std::os::unix::process::CommandExt;
370 let Some((name, value)) = self.env_pair(name) else {
371 return;
372 };
373 let fd = self.read.as_ref().map_or(-1, AsRawFd::as_raw_fd);
374 command.env(name, value);
375 unsafe {
379 command.pre_exec(move || {
380 let flags = libc::fcntl(fd, libc::F_GETFD);
381 if flags < 0 || libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC) != 0 {
382 return Err(std::io::Error::last_os_error());
383 }
384 Ok(())
385 });
386 }
387 }
388
389 fn write_fd(&self) -> Option<RawFd> {
390 self.write.as_ref().map(AsRawFd::as_raw_fd)
391 }
392 }
393
394 #[derive(Debug)]
400 pub struct MemberLifeline {
401 ends: PipeEnds,
402 }
403
404 impl MemberLifeline {
405 pub fn new() -> std::io::Result<Self> {
410 Ok(Self {
411 ends: PipeEnds::new()?,
412 })
413 }
414
415 pub fn hand_to(&self, command: &mut std::process::Command) {
422 self.ends.hand_to(FORMATION_LIFELINE_ENV, command);
423 }
424
425 pub fn spawned(&mut self) {
428 self.ends.read = None;
429 }
430
431 pub fn close(&mut self) {
434 self.ends.write = None;
435 }
436
437 pub fn write_fd(&self) -> Option<RawFd> {
439 self.ends.write_fd()
440 }
441 }
442
443 #[derive(Debug)]
449 pub struct ChildLifeline {
450 ends: PipeEnds,
451 }
452
453 impl ChildLifeline {
454 pub fn new() -> std::io::Result<Self> {
460 Ok(Self {
461 ends: PipeEnds::new()?,
462 })
463 }
464
465 pub fn hand_to(&self, command: &mut std::process::Command) {
472 self.ends.hand_to(SPAWNER_LIFELINE_ENV, command);
473 }
474
475 pub fn env_pair(&self) -> Option<(&'static str, String)> {
477 self.ends.env_pair(SPAWNER_LIFELINE_ENV)
478 }
479
480 pub fn spawned(&mut self) {
483 self.ends.read = None;
484 }
485
486 pub fn close(&mut self) {
489 self.ends.write = None;
490 }
491
492 pub fn write_fd(&self) -> Option<RawFd> {
494 self.ends.write_fd()
495 }
496 }
497
498 #[cfg(any(target_os = "linux", target_os = "android", target_os = "freebsd"))]
499 #[allow(unsafe_code)]
500 fn cloexec_pipe() -> std::io::Result<[OwnedFd; 2]> {
501 let mut fds = [0 as RawFd; 2];
502 let created = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) };
504 if created != 0 {
505 return Err(std::io::Error::last_os_error());
506 }
507 Ok(unsafe { [OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])] })
509 }
510
511 #[cfg(not(any(target_os = "linux", target_os = "android", target_os = "freebsd")))]
512 #[allow(unsafe_code)]
513 fn cloexec_pipe() -> std::io::Result<[OwnedFd; 2]> {
514 let mut fds = [0 as RawFd; 2];
515 let created = unsafe { libc::pipe(fds.as_mut_ptr()) };
517 if created != 0 {
518 return Err(std::io::Error::last_os_error());
519 }
520 let ends = unsafe { [OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])] };
522 for end in &ends {
523 set_cloexec(end.as_raw_fd())?;
524 }
525 Ok(ends)
526 }
527
528 #[cfg(test)]
529 #[allow(unsafe_code)]
530 mod tests {
531 use super::*;
532 use std::os::fd::IntoRawFd;
533 use std::sync::mpsc;
534 use std::time::Duration;
535
536 fn env_guard() -> std::sync::MutexGuard<'static, ()> {
539 crate::formation::FORMATION_ENV_LOCK
540 .lock()
541 .unwrap_or_else(|poisoned| poisoned.into_inner())
542 }
543
544 fn fd_flags(fd: RawFd) -> libc::c_int {
545 unsafe { libc::fcntl(fd, libc::F_GETFD) }
547 }
548
549 fn read_with(
552 value: &str,
553 taken: &AtomicBool,
554 ) -> Result<Option<FormationLifeline>, RuntimeError> {
555 std::env::set_var(FORMATION_LIFELINE_ENV, value);
556 let read = FormationLifeline::from_env_once(taken);
557 std::env::remove_var(FORMATION_LIFELINE_ENV);
558 read
559 }
560
561 fn read_spawner_with(
563 value: &str,
564 taken: &AtomicBool,
565 ) -> Result<Option<SpawnerLifeline>, RuntimeError> {
566 std::env::set_var(SPAWNER_LIFELINE_ENV, value);
567 let read = SpawnerLifeline::from_env_once(taken);
568 std::env::remove_var(SPAWNER_LIFELINE_ENV);
569 read
570 }
571
572 fn released(ends: &mut PipeEnds) -> RawFd {
575 ends.read.take().unwrap().into_raw_fd()
576 }
577
578 fn member_side() -> (RawFd, MemberLifeline) {
579 let mut lifeline = MemberLifeline::new().unwrap();
580 (released(&mut lifeline.ends), lifeline)
581 }
582
583 fn child_side() -> (RawFd, ChildLifeline) {
584 let mut lifeline = ChildLifeline::new().unwrap();
585 (released(&mut lifeline.ends), lifeline)
586 }
587
588 fn closed_fd() -> RawFd {
592 let limit = unsafe {
594 let mut limit: libc::rlimit = std::mem::zeroed();
595 assert_eq!(libc::getrlimit(libc::RLIMIT_NOFILE, &mut limit), 0);
596 limit.rlim_cur
597 };
598 let fd = RawFd::try_from(limit.saturating_sub(1)).unwrap_or(RawFd::MAX);
599 assert!(fd_flags(fd) < 0, "descriptor {fd} is open");
600 fd
601 }
602
603 #[test]
604 fn lifeline_from_env_sets_cloexec_and_a_second_read_is_refused() {
605 let _guard = env_guard();
606 let (fd, _launcher) = member_side();
607 unsafe {
609 libc::fcntl(fd, libc::F_SETFD, 0);
610 }
611 assert_eq!(fd_flags(fd) & libc::FD_CLOEXEC, 0);
612 let taken = AtomicBool::new(false);
613 let lifeline = read_with(&fd.to_string(), &taken).unwrap().unwrap();
614 assert_eq!(lifeline.fd.as_raw_fd(), fd);
615 assert_ne!(fd_flags(fd) & libc::FD_CLOEXEC, 0);
616
617 let again = read_with(&fd.to_string(), &taken).unwrap_err().to_string();
618 assert!(again.contains(FORMATION_LIFELINE_ENV), "{again}");
619 assert!(again.contains("already read"), "{again}");
620 }
621
622 #[test]
623 fn spawner_lifeline_from_env_sets_cloexec_and_a_second_read_is_refused() {
624 let _guard = env_guard();
625 let (fd, _parent) = child_side();
626 unsafe {
628 libc::fcntl(fd, libc::F_SETFD, 0);
629 }
630 let taken = AtomicBool::new(false);
631 let lifeline = read_spawner_with(&fd.to_string(), &taken).unwrap().unwrap();
632 assert_eq!(lifeline.fd.as_raw_fd(), fd);
633 assert_ne!(fd_flags(fd) & libc::FD_CLOEXEC, 0);
634
635 let again = read_spawner_with(&fd.to_string(), &taken)
636 .unwrap_err()
637 .to_string();
638 assert!(again.contains(SPAWNER_LIFELINE_ENV), "{again}");
639 assert!(again.contains("already read"), "{again}");
640 }
641
642 #[test]
645 fn spawner_and_formation_lifelines_do_not_share_a_once_flag() {
646 assert!(!std::ptr::eq(&SPAWNER_TAKEN, &FORMATION_TAKEN));
647 let _guard = env_guard();
648 let (spawner_fd, _parent) = child_side();
649 let (formation_fd, _launcher) = member_side();
650 std::env::set_var(SPAWNER_LIFELINE_ENV, spawner_fd.to_string());
651 std::env::set_var(FORMATION_LIFELINE_ENV, formation_fd.to_string());
652 let spawner_taken = AtomicBool::new(false);
653 let formation_taken = AtomicBool::new(false);
654 let spawner = SpawnerLifeline::from_env_once(&spawner_taken);
655 let formation = FormationLifeline::from_env_once(&formation_taken);
656 std::env::remove_var(SPAWNER_LIFELINE_ENV);
657 std::env::remove_var(FORMATION_LIFELINE_ENV);
658 assert_eq!(spawner.unwrap().unwrap().fd.as_raw_fd(), spawner_fd);
659 assert_eq!(formation.unwrap().unwrap().fd.as_raw_fd(), formation_fd);
660 }
661
662 #[test]
663 fn lifeline_absent_or_blank_is_none() {
664 let _guard = env_guard();
665 std::env::remove_var(FORMATION_LIFELINE_ENV);
666 std::env::remove_var(SPAWNER_LIFELINE_ENV);
667 let taken = AtomicBool::new(false);
668 assert!(FormationLifeline::from_env_once(&taken).unwrap().is_none());
669 assert!(read_with(" ", &taken).unwrap().is_none());
670 assert!(SpawnerLifeline::from_env_once(&taken).unwrap().is_none());
671 assert!(read_spawner_with(" ", &taken).unwrap().is_none());
672 assert!(
673 !taken.load(Ordering::SeqCst),
674 "an absent lifeline takes nothing"
675 );
676 }
677
678 #[allow(clippy::type_complexity)]
681 fn refusals() -> (
682 Vec<(String, &'static str)>,
683 (std::fs::File, std::fs::File, ChildLifeline),
684 ) {
685 let file = tempfile::tempfile().unwrap();
686 let null = std::fs::File::open("/dev/null").unwrap();
687 let launcher = ChildLifeline::new().unwrap();
688 let write = launcher.write_fd().unwrap();
689 let cases = vec![
690 ("three".to_string(), "not a decimal descriptor number"),
691 ("-4".to_string(), "not a decimal descriptor number"),
692 ("2".to_string(), "standard stream"),
693 (closed_fd().to_string(), "is not open"),
694 (null.as_raw_fd().to_string(), "a character device"),
695 (file.as_raw_fd().to_string(), "a regular file"),
696 (write.to_string(), "open for writing"),
697 ];
698 (cases, (file, null, launcher))
699 }
700
701 #[test]
702 fn lifeline_values_that_are_not_a_read_end_are_refused() {
703 let _guard = env_guard();
704 let (cases, _held) = refusals();
705 for (value, expected) in cases {
706 let taken = AtomicBool::new(false);
707 let refused = read_with(&value, &taken).unwrap_err().to_string();
708 assert!(refused.contains(FORMATION_LIFELINE_ENV), "{refused}");
709 assert!(refused.contains(expected), "{value}: {refused}");
710 if expected == "open for writing" {
711 assert!(refused.contains("a member holds only"), "{refused}");
712 }
713 }
714 }
715
716 #[test]
717 fn spawner_lifeline_values_that_are_not_a_read_end_are_refused() {
718 let _guard = env_guard();
719 let (cases, _held) = refusals();
720 for (value, expected) in cases {
721 let taken = AtomicBool::new(false);
722 let refused = read_spawner_with(&value, &taken).unwrap_err();
723 assert!(
724 matches!(refused, RuntimeError::SpawnerLifelineUnreadable { .. }),
725 "{refused:?}"
726 );
727 let refused = refused.to_string();
728 assert!(refused.contains(SPAWNER_LIFELINE_ENV), "{refused}");
729 assert!(refused.contains(expected), "{value}: {refused}");
730 if expected == "open for writing" {
731 assert!(refused.contains("a child holds only"), "{refused}");
732 }
733 }
734 }
735
736 fn handed(variable: &str, read: RawFd, write: RawFd) -> std::process::Command {
739 let mut command = std::process::Command::new("sh");
740 command.arg("-c").arg(format!(
741 "echo \"${variable}\"; \
742 if [ -e /dev/fd/{read} ]; then echo read; fi; \
743 if [ -e /dev/fd/{write} ]; then echo write; fi"
744 ));
745 command.stdout(std::process::Stdio::piped());
746 command
747 }
748
749 #[test]
750 fn lifeline_hand_to_gives_a_child_its_read_end_and_no_write_end() {
751 let mut lifeline = MemberLifeline::new().unwrap();
752 let (read, write) = (
753 lifeline.ends.read.as_ref().unwrap().as_raw_fd(),
754 lifeline.write_fd().unwrap(),
755 );
756 let mut command = handed(FORMATION_LIFELINE_ENV, read, write);
757 lifeline.hand_to(&mut command);
758 let child = command.spawn().unwrap();
759 lifeline.spawned();
760 let output = child.wait_with_output().unwrap();
761 let stdout = String::from_utf8_lossy(&output.stdout);
762 assert_eq!(
763 stdout.lines().collect::<Vec<_>>(),
764 [&read.to_string(), "read"]
765 );
766 assert!(
767 lifeline.ends.env_pair(FORMATION_LIFELINE_ENV).is_none(),
768 "the launcher released its read end"
769 );
770 assert!(lifeline.write_fd().is_some());
771 }
772
773 #[test]
774 fn spawner_lifeline_hand_to_gives_a_child_its_read_end_and_no_write_end() {
775 let mut lifeline = ChildLifeline::new().unwrap();
776 let (read, write) = (
777 lifeline.ends.read.as_ref().unwrap().as_raw_fd(),
778 lifeline.write_fd().unwrap(),
779 );
780 assert_eq!(
781 lifeline.env_pair(),
782 Some((SPAWNER_LIFELINE_ENV, read.to_string()))
783 );
784 let mut command = handed(SPAWNER_LIFELINE_ENV, read, write);
785 lifeline.hand_to(&mut command);
786 let child = command.spawn().unwrap();
787 lifeline.spawned();
788 let output = child.wait_with_output().unwrap();
789 let stdout = String::from_utf8_lossy(&output.stdout);
790 assert_eq!(
791 stdout.lines().collect::<Vec<_>>(),
792 [&read.to_string(), "read"]
793 );
794 assert!(
795 lifeline.env_pair().is_none(),
796 "the parent released its read end"
797 );
798 assert!(lifeline.write_fd().is_some());
799 }
800
801 fn watched(kind: &'static Kind, fd: RawFd) -> mpsc::Receiver<()> {
803 let fd = unsafe { OwnedFd::from_raw_fd(fd) };
805 let (tx, rx) = mpsc::channel();
806 kind.watch(fd, move || tx.send(()).unwrap()).unwrap();
807 rx
808 }
809
810 #[test]
811 fn lifeline_closing_the_write_end_fires_the_watcher_once() {
812 let _guard = env_guard();
813 let (fd, mut launcher) = member_side();
814 let taken = AtomicBool::new(false);
815 let lifeline = read_with(&fd.to_string(), &taken).unwrap().unwrap();
816 let (tx, rx) = mpsc::channel();
817 lifeline.watch(move || tx.send(()).unwrap()).unwrap();
818 assert!(rx.recv_timeout(Duration::from_millis(300)).is_err());
819 launcher.close();
820 rx.recv_timeout(Duration::from_secs(5))
821 .expect("EOF fires the watcher");
822 assert!(
823 matches!(
824 rx.recv_timeout(Duration::from_millis(300)),
825 Err(mpsc::RecvTimeoutError::Disconnected)
826 ),
827 "the watcher fires once and is gone"
828 );
829 }
830
831 #[test]
832 fn spawner_lifeline_closing_the_write_end_fires_the_watcher_once() {
833 let _guard = env_guard();
834 let (fd, mut parent) = child_side();
835 let taken = AtomicBool::new(false);
836 let lifeline = read_spawner_with(&fd.to_string(), &taken).unwrap().unwrap();
837 let (tx, rx) = mpsc::channel();
838 lifeline.watch(move || tx.send(()).unwrap()).unwrap();
839 assert!(rx.recv_timeout(Duration::from_millis(300)).is_err());
840 parent.close();
841 parent.close();
842 rx.recv_timeout(Duration::from_secs(5))
843 .expect("EOF fires the watcher");
844 assert!(
845 matches!(
846 rx.recv_timeout(Duration::from_millis(300)),
847 Err(mpsc::RecvTimeoutError::Disconnected)
848 ),
849 "the watcher fires once and is gone"
850 );
851 }
852
853 #[test]
856 fn spawner_lifeline_dropping_the_parent_side_fires_the_watcher() {
857 let (fd, parent) = child_side();
858 let rx = watched(&SPAWNER, fd);
859 assert!(rx.recv_timeout(Duration::from_millis(300)).is_err());
860 drop(parent);
861 rx.recv_timeout(Duration::from_secs(5))
862 .expect("dropping the write end fires the watcher");
863 }
864
865 #[test]
866 fn lifeline_eof_before_watch_fires_at_once() {
867 let _guard = env_guard();
868 let (fd, mut launcher) = member_side();
869 launcher.close();
870 let taken = AtomicBool::new(false);
871 let lifeline = read_with(&fd.to_string(), &taken).unwrap().unwrap();
872 let (tx, rx) = mpsc::channel();
873 lifeline.watch(move || tx.send(()).unwrap()).unwrap();
874 rx.recv_timeout(Duration::from_secs(5))
875 .expect("sticky EOF fires the watcher");
876 }
877
878 #[test]
879 fn spawner_lifeline_eof_before_watch_fires_at_once() {
880 let (fd, mut parent) = child_side();
881 parent.close();
882 let rx = watched(&SPAWNER, fd);
883 rx.recv_timeout(Duration::from_secs(5))
884 .expect("sticky EOF fires the watcher");
885 }
886
887 fn bytes_do_not_fire(kind: &'static Kind, fd: RawFd, write: RawFd) -> mpsc::Receiver<()> {
888 let rx = watched(kind, fd);
889 let bytes = [7u8; 64];
890 let written =
892 unsafe { libc::write(write, bytes.as_ptr().cast::<libc::c_void>(), bytes.len()) };
893 assert_eq!(written, 64);
894 assert!(
895 rx.recv_timeout(Duration::from_millis(500)).is_err(),
896 "bytes are not EOF"
897 );
898 rx
899 }
900
901 #[test]
902 fn lifeline_bytes_with_the_write_end_open_do_not_fire_the_watcher() {
903 let (fd, mut launcher) = member_side();
904 let rx = bytes_do_not_fire(&FORMATION, fd, launcher.write_fd().unwrap());
905 launcher.close();
906 rx.recv_timeout(Duration::from_secs(5))
907 .expect("EOF after the bytes fires");
908 }
909
910 #[test]
911 fn spawner_lifeline_bytes_with_the_write_end_open_do_not_fire_the_watcher() {
912 let (fd, mut parent) = child_side();
913 let rx = bytes_do_not_fire(&SPAWNER, fd, parent.write_fd().unwrap());
914 parent.close();
915 rx.recv_timeout(Duration::from_secs(5))
916 .expect("EOF after the bytes fires");
917 }
918
919 #[test]
920 fn the_watcher_threads_are_named_for_their_lifeline() {
921 assert_eq!(FORMATION.thread, "formation-lifeline");
922 assert_eq!(SPAWNER.thread, "spawner-lifeline");
923 }
924 }
925}