8.3 - יקיצה ומימוש executor בסיסי - הרצאה
מי קורא שוב לעתיד¶
בהרצאות הקודמות קראנו ל־poll בעצמנו, משום שידענו מראש מתי הדוגמה תוכל להתקדם. בתוכנית אמיתית איננו יודעים מתי תגיע תשובת רשת או מתי תתפנה משבצת בתור. אם נבדוק שוב ושוב, נבזבז זמן מעבד. אם נלך לישון בלי לרשום דרך להעיר אותנו, אנחנו עלולים להחמיץ בדיוק את האירוע שחיכינו לו.
המנוע שנבנה עכשיו מחזיק תור של משימות שאפשר לבדוק. מקור האירוע אינו מפעיל את העתיד ישירות, אלא מכניס את המשימה לתור. התהליכון שמריץ את המנוע מוציא משימה, בודק אותה ומחזיר לעצמו את השליטה. ההפרדה הזאת חשובה משום שהאירוע עשוי להגיע מתוך תהליכון אחר, בזמן שהעתיד עדיין נמצא בתוך בדיקה קודמת.
המונח יקיצה אינו מבטיח שהמשימה כבר יכולה להחזיר תוצאה. הוא אומר שיש סיבה לבדוק שוב. התוצאה עצמה נשמרת במקום אחר, והעתיד צריך לבדוק את התנאי מחדש. גם שתי יקיצות קרובות אינן מחייבות שתי תוצאות או שתי בדיקות נפרדות במנוע יעיל. בהמשך נלמד לאחד יקיצות; כעת נשתמש בתור פשוט כדי לראות את פרוטוקול הבעלות.
תור משימות שאפשר לקרוא¶
הדוגמה הבאה היא מנוע לימודי בעל תהליכון מפעיל אחד. הוא מקבל את כל המשימות לפני תחילת ההרצה, אינו מאפשר יצירת משימות מתוך המנוע, ומשתמש בתור בלתי מוגבל. אלה גבולות מכוונים של הדוגמה, ונטפל בהשלכות שלהם בהמשך הפרק.
use std::future::Future;
use std::pin::Pin;
use std::sync::{mpsc, Arc, Mutex};
use std::task::{Context, Poll, Wake, Waker};
type Job = Pin<Box<dyn Future<Output = ()> + Send>>;
struct Task {
job: Mutex<Option<Job>>,
ready: mpsc::Sender<Arc<Task>>,
}
impl Wake for Task {
fn wake(self: Arc<Self>) {
let _ = self.ready.send(self.clone());
}
}
struct Executor {
ready: mpsc::Sender<Arc<Task>>,
queue: mpsc::Receiver<Arc<Task>>,
remaining: usize,
}
impl Executor {
fn new() -> Self {
let (ready, queue) = mpsc::channel();
Self { ready, queue, remaining: 0 }
}
fn spawn(&mut self, future: impl Future<Output = ()> + Send + 'static) {
let task = Arc::new(Task {
job: Mutex::new(Some(Box::pin(future))),
ready: self.ready.clone(),
});
self.remaining += 1;
self.ready.send(task).unwrap();
}
fn run(mut self) {
while self.remaining != 0 {
let task = self.queue.recv().expect("ready queue disconnected");
let waker = Waker::from(task.clone());
let mut context = Context::from_waker(&waker);
let mut slot = task.job.lock().unwrap();
if let Some(future) = slot.as_mut() {
if future.as_mut().poll(&mut context).is_ready() {
*slot = None;
self.remaining -= 1;
}
}
}
}
}
המעטפת של המשימה שורדת כל עוד יש עותק של היקיצה, איבר בתור או בדיקה פעילה. המנעול מגן על המשבצת שבה העתיד נמצא. בדוגמה בעלת מפעיל יחיד אין שתי בדיקות בו־זמנית, אבל צורת האחסון מאפשרת ליקיצה לשתף את המשימה בבטחה עם תהליכון חיצוני. היקיצה עצמה אינה נועלת את משבצת העתיד, ולכן אפשר לבקש יקיצה מתוך poll בלי לנעול את אותו מנעול מחדש.
לאחר השלמה אנחנו מרוקנים את המשבצת. אם בתור כבר קיימת יקיצה נוספת לאותה משימה, המנוע יראה משבצת ריקה ולא יבדוק שוב עתיד שהושלם. המונה קטן רק כאשר עתיד קיים מחזיר השלמה. כך אירוע ישן אינו גורם לספירה כפולה או לקריאה אסורה נוספת.
אירוע שמגיע בזמן הרישום¶
נוסיף עתיד שמחכה לתהליכון חיצוני. המצב המשותף מכיל גם את התנאי וגם את היקיצה האחרונה. שני הנתונים מוגנים באותו מנעול, ולכן אין חלון שבו העתיד כבר החליט להמתין אבל היצרן עדיין אינו יכול למצוא את היקיצה שלו.
use std::thread;
use std::time::Duration;
struct State {
complete: bool,
waiter: Option<Waker>,
}
struct Delay {
state: Arc<Mutex<State>>,
}
impl Delay {
fn new(duration: Duration) -> Self {
let state = Arc::new(Mutex::new(State {
complete: false,
waiter: None,
}));
let worker_state = state.clone();
thread::spawn(move || {
thread::sleep(duration);
let waiter = {
let mut state = worker_state.lock().unwrap();
state.complete = true;
state.waiter.take()
};
if let Some(waiter) = waiter {
waiter.wake();
}
});
Self { state }
}
}
impl Future for Delay {
type Output = ();
fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> {
let mut state = self.state.lock().unwrap();
if state.complete {
Poll::Ready(())
} else {
state.waiter = Some(context.waker().clone());
Poll::Pending
}
}
}
fn main() {
let mut executor = Executor::new();
executor.spawn(async {
Delay::new(Duration::from_millis(30)).await;
println!("first completed");
});
executor.spawn(async {
Delay::new(Duration::from_millis(10)).await;
println!("second completed");
});
executor.run();
}
שמרו את שני חלקי הקוד באותו קובץ. הפקודות הבאות מיועדות להרצה מקומית.
התוצאה הצפויה היא שתי שורות השלמה ויציאה תקינה. בדרך כלל המשימה בעלת ההמתנה הקצרה תדפיס ראשונה, אבל לא נשתמש בסדר הזה כאינווריאנט של הבדיקה. המתזמן של מערכת ההפעלה אינו מבטיח שתהליכון מסוים ירוץ בדיוק בתום משך ההמתנה שביקש.
הוכחת מסלול היקיצה¶
אם העובד מסיים לפני הבדיקה הראשונה, העתיד ימצא תנאי אמיתי ויחזיר מיד השלמה. במקרה הזה אין צורך ביקיצה כלל. אם הבדיקה מגיעה קודם, היא שומרת את היקיצה לפני שחרור המנעול. העובד הבא שמקבל את המנעול יסמן השלמה וייקח את היקיצה. בשני המקרים נשמרת דרך להתקדם.
כעת נשנה בדמיון את המימוש כך שבדיקת התנאי ורישום היקיצה מתבצעים תחת שתי נעילות נפרדות. העובד יכול לסיים בין שתי הנעילות, לראות שאין עדיין ממתין ולצאת. אחריו העתיד ירשום יקיצה ויחזיר המתנה. התנאי כבר אמיתי, אבל איש אינו מתכוון להעיר את המשימה. זהו אובדן יקיצה, והוא אינו נפתר באמצעות הגדלת משך ההמתנה או הוספת הדפסות.
העובד קורא ליקיצה אחרי שחרור המנעול של המצב. כך המנוע יכול לבדוק את העתיד מיד בלי להיתקע מאחורי נעילה שהעובד עדיין מחזיק. בדוגמה שלנו היקיצה רק שולחת לתור, אך ממשק כללי אינו צריך להניח שכל מימוש יקיצה הוא זול או שאינו קורא לקוד אחר.
בעלות ברמה של מצביע גולמי¶
הממשק הבטוח Wake מבוסס על Arc. מתחתיו אפשר לבנות יקיצה באמצעות RawWaker וטבלת פעולות. הטבלה מתארת שכפול, יקיצה שצורכת בעלות, יקיצה דרך השאלה והשמדה. בכל פעולה צריך לדעת האם היא מחזיקה יחידת בעלות על הנתון שמאחורי המצביע.
מימוש שגוי של שכפול עלול לשחרר את המשימה בזמן שעדיין קיימת יקיצה שמצביעה אליה. מימוש שגוי של השמדה עלול לדלוף משימות או לשחרר אותן פעמיים. מימוש שמבוסס על מונה שאינו מתאים לתהליכונים עלול להיות שגוי גם אם הניסוי היחיד שלנו רץ בתהליכון אחד, משום שהטיפוס הרגיל Waker רשאי לעבור בין תהליכונים. לכן נעדיף את הממשק הבטוח כל עוד אין דרישה מדידה שמצדיקה מימוש גולמי.
התיעוד של היקיצה הגולמית ושל ממשק היקיצה הבטוח הוא חלק מסקירת החוזה. המנוע שכתבנו עדיין אינו מנוע לשימוש ייצור: אין בו מדיניות פאניקה, חסימת עומס, ביטול או תור מוגבל. עם זאת, אפשר כבר להסביר עבור כל המתנה מי מחזיק את המצב ומי רשאי לבקש בדיקה נוספת.