perdix 0.1.1

High-performance GPU-accelerated ring buffer for AI terminal multiplexing
Documentation 
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
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263

use crate::buffer::{Header, Slot};
use std::sync::atomic::Ordering;
use std::thread;
use std::time::{Duration, Instant};

pub struct CpuProducer {
    header: *mut Header,
    slots: *mut Slot,
    #[allow(dead_code)]
    #[allow(dead_code)]
    n_slots: usize,
    delay_ms: Option<u64>,
}

unsafe impl Send for CpuProducer {}

impl CpuProducer {
    pub fn new(
        header: *mut Header,
        slots: *mut Slot,
        n_slots: usize,
        delay_ms: Option<u64>,
    ) -> Self {
        Self {
            header,
            slots,
            n_slots,
            delay_ms,
        }
    }

    pub fn produce_messages(&mut self, n_msgs: u64) {
        let start = Instant::now();

        for i in 0..n_msgs {
            // Reserve sequence number atomically (mimics GPU atomicAdd)
            let seq = unsafe {
                (*self.header)
                    .producer
                    .write_idx
                    .fetch_add(1, Ordering::AcqRel)
            };

            // Calculate slot index with wrap-around
            let slot_idx = (seq & unsafe { (*self.header).config.wrap_mask }) as usize;

            // Write to slot
            unsafe {
                let slot = &mut *self.slots.add(slot_idx);

                // Create test payload
                let payload_str = format!("CPU Message #{:04} [seq={}]", i, seq);
                let payload_bytes = payload_str.as_bytes();
                let len = payload_bytes.len().min(192);

                // Copy payload
                slot.payload[..len].copy_from_slice(&payload_bytes[..len]);
                // Write non-atomic fields using volatile writes for unified memory
                std::ptr::write_volatile(&mut slot.len, len as u32);
                std::ptr::write_volatile(&mut slot.flags, 0);

                // Memory fence equivalent (ensures write visibility)
                std::sync::atomic::fence(Ordering::Release);

                // Publish sequence number (slot is ready)
                // Publish sequence using volatile write
                std::ptr::write_volatile(&mut slot.seq, seq);
            }

            // Optional delay to simulate processing time
            if let Some(delay) = self.delay_ms {
                thread::sleep(Duration::from_millis(delay));
            }

            // Progress indicator every 1000 messages
            if i > 0 && i % 1000 == 0 {
                let elapsed = start.elapsed();
                let rate = i as f64 / elapsed.as_secs_f64();
                println!("CPU Producer: {} messages, {:.0} msg/s", i, rate);
            }
        }

        let elapsed = start.elapsed();
        println!(
            "CPU Producer: Completed {} messages in {:.2}s ({:.0} msg/s)",
            n_msgs,
            elapsed.as_secs_f64(),
            n_msgs as f64 / elapsed.as_secs_f64()
        );
    }
}

pub fn start_cpu_producer(
    header: *mut Header,
    slots: *mut Slot,
    n_slots: usize,
    n_msgs: u64,
    delay_ms: Option<u64>,
) -> thread::JoinHandle<()> {
    // Convert pointers to usize to make them Send
    let header_addr = header as usize;
    let slots_addr = slots as usize;

    thread::spawn(move || {
        // Reconstruct pointers inside the thread
        let header = header_addr as *mut Header;
        let slots = slots_addr as *mut Slot;
        let mut producer = CpuProducer::new(header, slots, n_slots, delay_ms);
        producer.produce_messages(n_msgs);
    })
}

// Batch producer for higher throughput
pub struct BatchCpuProducer {
    header: *mut Header,
    slots: *mut Slot,
    _n_slots: usize,
    batch_size: usize,
}

unsafe impl Send for BatchCpuProducer {}

impl BatchCpuProducer {
    pub fn new(header: *mut Header, slots: *mut Slot, n_slots: usize, batch_size: usize) -> Self {
        Self {
            header,
            slots,
            _n_slots: n_slots,
            batch_size,
        }
    }

    pub fn produce_batch(&mut self, n_msgs: u64) {
        let start = Instant::now();
        let mut messages_produced = 0u64;

        while messages_produced < n_msgs {
            let batch_count = (n_msgs - messages_produced).min(self.batch_size as u64);

            // Reserve sequence numbers for entire batch atomically
            let base_seq = unsafe {
                (*self.header)
                    .producer
                    .write_idx
                    .fetch_add(batch_count, Ordering::AcqRel)
            };

            // Write batch of messages
            for i in 0..batch_count {
                let seq = base_seq + i;
                let slot_idx = (seq & unsafe { (*self.header).config.wrap_mask }) as usize;

                unsafe {
                    let slot = &mut *self.slots.add(slot_idx);

                    // Create payload
                    let payload_str =
                        format!("Batch CPU #{:04} [seq={}]", messages_produced + i, seq);
                    let payload_bytes = payload_str.as_bytes();
                    let len = payload_bytes.len().min(192);

                    slot.payload[..len].copy_from_slice(&payload_bytes[..len]);
                    // Write non-atomic fields using volatile writes
                    std::ptr::write_volatile(&mut slot.len, len as u32);
                    std::ptr::write_volatile(&mut slot.flags, 0x100); // Mark as batch-produced

                    // Memory fence
                    std::sync::atomic::fence(Ordering::Release);

                    // Publish sequence using volatile write
                    std::ptr::write_volatile(&mut slot.seq, seq);
                }
            }

            messages_produced += batch_count;

            // Progress update
            if messages_produced % 10000 == 0 {
                let elapsed = start.elapsed();
                let rate = messages_produced as f64 / elapsed.as_secs_f64();
                println!(
                    "Batch CPU Producer: {} messages, {:.0} msg/s",
                    messages_produced, rate
                );
            }
        }

        let elapsed = start.elapsed();
        println!(
            "Batch CPU Producer: Completed {} messages in {:.2}s ({:.0} msg/s)",
            n_msgs,
            elapsed.as_secs_f64(),
            n_msgs as f64 / elapsed.as_secs_f64()
        );
    }
}

// Multi-threaded CPU producer for maximum throughput
pub fn start_parallel_cpu_producer(
    header: *mut Header,
    slots: *mut Slot,
    n_slots: usize,
    n_msgs: u64,
    n_threads: usize,
) -> Vec<thread::JoinHandle<()>> {
    let msgs_per_thread = n_msgs / n_threads as u64;
    let remainder = n_msgs % n_threads as u64;

    // Convert pointers to usize for thread safety
    let header_addr = header as usize;
    let slots_addr = slots as usize;

    (0..n_threads)
        .map(|thread_id| {
            let header_addr = header_addr;
            let slots_addr = slots_addr;
            let n_slots = n_slots;
            let thread_msgs = if thread_id == 0 {
                msgs_per_thread + remainder
            } else {
                msgs_per_thread
            };

            thread::spawn(move || {
                let header = header_addr as *mut Header;
                let slots = slots_addr as *mut Slot;
                let mut producer = CpuProducer::new(header, slots, n_slots, None);
                println!(
                    "CPU Producer thread {} starting, producing {} messages",
                    thread_id, thread_msgs
                );
                producer.produce_messages(thread_msgs);
                println!("CPU Producer thread {} completed", thread_id);
            })
        })
        .collect()
}

// Performance testing utilities
pub fn benchmark_cpu_producer(
    header: *mut Header,
    slots: *mut Slot,
    n_slots: usize,
    n_msgs: u64,
) -> Duration {
    let start = Instant::now();
    let mut producer = CpuProducer::new(header, slots, n_slots, None);
    producer.produce_messages(n_msgs);
    start.elapsed()
}

pub fn benchmark_batch_producer(
    header: *mut Header,
    slots: *mut Slot,
    n_slots: usize,
    n_msgs: u64,
    batch_size: usize,
) -> Duration {
    let start = Instant::now();
    let mut producer = BatchCpuProducer::new(header, slots, n_slots, batch_size);
    producer.produce_batch(n_msgs);
    start.elapsed()
}