פתרון פרויקט צינור מדיה עם בעלות מוגבלת¶
מסלול הנתונים¶
הפתרון מפעיל את כלי aplay ומוסר לו זרם גולמי בצינור. כלי זה משתמש בממשק השמע של לינוקס. אנחנו מנהלים את יצירת הנתונים, גבול הזיכרון, כתיבה חלקית ומחזור חיי תהליך הילד. ההפרדה מאפשרת להתמקד בחוזה ההעברה, ובהמשך להחליף את המתאם בספריית שמע בלי לשנות את בעלות הבאפרים.
ניצור מיזם ראסט עם תלות ב־libc. התוכנית הבאה היא קובץ ראשי מלא. היא דורשת את כלי הניגון בהפצה ומקבלת שם התקן וזמן האטה במילישניות. ברירת המחדל אינה משמיעה קול.
use std::io;
use std::os::fd::AsRawFd;
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, Instant};
const FRAMES: usize = 480;
const BYTES: usize = FRAMES * 2 * 2;
const PERIODS: u64 = 200;
struct Block { sequence: u64, due: Instant, bytes: Vec<u8> }
fn set_nonblocking(fd: i32) -> io::Result<()> {
let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
if flags < 0 || unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } < 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
fn write_period(fd: i32, bytes: &[u8], budget: Duration) -> io::Result<()> {
let deadline = Instant::now() + budget;
let mut offset = 0;
while offset < bytes.len() {
if Instant::now() >= deadline {
return Err(io::Error::new(io::ErrorKind::TimedOut, "audio pipe stalled"));
}
let count = unsafe {
libc::write(fd, bytes[offset..].as_ptr().cast(), bytes.len() - offset)
};
if count > 0 { offset += count as usize; continue; }
if count == 0 { return Err(io::ErrorKind::WriteZero.into()); }
let error = io::Error::last_os_error();
if error.kind() == io::ErrorKind::Interrupted { continue; }
if error.kind() != io::ErrorKind::WouldBlock { return Err(error); }
let mut poll = libc::pollfd { fd, events: libc::POLLOUT, revents: 0 };
let result = unsafe { libc::poll(&mut poll, 1, 10) };
if result < 0 {
let error = io::Error::last_os_error();
if error.kind() != io::ErrorKind::Interrupted { return Err(error); }
}
}
Ok(())
}
fn generate(sequence: u64, bytes: &mut [u8]) {
for frame in 0..FRAMES {
let sample_index = sequence * FRAMES as u64 + frame as u64;
let phase = sample_index as f64 * 440.0 * std::f64::consts::TAU / 48000.0;
let value = (phase.sin() * 2000.0) as i16;
let sample = value.to_le_bytes();
let offset = frame * 4;
bytes[offset..offset + 2].copy_from_slice(&sample);
bytes[offset + 2..offset + 4].copy_from_slice(&sample);
}
}
fn finish(child: &mut Child, successful_stream: bool) -> io::Result<()> {
let deadline = Instant::now() + Duration::from_secs(2);
if successful_stream {
while Instant::now() < deadline {
if let Some(status) = child.try_wait()? {
return if status.success() { Ok(()) }
else { Err(io::Error::other(format!("player failed: {status}"))) };
}
thread::sleep(Duration::from_millis(10));
}
}
let _ = child.kill();
let _ = child.wait();
Err(io::Error::new(io::ErrorKind::TimedOut, "player terminated after failure or deadline"))
}
fn run(child: &mut Child, slow: Duration) -> io::Result<()> {
let input = child.stdin.take().ok_or_else(|| io::Error::other("missing pipe"))?;
set_nonblocking(input.as_raw_fd())?;
let (free_sender, free_receiver) = mpsc::sync_channel::<Vec<u8>>(4);
let (ready_sender, ready_receiver) = mpsc::sync_channel::<Block>(4);
for _ in 0..4 { free_sender.send(vec![0; BYTES]).unwrap(); }
let stop = Arc::new(AtomicBool::new(false));
let producer_stop = stop.clone();
let producer = thread::spawn(move || {
let origin = Instant::now();
let (mut accepted, mut dropped) = (0u64, 0u64);
for sequence in 0..PERIODS {
if producer_stop.load(Ordering::Relaxed) { break; }
let due = origin + Duration::from_millis(sequence * 10);
if let Some(delay) = due.checked_duration_since(Instant::now()) {
thread::sleep(delay);
}
let mut bytes = match free_receiver.try_recv() {
Ok(bytes) => bytes,
Err(mpsc::TryRecvError::Empty) => { dropped += 1; continue; }
Err(mpsc::TryRecvError::Disconnected) => break,
};
generate(sequence, &mut bytes);
let block = Block { sequence, due, bytes };
match ready_sender.try_send(block) {
Ok(()) => accepted += 1,
Err(mpsc::TrySendError::Disconnected(_)) => break,
Err(mpsc::TrySendError::Full(_)) => panic!("four-buffer ownership invariant"),
}
}
(accepted, dropped)
});
let mut delivered = 0u64;
let mut previous = None;
let mut max_age = Duration::ZERO;
let result: io::Result<()> = (|| {
loop {
let block = match ready_receiver.recv_timeout(Duration::from_millis(20)) {
Ok(block) => block,
Err(mpsc::RecvTimeoutError::Disconnected) => break,
Err(mpsc::RecvTimeoutError::Timeout) => {
if child.try_wait()?.is_some() {
return Err(io::Error::new(io::ErrorKind::BrokenPipe, "player exited"));
}
continue;
}
};
if let Some(old) = previous { assert!(block.sequence > old); }
previous = Some(block.sequence);
max_age = max_age.max(block.due.elapsed());
thread::sleep(slow);
write_period(input.as_raw_fd(), &block.bytes, Duration::from_millis(200))?;
delivered += 1;
let _ = free_sender.try_send(block.bytes);
}
Ok(())
})();
stop.store(true, Ordering::Relaxed);
drop(ready_receiver);
drop(free_sender);
let (accepted, dropped) = producer.join().map_err(|_| io::Error::other("producer panic"))?;
drop(input);
println!("accepted={accepted} delivered={delivered} dropped={dropped} max_age_us={}",
max_age.as_micros());
if result.is_ok() {
assert_eq!(accepted + dropped, PERIODS);
assert_eq!(accepted, delivered);
}
result
}
fn main() -> io::Result<()> {
let device = std::env::args().nth(1).unwrap_or_else(|| "null".into());
let slow_ms: u64 = std::env::args().nth(2).unwrap_or_else(|| "0".into())
.parse().map_err(|_| io::Error::other("invalid delay"))?;
if slow_ms > 100 { return Err(io::Error::other("delay limit is 100 ms")); }
let mut child = Command::new("aplay")
.args(["-q", "-D", &device, "-t", "raw", "-f", "S16_LE", "-r", "48000", "-c", "2"])
.stdin(Stdio::piped()).stdout(Stdio::null()).stderr(Stdio::inherit()).spawn()?;
println!("player_pid={}", child.id());
let result = run(&mut child, Duration::from_millis(slow_ms));
let cleanup = finish(&mut child, result.is_ok());
result?;
cleanup
}
מדוע המאגר נשאר מוגבל¶
כל אחד מארבעת הווקטורים נמצא בדיוק במקום אחד: בתור הפנויים, אצל המייצר, בתור המוכנים או אצל הצרכן. לתור המוכנים יש מקום לארבעה ערכים, אבל אם המייצר כבר מחזיק אחד מהם, לא יכולים להימצא בו ארבעת האחרים. לכן כשל של תור מלא הוא שבירה של אינווריאנט ולא מצב עומס רגיל במודל הזה.
אחרי סיום המייצר תור הפנויים יכול להיות מנותק. החזרת באפר אליו אינה נחוצה עוד, והווקטור משתחרר כאשר ניסיון ההחזרה נכשל. בשגיאת כתיבה הצרכן מפסיק, סוגר את ערוצי העבודה ומודיע למייצר לעצור. המייצר אינו ממתין באופן חוסם לערוץ ולכן אינו תלוי בצרכן שכבר נכשל כדי לסיים.
הגבול של תוכן השמע הוא ארבע כפול אלף תשע מאות ועשרים בתים. בנוסף קיימים מטא נתונים, מטמוני צינור, זיכרון תהליך הילד ובאפרים של שכבת השמע. לא נציג את גבול המאגר כגבול הזיכרון של כל המערכת.
הרצה ובדיקת כשל¶
נריץ את שתי הדוגמאות הבאות ונשמור את המונים שהתקבלו. בהרצה השנייה צפויות השמטות בגלל צרכן שאיטי מקצב ההפקה. אין מספר קבוע של השמטות שנכון לכל מתזמן.
לבדיקת סיום מוקדם נשלח אות סיום רק למזהה תהליך הניגון שהתוכנית שלנו הדפיסה בזמן ההרצה. נוודא שההורה מחזיר שגיאה ואוסף את הילד. לא נשתמש בפקודה שמסיימת את כל תהליכי הניגון במחשב.
להתקן שמע אמיתי נבחר שם מתוך aplay -L בהתאם לסביבת המעבדה. נמדוד בנפרד את התנהגות השרת, ההתקן והשעון שלו. התקן התוכנה אינו מדמה רעב חומרה או זמן ניגון, ולכן הצלחה בו מוכיחה את מסלול ההעברה המצומצם בלבד.
שיפור המדיניות¶
השמטת תקופה חדשה גורמת לכך שזרם הבתים אינו שומר בדיוק את ציר הזמן המקורי, ויכולה ליצור הפרעות נשמעות. עבור מוצר נבחר חוזה אחר: סיום הזרם כשאיכות אינה נשמרת, השלמת שקט בזמן הנכון או התאמת קצב לפי שעון ההתקן. לא נוסיף תור בלתי מוגבל כדי להסתיר את הבעיה.
הרחבה לספריית שמע ישירה תחליף את תהליך הילד במתאם שמקבל באפרים ומחזיר אותם לפי חוזה השלמה. לפני השינוי נשמור את מבחני הבעלות, הסגירה וההאטה. כך נוכל להשוות שני מסלולים בלי לוותר על הראיות שכבר בנינו.