פתרון פרויקט תור אירועים בקרנל

חוזה הזיכרון והבעלות

הפתרון הבא משתמש בתור גלובלי קבוע, ולכן אינו מקצה זיכרון במהלך קריאה וכתיבה. הנעילה מגנה על הראש, הכמות, הרצף והמצב הסופי. תנאי המתנה נקראים באמצעות גישות יחידות, אך לאחר ההתעוררות מתבצעת בדיקה חוזרת תחת הנעילה. ההתעוררות היא הזמנה לבדוק מצב, ואינה הקצאה של אירוע לקורא מסוים.

הקוד המלא נשמר בקובץ events.c ונבנה כמו המודולים הקודמים עם obj-m += events.o. הממשק נגיש למנהל באורח המעבדה. בחירה בהרשאות רחבות יותר מחייבת מדיניות סמכות מפורשת לפעולת ההפסקה הגלובלית.

#include <linux/module.h>
#include <linux/miscdevice.h>
#include <linux/fs.h>
#include <linux/mutex.h>
#include <linux/poll.h>
#include <linux/uaccess.h>
#include <linux/ioctl.h>
#include <linux/byteorder/generic.h>

#define CAPACITY 32
#define AMIT_STOP _IO('A', 1)
struct event { __le64 sequence; __le64 value; };
static struct event entries[CAPACITY];
static DEFINE_MUTEX(lock);
static DECLARE_WAIT_QUEUE_HEAD(waiters);
static unsigned int head, count;
static u64 sequence;
static bool stopped;

static ssize_t events_read(struct file *file, char __user *buffer,
                           size_t length, loff_t *position)
{
    int error;
    if (!length)
        return 0;
    if (length < sizeof(struct event))
        return -EMSGSIZE;
    for (;;) {
        if (mutex_lock_interruptible(&lock))
            return -ERESTARTSYS;
        if (count)
            break;
        if (stopped) {
            mutex_unlock(&lock);
            return 0;
        }
        mutex_unlock(&lock);
        if (file->f_flags & O_NONBLOCK)
            return -EAGAIN;
        error = wait_event_interruptible(waiters,
                    READ_ONCE(count) || READ_ONCE(stopped));
        if (error)
            return error;
    }
    if (copy_to_user(buffer, &entries[head], sizeof(struct event))) {
        mutex_unlock(&lock);
        return -EFAULT;
    }
    head = (head + 1) % CAPACITY;
    WRITE_ONCE(count, count - 1);
    mutex_unlock(&lock);
    wake_up_interruptible_poll(&waiters, EPOLLOUT | EPOLLWRNORM);
    return sizeof(struct event);
}

static ssize_t events_write(struct file *file, const char __user *buffer,
                            size_t length, loff_t *position)
{
    __le64 value;
    unsigned int tail;
    int error;
    if (length != sizeof(value))
        return -EINVAL;
    if (copy_from_user(&value, buffer, sizeof(value)))
        return -EFAULT;
    for (;;) {
        if (mutex_lock_interruptible(&lock))
            return -ERESTARTSYS;
        if (stopped) {
            mutex_unlock(&lock);
            return -ENODEV;
        }
        if (count < CAPACITY)
            break;
        mutex_unlock(&lock);
        if (file->f_flags & O_NONBLOCK)
            return -EAGAIN;
        error = wait_event_interruptible(waiters,
                    READ_ONCE(count) < CAPACITY || READ_ONCE(stopped));
        if (error)
            return error;
    }
    if (sequence == U64_MAX) {
        mutex_unlock(&lock);
        return -EOVERFLOW;
    }
    tail = (head + count) % CAPACITY;
    entries[tail].sequence = cpu_to_le64(++sequence);
    entries[tail].value = value;
    WRITE_ONCE(count, count + 1);
    mutex_unlock(&lock);
    wake_up_interruptible_poll(&waiters, EPOLLIN | EPOLLRDNORM);
    return sizeof(value);
}

static __poll_t events_poll(struct file *file, poll_table *table)
{
    __poll_t result = 0;
    poll_wait(file, &waiters, table);
    mutex_lock(&lock);
    if (count)
        result |= EPOLLIN | EPOLLRDNORM;
    if (!stopped && count < CAPACITY)
        result |= EPOLLOUT | EPOLLWRNORM;
    if (stopped)
        result |= EPOLLHUP;
    mutex_unlock(&lock);
    return result;
}

static long events_ioctl(struct file *file, unsigned int command,
                         unsigned long argument)
{
    if (command != AMIT_STOP)
        return -ENOTTY;
    mutex_lock(&lock);
    WRITE_ONCE(stopped, true);
    mutex_unlock(&lock);
    wake_up_interruptible_poll(&waiters, EPOLLIN | EPOLLOUT | EPOLLHUP);
    return 0;
}

static const struct file_operations operations = {
    .owner = THIS_MODULE,
    .open = nonseekable_open,
    .read = events_read,
    .write = events_write,
    .poll = events_poll,
    .unlocked_ioctl = events_ioctl,
    .compat_ioctl = events_ioctl,
    .llseek = no_llseek,
};
static struct miscdevice device = {
    .minor = MISC_DYNAMIC_MINOR,
    .name = "amit_events",
    .fops = &operations,
    .mode = 0600,
};
static int __init events_init(void) { return misc_register(&device); }
static void __exit events_exit(void) { misc_deregister(&device); }
module_init(events_init);
module_exit(events_exit);
MODULE_LICENSE("GPL");

לקוח עם בדיקות חוזה

הלקוח הבא משתמש בספריית libc לצורך דגל הפתיחה ופעולת הבקרה. נשמור אותו במיזם ראסט רגיל עם תלות זו. ערך הבקרה המספרי מתאים לקידוד של יעד המעבדה, ולא נציג אותו כקבוע נייד לכל ארכיטקטורה.

use std::fs::OpenOptions;
use std::io::{self, Read, Write};
use std::os::fd::AsRawFd;
use std::os::unix::fs::OpenOptionsExt;

fn main() -> io::Result<()> {
    let mut file = OpenOptions::new().read(true).write(true)
        .custom_flags(libc::O_NONBLOCK).open("/dev/amit_events")?;
    let mut event = [0u8; 16];
    assert_eq!(file.read(&mut event).unwrap_err().kind(),
               io::ErrorKind::WouldBlock);
    for value in 100u64..132 {
        assert_eq!(file.write(&value.to_le_bytes())?, 8);
    }
    assert_eq!(file.write(&0u64.to_le_bytes()).unwrap_err().kind(),
               io::ErrorKind::WouldBlock);
    let mut short = [0u8; 8];
    assert_eq!(file.read(&mut short).unwrap_err().raw_os_error(),
               Some(libc::EMSGSIZE));
    if unsafe { libc::ioctl(file.as_raw_fd(), 0x4101) } < 0 {
        return Err(io::Error::last_os_error());
    }
    for expected in 1u64..=32 {
        assert_eq!(file.read(&mut event)?, 16);
        let sequence = u64::from_le_bytes(event[..8].try_into().unwrap());
        let value = u64::from_le_bytes(event[8..].try_into().unwrap());
        assert_eq!(sequence, expected);
        assert_eq!(value, expected + 99);
    }
    assert_eq!(file.read(&mut event)?, 0);
    assert_eq!(file.write(&1u64.to_le_bytes()).unwrap_err().raw_os_error(),
               Some(libc::ENODEV));
    println!("all contract checks passed");
    Ok(())
}

הלקוח מניח מודול שזה עתה נטען וללא לקוחות נוספים. לאחר הבדיקה נטען מחדש לפני תרחיש אחר. עבור אורח האתחול המינימלי נבנה לקוח סטטי ליעד x86_64-unknown-linux-musl, נוודא באמצעות file שאין צורך במפרש דינמי ונעתיק אותו לתמונת האתחול. אם סביבת הבנייה דורשת כלי קישור ליעד זה, נתקין ונגדיר אותם במכונת הבנייה. חלופה תקינה היא אורח הפצה רגיל עם הספריות הנדרשות, ולא העתקת קובץ דינמי לבדו למערכת מינימלית.

rustup target add x86_64-unknown-linux-musl
cargo build --release --target x86_64-unknown-linux-musl
file target/x86_64-unknown-linux-musl/release/events-client

בדיקות מקביליות ומוכנות

לבדיקת חסימה נפתח מתאר רגיל לקורא ותהליך נפרד לכותב. הקורא ימתין בתור ריק ויתעורר אחרי כתיבה. לבדיקת מוכנות נרשום את המתאר ב־epoll, נכתוב ואז נקרא עד שגיאת מוכנות. לאחר הפסקה נתייחס לתלייה יחד עם אפשרות קלט, מפני שאירועים שכבר נשמרו עדיין זמינים.

לבדיקת שני כותבים נשלח ערכים שמכילים מזהה כותב ומונה מקומי. הקורא יוודא שהרצף הגלובלי עולה ללא חורים ושכל ערך מופיע פעם אחת. אין דרישה לסדר קבוע בין הכותבים, מפני שסדר רכישת הנעילה אינו מובטח. בעת תור מלא הכותבים החוסמים ימתינו, ולכן חייב לפעול קורא במקביל.

מגבלות והרחבה מוצדקת

הפתרון מחזיק נעילה ישנה בזמן העתקה למשתמש. הדבר מותר מבחינת הקשר הביצוע, אך תהליך יכול לעכב את ההעתקה ולחסום את כלל התור. זהו מחיר של חוזה שבו אירוע אינו נצרך אם ההעתקה נכשלת. שיפור אפשרי הוא העברת האירוע למצב שמור לכל פתיחה והגדרת סמנטיקת ניסיון חוזר, אך אז חייבים לטפל בסגירת הפתיחה בזמן שקיים אירוע שמור.

כשל העתקה עשוי כבר לשנות חלק מהבאפר של המשתמש, למרות שהאירוע נשאר בתור. הלקוח חייב להתעלם מתוכן הבאפר כאשר הקריאה מחזירה שגיאה. בפריקה רגילה הפניות הקובץ מחזיקות את קוד המודול בזכות שדה הבעלות. אין בכך פתרון לניתוק חומרה אמיתית, והפרויקט אינו משתמש בפריקה כפויה.

הראיה הסופית מורכבת מלוגיקת תור שנבדקה, חוזה בינארי מפורש, בדיקות שגיאה ותיעוד חיי המשאבים. פלט הצלחה של הלקוח הבודד מכסה חלק מהחוזה בלבד. דוח ההגשה יכלול בנפרד את בדיקות המקביליות, המוכנות וההעתקה שנערכו בפועל.