thread-share 0.1.6

A Rust library for safe and efficient data sharing between threads with zero-copy operations, change detection, and enhanced thread management.
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
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352

use std::time::{Duration, Instant};
use std::collections::HashMap;
use thread_share::{enhanced_share, spawn_workers};
use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

/// HTML helper functions for generating simple pages
fn render_page(data: &HashMap<String, String>) -> String {
    let mut html = String::new();
    for (key, value) in data {
        html.push_str(&format!("<p><strong>{}:</strong> {}</p>", key, value));
    }
    
    format!(
        r#"<!DOCTYPE html>
<html>
<head>
    <title>Server</title>
</head>
<body>
{}
</body>
</html>"#,
        html
    )
}

/// Async HTTP server using tokio and ThreadShare
/// 
/// This example demonstrates how to use ThreadShare with tokio for
/// building high-performance async HTTP servers with shared state.
#[tokio::main]
async fn main() {
    println!("=== Tokio HTTP Server with ThreadShare ===");
    println!("Building async HTTP server with shared state...\n");

    // Create shared server state
    let server = enhanced_share!(AsyncHttpServer {
        port: 8081,
        is_running: true,
        requests_handled: 0,
        active_connections: 0,
        start_time: Instant::now(),
    });

    // Create shared visit counter
    let visits = enhanced_share!(0);

    println!("🚀 Starting Tokio HTTP server on port 8081...");
    println!("📱 Server URLs:");
    println!("   • Main page: http://127.0.0.1:8081/");
    println!("   • Status: http://127.0.0.1:8081/status");
    println!("   • Health: http://127.0.0.1:8081/health");
    println!("   • Metrics: http://127.0.0.1:8081/metrics");
    println!();

    // Start all async workers with spawn_workers!
    let visits_clone = visits.clone();
    let manager = spawn_workers!(server, {
        server_main: move |server: thread_share::ThreadShare<AsyncHttpServer>| {
            println!("🌐 Tokio server main worker started");
            
            // Create runtime for this worker
            let rt = tokio::runtime::Runtime::new().unwrap();
            
            rt.block_on(async {
                // Start the server
                server.update(|s| s.start().expect("Failed to start server"));
                
                // Create TCP listener
                let port = server.get().port;
                let listener = TcpListener::bind(format!("127.0.0.1:{}", port))
                    .await
                    .expect("Failed to bind");
                
                println!("🔌 Listening on http://127.0.0.1:{}", port);
                
                // Accept connections
                loop {
                    match listener.accept().await {
                        Ok((stream, _addr)) => {
                            // Increment connection counter
                            server.update(|s| s.increment_connections());
                            
                            // Spawn task to handle connection
                            let server_clone = server.clone();
                            let visits_clone = visits_clone.clone();
                            
                            tokio::spawn(async move {
                                handle_connection(stream, server_clone, visits_clone).await;
                            });
                        }
                        Err(e) => {
                            eprintln!("❌ Connection failed: {}", e);
                        }
                    }
                    
                    // Check if server should stop
                    if !server.get().is_running {
                        break;
                    }
                }
                
                println!("🌐 Tokio server main worker finished");
            });
        }
    });

    // Demonstrate worker management
    println!("🔧 Worker Manager Demo:");
    println!("📋 Worker names: {:?}", manager.get_worker_names());
    println!("🔢 Active workers: {}", manager.active_workers());

    // Add stats monitor worker programmatically
    println!("\n➕ Adding stats monitor worker programmatically...");
    let server_clone = server.clone();
    let stats_handle = std::thread::spawn(move || {
        println!("📊 Stats monitor worker started");
        
        // Create runtime for this worker
        let rt = tokio::runtime::Runtime::new().unwrap();
        
        rt.block_on(async {
            // Monitor server statistics every 3 seconds
            for _i in 1..=20 { // 20 iterations * 3 seconds = 1 minute
                let current_server = server_clone.get();
                
                if current_server.is_running {
                    println!("📊 Tokio Server Stats | Port: {} | Requests: {} | Connections: {} | Uptime: {}", 
                        current_server.port,
                        current_server.requests_handled,
                        current_server.active_connections,
                        current_server.get_uptime()
                    );
                } else {
                    println!("📊 Server stopped, stats monitor exiting");
                    break;
                }
                
                tokio::time::sleep(Duration::from_secs(3)).await;
            }
            
            // Stop server after 1 minute by stopping the main worker
            println!("⏰ Stopping tokio server after 1 minute...");
            server_clone.update(|s| s.stop());
            println!("📊 Stats monitor worker finished");
        });
    });

    // Add the stats monitor to the manager
    if let Err(e) = manager.add_worker("stats_monitor", stats_handle) {
        println!("❌ Failed to add stats monitor: {}", e);
    } else {
        println!("✅ Stats monitor worker added successfully");
    }

    println!("📋 Updated worker names: {:?}", manager.get_worker_names());
    println!("🔢 Updated active workers: {}", manager.active_workers());

    // Wait for all workers to complete
    manager.join_all().expect("Failed to join tokio workers");
    
    println!("✅ Tokio HTTP server completed successfully!");
}

/// Handle individual TCP connection
async fn handle_connection(
    mut stream: TcpStream,
    server: thread_share::ThreadShare<AsyncHttpServer>,
    visits: thread_share::EnhancedThreadShare<u32>,
) {
    let mut buffer = [0; 1024];
    
    // Read request
    let n = match stream.read(&mut buffer).await {
        Ok(n) if n == 0 => return, // Connection closed
        Ok(n) => n,
        Err(e) => {
            eprintln!("❌ Error reading from stream: {}", e);
            return;
        }
    };
    
    let request = String::from_utf8_lossy(&buffer[..n]);
    let request_lines: Vec<&str> = request.lines().collect();
    
    if request_lines.is_empty() {
        return;
    }
    
    let request_line = request_lines[0];
    let parts: Vec<&str> = request_line.split_whitespace().collect();
    
    if parts.len() < 2 {
        return;
    }
    
    let method = parts[0];
    let path = parts[1];
    
    // Increment request counter
    server.update(|s| s.increment_requests());
    
    // Handle different routes
    let (status_line, content, content_type) = match (method, path) {
        ("GET", "/") => {
            // Increment visit counter for main page
            visits.update(|v| *v += 1);
            let visit_count = visits.get();
            
            (
                "HTTP/1.1 200 OK",
                render_page(&{
                    let mut map = HashMap::new();
                    map.insert("Page Visits".to_string(), visit_count.to_string());
                    map.insert("Total Requests".to_string(), server.get().requests_handled.to_string());
                    map.insert("Active Connections".to_string(), server.get().active_connections.to_string());
                    map.insert("Uptime".to_string(), server.get().get_uptime());
                    map
                }),
                "text/html; charset=utf-8"
            )
        }
        
        ("GET", "/status") => {
            let server_data = server.get();
            (
                "HTTP/1.1 200 OK",
                render_page(&{
                    let mut map = HashMap::new();
                    map.insert("Status".to_string(), if server_data.is_running { "Running".to_string() } else { "Stopped".to_string() });
                    map.insert("Port".to_string(), server_data.port.to_string());
                    map.insert("Uptime".to_string(), server_data.get_uptime());
                    map.insert("Requests".to_string(), server_data.requests_handled.to_string());
                    map.insert("Connections".to_string(), server_data.active_connections.to_string());
                    map
                }),
                "text/html; charset=utf-8"
            )
        }
        
        ("GET", "/health") => {
            (
                "HTTP/1.1 200 OK",
                render_page(&{
                    let mut map = HashMap::new();
                    map.insert("Health".to_string(), "OK".to_string());
                    map
                }),
                "text/html; charset=utf-8"
            )
        }
        
        ("GET", "/metrics") => {
            let server_data = server.get();
            let visit_count = visits.get();
            
            (
                "HTTP/1.1 200 OK",
                render_page(&{
                    let mut map = HashMap::new();
                    map.insert("Requests".to_string(), server_data.requests_handled.to_string());
                    map.insert("Connections".to_string(), server_data.active_connections.to_string());
                    map.insert("Page Visits".to_string(), visit_count.to_string());
                    map.insert("Uptime".to_string(), format!("{}s", server_data.start_time.elapsed().as_secs()));
                    map
                }),
                "text/html; charset=utf-8"
            )
        }
        
        _ => {
            (
                "HTTP/1.1 404 Not Found",
                render_page(&{
                    let mut map = HashMap::new();
                    map.insert("Error".to_string(), "404 - Page not found".to_string());
                    map
                }),
                "text/html; charset=utf-8"
            )
        }
    };
    
    // Build response
    let response = format!(
        "{}\r\nContent-Type: {}\r\nContent-Length: {}\r\n\r\n{}",
        status_line,
        content_type,
        content.len(),
        content
    );
    
    // Write response
    if let Err(e) = stream.write_all(response.as_bytes()).await {
        eprintln!("❌ Error writing to stream: {}", e);
    }
    
    // Decrement connection counter
    server.update(|s| s.decrement_connections());
}

/// Async HTTP server structure
#[derive(Clone, Debug)]
struct AsyncHttpServer {
    port: u16,
    is_running: bool,
    requests_handled: u64,
    active_connections: u32,
    start_time: Instant,
}

impl AsyncHttpServer {
    fn start(&mut self) -> Result<(), String> {
        self.is_running = true;
        self.start_time = Instant::now();
        Ok(())
    }
    
    fn stop(&mut self) {
        self.is_running = false;
    }
    
    fn increment_requests(&mut self) {
        self.requests_handled += 1;
    }
    
    fn increment_connections(&mut self) {
        self.active_connections += 1;
    }
    
    fn decrement_connections(&mut self) {
        if self.active_connections > 0 {
            self.active_connections -= 1;
        }
    }
    
    fn get_uptime(&self) -> String {
        let duration = self.start_time.elapsed();
        let seconds = duration.as_secs();
        let minutes = seconds / 60;
        let hours = minutes / 60;
        
        if hours > 0 {
            format!("{}h {}m {}s", hours, minutes % 60, seconds % 60)
        } else if minutes > 0 {
            format!("{}m {}s", minutes, seconds % 60)
        } else {
            format!("{}s", seconds)
        }
    }
}