I/O Multiplexing

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

// Conceptual I/O multiplexing pattern
int server_loop(int listen_fd, int *clients, int num_clients) {
    while (1) {
        // Build set of all descriptors we care about
        fd_set interest_set;
        build_interest_set(&interest_set, listen_fd, clients, num_clients);
        
        // Block until ANY descriptor is ready (or timeout)
        // This is the key - ONE system call, watches EVERYTHING
        int num_ready = select(max_fd + 1, &interest_set, NULL, NULL, NULL);
        
        if (num_ready < 0) {
            handle_error();
            continue;
        }
        
        // Check which descriptors are ready
        // Only iterate over the ones we know are ready
        
        if (FD_ISSET(listen_fd, &interest_set)) {
            // New connection ready to accept
            int new_client = accept(listen_fd, NULL, NULL);  // Won't block!
            add_client(new_client);
        }
        
        for (int i = 0; i < num_clients; i++) {
            if (FD_ISSET(clients[i], &interest_set)) {
                // This client has data ready - read won't block
                handle_client_data(clients[i]);
            }
        }
    }
}

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

// The readiness model workflow
 
// 1. Wait for readiness (this blocks)
wait_for_readiness(fd);
 
// 2. Perform I/O (this does NOT block because we know it's ready)
ssize_t n = read(fd, buf, sizeof(buf));
 
// 3. Handle the result
if (n > 0) {
    // Got data
    process(buf, n);
} else if (n == 0) {
    // EOF
    close(fd);
} else {
    // Error (shouldn't be EAGAIN since we checked readiness)
    handle_error();
}
 
/*
 * Important: Readiness doesn't guarantee you'll get all data you want.
 * If you need 1000 bytes but only 100 are ready, you'll read 100.
 * For complete messages, you often need to:
 * 1. Accumulate data across multiple ready notifications
 * 2. Parse to find message boundaries
 * 3. Process complete messages only
 */

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

#include <sys/epoll.h>
#include <unistd.h>
#include <stdio.h>
 
#define MAX_EVENTS 64
 
typedef void (*event_handler)(int fd, void *data);
 
struct event_data {
    event_handler read_handler;
    event_handler write_handler;
    void *user_data;
};
 
/**
 * A simple event loop using epoll
 */
void event_loop(int epoll_fd) {
    struct epoll_event events[MAX_EVENTS];
    
    while (1) {
        // Wait for events (this is where we spend most time blocked)
        int num_events = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
        
        if (num_events < 0) {
            if (errno == EINTR) continue;  // Signal interrupted us
            perror("epoll_wait");
            break;
        }
        
        // Dispatch events to handlers
        for (int i = 0; i < num_events; i++) {
            struct event_data *data = events[i].data.ptr;
            
            if (events[i].events & EPOLLIN) {
                // Read event
                data->read_handler(events[i].data.fd, data->user_data);
            }
            
            if (events[i].events & EPOLLOUT) {
                // Write event
                data->write_handler(events[i].data.fd, data->user_data);
            }
            
            if (events[i].events & (EPOLLERR | EPOLLHUP)) {
                // Error or hangup
                handle_error(events[i].data.fd);
            }
        }
        
        // Process timers, deferred work, etc.
        process_timers();
        process_deferred();
    }
}

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

/*
 * Level-Triggered vs Edge-Triggered Behavior
 * 
 * Scenario: 100 bytes arrive on a socket
 * 
 * LEVEL-TRIGGERED:
 * ────────────────────────────────────────────
 * 1. epoll_wait() returns: socket is readable
 * 2. You read 50 bytes
 * 3. epoll_wait() returns AGAIN: socket still readable (50 bytes left)
 * 4. You read 50 bytes
 * 5. epoll_wait() blocks: no more data
 * 
 * LT is forgiving: read what you want, come back later.
 * 
 * 
 * EDGE-TRIGGERED:
 * ────────────────────────────────────────────
 * 1. epoll_wait() returns: socket became readable (edge!)
 * 2. You read 50 bytes
 * 3. epoll_wait() BLOCKS: no new edge occurred!
 * 4. You're stuck - 50 bytes remain unread forever!
 * 
 * ET requires: read until EAGAIN, every time.
 */
 
// CORRECT edge-triggered read pattern
void et_read_handler(int fd) {
    char buffer[4096];
    
    // MUST loop until EAGAIN - can't leave data!
    while (1) {
        ssize_t n = read(fd, buffer, sizeof(buffer));
        
        if (n > 0) {
            // Got data, process it
            process_data(buffer, n);
            continue;  // There might be more!
        }
        
        if (n == 0) {
            // EOF - connection closed
            close(fd);
            return;
        }
        
        // n < 0
        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            // No more data RIGHT NOW - safe to return
            // epoll will notify us when more arrives
            return;
        }
        
        // Actual error
        handle_error(fd);
        return;
    }
}
 
// WRONG edge-triggered read - WILL LOSE DATA
void et_read_handler_BROKEN(int fd) {
    char buffer[4096];
    
    // Only reads once!
    ssize_t n = read(fd, buffer, sizeof(buffer));
    if (n > 0) {
        process_data(buffer, n);
    }
    // BUG: If more than 4096 bytes arrived, we lose the rest!
    // Edge-triggered won't notify again for data already present.
}

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

#include <sys/epoll.h>
#include <sys/signalfd.h>
#include <sys/timerfd.h>
#include <signal.h>
#include <unistd.h>
 
/**
 * Unified event handling: sockets, signals, timers
 * All through epoll!
 */
int setup_unified_event_loop() {
    int epoll_fd = epoll_create1(0);
    
    // 1. Add listening socket
    int listen_fd = create_listen_socket(8080);
    add_to_epoll(epoll_fd, listen_fd, EPOLLIN);
    
    // 2. Add signal handling via signalfd
    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGINT);
    sigaddset(&mask, SIGTERM);
    sigprocmask(SIG_BLOCK, &mask, NULL);  // Block normal delivery
    
    int signal_fd = signalfd(-1, &mask, SFD_NONBLOCK);
    add_to_epoll(epoll_fd, signal_fd, EPOLLIN);
    
    // 3. Add periodic timer
    int timer_fd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
    struct itimerspec timer_spec = {
        .it_interval = { .tv_sec = 1, .tv_nsec = 0 },  // 1 second repeat
        .it_value = { .tv_sec = 1, .tv_nsec = 0 }
    };
    timerfd_settime(timer_fd, 0, &timer_spec, NULL);
    add_to_epoll(epoll_fd, timer_fd, EPOLLIN);
    
    // 4. Add stdin for interactive commands
    add_to_epoll(epoll_fd, STDIN_FILENO, EPOLLIN);
    
    // Now epoll_wait handles:
    // - New network connections
    // - Shutdown signals (SIGINT, SIGTERM)
    // - Periodic housekeeping (timer)
    // - User commands (stdin)
    // All in one unified loop!
    
    return epoll_fd;
}
 
void event_loop(int epoll_fd) {
    struct epoll_event events[64];
    
    while (running) {
        int n = epoll_wait(epoll_fd, events, 64, -1);
        
        for (int i = 0; i < n; i++) {
            int fd = events[i].data.fd;
            
            if (fd == signal_fd) {
                // Read signal info
                struct signalfd_siginfo info;
                read(signal_fd, &info, sizeof(info));
                printf("Received signal %d
", info.ssi_signo);
                running = 0;  // Shutdown
                
            } else if (fd == timer_fd) {
                // Timer expired - do periodic work
                uint64_t expirations;
                read(timer_fd, &expirations, sizeof(expirations));
                do_housekeeping();
                
            } else if (fd == listen_fd) {
                // New connection
                accept_connection(listen_fd);
                
            } else if (fd == STDIN_FILENO) {
                // User command
                handle_user_input();
                
            } else {
                // Client data
                handle_client(fd);
            }
        }
    }
}