thread-local io backend

README.md

@@ -24,7 +24,7 @@
 - **TLS/HTTPS**: OpenSSL-based with ALPN and certificate verification
 - **Header-Only Library** for easy integration
 - **Debugging Tools**: GDB/LLDB extensions and pstack-like CLI
-- **Comprehensive Testing** with Catch2 and ASAN
+- **Comprehensive Testing** with Catch2, ASAN, and TSAN
 - **Integrated Logging** with fmtlib
 - **CI/CD** with GitHub Actions
 
@@ -59,6 +59,9 @@ ctest --output-on-failure
 
 # ASAN tests (memory safety)
 ./build/tests/elio_tests_asan
+
+# TSAN tests (thread safety)
+./build/tests/elio_tests_tsan
 ```
 
 ### Your First Coroutine
@@ -179,6 +182,7 @@ outer() -> middle() -> inner()
 The scheduler manages a pool of worker threads, each with a local task queue. Key features:
 - **Lock-free operations**: Chase-Lev deques for optimal performance
 - **Work stealing**: Idle threads steal tasks from busy threads
+- **Per-worker I/O context**: Each worker has its own io_uring/epoll backend for thread-safe I/O
 - **Dynamic sizing**: Adjust thread count at runtime
 - **Load balancing**: Automatic task distribution
 
@@ -343,7 +347,8 @@ Elio includes comprehensive tests:
 
 - **Unit Tests**: Test each component in isolation
 - **Integration Tests**: Test components working together
-- **ASAN Tests**: Detect memory errors
+- **ASAN Tests**: Detect memory errors (use-after-free, buffer overflow, etc.)
+- **TSAN Tests**: Detect data races and thread safety issues
 
 ```bash
 # Run all tests
@@ -354,7 +359,8 @@ ctest --test-dir build --output-on-failure
 ./build/tests/elio_tests "[integration]"
 
 # Run with sanitizers
-./build/tests/elio_tests_asan
+./build/tests/elio_tests_asan    # Memory safety
+./build/tests/elio_tests_tsan    # Thread safety
 ```
 
 ## Performance

examples/async_file_io.cpp

@@ -22,8 +22,6 @@ using namespace elio::runtime;
 
 /// Async file copy using io_uring/epoll
 task<bool> async_copy_file(const std::string& src_path, const std::string& dst_path) {
-    auto& ctx = io::default_io_context();
-
     // Open source file
     int src_fd = open(src_path.c_str(), O_RDONLY);
     if (src_fd < 0) {
@@ -62,7 +60,7 @@ task<bool> async_copy_file(const std::string& src_path, const std::string& dst_p
         size_t to_read = std::min(BUFFER_SIZE, file_size - total_copied);
 
         // Async read from source
-        auto read_result = co_await io::async_read(ctx, src_fd, buffer.data(), to_read, offset);
+        auto read_result = co_await io::async_read(src_fd, buffer.data(), to_read, offset);
 
         if (read_result.result <= 0) {
             if (read_result.result == 0) {
@@ -77,7 +75,7 @@ task<bool> async_copy_file(const std::string& src_path, const std::string& dst_p
         size_t bytes_read = read_result.result;
 
         // Async write to destination
-        auto write_result = co_await io::async_write(ctx, dst_fd, buffer.data(), bytes_read, offset);
+        auto write_result = co_await io::async_write(dst_fd, buffer.data(), bytes_read, offset);
 
         if (write_result.result <= 0) {
             std::cerr << "Write error: " << strerror(-write_result.result) << std::endl;
@@ -98,8 +96,8 @@ task<bool> async_copy_file(const std::string& src_path, const std::string& dst_p
     auto duration_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
 
     // Close files
-    co_await io::async_close(ctx, src_fd);
-    co_await io::async_close(ctx, dst_fd);
+    co_await io::async_close(src_fd);
+    co_await io::async_close(dst_fd);
 
     std::cout << std::endl;
     std::cout << "Copy completed in " << duration_ms << " ms" << std::endl;
@@ -114,8 +112,6 @@ task<bool> async_copy_file(const std::string& src_path, const std::string& dst_p
 
 /// Concurrent file read demonstration
 task<void> concurrent_read_demo(const std::vector<std::string>& files) {
-    auto& ctx = io::default_io_context();
-
     std::cout << "Reading " << files.size() << " files concurrently..." << std::endl;
 
     struct FileInfo {
@@ -148,7 +144,7 @@ task<void> concurrent_read_demo(const std::vector<std::string>& files) {
     // Read all files (would be more concurrent with multiple coroutines)
     size_t total_bytes = 0;
     for (auto& info : file_infos) {
-        auto result = co_await io::async_read(ctx, info.fd, info.buffer.data(), info.size, 0);
+        auto result = co_await io::async_read(info.fd, info.buffer.data(), info.size, 0);
         if (result.result > 0) {
             total_bytes += result.result;
             std::cout << "  Read " << result.result << " bytes from " << info.path << std::endl;
@@ -166,13 +162,11 @@ task<void> concurrent_read_demo(const std::vector<std::string>& files) {
 
 /// Benchmark async vs sync file I/O
 task<void> benchmark_io(size_t file_size_mb) {
-    auto& ctx = io::default_io_context();
-
     const std::string test_file = "/tmp/elio_benchmark_test.dat";
     size_t file_size = file_size_mb * 1024 * 1024;
 
     std::cout << "Benchmark: " << file_size_mb << " MB file" << std::endl;
-    std::cout << "I/O Backend: " << ctx.get_backend_name() << std::endl;
+    std::cout << "I/O Backend: " << io::current_io_context().get_backend_name() << std::endl;
     std::cout << std::endl;
 
     // Create test file
@@ -201,7 +195,7 @@ task<void> benchmark_io(size_t file_size_mb) {
         size_t total_read = 0;
         int64_t offset = 0;
         while (total_read < file_size) {
-            auto result = co_await io::async_read(ctx, fd, buffer.data(), BUFFER_SIZE, offset);
+            auto result = co_await io::async_read(fd, buffer.data(), BUFFER_SIZE, offset);
             if (result.result <= 0) break;
             total_read += result.result;
             offset += result.result;
@@ -261,9 +255,6 @@ int main(int argc, char* argv[]) {
 
     scheduler sched(2);
 
-    // Set the I/O context so workers can poll for I/O completions
-    sched.set_io_context(&io::default_io_context());
-
     sched.start();
 
     std::atomic<bool> done{false};

examples/http2_client.cpp

@@ -151,7 +151,6 @@ int main(int argc, char* argv[]) {
 
     // Create scheduler
     scheduler sched(2);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Run appropriate mode

examples/http_client.cpp

@@ -236,7 +236,6 @@ int main(int argc, char* argv[]) {
 
     // Create scheduler
     scheduler sched(2);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Run appropriate mode

examples/http_server.cpp

@@ -258,7 +258,6 @@ int main(int argc, char* argv[]) {
 
     // Create scheduler
     scheduler sched(4);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Spawn signal handler coroutine

examples/rpc_client_example.cpp

@@ -336,11 +336,9 @@ task<void> run_demo(tcp_rpc_client::ptr client) {
 }
 
 task<void> client_main(const char* host, uint16_t port) {
-    auto& ctx = io::default_io_context();
-
     std::cout << "Connecting to " << host << ":" << port << "..." << std::endl;
 
-    auto client = co_await tcp_rpc_client::connect(ctx, host, port);
+    auto client = co_await tcp_rpc_client::connect(host, port);
     if (!client) {
         std::cerr << "Failed to connect to server" << std::endl;
         co_return;
@@ -368,7 +366,6 @@ int main(int argc, char* argv[]) {
 
     // Create and start scheduler
     scheduler sched(2);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Run client

examples/rpc_server_example.cpp

@@ -318,7 +318,6 @@ int main(int argc, char* argv[]) {
 
     // Create and start scheduler
     scheduler sched(4);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Spawn signal handler coroutine

examples/signal_handling.cpp

@@ -139,10 +139,6 @@ int main() {
     // Create scheduler with worker threads
     scheduler sched(4);
 
-    // Set up I/O context for async operations
-    io::io_context ctx;
-    sched.set_io_context(&ctx);
-
     sched.start();
 
     // Spawn main task
@@ -151,18 +147,12 @@ int main() {
 
     // Run until shutdown
     while (g_running) {
-        ctx.poll(std::chrono::milliseconds(10));
         std::this_thread::sleep_for(std::chrono::milliseconds(10));
     }
 
     // Give coroutines time to clean up
     std::this_thread::sleep_for(std::chrono::milliseconds(200));
 
-    // Poll any remaining I/O
-    for (int i = 0; i < 10 && ctx.has_pending(); ++i) {
-        ctx.poll(std::chrono::milliseconds(10));
-    }
-
     sched.shutdown();
 
     ELIO_LOG_INFO("Application shutdown complete");

examples/sse_client.cpp

@@ -235,7 +235,6 @@ int main(int argc, char* argv[]) {
 
     // Create scheduler
     runtime::scheduler sched(2);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Run client based on mode

examples/sse_server.cpp

@@ -366,7 +366,6 @@ int main(int argc, char* argv[]) {
 
     // Create scheduler
     runtime::scheduler sched(4);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Spawn signal handler coroutine

examples/tcp_echo_client.cpp

@@ -28,13 +28,10 @@ using namespace elio::net;
 
 /// Client coroutine - connects, sends messages, receives responses
 task<int> client_main(std::string_view host, uint16_t port) {
-    // Use the default io_context which is polled by scheduler workers
-    auto& ctx = io::default_io_context();
-
     ELIO_LOG_INFO("Connecting to {}:{}...", host, port);
 
     // Connect to server
-    auto stream_result = co_await tcp_connect(ctx, host, port);
+    auto stream_result = co_await tcp_connect(host, port);
 
     if (!stream_result) {
         ELIO_LOG_ERROR("Connection failed: {}", strerror(errno));
@@ -81,12 +78,9 @@ task<int> client_main(std::string_view host, uint16_t port) {
 
 /// Non-interactive benchmark mode
 task<int> benchmark_main(std::string_view host, uint16_t port, int iterations) {
-    // Use the default io_context which is polled by scheduler workers
-    auto& ctx = io::default_io_context();
-
     ELIO_LOG_INFO("Connecting to {}:{} for benchmark...", host, port);
 
-    auto stream_result = co_await tcp_connect(ctx, host, port);
+    auto stream_result = co_await tcp_connect(host, port);
     if (!stream_result) {
         ELIO_LOG_ERROR("Connection failed: {}", strerror(errno));
         co_return 1;
@@ -189,9 +183,6 @@ int main(int argc, char* argv[]) {
     // Create scheduler
     scheduler sched(2);
 
-    // Set the I/O context so workers can poll for I/O completions
-    sched.set_io_context(&io::default_io_context());
-
     sched.start();
 
     int result = 0;

examples/tcp_echo_server.cpp

@@ -199,9 +199,6 @@ int main(int argc, char* argv[]) {
     // Create scheduler with worker threads
     scheduler sched(4);
 
-    // Set the I/O context so workers can poll for I/O completions
-    sched.set_io_context(&io::default_io_context());
-
     sched.start();
 
     // Spawn signal handler coroutine

examples/uds_echo_client.cpp

@@ -21,13 +21,10 @@ using namespace elio::net;
 
 /// Client coroutine - connects, sends messages, receives responses
 task<int> client_main(const unix_address& addr) {
-    // Use the default io_context which is polled by scheduler workers
-    auto& ctx = io::default_io_context();
-
     ELIO_LOG_INFO("Connecting to {}...", addr.to_string());
 
     // Connect to server
-    auto stream_result = co_await uds_connect(ctx, addr);
+    auto stream_result = co_await uds_connect(addr);
 
     if (!stream_result) {
         ELIO_LOG_ERROR("Connection failed: {}", strerror(errno));
@@ -74,12 +71,9 @@ task<int> client_main(const unix_address& addr) {
 
 /// Non-interactive benchmark mode
 task<int> benchmark_main(const unix_address& addr, int iterations) {
-    // Use the default io_context which is polled by scheduler workers
-    auto& ctx = io::default_io_context();
-
     ELIO_LOG_INFO("Connecting to {} for benchmark...", addr.to_string());
 
-    auto stream_result = co_await uds_connect(ctx, addr);
+    auto stream_result = co_await uds_connect(addr);
     if (!stream_result) {
         ELIO_LOG_ERROR("Connection failed: {}", strerror(errno));
         co_return 1;
@@ -179,9 +173,6 @@ int main(int argc, char* argv[]) {
     // Create scheduler
     scheduler sched(2);
 
-    // Set the I/O context so workers can poll for I/O completions
-    sched.set_io_context(&io::default_io_context());
-
     sched.start();
 
     int result = 0;

examples/uds_echo_server.cpp

@@ -162,9 +162,6 @@ int main(int argc, char* argv[]) {
     // Create scheduler with worker threads
     scheduler sched(4);
 
-    // Set the I/O context so workers can poll for I/O completions
-    sched.set_io_context(&io::default_io_context());
-
     sched.start();
 
     // Spawn signal handler coroutine

examples/websocket_client.cpp

@@ -258,7 +258,6 @@ int main(int argc, char* argv[]) {
 
     // Create scheduler
     runtime::scheduler sched(2);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Run client

examples/websocket_server.cpp

@@ -309,7 +309,6 @@ int main(int argc, char* argv[]) {
 
     // Create scheduler
     runtime::scheduler sched(4);
-    sched.set_io_context(&io::default_io_context());
     sched.start();
 
     // Spawn signal handler coroutine

include/elio/http/http.hpp

@@ -52,7 +52,6 @@ namespace http {
 ///     server srv(std::move(r));
 ///     
 ///     runtime::scheduler sched(4);
-///     sched.set_io_context(&io::default_io_context());
 ///     sched.start();
 ///     
 ///     auto task = srv.listen(net::ipv4_address(8080), io::default_io_context(), sched);

include/elio/http/http2_client.hpp

@@ -197,7 +197,7 @@ class h2_client {
 
         // Create new HTTP/2 connection
         // First establish TCP connection
-        auto tcp_result = co_await net::tcp_connect(*io_ctx_, host, port);
+        auto tcp_result = co_await net::tcp_connect(host, port);
         if (!tcp_result) {
             ELIO_LOG_ERROR("Failed to connect to {}:{}", host, port);
             co_return std::nullopt;