Read-Ahead: Anticipatory Data Loading

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
89
90
91
92
93

/* Metrics tracked for adaptive read-ahead decisions */
 
struct ra_metrics {
    /* Cache effectiveness */
    atomic64_t pages_prefetched;     /* Total pages prefetched */
    atomic64_t pages_used;           /* Prefetched pages actually read */
    atomic64_t cache_hits;           /* Reads served from cache */
    atomic64_t cache_misses;         /* Reads that required I/O */
    
    /* Memory pressure */
    atomic_t   evictions_while_valid;/* Pages evicted before read */
    atomic_t   memory_pressure_events;
    
    /* Timing */
    u64        last_read_time_ns;    /* Time of last read */
    u64        cumulative_wait_ns;   /* Total I/O wait time */
    u64        sample_period_ns;     /* Measurement window */
    
    /* Pattern tracking */
    int        pattern_breaks;       /* Sequential pattern disruptions */
    int        pattern_stable_count; /* Consecutive stable measurements */
};
 
/* Calculate prefetch utilization (0.0 to 1.0) */
static inline float calc_utilization(struct ra_metrics *m)
{
    u64 prefetched = atomic64_read(&m->pages_prefetched);
    u64 used = atomic64_read(&m->pages_used);
    
    if (prefetched == 0) return 1.0;  /* Assume perfect if no data */
    return (float)used / (float)prefetched;
}
 
/* Calculate cache hit ratio (0.0 to 1.0) */
static inline float calc_hit_ratio(struct ra_metrics *m)
{
    u64 hits = atomic64_read(&m->cache_hits);
    u64 misses = atomic64_read(&m->cache_misses);
    u64 total = hits + misses;
    
    if (total == 0) return 1.0;
    return (float)hits / (float)total;
}
 
/* Calculate read rate (pages per second) */
static inline u64 calc_read_rate(struct ra_metrics *m, u64 pages_read)
{
    u64 elapsed_ns = m->sample_period_ns;
    if (elapsed_ns == 0) return 0;
    
    /* Pages per second = pages * (ns_per_sec / elapsed_ns) */
    return (pages_read * NSEC_PER_SEC) / elapsed_ns;
}
 
/* Determine if memory pressure is high */
static inline bool is_memory_pressure_high(struct ra_metrics *m)
{
    /* If prefetched pages are being evicted before use,
     * we're consuming too much memory */
    return atomic_read(&m->evictions_while_valid) > 0;
}
 
/* Make adaptive decision based on metrics */
enum ra_adjustment decide_adjustment(struct ra_metrics *m)
{
    float utilization = calc_utilization(m);
    float hit_ratio = calc_hit_ratio(m);
    bool memory_pressure = is_memory_pressure_high(m);
    
    /* Decision matrix */
    if (memory_pressure) {
        /* Memory pressure overrides everything - shrink */
        return RA_SHRINK_AGGRESSIVE;
    }
    
    if (hit_ratio < 0.80 && utilization > 0.90) {
        /* Not prefetching enough - grow window */
        return RA_GROW;
    }
    
    if (utilization < 0.50) {
        /* Wasting prefetch - shrink window */
        return RA_SHRINK;
    }
    
    if (hit_ratio > 0.95 && utilization > 0.80) {
        /* Sweet spot - maintain */
        return RA_MAINTAIN;
    }
    
    /* Default: slight growth bias for streaming workloads */
    return RA_MAINTAIN;
}

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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

/* AIMD-based adaptive read-ahead algorithm */
 
struct aimd_ra_state {
    unsigned int window;       /* Current window size (pages) */
    unsigned int max_window;   /* Maximum allowed window */
    unsigned int min_window;   /* Minimum window size */
    
    /* AIMD parameters */
    unsigned int additive_increment;  /* Constant increase step */
    unsigned int multiplicative_factor_num;  /* Decrease fraction numerator */
    unsigned int multiplicative_factor_den;  /* Decrease fraction denominator */
    
    /* State tracking */
    int consecutive_successes;
    int consecutive_failures;
};
 
#define DEFAULT_ADDITIVE_INCREMENT  16   /* Grow by 64KB (16 pages) */
#define DEFAULT_MD_NUMERATOR        1    /* Shrink to 1/2 */
#define DEFAULT_MD_DENOMINATOR      2
 
void init_aimd_state(struct aimd_ra_state *state, unsigned int max)
{
    state->window = 32;  /* Start conservative: 128KB */
    state->max_window = max;
    state->min_window = 4;  /* Never go below 16KB */
    
    state->additive_increment = DEFAULT_ADDITIVE_INCREMENT;
    state->multiplicative_factor_num = DEFAULT_MD_NUMERATOR;
    state->multiplicative_factor_den = DEFAULT_MD_DENOMINATOR;
    
    state->consecutive_successes = 0;
    state->consecutive_failures = 0;
}
 
/* Called when prefetching was successful (application used prefetched data) */
void aimd_success(struct aimd_ra_state *state)
{
    state->consecutive_failures = 0;
    state->consecutive_successes++;
    
    /*
     * Additive Increase: Grow linearly
     * Only increase if we've had consistent success (hysteresis)
     */
    if (state->consecutive_successes >= 3) {
        state->window += state->additive_increment;
        
        /* Cap at maximum */
        if (state->window > state->max_window) {
            state->window = state->max_window;
        }
        
        /* Reset counter but not fully - maintain some memory */
        state->consecutive_successes = 1;
    }
}
 
/* Called when prefetching failed (e.g., cache miss, memory pressure) */
void aimd_failure(struct aimd_ra_state *state)
{
    state->consecutive_successes = 0;
    state->consecutive_failures++;
    
    /*
     * Multiplicative Decrease: Shrink by fraction
     * Immediate response to failure (no hysteresis needed)
     */
    state->window = (state->window * state->multiplicative_factor_num) /
                     state->multiplicative_factor_den;
    
    /* Enforce minimum */
    if (state->window < state->min_window) {
        state->window = state->min_window;
    }
}
 
/*
 * AIMD Window Evolution Example:
 * 
 * Starting window: 32 pages (128KB)
 * Increment: 16 pages, Decrease factor: 1/2
 * 
 * Event sequence:
 * 1. Success x3 -> 32 + 16 = 48 pages
 * 2. Success x3 -> 48 + 16 = 64 pages
 * 3. Success x3 -> 64 + 16 = 80 pages
 * 4. Failure    -> 80 * 1/2 = 40 pages (immediate halving)
 * 5. Success x3 -> 40 + 16 = 56 pages
 * ...
 * 
 * The algorithm converges to a stable window size that balances
 * aggressive prefetching against resource constraints.
 */
 
/* Calculate optimal AIMD parameters based on workload */
void tune_aimd_parameters(struct aimd_ra_state *state,
                          unsigned int target_latency_ms,
                          unsigned int storage_latency_ms)
{
    /*
     * For fast storage, we can be more aggressive:
     * - Larger increments (grow faster)
     * - Smaller decrements (don't shrink as much)
     */
    if (storage_latency_ms < 1) {  /* NVMe SSD */
        state->additive_increment = 32;  /* 128KB steps */
        state->multiplicative_factor_num = 2;
        state->multiplicative_factor_den = 3;  /* Shrink to 2/3 */
    }
    else if (storage_latency_ms < 10) {  /* SATA SSD */
        state->additive_increment = 16;  /* 64KB steps */
        state->multiplicative_factor_num = 1;
        state->multiplicative_factor_den = 2;  /* Shrink to 1/2 */
    }
    else {  /* HDD */
        state->additive_increment = 32;  /* 128KB steps (amortize seek) */
        state->multiplicative_factor_num = 1;
        state->multiplicative_factor_den = 2;
    }
}

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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

/* Linux adaptive read-ahead mechanisms */
 
/*
 * Mechanism 1: mmap_miss - Memory pressure adaptation
 * From mm/readahead.c
 */
static void ra_adjust_for_mmap(struct file_ra_state *ra,
                                struct address_space *mapping)
{
    if (ra->mmap_miss > 0) {
        /*
         * Pages were prefetched but evicted before use.
         * This indicates either:
         *   1. Memory pressure - system needs pages elsewhere
         *   2. Over-prefetching - window is too large
         * 
         * Either way, reduce the window.
         */
        ra->ra_pages = max(ra->ra_pages / 2, MIN_RA_PAGES);
        ra->mmap_miss--;
    }
}
 
/*
 * Mechanism 2: Context-based initial size
 * Different contexts get different initial windows
 */
static unsigned long get_context_init_ra_size(struct file *filp,
                                               unsigned long max)
{
    /* Memory-mapped access: more conservative */
    if (filp->f_op->mmap) {
        return max / 4;
    }
    
    /* Regular read(): standard initial size */
    return min(32UL, max);
}
 
/*
 * Mechanism 3: Async readahead rate limiting
 * Prevent async readahead from overwhelming the system
 */
static bool should_throttle_readahead(struct backing_dev_info *bdi)
{
    /*
     * Check if there are too many pages in flight
     * This prevents prefetching from consuming all I/O bandwidth
     */
    unsigned long nr_pages_in_flight = bdi_stat(bdi, BDI_WRITEBACK);
    unsigned long threshold = bdi->ra_pages * 2;
    
    return nr_pages_in_flight > threshold;
}
 
/*
 * Mechanism 4: Per-device adaptation
 * Read-ahead max is set per block device
 */
void adapt_for_device_characteristics(struct block_device *bdev,
                                       struct file_ra_state *ra)
{
    struct request_queue *q = bdev_get_queue(bdev);
    
    /* Rotational devices (HDDs) benefit from larger read-ahead */
    if (blk_queue_nonrot(q)) {
        /* SSD: smaller window is fine, random is cheap */
        ra->ra_pages = min(ra->ra_pages, 256UL);  /* 1MB max */
    } else {
        /* HDD: larger window amortizes seek time */
        ra->ra_pages = min(ra->ra_pages, 512UL);  /* 2MB max */
    }
}
 
/*
 * Mechanism 5: Application hints override
 * posix_fadvise() allows applications to guide adaptation
 */
void apply_fadvise_hint(struct file *filp, int advice)
{
    struct file_ra_state *ra = &filp->f_ra;
    
    switch (advice) {
        case POSIX_FADV_SEQUENTIAL:
            /* App declares sequential: double the max window */
            ra->ra_pages = min(ra->ra_pages * 2, MAX_RA_PAGES);
            break;
            
        case POSIX_FADV_RANDOM:
            /* App declares random: disable prefetching */
            ra->ra_pages = 0;
            break;
            
        case POSIX_FADV_WILLNEED:
            /* App knows what it needs: prefetch immediately */
            /* (Handled separately via explicit prefetch call) */
            break;
            
        case POSIX_FADV_DONTNEED:
            /* App is done: can evict prefetched pages */
            invalidate_mapping_pages(filp->f_mapping, 0, -1);
            break;
    }
}

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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

/* Workload-aware adaptive prefetching */
 
enum workload_class {
    WORKLOAD_UNKNOWN,
    WORKLOAD_STREAMING,      /* Constant-rate sequential */
    WORKLOAD_BATCH_SEQ,      /* Variable-rate sequential */
    WORKLOAD_SCAN_FILTER,    /* Sequential with skips */
    WORKLOAD_RANDOM,         /* Random access */
    WORKLOAD_STRIDED,        /* Regular strided pattern */
    WORKLOAD_MIXED           /* Combination patterns */
};
 
struct workload_detector {
    /* Pattern metrics */
    u64 seq_reads;
    u64 random_reads;
    u64 total_reads;
    u64 stride_sum;
    u64 stride_variance;
    
    /* Rate metrics */
    u64 bytes_per_second;
    u64 reads_per_second;
    u64 avg_read_size;
    
    /* Classification */
    enum workload_class current_class;
    int classification_confidence;
};
 
/* Classify workload based on observed metrics */
enum workload_class classify_workload(struct workload_detector *wd)
{
    float seq_ratio = (float)wd->seq_reads / max(wd->total_reads, 1);
    
    /* High sequential ratio and consistent rate = streaming */
    if (seq_ratio > 0.95 && wd->stride_variance < 0.1) {
        wd->classification_confidence = 95;
        return WORKLOAD_STREAMING;
    }
    
    /* High sequential but variable rate = batch sequential */
    if (seq_ratio > 0.90) {
        wd->classification_confidence = 85;
        return WORKLOAD_BATCH_SEQ;
    }
    
    /* Moderate sequential with regular gaps = scan with filter */
    if (seq_ratio > 0.60 && wd->stride_variance < 0.3) {
        wd->classification_confidence = 75;
        return WORKLOAD_SCAN_FILTER;
    }
    
    /* Low sequential but very regular strides = strided */
    if (seq_ratio < 0.50 && wd->stride_variance < 0.1 && 
        wd->stride_sum / max(wd->total_reads, 1) > 0) {
        wd->classification_confidence = 70;
        return WORKLOAD_STRIDED;
    }
    
    /* Low sequential, irregular = random */
    if (seq_ratio < 0.30) {
        wd->classification_confidence = 80;
        return WORKLOAD_RANDOM;
    }
    
    /* Everything else = mixed */
    wd->classification_confidence = 50;
    return WORKLOAD_MIXED;
}
 
/* Get optimal prefetch parameters for workload */
struct prefetch_params get_optimal_params(enum workload_class wc)
{
    struct prefetch_params params = {0};
    
    switch (wc) {
        case WORKLOAD_STREAMING:
            params.max_window = 2048;  /* 8MB window */
            params.growth_factor = 2;   /* Fast exponential */
            params.async_trigger = 50;  /* Trigger at 50% */
            params.min_window = 128;    /* Never go too small */
            break;
            
        case WORKLOAD_BATCH_SEQ:
            params.max_window = 512;   /* 2MB window */
            params.growth_factor = 2;
            params.async_trigger = 60;
            params.min_window = 32;
            break;
            
        case WORKLOAD_SCAN_FILTER:
            params.max_window = 256;   /* 1MB window */
            params.growth_factor = 1.5; /* Slower growth */
            params.async_trigger = 70;  /* Earlier trigger */
            params.min_window = 32;
            break;
            
        case WORKLOAD_STRIDED:
            params.max_window = 128;   /* 512KB */
            params.stride_prefetch = true;
            params.prefetch_strides = 4;
            break;
            
        case WORKLOAD_RANDOM:
            params.max_window = 0;     /* No prefetch */
            break;
            
        case WORKLOAD_MIXED:
        default:
            params.max_window = 128;   /* Conservative */
            params.growth_factor = 1.5;
            params.async_trigger = 60;
            params.min_window = 16;
    }
    
    return params;
}

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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

"""
Machine Learning-based Read-Ahead Prediction
 
This example demonstrates an LSTM-based approach to predicting
file access patterns for optimizing prefetching decisions.
"""
 
import numpy as np
import torch
import torch.nn as nn
from collections import deque
 
class AccessPatternPredictor(nn.Module):
    """LSTM-based model to predict next block accesses."""
    
    def __init__(self, block_space_size: int, embedding_dim: int = 32,
                 hidden_dim: int = 64, num_layers: int = 2):
        super().__init__()
        
        # Embed block numbers into dense vectors
        self.embedding = nn.Embedding(block_space_size, embedding_dim)
        
        # LSTM to capture sequential patterns
        self.lstm = nn.LSTM(
            input_size=embedding_dim,
            hidden_size=hidden_dim,
            num_layers=num_layers,
            batch_first=True,
            dropout=0.1
        )
        
        # Output: probability distribution over next blocks
        self.fc = nn.Linear(hidden_dim, block_space_size)
        
    def forward(self, block_sequence: torch.Tensor):
        """
        Args:
            block_sequence: [batch, seq_len] tensor of block numbers
        Returns:
            [batch, block_space_size] probabilities for next block
        """
        embedded = self.embedding(block_sequence)
        lstm_out, _ = self.lstm(embedded)
        last_hidden = lstm_out[:, -1, :]
        logits = self.fc(last_hidden)
        return torch.softmax(logits, dim=-1)
 
 
class AdaptivePrefetcher:
    """Adaptive prefetcher using ML predictions."""
    
    def __init__(self, model: AccessPatternPredictor, 
                 history_len: int = 10,
                 prefetch_threshold: float = 0.2):
        self.model = model
        self.model.eval()
        self.history = deque(maxlen=history_len)
        self.prefetch_threshold = prefetch_threshold
        
        # Adaptation metrics
        self.correct_predictions = 0
        self.total_predictions = 0
        
    def record_access(self, block_num: int):
        """Record a block access for pattern learning."""
        self.history.append(block_num)
        
    def predict_prefetch_candidates(self, top_k: int = 5) -> list:
        """Predict which blocks to prefetch."""
        if len(self.history) < 3:
            # Not enough history - use simple sequential prediction
            if self.history:
                last = self.history[-1]
                return [last + i for i in range(1, top_k + 1)]
            return []
        
        # Convert history to tensor
        seq = torch.tensor([list(self.history)]).long()
        
        with torch.no_grad():
            probs = self.model(seq)[0]  # [block_space_size]
            
        # Get top-k predictions above threshold
        top_probs, top_indices = torch.topk(probs, k=top_k)
        
        candidates = []
        for prob, idx in zip(top_probs.tolist(), top_indices.tolist()):
            if prob >= self.prefetch_threshold:
                candidates.append((idx, prob))
        
        return candidates
    
    def update_accuracy(self, prefetched: list, actual_access: int):
        """Track prediction accuracy for adaptation."""
        self.total_predictions += 1
        if actual_access in [p[0] for p in prefetched]:
            self.correct_predictions += 1
        
        # Adapt threshold based on accuracy
        accuracy = self.correct_predictions / max(self.total_predictions, 1)
        
        if accuracy < 0.5:
            # Low accuracy: raise threshold (be more conservative)
            self.prefetch_threshold = min(0.5, self.prefetch_threshold + 0.05)
        elif accuracy > 0.8:
            # High accuracy: lower threshold (prefetch more)
            self.prefetch_threshold = max(0.1, self.prefetch_threshold - 0.02)
 
 
# Training loop (simplified)
def train_predictor(access_traces: list, epochs: int = 100):
    """Train the access pattern predictor on historical traces."""
    
    block_space_size = max(max(trace) for trace in access_traces) + 1
    model = AccessPatternPredictor(block_space_size)
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    criterion = nn.CrossEntropyLoss()
    
    model.train()
    for epoch in range(epochs):
        total_loss = 0
        for trace in access_traces:
            for i in range(len(trace) - 10):
                # Input: 10 consecutive blocks, Target: next block
                seq = torch.tensor([trace[i:i+10]]).long()
                target = torch.tensor([trace[i+10]]).long()
                
                optimizer.zero_grad()
                probs = model(seq)
                loss = criterion(probs, target)
                loss.backward()
                optimizer.step()
                total_loss += loss.item()
        
        if epoch % 10 == 0:
            print(f"Epoch {epoch}, Loss: {total_loss/len(access_traces):.4f}")
    
    return model

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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216

/* Complete adaptive read-ahead implementation */
 
#include <stdatomic.h>
#include <stdbool.h>
 
struct adaptive_ra {
    /* Core window state */
    size_t window_size;
    size_t max_window;
    size_t min_window;
    off_t current_pos;
    off_t window_start;
    
    /* Metrics for adaptation */
    atomic_uint_fast64_t prefetch_count;
    atomic_uint_fast64_t hit_count;
    atomic_uint_fast64_t miss_count;
    atomic_uint_fast32_t pressure_events;
    
    /* Rate tracking */
    struct timespec last_access;
    size_t bytes_since_last;
    double smoothed_rate;
    
    /* AIMD state */
    int consecutive_hits;
    int consecutive_misses;
    
    /* Pattern classification */
    int seq_streak;
    off_t last_stride;
    int stride_count;
};
 
/* Initialize adaptive state */
void adaptive_ra_init(struct adaptive_ra *ra, size_t device_max_ra)
{
    ra->window_size = 32 * 4096;  /* 128KB initial */
    ra->max_window = device_max_ra;
    ra->min_window = 4 * 4096;    /* 16KB minimum */
    ra->current_pos = 0;
    ra->window_start = 0;
    
    atomic_init(&ra->prefetch_count, 0);
    atomic_init(&ra->hit_count, 0);
    atomic_init(&ra->miss_count, 0);
    atomic_init(&ra->pressure_events, 0);
    
    ra->smoothed_rate = 0;
    ra->consecutive_hits = 0;
    ra->consecutive_misses = 0;
    ra->seq_streak = 0;
}
 
/* Calculate current metrics */
struct ra_metrics {
    double hit_ratio;
    double utilization;
    double rate_mbps;
    bool memory_pressure;
};
 
struct ra_metrics calculate_metrics(struct adaptive_ra *ra)
{
    struct ra_metrics m;
    
    uint64_t hits = atomic_load(&ra->hit_count);
    uint64_t misses = atomic_load(&ra->miss_count);
    uint64_t prefetched = atomic_load(&ra->prefetch_count);
    
    m.hit_ratio = (double)hits / fmax(hits + misses, 1);
    m.utilization = (double)hits / fmax(prefetched, 1);
    m.rate_mbps = ra->smoothed_rate / (1024 * 1024);
    m.memory_pressure = atomic_load(&ra->pressure_events) > 0;
    
    return m;
}
 
/* Make adaptation decision */
enum ra_action {
    RA_GROW,
    RA_SHRINK,  
    RA_SHRINK_FAST,
    RA_MAINTAIN
};
 
enum ra_action decide_action(struct adaptive_ra *ra, 
                              struct ra_metrics *m)
{
    /* Priority 1: Memory pressure always shrinks */
    if (m->memory_pressure) {
        atomic_fetch_sub(&ra->pressure_events, 1);
        return RA_SHRINK_FAST;
    }
    
    /* Priority 2: Very low utilization = over-prefetching */
    if (m->utilization < 0.4) {
        return RA_SHRINK;
    }
    
    /* Priority 3: High miss ratio = under-prefetching */
    if (m->hit_ratio < 0.85 && m->utilization > 0.7) {
        return RA_GROW;
    }
    
    /* Priority 4: Sequential streak bonus */
    if (ra->seq_streak >= 5 && m->hit_ratio > 0.95) {
        return RA_GROW;
    }
    
    return RA_MAINTAIN;
}
 
/* Apply the adaptation */
void apply_action(struct adaptive_ra *ra, enum ra_action action)
{
    switch (action) {
        case RA_GROW:
            /* AIMD additive increase */
            ra->consecutive_hits++;
            if (ra->consecutive_hits >= 3) {
                size_t increment = 16 * 4096;  /* 64KB */
                ra->window_size += increment;
                if (ra->window_size > ra->max_window) {
                    ra->window_size = ra->max_window;
                }
                ra->consecutive_hits = 0;
            }
            ra->consecutive_misses = 0;
            break;
            
        case RA_SHRINK:
            /* AIMD multiplicative decrease */
            ra->window_size = (ra->window_size * 3) / 4;  /* 75% */
            if (ra->window_size < ra->min_window) {
                ra->window_size = ra->min_window;
            }
            ra->consecutive_hits = 0;
            ra->consecutive_misses++;
            break;
            
        case RA_SHRINK_FAST:
            /* Aggressive shrink for memory pressure */
            ra->window_size /= 2;
            if (ra->window_size < ra->min_window) {
                ra->window_size = ra->min_window;
            }
            ra->consecutive_hits = 0;
            ra->consecutive_misses++;
            break;
            
        case RA_MAINTAIN:
            /* Keep current size */
            break;
    }
}
 
/* Main entry point: called on each read */
size_t adaptive_ra_get_prefetch_size(struct adaptive_ra *ra,
                                      off_t offset,
                                      size_t read_size)
{
    /* Update rate tracking */
    struct timespec now;
    clock_gettime(CLOCK_MONOTONIC, &now);
    double elapsed = (now.tv_sec - ra->last_access.tv_sec) +
                     (now.tv_nsec - ra->last_access.tv_nsec) / 1e9;
    
    if (elapsed > 0.001) {  /* Avoid division by small numbers */
        double current_rate = ra->bytes_since_last / elapsed;
        /* Exponential moving average */
        ra->smoothed_rate = 0.7 * ra->smoothed_rate + 0.3 * current_rate;
        ra->bytes_since_last = 0;
        ra->last_access = now;
    }
    ra->bytes_since_last += read_size;
    
    /* Update sequential streak */
    if (offset == ra->current_pos) {
        ra->seq_streak++;
    } else {
        ra->seq_streak = 0;
    }
    ra->current_pos = offset + read_size;
    
    /* Calculate metrics and decide action */
    struct ra_metrics m = calculate_metrics(ra);
    enum ra_action action = decide_action(ra, &m);
    apply_action(ra, action);
    
    /* Return prefetch recommendation */
    if (ra->seq_streak >= 2) {
        return ra->window_size;
    } else {
        /* Not clearly sequential - be conservative */
        return ra->min_window;
    }
}
 
/* Called by memory subsystem on pressure */
void adaptive_ra_pressure_event(struct adaptive_ra *ra)
{
    atomic_fetch_add(&ra->pressure_events, 1);
}
 
/* Called when prefetched data is used */
void adaptive_ra_record_hit(struct adaptive_ra *ra)
{
    atomic_fetch_add(&ra->hit_count, 1);
}
 
/* Called when read required I/O (prefetch missed) */
void adaptive_ra_record_miss(struct adaptive_ra *ra)
{
    atomic_fetch_add(&ra->miss_count, 1);
}