Process Queues: Managing System Resources

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

/* Conceptual device queue structure */
struct device_queue {
    /* Device identification */
    struct device *dev;          /* Associated device */
    char name[32];               /* Device name (e.g., "sda") */
    
    /* Queue of pending I/O requests */
    struct list_head request_queue;   /* I/O requests pending */
    struct request *current_request;  /* Currently serviced request */
    
    /* Queue of waiting processes (may be in request structures) */
    int waiting_count;                /* Number of blocked processes */
    
    /* Queue statistics */
    unsigned long total_requests;     /* Total requests serviced */
    unsigned long total_wait_time;    /* Cumulative wait time */
    
    /* Scheduling policy */
    struct elevator_type *elevator;   /* I/O scheduler */
    
    /* Synchronization */
    spinlock_t queue_lock;
    wait_queue_head_t wait;           /* Wait queue for blocking */
};
 
/* Individual I/O request structure */
struct request {
    struct list_head list;       /* Queue linkage */
    
    /* Request details */
    sector_t sector;             /* Starting sector (for block devices) */
    unsigned int nr_sectors;     /* Number of sectors */
    enum req_op operation;       /* READ, WRITE, FLUSH, etc. */
    
    /* Buffers */
    void *buffer;                /* Data buffer */
    struct bio *bio;             /* Bio chain (Linux) */
    
    /* Associated process */
    struct task_struct *task;    /* Process waiting for this request */
    
    /* Timing */
    unsigned long start_time;    /* When request was submitted */
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

/* Linux request queue structure (simplified) */
struct request_queue {
    /* Request lists */
    struct list_head queue_head;     /* List of pending requests */
    
    /* Queue limits */
    unsigned int nr_requests;        /* Max requests in queue */
    unsigned int nr_congestion_on;   /* Congestion threshold */
    
    /* I/O scheduler (elevator) */
    struct elevator_queue *elevator;
    
    /* Make request function - entry point from filesystem */
    make_request_fn *make_request_fn;
    
    /* Per-queue statistics */
    struct queue_limits limits;
    
    /* Backing device info */
    struct backing_dev_info *backing_dev_info;
    
    /* Synchronization */
    spinlock_t queue_lock;
};
 
/* Submit I/O request and potentially block */
void submit_bio_wait(struct bio *bio) {
    struct completion done;
    
    init_completion(&done);
    bio->bi_private = &done;
    bio->bi_end_io = bio_complete;
    
    /* Submit to device queue */
    submit_bio(bio);
    
    /* Block current process until I/O completes */
    wait_for_completion(&done);
}
 
/* Called by device driver when I/O completes */
void bio_complete(struct bio *bio) {
    struct completion *done = bio->bi_private;
    
    /* Wake up the waiting process */
    complete(done);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

/* Deadline scheduler maintains four queues */
struct deadline_data {
    /* Sorted queues (by sector number) for seeking efficiency */
    struct rb_root sort_list[2];    /* [READ] and [WRITE] sorted lists */
    
    /* FIFO queues (by time) for deadline enforcement */
    struct list_head fifo_list[2];  /* [READ] and [WRITE] deadline lists */
    
    /* Batching state */
    int fifo_batch;                 /* How many to service before checking other queue */
    int writes_starved;             /* Count of times reads preempted writes */
    
    /* Tunable deadlines */
    int read_expire;                /* Read deadline: 500ms default */
    int write_expire;               /* Write deadline: 5000ms default */
    
    /* Which direction we're currently servicing */
    int data_dir;                   /* READ or WRITE */
    sector_t last_sector;           /* Last sector serviced (for batching) */
};
 
/* Deadline scheduler dispatch logic */
struct request *deadline_dispatch(struct deadline_data *dd) {
    struct request *rq;
    int data_dir = dd->data_dir;
    
    /* Check if there's an expired request (deadline missed) */
    rq = deadline_check_fifo(dd, READ);  /* Reads have priority */
    if (rq)
        return rq;  /* Must service this immediately */
    
    rq = deadline_check_fifo(dd, WRITE);
    if (rq)
        return rq;
    
    /* No deadline pressure - batch requests in current direction */
    if (deadline_has_work(dd, data_dir)) {
        /* Get next request in sector order for efficient seeking */
        rq = deadline_next_request(dd, data_dir);
        if (rq)
            return rq;
    }
    
    /* Switch direction if other direction has work */
    if (deadline_has_work(dd, !data_dir)) {
        dd->data_dir = !data_dir;
        rq = deadline_next_request(dd, !data_dir);
        if (rq)
            return rq;
    }
    
    return NULL;  /* No work */
}
 
/* Check if any request in this queue has exceeded its deadline */
struct request *deadline_check_fifo(struct deadline_data *dd, int dir) {
    struct request *rq = list_first_entry_or_null(
        &dd->fifo_list[dir], struct request, queuelist);
    
    if (rq) {
        unsigned long deadline = dir == READ ? 
            dd->read_expire : dd->write_expire;
        
        if (time_after(jiffies, rq->start_time + deadline)) {
            /* This request has waited too long */
            return rq;
        }
    }
    
    return NULL;  /* No expired requests */
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

/* Wait queue head - one per condition/event */
struct wait_queue_head {
    spinlock_t lock;
    struct list_head head;  /* List of waiting tasks */
};
typedef struct wait_queue_head wait_queue_head_t;
 
/* Wait queue entry - one per waiting task */
struct wait_queue_entry {
    unsigned int flags;
    void *private;           /* Usually the task_struct pointer */
    wait_queue_func_t func;  /* Function to call on wake */
    struct list_head entry;  /* List linkage */
};
 
/* Initialize a wait queue head */
#define DECLARE_WAIT_QUEUE_HEAD(name) \
    struct wait_queue_head name = { \
        .lock = __SPIN_LOCK_UNLOCKED(name.lock), \
        .head = LIST_HEAD_INIT(name.head) \
    }
 
/* Initialize a wait queue entry */
#define DECLARE_WAITQUEUE(name, tsk) \
    struct wait_queue_entry name = { \
        .private = tsk, \
        .func = default_wake_function, \
        .entry = LIST_HEAD_INIT(name.entry) \
    }
 
/* Adding/removing from wait queue */
void add_wait_queue(struct wait_queue_head *wq_head,
                    struct wait_queue_entry *wq_entry) {
    unsigned long flags;
    spin_lock_irqsave(&wq_head->lock, flags);
    list_add(&wq_entry->entry, &wq_head->head);
    spin_unlock_irqrestore(&wq_head->lock, flags);
}
 
void remove_wait_queue(struct wait_queue_head *wq_head,
                       struct wait_queue_entry *wq_entry) {
    unsigned long flags;
    spin_lock_irqsave(&wq_head->lock, flags);
    list_del(&wq_entry->entry);
    spin_unlock_irqrestore(&wq_head->lock, flags);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

/* Correct blocking pattern - avoids lost wakeup race */
int wait_for_io_completion(struct io_request *req) {
    DECLARE_WAITQUEUE(wait, current);
    int result;
    
    /* Add ourselves to wait queue BEFORE checking condition */
    add_wait_queue(&req->wait_queue, &wait);
    
    while (1) {
        /* Set state to INTERRUPTIBLE or UNINTERRUPTIBLE */
        set_current_state(TASK_INTERRUPTIBLE);
        
        /* Check if condition is already satisfied */
        if (io_request_complete(req))
            break;
        
        /* Check for signals (if INTERRUPTIBLE) */
        if (signal_pending(current)) {
            result = -ERESTARTSYS;
            goto out;
        }
        
        /* Actually sleep - scheduler will run another task */
        schedule();
        
        /* We've been woken up - loop back to check condition */
    }
    
    result = req->result;
    
out:
    /* Reset state and remove from wait queue */
    set_current_state(TASK_RUNNING);
    remove_wait_queue(&req->wait_queue, &wait);
    
    return result;
}
 
/* Wake up waiting process when I/O completes (interrupt context) */
void complete_io_request(struct io_request *req) {
    /* Mark request complete */
    req->complete = true;
    req->result = 0;  /* Success */
    
    /* Wake up waiters */
    wake_up(&req->wait_queue);
}
 
/* Convenience macros hide this complexity */
/* wait_event(wq, condition) - uninterruptible wait */
/* wait_event_interruptible(wq, condition) - can be signaled */
/* wait_event_timeout(wq, condition, timeout) - with timeout */
 
int read_from_device(struct device *dev, void *buf, size_t count) {
    struct io_request req;
    
    init_io_request(&req, READ, buf, count);
    submit_io_request(dev, &req);
    
    /* This macro handles all the sleep/wake logic */
    return wait_event_interruptible(req.wait_queue, 
                                     io_request_complete(&req));
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

/* Device driver interrupt handler */
irqreturn_t disk_interrupt_handler(int irq, void *dev_id) {
    struct block_device *bdev = dev_id;
    struct request *req;
    
    /* Acknowledge interrupt to hardware */
    ack_device_interrupt(bdev);
    
    /* Get the completed request */
    req = bdev->current_request;
    if (!req)
        return IRQ_NONE;  /* Spurious interrupt */
    
    /* Transfer data if this was a read */
    if (req->operation == READ) {
        copy_from_device_buffer(req->buffer, bdev, req->nr_sectors);
    }
    
    /* Mark request complete */
    req->error = 0;  /* Success */
    
    /* Complete the request - this wakes the waiting process */
    blk_mq_complete_request(req);
    
    /* Start next request if any */
    bdev->current_request = get_next_request(bdev);
    if (bdev->current_request)
        start_device_operation(bdev);
    
    return IRQ_HANDLED;
}
 
/* Request completion path */
void blk_mq_complete_request(struct request *rq) {
    /* May need to defer to softirq if in hard irq context */
    if (in_hardirq()) {
        /* Schedule completion to run in softirq context */
        raise_softirq(BLOCK_SOFTIRQ);
        return;
    }
    
    /* Call the request's completion function */
    rq->end_io(rq, rq->error);
}
 
/* Completion callback set by the submitting code */
void bio_end_io(struct bio *bio) {
    struct task_struct *waiter = bio->bi_private;
    
    /* Record any errors */
    bio->bi_status = BLK_STS_OK;
    
    /* Wake up the process that submitted this I/O */
    wake_up_process(waiter);
}
 
/* wake_up_process implementation */
int wake_up_process(struct task_struct *p) {
    return try_to_wake_up(p, TASK_NORMAL, 0);
}
 
int try_to_wake_up(struct task_struct *p, 
                   unsigned int state, 
                   int wake_flags) {
    unsigned long flags;
    struct rq *rq;
    
    /* Get the run queue this task is associated with */
    raw_spin_lock_irqsave(&p->pi_lock, flags);
    
    /* Check if task is in a wakeable state */
    if (!(p->state & state))
        goto out_unlock;
    
    /* Mark task as runnable */
    p->state = TASK_RUNNING;
    
    /* Add to ready queue */
    rq = task_rq(p);
    activate_task(rq, p, ENQUEUE_WAKEUP);
    
    /* Check if we should preempt current task */
    check_preempt_curr(rq, p, wake_flags);
    
out_unlock:
    raw_spin_unlock_irqrestore(&p->pi_lock, flags);
    return 0;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

/* epoll: Efficiently wait on thousands of file descriptors */
#include <sys/epoll.h>
 
int main() {
    /* Create epoll instance */
    int epfd = epoll_create1(0);
    
    /* Add file descriptors to monitor */
    struct epoll_event ev;
    ev.events = EPOLLIN;  /* Wait for read readiness */
    ev.data.fd = socket_fd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, socket_fd, &ev);
    
    ev.data.fd = file_fd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, file_fd, &ev);
    
    /* Wait for events */
    struct epoll_event events[MAX_EVENTS];
    while (1) {
        /* Block until at least one fd is ready */
        int nfds = epoll_wait(epfd, events, MAX_EVENTS, -1);
        
        /* Process ready file descriptors */
        for (int i = 0; i < nfds; i++) {
            if (events[i].data.fd == socket_fd) {
                handle_socket_data();
            } else if (events[i].data.fd == file_fd) {
                handle_file_data();
            }
        }
    }
}
 
/* Under the hood, epoll uses a red-black tree + ready list */
struct eventpoll {
    /* Lock protecting this structure */
    spinlock_t lock;
    
    /* RB-tree of monitored file descriptors */
    struct rb_root rbr;
    
    /* List of file descriptors ready for userspace */
    struct list_head rdllist;
    
    /* Wait queue for epoll_wait callers */
    wait_queue_head_t wq;
};
 
/* When any monitored fd becomes ready:
 * 1. fd's poll callback triggers
 * 2. Epitem is added to rdllist
 * 3. Process waiting in epoll_wait is woken
 * 4. Ready fds returned to userspace
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

/* io_uring: Ring-buffer based async I/O */
#include <liburing.h>
 
int main() {
    struct io_uring ring;
    struct io_uring_sqe *sqe;
    struct io_uring_cqe *cqe;
    
    /* Initialize ring */
    io_uring_queue_init(32, &ring, 0);
    
    /* Get a submission queue entry */
    sqe = io_uring_get_sqe(&ring);
    
    /* Prepare a read operation (doesn't submit yet) */
    io_uring_prep_read(sqe, fd, buffer, sizeof(buffer), 0);
    sqe->user_data = 123;  /* Tag for completion matching */
    
    /* Submit all prepared ops to kernel */
    io_uring_submit(&ring);
    
    /* Wait for completion */
    io_uring_wait_cqe(&ring, &cqe);
    
    /* Process result */
    if (cqe->res >= 0) {
        printf("Read %d bytes\n", cqe->res);
    }
    
    /* Mark completion as consumed */
    io_uring_cqe_seen(&ring, cqe);
    
    io_uring_queue_exit(&ring);
}
 
/*
 * io_uring advantage: batched submissions
 * 
 * Traditional approach: 1000 reads = 1000 syscalls
 * io_uring approach: 1000 reads = ~1-2 syscalls
 * 
 * Also supports:
 * - Polling mode (no interrupts needed)
 * - Linked operations
 * - Fixed buffers (no copy overhead)
 */