diff --git a/README.md b/README.md index 24b8d23..01ebdb8 100644 --- a/README.md +++ b/README.md @@ -71,21 +71,21 @@ cd ../benchmarks DURATION=30s THREADS=4 CONNECTIONS=100 ./benchmark.sh ``` -Output (non-ASAN Release build): +Output (non-ASAN Release build, generated Thu Jun 11 20:26:47 MSK 2026): #### GET /echo | Server | Latency avg | Latency stdev | Latency max | +/- stdev | Requests/sec | Transfer/sec | | --- | ---: | ---: | ---: | ---: | ---: | ---: | -| C++ coroutine server | 132.16us | 66.78us | 4.97ms | 80.82% | 439792.17 | 29.78MB | -| Rust Tokio server | 150.12us | 119.34us | 5.53ms | 94.01% | 506996.06 | 40.61MB | +| C++ coroutine server | 100.04us | 50.46us | 5.11ms | 87.62% | 562133.10 | 45.03MB | +| Rust Tokio server | 131.02us | 105.52us | 4.47ms | 94.77% | 559214.10 | 44.80MB | #### POST /echo | Server | Latency avg | Latency stdev | Latency max | +/- stdev | Requests/sec | Transfer/sec | | --- | ---: | ---: | ---: | ---: | ---: | ---: | -| C++ coroutine server | 149.98us | 77.14us | 5.27ms | 81.01% | 402624.34 | 29.57MB | -| Rust Tokio server | 168.98us | 138.11us | 5.80ms | 91.79% | 465844.80 | 39.98MB | +| C++ coroutine server | 107.32us | 47.45us | 3.89ms | 84.16% | 519114.00 | 44.56MB | +| Rust Tokio server | 157.70us | 134.03us | 7.43ms | 92.10% | 510284.18 | 43.80MB | ## Pull Request Checks diff --git a/server/CTestTestfile.cmake b/server/CTestTestfile.cmake new file mode 100644 index 0000000..f4c5f14 --- /dev/null +++ b/server/CTestTestfile.cmake @@ -0,0 +1,10 @@ +# CMake generated Testfile for +# Source directory: /home/shfdis/httpserver/server +# Build directory: /home/shfdis/httpserver/server +# +# This file includes the relevant testing commands required for +# testing this directory and lists subdirectories to be tested as well. +add_test(http_parser_test "/home/shfdis/httpserver/server/tests/http_parser_test") +set_tests_properties(http_parser_test PROPERTIES _BACKTRACE_TRIPLES "/home/shfdis/httpserver/server/CMakeLists.txt;64;add_test;/home/shfdis/httpserver/server/CMakeLists.txt;0;") +add_test(http_integration_test "/home/shfdis/httpserver/server/tests/http_integration_test") +set_tests_properties(http_integration_test PROPERTIES _BACKTRACE_TRIPLES "/home/shfdis/httpserver/server/CMakeLists.txt;71;add_test;/home/shfdis/httpserver/server/CMakeLists.txt;0;") diff --git a/server/DartConfiguration.tcl b/server/DartConfiguration.tcl new file mode 100644 index 0000000..1e6edf5 --- /dev/null +++ b/server/DartConfiguration.tcl @@ -0,0 +1,109 @@ +# This file is configured by CMake automatically as DartConfiguration.tcl +# If you choose not to use CMake, this file may be hand configured, by +# filling in the required variables. + + +# Configuration directories and files +SourceDirectory: /home/shfdis/httpserver/server +BuildDirectory: /home/shfdis/httpserver/server + +# Where to place the cost data store +CostDataFile: + +# Site is something like machine.domain, i.e. pragmatic.crd +Site: archlinux + +# Build name is osname-revision-compiler, i.e. Linux-2.4.2-2smp-c++ +BuildName: Linux-c++ + +# Subprojects +LabelsForSubprojects: + +# Submission information +SubmitURL: http:// +SubmitInactivityTimeout: + +# Dashboard start time +NightlyStartTime: 00:00:00 EDT + +# Commands for the build/test/submit cycle +ConfigureCommand: "/usr/bin/cmake" "/home/shfdis/httpserver/server" +MakeCommand: /usr/bin/cmake --build . --config "${CTEST_CONFIGURATION_TYPE}" +DefaultCTestConfigurationType: Release + +# version control +UpdateVersionOnly: + +# CVS options +# Default is "-d -P -A" +CVSCommand: +CVSUpdateOptions: + +# Subversion options +SVNCommand: +SVNOptions: +SVNUpdateOptions: + +# Git options +GITCommand: +GITInitSubmodules: +GITUpdateOptions: +GITUpdateCustom: + +# Perforce options +P4Command: +P4Client: +P4Options: +P4UpdateOptions: +P4UpdateCustom: + +# Generic update command +UpdateCommand: +UpdateOptions: +UpdateType: + +# Compiler info +Compiler: /usr/bin/c++ +CompilerVersion: 16.1.1 + +# Dynamic analysis (MemCheck) +PurifyCommand: +ValgrindCommand: +ValgrindCommandOptions: +DrMemoryCommand: +DrMemoryCommandOptions: +CudaSanitizerCommand: +CudaSanitizerCommandOptions: +MemoryCheckType: +MemoryCheckSanitizerOptions: +MemoryCheckCommand: MEMORYCHECK_COMMAND-NOTFOUND +MemoryCheckCommandOptions: +MemoryCheckSuppressionFile: + +# Coverage +CoverageCommand: /usr/bin/gcov +CoverageExtraFlags: -l + +# Testing options +# TimeOut is the amount of time in seconds to wait for processes +# to complete during testing. After TimeOut seconds, the +# process will be summarily terminated. +# Currently set to 25 minutes +TimeOut: 1500 + +# During parallel testing CTest will not start a new test if doing +# so would cause the system load to exceed this value. +TestLoad: + +TLSVerify: +TLSVersion: + +UseLaunchers: +CurlOptions: +# warning, if you add new options here that have to do with submit, +# you have to update cmCTestSubmitCommand.cxx + +# For CTest submissions that timeout, these options +# specify behavior for retrying the submission +CTestSubmitRetryDelay: 5 +CTestSubmitRetryCount: 3 diff --git a/server/Testing/Temporary/CTestCostData.txt b/server/Testing/Temporary/CTestCostData.txt new file mode 100644 index 0000000..6339b36 --- /dev/null +++ b/server/Testing/Temporary/CTestCostData.txt @@ -0,0 +1,3 @@ +http_parser_test 9 0.000310016 +http_integration_test 9 0.000431396 +--- diff --git a/server/Testing/Temporary/LastTest.log b/server/Testing/Temporary/LastTest.log new file mode 100644 index 0000000..7b5486f --- /dev/null +++ b/server/Testing/Temporary/LastTest.log @@ -0,0 +1,33 @@ +Start testing: Jun 11 20:24 MSK +---------------------------------------------------------- +1/2 Testing: http_parser_test +1/2 Test: http_parser_test +Command: "/home/shfdis/httpserver/server/tests/http_parser_test" +Directory: /home/shfdis/httpserver/server +"http_parser_test" start time: Jun 11 20:24 MSK +Output: +---------------------------------------------------------- + +Test time = 0.00 sec +---------------------------------------------------------- +Test Passed. +"http_parser_test" end time: Jun 11 20:24 MSK +"http_parser_test" time elapsed: 00:00:00 +---------------------------------------------------------- + +2/2 Testing: http_integration_test +2/2 Test: http_integration_test +Command: "/home/shfdis/httpserver/server/tests/http_integration_test" +Directory: /home/shfdis/httpserver/server +"http_integration_test" start time: Jun 11 20:24 MSK +Output: +---------------------------------------------------------- + +Test time = 0.00 sec +---------------------------------------------------------- +Test Passed. +"http_integration_test" end time: Jun 11 20:24 MSK +"http_integration_test" time elapsed: 00:00:00 +---------------------------------------------------------- + +End testing: Jun 11 20:24 MSK diff --git a/server/include/co_future.h b/server/include/co_future.h index 5dcc1a4..3e28234 100644 --- a/server/include/co_future.h +++ b/server/include/co_future.h @@ -2,10 +2,10 @@ #include #include +#include #include #include #include -#include #include #include #include @@ -15,6 +15,8 @@ namespace HTTP { +class IOUring; + template class CoPromise; @@ -29,21 +31,76 @@ class CoFuture { friend class CoPromise; template friend class CoFutureAwaiter; + friend class IOUring; + + using StoredValue = std::conditional_t, bool, T>; struct ControlBlock { - std::mutex mutex; std::atomic_bool ready{false}; + std::atomic_bool satisfied{false}; std::atomic_bool futureRetrieved{false}; + std::atomic_uint waiters{0}; std::conditional_t, bool, std::optional> value; std::exception_ptr exception; using Callback = std::function; - std::atomic> continuation{nullptr}; + std::atomic_uintptr_t continuation{0}; }; std::shared_ptr control_; explicit CoFuture(std::shared_ptr control) : control_(std::move(control)) {} + static constexpr uintptr_t kCallbackTag = 1; + + static void RunContinuation(uintptr_t continuation) { + if (continuation == 0) { + return; + } + if ((continuation & kCallbackTag) != 0) { + std::unique_ptr callback( + reinterpret_cast( + continuation & ~kCallbackTag)); + (*callback)(); + return; + } + auto handle = + std::coroutine_handle<>::from_address(reinterpret_cast(continuation)); + if (handle && !handle.done()) { + handle.resume(); + } + } + + static void PublishReady(const std::shared_ptr &control) { + control->ready.store(true, std::memory_order_release); + auto continuation = control->continuation.exchange(0, std::memory_order_acq_rel); + if (control->waiters.load(std::memory_order_acquire) > 0) { + control->ready.notify_all(); + } + RunContinuation(continuation); + } + + static void CompleteValue(const std::shared_ptr &control, + StoredValue value = StoredValue{}) { + if (control->satisfied.exchange(true, std::memory_order_acq_rel)) { + throw std::runtime_error("Promise already satisfied"); + } + if constexpr (std::is_void_v) { + control->value = true; + } else { + control->value = std::move(value); + } + PublishReady(control); + } + + static void CompleteException(const std::shared_ptr &control, + std::exception_ptr exception) { + if (control->satisfied.exchange(true, std::memory_order_acq_rel)) { + throw std::runtime_error("Promise already satisfied"); + } + control->exception = std::move(exception); + PublishReady(control); + } + public: CoFuture() = default; @@ -58,9 +115,12 @@ class CoFuture { throw std::logic_error("Invalid future"); } - if (!control_->ready.load(std::memory_order_acquire)) { - control_->mutex.lock(); - control_->mutex.unlock(); + while (!control_->ready.load(std::memory_order_acquire)) { + control_->waiters.fetch_add(1, std::memory_order_acq_rel); + if (!control_->ready.load(std::memory_order_acquire)) { + control_->ready.wait(false, std::memory_order_acquire); + } + control_->waiters.fetch_sub(1, std::memory_order_acq_rel); } if (control_->exception) { @@ -93,25 +153,30 @@ class CoFuture { nextPromise->SetException(std::current_exception()); } }; - auto callback = std::make_shared(std::move(run)); + auto callback = + std::make_unique(std::move(run)); if (control_->ready.load(std::memory_order_acquire)) { (*callback)(); return nextFuture; } - std::shared_ptr expected = nullptr; - if (!control_->continuation.compare_exchange_strong(expected, callback, + uintptr_t raw = reinterpret_cast(callback.get()); + assert((raw & kCallbackTag) == 0); + raw |= kCallbackTag; + uintptr_t expected = 0; + if (!control_->continuation.compare_exchange_strong(expected, raw, std::memory_order_acq_rel, std::memory_order_acquire)) { throw std::runtime_error("Tried to set the subscriber second time"); } + callback.release(); if (control_->ready.load(std::memory_order_acquire)) { - expected = callback; - if (control_->continuation.compare_exchange_strong(expected, nullptr, + expected = raw; + if (control_->continuation.compare_exchange_strong(expected, 0, std::memory_order_acq_rel, std::memory_order_acquire)) { - (*callback)(); + RunContinuation(raw); } } return nextFuture; @@ -123,13 +188,12 @@ class CoPromise { using ControlBlock = typename CoFuture::ControlBlock; std::shared_ptr control_; - std::unique_lock producerLock_; template friend class CoFuture; public: - CoPromise() : control_(std::make_shared()), producerLock_(control_->mutex) {} + CoPromise() : control_(std::make_shared()) {} CoPromise(const CoPromise &) = delete; CoPromise &operator=(const CoPromise &) = delete; @@ -152,35 +216,13 @@ class CoPromise { CoFuture getFuture() { return GetFuture(); } void Set(T value) { - if (!producerLock_.owns_lock()) { - throw std::runtime_error("Promise already satisfied"); - } - - control_->value = std::move(value); - control_->ready.store(true, std::memory_order_release); - auto continuation = control_->continuation.exchange(nullptr, std::memory_order_acq_rel); - producerLock_.unlock(); - - if (continuation) { - (*continuation)(); - } + CoFuture::CompleteValue(control_, std::move(value)); } void set(T value) { Set(std::move(value)); } void SetException(std::exception_ptr exception) { - if (!producerLock_.owns_lock()) { - throw std::runtime_error("Promise already satisfied"); - } - - control_->exception = std::move(exception); - control_->ready.store(true, std::memory_order_release); - auto continuation = control_->continuation.exchange(nullptr, std::memory_order_acq_rel); - producerLock_.unlock(); - - if (continuation) { - (*continuation)(); - } + CoFuture::CompleteException(control_, std::move(exception)); } void setException(std::exception_ptr exception) { SetException(std::move(exception)); } @@ -191,10 +233,9 @@ class CoPromise { using ControlBlock = typename CoFuture::ControlBlock; std::shared_ptr control_; - std::unique_lock producerLock_; public: - CoPromise() : control_(std::make_shared()), producerLock_(control_->mutex) {} + CoPromise() : control_(std::make_shared()) {} CoPromise(const CoPromise &) = delete; CoPromise &operator=(const CoPromise &) = delete; @@ -217,30 +258,13 @@ class CoPromise { CoFuture getFuture() { return GetFuture(); } void Set() { - if (!producerLock_.owns_lock()) { - throw std::runtime_error("Promise already satisfied"); - } - control_->ready.store(true, std::memory_order_release); - auto continuation = control_->continuation.exchange(nullptr, std::memory_order_acq_rel); - producerLock_.unlock(); - if (continuation) { - (*continuation)(); - } + CoFuture::CompleteValue(control_, true); } void set() { Set(); } void SetException(std::exception_ptr exception) { - if (!producerLock_.owns_lock()) { - throw std::runtime_error("Promise already satisfied"); - } - control_->exception = std::move(exception); - control_->ready.store(true, std::memory_order_release); - auto continuation = control_->continuation.exchange(nullptr, std::memory_order_acq_rel); - producerLock_.unlock(); - if (continuation) { - (*continuation)(); - } + CoFuture::CompleteException(control_, std::move(exception)); } void setException(std::exception_ptr exception) { SetException(std::move(exception)); } @@ -261,27 +285,22 @@ class CoFutureAwaiter { } bool await_suspend(std::coroutine_handle<> awaiting) { - auto callback = std::make_shared( - [awaiting]() mutable { - if (awaiting && !awaiting.done()) { - awaiting.resume(); - } - }); - if (control_->ready.load(std::memory_order_acquire)) { return false; } - std::shared_ptr expected = nullptr; - if (!control_->continuation.compare_exchange_strong(expected, callback, + uintptr_t raw = reinterpret_cast(awaiting.address()); + assert((raw & CoFuture::kCallbackTag) == 0); + uintptr_t expected = 0; + if (!control_->continuation.compare_exchange_strong(expected, raw, std::memory_order_acq_rel, std::memory_order_acquire)) { throw std::runtime_error("Tried to set the subscriber second time"); } if (control_->ready.load(std::memory_order_acquire)) { - expected = callback; - if (control_->continuation.compare_exchange_strong(expected, nullptr, + expected = raw; + if (control_->continuation.compare_exchange_strong(expected, 0, std::memory_order_acq_rel, std::memory_order_acquire)) { return false; diff --git a/server/include/io_uring.h b/server/include/io_uring.h index 3d33d6a..3e32985 100644 --- a/server/include/io_uring.h +++ b/server/include/io_uring.h @@ -4,11 +4,9 @@ #include #include #include -#include #include #include #include -#include #include #include #define QUEUE_DEPTH 1024 @@ -24,21 +22,25 @@ class IOUring { struct Entry { OpType type; int fd; - std::optional toRead; - std::function complete; + char *toRead{nullptr}; + std::shared_ptr::ControlBlock> control; std::string_view writeData; size_t writeOffset{0}; size_t writeLen{0}; }; struct SqeData { - std::function complete; + std::shared_ptr::ControlBlock> control; std::string_view writeData; size_t writeOffset{0}; size_t writeLen{0}; + bool inUse{false}; }; std::deque queue_; + std::array sqeData_{}; io_uring ring_; std::atomic_uint64_t inProcess_ = 0; + size_t nextSqeData_{0}; + SqeData *AcquireSqeData(); void ProcessCalls(); void AddEntries(); @@ -47,13 +49,15 @@ class IOUring { ~IOUring(); IOUring(); IOUring &operator=(IOUring &&rhs); - void Read(int fileDescriptor, std::array &buffer, std::function complete); + void Read(int fileDescriptor, std::array &buffer, + std::shared_ptr::ControlBlock> control); CoFuture ReadAsync(int fileDescriptor, std::array &buffer); void Write(int fileDescriptor, std::string_view data, size_t offset, size_t len, - std::function complete); + std::shared_ptr::ControlBlock> control); CoFuture WriteAsync(int fileDescriptor, std::string_view data, size_t offset, size_t len); - void Accept(int fileDescriptor, std::function complete); + void Accept(int fileDescriptor, + std::shared_ptr::ControlBlock> control); CoFuture AcceptAsync(int fileDescriptor); }; } diff --git a/server/include/request_data.h b/server/include/request_data.h index c797ff1..05fcd88 100644 --- a/server/include/request_data.h +++ b/server/include/request_data.h @@ -11,9 +11,59 @@ enum Method { GET, PUT, POST, PATCH, DELETE, HEAD, OPTIONS, UNKNOWN }; constexpr size_t kRoutableMethodCount = 7; +struct QueryParams { + using Storage = std::vector>; + using iterator = Storage::iterator; + using const_iterator = Storage::const_iterator; + + Storage values; + + void clear() { values.clear(); } + void reserve(size_t count) { values.reserve(count); } + iterator begin() { return values.begin(); } + iterator end() { return values.end(); } + const_iterator begin() const { return values.begin(); } + const_iterator end() const { return values.end(); } + + iterator find(std::string_view name) { + for (auto it = values.begin(); it != values.end(); ++it) { + if (it->first == name) { + return it; + } + } + return values.end(); + } + + const_iterator find(std::string_view name) const { + for (auto it = values.begin(); it != values.end(); ++it) { + if (it->first == name) { + return it; + } + } + return values.end(); + } + + std::string &operator[](std::string_view name) { + auto it = find(name); + if (it != end()) { + return it->second; + } + values.emplace_back(name, std::string{}); + return values.back().second; + } + + void Set(std::string_view name, std::string_view value) { + auto it = find(name); + if (it != end()) { + it->second.assign(value); + return; + } + values.emplace_back(name, value); + } +}; + struct RequestData { - std::unordered_map headers; - std::unordered_map params; + QueryParams params; std::unordered_map trailers; std::vector> rawHeaders; std::vector urlVariables; @@ -28,10 +78,30 @@ struct RequestData { bool connectionKeepAlive{false}; std::optional Header(std::string_view name) const; + void Clear() { + params.clear(); + trailers.clear(); + rawHeaders.clear(); + urlVariables.clear(); + method = UNKNOWN; + methodToken.clear(); + target.clear(); + path.clear(); + query.clear(); + version.clear(); + body.clear(); + connectionClose = false; + connectionKeepAlive = false; + } }; struct ResponseData { std::unordered_map headers; std::string body; unsigned short status{200}; + void Clear() { + headers.clear(); + body.clear(); + status = 200; + } }; } // namespace HTTP diff --git a/server/src/http_parser.cpp b/server/src/http_parser.cpp index 99d8287..7c6c1ee 100644 --- a/server/src/http_parser.cpp +++ b/server/src/http_parser.cpp @@ -259,16 +259,7 @@ HttpParseResult parse_field_line(std::string_view line, std::string_view &name, void add_header(RequestData &request, std::string_view name, std::string_view value) { - std::string storedName(name); - std::string storedValue(value); - request.rawHeaders.emplace_back(storedName, storedValue); - auto existing = request.headers.find(storedName); - if (existing == request.headers.end()) { - request.headers.emplace(std::move(storedName), std::move(storedValue)); - } else { - existing->second += ", "; - existing->second += storedValue; - } + request.rawHeaders.emplace_back(std::string(name), std::string(value)); } void add_trailer(RequestData &request, std::string_view name, @@ -300,11 +291,7 @@ void parse_query_params(RequestData &request) { eq == std::string_view::npos ? std::string_view() : part.substr(eq + 1); if (!name.empty()) { - auto [it, inserted] = - request.params.emplace(std::string(name), std::string(value)); - if (!inserted) { - it->second.assign(value); - } + request.params.Set(name, value); } } if (amp == std::string::npos) { @@ -492,26 +479,20 @@ std::optional RequestData::Header(std::string_view name) const { if (!result.empty()) { return result; } - - for (const auto &[fieldName, value] : headers) { - if (iequals(fieldName, name)) { - return value; - } - } return std::nullopt; } HttpParserState::HttpParserState() { line_.reserve(256); pending_.reserve(256); - current_.headers.reserve(8); + current_.params.reserve(4); current_.rawHeaders.reserve(8); } void HttpParserState::ResetForNext() { state_ = State::StartLine; - current_ = RequestData{}; - current_.headers.reserve(8); + current_.Clear(); + current_.params.reserve(4); current_.rawHeaders.reserve(8); line_.clear(); sawCr_ = false; @@ -1003,7 +984,7 @@ void HttpParserState::MarkContinueSent() { HttpParseResult HttpParserState::ParseNext(RequestData &request) { if (state_ == State::Complete) { - request = std::move(current_); + std::swap(request, current_); ResetForNext(); return Complete(); } @@ -1024,7 +1005,7 @@ HttpParseResult HttpParserState::ParseNext(RequestData &request) { pending_.append(std::string_view(data).substr(consumed)); } if (state_ == State::Complete) { - request = std::move(current_); + std::swap(request, current_); ResetForNext(); return Complete(); } diff --git a/server/src/io_uring.cpp b/server/src/io_uring.cpp index 6fc8821..4bf2342 100644 --- a/server/src/io_uring.cpp +++ b/server/src/io_uring.cpp @@ -19,6 +19,18 @@ IOUring::IOUring() { } } +IOUring::SqeData *IOUring::AcquireSqeData() { + for (size_t i = 0; i < sqeData_.size(); ++i) { + SqeData &data = sqeData_[(nextSqeData_ + i) % sqeData_.size()]; + if (!data.inUse) { + data.inUse = true; + nextSqeData_ = (nextSqeData_ + i + 1) % sqeData_.size(); + return &data; + } + } + return nullptr; +} + void IOUring::AddEntries() { size_t fdsSize = 0; for (int count = 0; count < QUEUE_DEPTH && !queue_.empty(); count++) { @@ -30,13 +42,19 @@ void IOUring::AddEntries() { queue_.push_front(std::move(entry)); break; } - - SqeData *sqeData = new SqeData{ - std::move(entry.complete), std::move(entry.writeData), entry.writeOffset, entry.writeLen}; + + SqeData *sqeData = AcquireSqeData(); + if (sqeData == nullptr) { + queue_.push_front(std::move(entry)); + break; + } + sqeData->control = std::move(entry.control); + sqeData->writeData = entry.writeData; + sqeData->writeOffset = entry.writeOffset; + sqeData->writeLen = entry.writeLen; if (entry.type == IOUring::READ) [[likely]] { - io_uring_prep_read(sqEntry, entry.fd, entry.toRead.value(), - kReadBufferSize, 0); + io_uring_prep_read(sqEntry, entry.fd, entry.toRead, kReadBufferSize, 0); } else if (entry.type == IOUring::ACCEPT) { io_uring_prep_accept(sqEntry, entry.fd, nullptr, nullptr, 0); } else [[likely]] { @@ -75,7 +93,8 @@ void IOUring::Poll() { } void IOUring::Write(int fileDescriptor, std::string_view data, size_t offset, - size_t len, std::function complete) { + size_t len, + std::shared_ptr::ControlBlock> control) { if (fileDescriptor < 0) { throw std::runtime_error("Invalid file descriptor"); } @@ -85,14 +104,14 @@ void IOUring::Write(int fileDescriptor, std::string_view data, size_t offset, entry.writeData = data; entry.writeOffset = offset; entry.writeLen = len; - entry.complete = std::move(complete); + entry.control = std::move(control); queue_.push_back(std::move(entry)); AddEntries(); } void IOUring::Read(int fileDescriptor, std::array &buffer, - std::function complete) { + std::shared_ptr::ControlBlock> control) { if (fileDescriptor < 0) { throw std::runtime_error("Invalid file descriptor"); } @@ -100,49 +119,44 @@ void IOUring::Read(int fileDescriptor, entry.type = IOUring::READ; entry.fd = fileDescriptor; entry.toRead = buffer.begin(); - entry.complete = std::move(complete); + entry.control = std::move(control); queue_.push_back(entry); } CoFuture IOUring::ReadAsync( int fileDescriptor, std::array &buffer) { - auto promise = std::make_shared>(); - auto future = promise->GetFuture(); - Read(fileDescriptor, buffer, [promise](int result) { - promise->Set(result); - }); - return future; + auto control = std::make_shared::ControlBlock>(); + control->futureRetrieved.store(true, std::memory_order_release); + Read(fileDescriptor, buffer, control); + return CoFuture(std::move(control)); } -void IOUring::Accept(int fileDescriptor, std::function complete) { +void IOUring::Accept( + int fileDescriptor, std::shared_ptr::ControlBlock> control) { if (fileDescriptor < 0) { throw std::runtime_error("Invalid file descriptor"); } Entry entry; entry.type = IOUring::ACCEPT; entry.fd = fileDescriptor; - entry.complete = std::move(complete); + entry.control = std::move(control); queue_.push_back(entry); AddEntries(); } CoFuture IOUring::AcceptAsync(int fileDescriptor) { - auto promise = std::make_shared>(); - auto future = promise->GetFuture(); - Accept(fileDescriptor, [promise](int result) { - promise->Set(result); - }); - return future; + auto control = std::make_shared::ControlBlock>(); + control->futureRetrieved.store(true, std::memory_order_release); + Accept(fileDescriptor, control); + return CoFuture(std::move(control)); } CoFuture IOUring::WriteAsync(int fileDescriptor, std::string_view data, size_t offset, size_t len) { - auto promise = std::make_shared>(); - auto future = promise->GetFuture(); - Write(fileDescriptor, data, offset, len, [promise](int result) { - promise->Set(result); - }); - return future; + auto control = std::make_shared::ControlBlock>(); + control->futureRetrieved.store(true, std::memory_order_release); + Write(fileDescriptor, data, offset, len, control); + return CoFuture(std::move(control)); } void IOUring::ProcessCalls() { @@ -164,13 +178,16 @@ void IOUring::ProcessCalls() { } int result = cqEntry->res; - auto complete = std::move(sqeData->complete); + auto control = std::move(sqeData->control); - delete sqeData; + sqeData->writeData = {}; + sqeData->writeOffset = 0; + sqeData->writeLen = 0; + sqeData->inUse = false; io_uring_cqe_seen(&ring_, cqEntry); - if (complete) { - complete(result); + if (control) { + CoFuture::CompleteValue(control, result); } } diff --git a/server/src/server.cpp b/server/src/server.cpp index d6f2663..5c304b7 100644 --- a/server/src/server.cpp +++ b/server/src/server.cpp @@ -317,6 +317,20 @@ CoFuture Server::WriteResponse(IOUring &ring, int connectionFD, const RequestData &request, bool keepAlive, std::string &buffer) { const unsigned short status = NormalizeStatus(data.status); + if (status == 200 && keepAlive && request.version != "HTTP/1.0" && + request.method != HEAD && data.headers.empty()) { + buffer.clear(); + buffer.reserve(72 + data.body.size()); + buffer += "HTTP/1.1 200 OK\r\nDate: "; + buffer += HttpDate(); + buffer += "\r\nContent-Length: "; + AppendUnsigned(buffer, data.body.size()); + buffer += "\r\n\r\n"; + buffer += data.body; + co_await WriteRaw(ring, connectionFD, buffer); + co_return; + } + const bool sendBody = ShouldSendBody(status, request.method); buffer.clear(); buffer.reserve(160 + data.body.size()); @@ -362,12 +376,14 @@ CoFuture Server::WriteResponse(IOUring &ring, int connectionFD, CoFuture Server::Process(IOUring &ring, int connectionFD) { HttpRequestParser parser(ring, connectionFD); + RequestData request; + ResponseData response; std::string writeBuffer; writeBuffer.reserve(256); while (true) { - RequestData request; - ResponseData response; + request.Clear(); + response.Clear(); bool keepAlive = true; bool mustClose = false; diff --git a/server/tests/http_integration_test b/server/tests/http_integration_test index a31ab43..c49477d 100755 Binary files a/server/tests/http_integration_test and b/server/tests/http_integration_test differ diff --git a/server/tests/http_parser_test b/server/tests/http_parser_test index 2b0c5bf..3d15d32 100755 Binary files a/server/tests/http_parser_test and b/server/tests/http_parser_test differ