diff --git a/CMakeLists.txt b/CMakeLists.txt index 21624dc68..e7068f1e4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -57,6 +57,7 @@ option(CROW_RETURNS_OK_ON_HTTP_OPTIONS_REQUEST option(CROW_ENABLE_SSL "Enable Crow's SSL feature for supporting https" OFF) option(CROW_ENABLE_COMPRESSION "Enable Crow's Compression feature for supporting compressed http content" OFF) +option(CROW_ENABLE_TSAN "Enable ThreadSanitizer" OFF) if(CROW_GENERATE_SBOM OR CROW_BUILD_TESTS) include(${CMAKE_CURRENT_SOURCE_DIR}/cmake/CPM.cmake) @@ -82,6 +83,16 @@ if(CROW_GENERATE_SBOM) ) endif() +if(CROW_ENABLE_TSAN) + if(CMAKE_CXX_COMPILER_ID STREQUAL "Clang") + message(STATUS "ThreadSanitizer enabled") + add_compile_options(-fsanitize=thread -fno-omit-frame-pointer) + add_link_options(-fsanitize=thread) + else() + message(WARNING "ThreadSanitizer is exclusive to Clang and cannot be used with any other compilers") + endif() +endif() + ##################################### # Define Targets ##################################### diff --git a/include/crow/app.h b/include/crow/app.h index f25ab0071..923d91dd7 100644 --- a/include/crow/app.h +++ b/include/crow/app.h @@ -619,6 +619,7 @@ namespace crow void close_websockets() { + std::lock_guard lock{websockets_mutex_}; for (auto websocket : websockets_) { CROW_LOG_INFO << "Quitting Websocket: " << websocket; @@ -629,11 +630,13 @@ namespace crow void add_websocket(std::shared_ptr conn) { + std::lock_guard lock{websockets_mutex_}; websockets_.push_back(conn); } void remove_websocket(std::shared_ptr conn) { + std::lock_guard lock{websockets_mutex_}; websockets_.erase(std::remove(websockets_.begin(), websockets_.end(), conn), websockets_.end()); } @@ -846,6 +849,7 @@ namespace crow bool server_started_{false}; std::condition_variable cv_started_; std::mutex start_mutex_; + std::mutex websockets_mutex_; ///< \brief mutex to protect websockets_ std::vector> websockets_; }; diff --git a/tests/unit_tests/test_websocket.cpp b/tests/unit_tests/test_websocket.cpp index 48b6087e5..e856f957a 100644 --- a/tests/unit_tests/test_websocket.cpp +++ b/tests/unit_tests/test_websocket.cpp @@ -1,6 +1,8 @@ #include "catch2/catch_all.hpp" #include "crow.h" +#include +#include using namespace std; using namespace crow; @@ -649,4 +651,90 @@ TEST_CASE("mirror_websocket_subprotocols", "[websocket]") } app.stop(); -} \ No newline at end of file +} + +TEST_CASE("multithreaded_websockets_open_close", "[websocket]") +{ + static std::string http_message = + "GET /ws HTTP/1.1\r\n" + "Connection: keep-alive, Upgrade\r\n" + "upgrade: websocket\r\n" + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + "Sec-WebSocket-Version: 13\r\n" + "Host: localhost\r\n" + "\r\n"; + + CROW_LOG_INFO << "Setting up app!\n"; + SimpleApp app; + + CROW_WEBSOCKET_ROUTE(app, "/ws") + .onaccept([&](const crow::request& req, void**) { + CROW_LOG_INFO << "Accepted websocket with URL " << req.url; + return true; + }) + .onopen([&](websocket::connection&) { + CROW_LOG_INFO << "Connected websocket"; + }) + .onmessage([&](websocket::connection&, const std::string&, bool) {}) + .onclose([&](websocket::connection& conn, const std::string&, uint16_t) { + // There should just be one connection + CHECK_FALSE(conn.get_remote_ip().empty()); + CROW_LOG_INFO << "Closing websocket"; + }); + + app.validate(); + + CROW_LOG_WARNING << "Starting app!\n"; + auto _ = app.bindaddr(LOCALHOST_ADDRESS).port(45453).run_async(); + app.wait_for_server_start(); + CROW_LOG_WARNING << "App started!\n"; + asio::io_context ic; + + const auto thread_function = [&]() + { + asio::ip::tcp::socket c(ic); + c.connect(asio::ip::tcp::endpoint( + asio::ip::make_address(LOCALHOST_ADDRESS), 45453)); + + CROW_LOG_WARNING << "Connected!\n"; + + char buf[2048]; + + //----------Handshake---------- + { + std::fill_n(buf, 2048, 0); + c.send(asio::buffer(http_message)); + + c.receive(asio::buffer(buf, 2048)); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + + //----------Close websocket---------- + std::fill_n(buf, 2048, 0); + // Close message with, len = 2, status code = 1000 + char close_message[5]("\x88\x02\x03\xE8"); + c.send(asio::buffer(close_message, 4)); + c.receive(asio::buffer(buf, 2048)); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + }; + + const auto multiple_run_thread_function = [&](const std::size_t count){ + for(std::size_t i = 0; i < count; ++i) { + thread_function(); + } + }; + + constexpr std::size_t threads_count = 3; + constexpr std::size_t thread_run = 10; + + std::vector threads; + for(std::size_t i = 0; i < threads_count; ++i) { + threads.emplace_back(multiple_run_thread_function, thread_run); + } + for(std::thread& thread : threads) { + thread.join(); + } + + CROW_LOG_WARNING << "Stopping app!\n"; + app.stop(); +}