diff --git a/Common/Cpp/Sockets/AbstractServerSocket.h b/Common/Cpp/Sockets/AbstractServerSocket.h index 8f9e13aacb..a3b53c921b 100644 --- a/Common/Cpp/Sockets/AbstractServerSocket.h +++ b/Common/Cpp/Sockets/AbstractServerSocket.h @@ -78,7 +78,10 @@ class AbstractServerSocket{ // State::LISTENING on success or State::NOT_RUNNING on failure. virtual void start(const std::string& address, uint16_t port) = 0; - // Closes the active connection, if any, and keeps listening. + // Closes the active connection, if any, and keeps listening. Unlike + // "close()", this does not wait for the worker threads, so it is safe + // to call from within a listener callback. "Listener::on_disconnect" + // is not called for the connection closed this way. virtual void close_connection() = 0; // Stops the server: it stops accepting new connections, closes the diff --git a/Common/Cpp/Sockets/ClientSocket_POSIX.cpp b/Common/Cpp/Sockets/ClientSocket_POSIX.cpp index 8e46b363a4..caf9d17773 100644 --- a/Common/Cpp/Sockets/ClientSocket_POSIX.cpp +++ b/Common/Cpp/Sockets/ClientSocket_POSIX.cpp @@ -133,10 +133,11 @@ size_t ClientSocket_POSIX::send(const void* data, size_t bytes){ size_t current = std::min(bytes, BLOCK_SIZE); ssize_t current_sent = ::send(socket, ptr, (int)current, MSG_DONTWAIT | MSG_NOSIGNAL); if (current_sent != -1){ + // send() may return after writing only part of the + // buffer (macOS does this when the kernel's send + // buffer is full); loop until everything has been + // written, the same way the Qt backend does. sent += current_sent; - if ((size_t)current_sent < current){ - return sent; - } ptr += current_sent; bytes -= current_sent; continue; diff --git a/Common/Cpp/Sockets/ClientSocket_Qt.h b/Common/Cpp/Sockets/ClientSocket_Qt.h index c8d235faef..c944ec2b6f 100644 --- a/Common/Cpp/Sockets/ClientSocket_Qt.h +++ b/Common/Cpp/Sockets/ClientSocket_Qt.h @@ -9,7 +9,6 @@ #include #include -#include #include #include "Common/Cpp/Concurrency/Mutex.h" #include "Common/Cpp/Concurrency/ConditionVariable.h" @@ -32,7 +31,7 @@ class ClientSocket_Qt final : public QObject, public AbstractClientSocket{ bool done = false; Mutex lock; - std::condition_variable cv; + ConditionVariable cv; }; diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.cpp b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp new file mode 100644 index 0000000000..cf375f1d49 --- /dev/null +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp @@ -0,0 +1,445 @@ +/* Server Socket (POSIX) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#if !_WIN32 + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "Common/Cpp/Concurrency/ThreadPool.h" +#include "ServerSocket_POSIX.h" + +namespace PokemonAutomation{ + +ServerSocket_POSIX::ServerSocket_POSIX(ThreadPool& thread_pool) + : m_thread_pool(thread_pool) +{ + // Ignore SIGPIPE. Handle errors via errno instead + signal(SIGPIPE, SIG_IGN); +} + +void ServerSocket_POSIX::start(const std::string& address, uint16_t port){ + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING + || m_starting){ + return; + } + m_starting = true; + } + + // Use the plain dispatch() rather than dispatch_now_blocking(): + // start() may be called from a thread that is itself running on + // this pool, and blocking here until a pool thread is free could + // deadlock a capped pool. The returned AsyncTask still lets + // close() join the listener thread. + try{ + AsyncTask task = m_thread_pool.dispatch([=, this]{ + accept_thread_loop(address, port); + }); + { + // m_task_lock, not m_lock: close() only reads the slot + // after it has waited for m_starting to clear, so this + // assignment is the only writer besides close(). + std::lock_guard lg(m_task_lock); + m_accept_thread = std::move(task); + } + }catch (...){ + { + std::lock_guard lg(m_lock); + m_starting = false; + m_cv.notify_all(); + } + m_listeners.run_method(&Listener::on_listen_finished, "Failed to start the listener thread."); + throw; + } + + { + std::lock_guard lg(m_lock); + m_starting = false; + m_cv.notify_all(); + } +} + +void ServerSocket_POSIX::close_connection(){ + int socket; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + return; + } + m_close_requested = true; + m_state.store(State::LISTENING, std::memory_order_release); + socket = m_client; + } + if (socket != -1){ + // Tells the connection thread to stop; it closes the socket + // itself. This is safe to call from within a listener callback. + ::shutdown(socket, SHUT_RDWR); + } +} + +void ServerSocket_POSIX::close() noexcept{ + { + std::unique_lock lg(m_lock); + // Wait for an in-progress start() to publish the listener task + // before we take the task slots. + m_cv.wait(lg, [&]{ return !m_starting; }); + if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING){ + m_close_requested = true; + m_state.store(State::DESTRUCTING, std::memory_order_release); + m_closing++; + } + } + + bool drain = false; + { + std::lock_guard lg(m_lock); + drain = (m_closing == 1); + } + if (!drain){ + // Either nothing was running (m_closing == 0), or another + // close() is already draining the threads (m_closing > 1). + // In the second case, wait for it to finish. + std::unique_lock lg(m_lock); + m_cv.wait(lg, [&]{ + return m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING; + }); + return; + } + + AsyncTask accept_thread; + AsyncTask client_thread; + { + // m_task_lock, not m_lock: moving the tasks out destroys the + // old AsyncTask objects, and their destructors wait for the + // tasks to finish - which they can only do if they can acquire + // m_lock. + std::lock_guard lg(m_task_lock); + accept_thread = std::move(m_accept_thread); + m_accept_thread = AsyncTask(); + client_thread = std::move(m_client_thread); + m_client_thread = AsyncTask(); + } + + // The threads notice the new state within a few milliseconds and + // close their own sockets. + accept_thread.wait_and_ignore_exceptions(); + client_thread.wait_and_ignore_exceptions(); + + { + std::lock_guard lg(m_lock); + m_state.store(State::NOT_RUNNING, std::memory_order_release); + m_close_requested = false; + m_closing = 0; + m_cv.notify_all(); + } +} + +size_t ServerSocket_POSIX::send(const void* data, size_t bytes){ + int socket; + { + std::lock_guard lg(m_lock); + socket = + (m_state.load(std::memory_order_relaxed) == State::CONNECTED) ? m_client : -1; + } + if (socket == -1){ + return 0; + } + + const char* ptr = (const char*)data; + size_t sent = 0; + + while (bytes > 0){ + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + return sent; + } + } + + constexpr size_t BLOCK_SIZE = 1 << 30; + size_t current = std::min(BLOCK_SIZE, bytes); + ssize_t current_sent = ::send(socket, ptr, current, MSG_DONTWAIT | MSG_NOSIGNAL); + if (current_sent != -1){ + sent += (size_t)current_sent; + ptr += current_sent; + bytes -= (size_t)current_sent; + continue; + } + + int error = errno; + switch (error){ + case EAGAIN: +#if EAGAIN != EWOULDBLOCK + case EWOULDBLOCK: +#endif + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + continue; + default: + // EPIPE, ECONNRESET, or any other error: the connection is + // gone. + return sent; + } + } + + return sent; +} + +void ServerSocket_POSIX::accept_thread_loop(std::string address, uint16_t port){ + int listen_socket = ::socket(AF_INET, SOCK_STREAM, 0); + if (listen_socket == -1){ + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to create the socket: " + std::string(strerror(errno)) + ); + return; + } + + int yes = 1; + setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)); + + sockaddr_in server; + std::memset(&server, 0, sizeof(server)); + server.sin_family = AF_INET; + server.sin_port = htons(port); + server.sin_addr.s_addr = inet_addr(address.c_str()); + + if (bind(listen_socket, (struct sockaddr*)&server, sizeof(server)) == -1){ + int error = errno; + ::close(listen_socket); + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to bind the socket: " + std::string(strerror(error)) + ); + return; + } + + if (::listen(listen_socket, 16) == -1){ + int error = errno; + ::close(listen_socket); + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to listen on the socket: " + std::string(strerror(error)) + ); + return; + } + + std::string start_error; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) == State::DESTRUCTING){ + // close() won the race. + start_error = "The server was closed before it could start listening."; + }else{ + m_listen = listen_socket; + listen_socket = -1; + m_state.store(State::LISTENING, std::memory_order_release); + } + } + if (listen_socket != -1){ + ::close(listen_socket); + } + + m_listeners.run_method(&Listener::on_listen_finished, start_error); + if (!start_error.empty()){ + return; + } + + // Accept loop. Runs for the whole lifetime of the server: it keeps + // polling while a connection is active, and rejects new clients in + // the meantime. + while (true){ + int listen; + { + std::lock_guard lg(m_lock); + State state = m_state.load(std::memory_order_relaxed); + if (state == State::DESTRUCTING || state == State::NOT_RUNNING){ + break; + } + listen = m_listen; + } + + pollfd pf; + pf.fd = listen; + pf.events = POLLIN; + pf.revents = 0; + int ret = ::poll(&pf, 1, 10); + if (ret == -1){ + if (errno == EINTR){ + continue; + } + break; + } + if (ret == 0){ + continue; + } + + int client_socket = ::accept(listen, nullptr, nullptr); + if (client_socket == -1){ + continue; + } + + bool rejected = false; + bool accepted = false; + { + std::lock_guard lg(m_lock); + State state = m_state.load(std::memory_order_relaxed); + if (state == State::DESTRUCTING || state == State::NOT_RUNNING){ + ::close(client_socket); + break; + } + if (state == State::CONNECTED){ + // A client is already connected: reject this one. + ::close(client_socket); + rejected = true; + }else{ + fcntl(client_socket, F_SETFL, O_NONBLOCK); + m_client = client_socket; + m_close_requested = false; + m_state.store(State::CONNECTED, std::memory_order_release); + accepted = true; + } + } + + if (accepted){ + // Start the connection thread without holding m_lock. + // The assignment to m_client_thread destroys the previous + // connection task's AsyncTask, whose destructor waits for + // that task to finish - and that task may be blocked on + // m_lock. + try{ + AsyncTask task = m_thread_pool.dispatch([this]{ + connection_thread_loop(); + }); + std::lock_guard lg(m_task_lock); + m_client_thread = std::move(task); + }catch (...){ + // The connection thread could not be started. Drop + // this connection and keep listening. + std::lock_guard lg(m_lock); + ::close(client_socket); + m_client = -1; + m_state.store(State::LISTENING, std::memory_order_release); + } + } + + if (rejected){ + m_listeners.run_method(&Listener::on_connect_rejected); + } + } + + int listen; + { + std::lock_guard lg(m_lock); + listen = m_listen; + m_listen = -1; + } + if (listen != -1){ + ::close(listen); + } +} + +void ServerSocket_POSIX::connection_thread_loop(){ + m_listeners.run_method(&Listener::on_thread_start); + + int socket; + bool active; + { + std::lock_guard lg(m_lock); + socket = m_client; + active = (m_state.load(std::memory_order_relaxed) == State::CONNECTED); + if (!active && m_client == socket){ + m_client = -1; + } + } + if (!active){ + // The connection was closed before this thread started. + ::close(socket); + return; + } + + m_listeners.run_method(&Listener::on_connect); + + constexpr size_t BUFFER_SIZE = 4096; + char buffer[BUFFER_SIZE]; + std::string error; + + while (true){ + ssize_t bytes; + int error_no = 0; + bool no_data; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + break; + } + bytes = ::recv(socket, buffer, BUFFER_SIZE, 0); + no_data = (bytes == -1); + if (no_data){ + error_no = errno; + no_data = (error_no == EAGAIN); +#if EAGAIN != EWOULDBLOCK + no_data = no_data || (error_no == EWOULDBLOCK); +#endif + } + } + + if (bytes > 0){ + m_listeners.run_method(&Listener::on_receive_data, buffer, bytes); + continue; + } + + if (no_data){ + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + continue; + } + + if (bytes == 0){ + error = "Connection closed by client."; + }else{ + switch (error_no){ + case EINTR: + continue; + default: + error = "Receive error (errno " + std::to_string(error_no) + "): " + std::string(strerror(error_no)); + break; + } + } + break; + } + + bool report_disconnect; + { + std::lock_guard lg(m_lock); + report_disconnect = false; + if (m_client == socket){ + // Still the active connection (or it was closed before a + // new one took over): we own the socket and the state. + m_client = -1; + if (m_state.load(std::memory_order_relaxed) == State::CONNECTED){ + m_state.store(State::LISTENING, std::memory_order_release); + report_disconnect = !m_close_requested; + } + } + } + ::close(socket); + if (report_disconnect){ + m_listeners.run_method(&Listener::on_disconnect, error); + } +} + +} + +#endif diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.h b/Common/Cpp/Sockets/ServerSocket_POSIX.h new file mode 100644 index 0000000000..7626cb64de --- /dev/null +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.h @@ -0,0 +1,80 @@ +/* Server Socket (POSIX) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#ifndef PokemonAutomation_ServerSocket_POSIX_H +#define PokemonAutomation_ServerSocket_POSIX_H + +#include +#include +#include "Common/Cpp/Concurrency/AsyncTask.h" +#include "Common/Cpp/Concurrency/ConditionVariable.h" +#include "Common/Cpp/Concurrency/Mutex.h" +#include "AbstractServerSocket.h" + + +namespace PokemonAutomation{ + +class ThreadPool; + + +class ServerSocket_POSIX final : public AbstractServerSocket{ +public: + + ServerSocket_POSIX(ThreadPool& thread_pool); + + + virtual ~ServerSocket_POSIX(){ + close(); + } + + + + virtual void start(const std::string& address, uint16_t port) override; + + + virtual void close_connection() override; + + + + virtual void close() noexcept override; + + + virtual size_t send(const void* data, size_t bytes) override; + + +private: + + + void accept_thread_loop(std::string address, uint16_t port); + + + void connection_thread_loop(); + +private: + ThreadPool& m_thread_pool; + Mutex m_lock; + ConditionVariable m_cv; + bool m_starting = false; + int m_closing = 0; + + // Protects m_accept_thread and m_client_thread. Kept separate from + // m_lock: assigning to a task slot destroys the previous task's + // AsyncTask, and its destructor waits for that task to finish - which + // requires m_lock, so the two locks must never be held at once. + Mutex m_task_lock; + + int m_listen = -1; + int m_client = -1; + bool m_close_requested = false; + + AsyncTask m_accept_thread; + AsyncTask m_client_thread; +}; + + + +} +#endif diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.cpp b/Common/Cpp/Sockets/ServerSocket_Qt.cpp new file mode 100644 index 0000000000..9bae4f48da --- /dev/null +++ b/Common/Cpp/Sockets/ServerSocket_Qt.cpp @@ -0,0 +1,388 @@ +/* Server Socket (Qt) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#include +#include +#include +#include +#include "Common/Qt/QtThreadPool.h" +#include "Common/Qt/GlobalThreadPoolsQt.h" +#include "ServerSocket_Qt.h" + +namespace PokemonAutomation{ + +void ServerSocket_Qt::start(const std::string& address, uint16_t port){ + bool starting; + { + std::lock_guard lg(m_lock); + starting = + (m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING) + && m_server == nullptr + && !m_starting; + m_starting = true; + } + if (!starting){ + return; + } + + m_listen_error = ""; + QTcpServer* server = nullptr; + try{ + server = static_cast( + GlobalThreadPools::qt_event_threadpool().add_object( + [this, address, port]{ + return make_server(address, port); + } + ) + ); + } + catch (...){ + { + std::lock_guard lg(m_lock); + m_starting = false; + m_cv.notify_all(); + } + m_listeners.run_method(&Listener::on_listen_finished, "Failed to start the listener."); + throw; + } + bool listening; + { + std::lock_guard lg(m_lock); + m_starting = false; + m_server = server; + if (m_state.load(std::memory_order_relaxed) == State::DESTRUCTING){ + // "close()" is already in progress: it will destroy the + // server we just created. Leave the state for it. + }else{ + listening = m_listen_error.empty(); + m_state.store( + listening ? State::LISTENING : State::NOT_RUNNING, + std::memory_order_release + ); + } + m_cv.notify_all(); + } + + m_listeners.run_method(&Listener::on_listen_finished, m_listen_error); +} + +void ServerSocket_Qt::close_connection(){ + QThread* event_thread; + { + std::lock_guard lg(m_lock); + event_thread = m_event_thread; + if (m_server == nullptr){ + // Not running. + return; + } + } + + QSharedPointer completion = QSharedPointer::create(); + emit internal_close_connection(completion); + + // If called from a thread other than the server's event thread, the + // call is queued: wait until it has run, so the state is up to date + // when this returns (same as the POSIX backend). If called from the + // event thread (e.g. a listener callback), the call ran directly and + // this does not block. + // + // The shared ownership makes the timeout safe: if it expires, the + // queued call's own copy keeps the completion alive until it runs. + if (QThread::currentThread() != event_thread){ + std::unique_lock lg(completion->lock); + completion->cv.wait_for( + lg, + std::chrono::seconds(5), + [&completion]{ return completion->done; } + ); + } +} +void ServerSocket_Qt::close() noexcept{ + QTcpServer* server; + { + std::lock_guard lg(m_lock); + if (m_server == nullptr && !m_starting){ + // Never started, or already closed. + return; + } + m_close_requested = true; + m_state.store(State::DESTRUCTING, std::memory_order_release); + } + + // Wait for an in-flight "start()" to finish so we destroy the + // server it creates, and for in-flight "send()" calls to finish so + // we don't destroy the server out from under them. + { + std::unique_lock lg(m_lock); + m_cv.wait(lg, [this]{ + return !m_starting + && m_sends_in_flight.load(std::memory_order_acquire) == 0; + }); + server = m_server; + m_server = nullptr; + m_socket = nullptr; + } + if (server == nullptr){ + // "start()" failed or was never reached: nothing to destroy. + std::lock_guard lg(m_lock); + m_state.store(State::NOT_RUNNING, std::memory_order_release); + m_close_requested = false; + return; + } + + // Destroy the server on its event thread. This blocks until the + // destruction has happened. + GlobalThreadPools::qt_event_threadpool().remove_object(server); + + std::lock_guard lg(m_lock); + m_state.store(State::NOT_RUNNING, std::memory_order_release); + m_close_requested = false; +} + +size_t ServerSocket_Qt::send(const void* data, size_t bytes){ + QTcpSocket* socket; + { + std::lock_guard lg(m_lock); + socket = + (m_state.load(std::memory_order_relaxed) == State::CONNECTED) ? m_socket : nullptr; + } + if (socket == nullptr){ + return 0; + } + + QSharedPointer send_data = QSharedPointer::create(); + send_data->packet.assign((const char*)data, bytes); + + emit internal_send(send_data); + + // Wait for the server's event thread to finish writing the packet. + // The timeout covers the case where "close()" destroys the server + // before the queued call ever runs, in which case it will never + // complete on its own; the shared ownership keeps the packet alive + // until the queued call runs (or is discarded with the server). + { + std::unique_lock lg(send_data->lock); + send_data->cv.wait_for( + lg, + std::chrono::seconds(5), + [&send_data]{ return send_data->done; } + ); + } + + return send_data->bytes_sent; +} + +void ServerSocket_Qt::on_new_connection(){ + QTcpServer* server; + { + std::lock_guard lg(m_lock); + server = m_server; + } + if (server == nullptr){ + return; + } + + while (true){ + QTcpSocket* sock = server->nextPendingConnection(); + if (sock == nullptr){ + break; + } + // The connection socket lives with the server, so "close()" + // destroys it on the event thread. + sock->setParent(server); + + bool accepted; + bool rejected; + { + std::lock_guard lg(m_lock); + State state = m_state.load(std::memory_order_relaxed); + if (state == State::CONNECTED){ + // A client is already connected: reject this one. + accepted = false; + rejected = true; + }else if (state != State::LISTENING){ + // The server is shutting down. + accepted = false; + rejected = false; + }else{ + m_socket = sock; + m_close_requested = false; + m_state.store(State::CONNECTED, std::memory_order_release); + accepted = true; + rejected = false; + } + } + + if (!accepted){ + if (rejected){ + m_listeners.run_method(&Listener::on_connect_rejected); + } + sock->abort(); + continue; + } + + QObject::connect( + sock, &QAbstractSocket::readyRead, + sock, [this, sock]{ + // readyRead only fires once per delivery, and more than + // one buffer's worth of data may be waiting, so read + // until the socket has no more data available. + while (true){ + qint64 available = sock->bytesAvailable(); + if (available <= 0){ + break; + } + + constexpr size_t BUFFER_SIZE = 4096; + char buffer[BUFFER_SIZE]; + qint64 bytes = sock->read(buffer, std::min(BUFFER_SIZE, available)); + if (bytes <= 0){ + break; + } + + m_listeners.run_method(&Listener::on_receive_data, buffer, bytes); + } + } + ); + QObject::connect( + sock, &QAbstractSocket::disconnected, + sock, [this, sock]{ + connection_lost(sock, "Connection closed by client."); + } + ); + QObject::connect( + sock, &QAbstractSocket::errorOccurred, + sock, [this, sock](QAbstractSocket::SocketError){ + connection_lost(sock, sock->errorString().toStdString()); + } + ); + + m_listeners.run_method(&Listener::on_connect); + } +} + +void ServerSocket_Qt::connection_lost(QTcpSocket* sock, const std::string& error_message){ + bool report; + { + std::lock_guard lg(m_lock); + if (m_socket != sock){ + // Already replaced or closed; the first callback wins. + return; + } + report = + !m_close_requested + && m_state.load(std::memory_order_relaxed) == State::CONNECTED; + m_socket = nullptr; + if (m_state.load(std::memory_order_relaxed) == State::CONNECTED){ + m_state.store(State::LISTENING, std::memory_order_release); + } + } + + if (report){ + m_listeners.run_method(&Listener::on_disconnect, error_message); + } +} + +void ServerSocket_Qt::on_close_connection(QSharedPointer completion){ + QTcpSocket* sock; + { + std::lock_guard lg(m_lock); + if (m_socket != nullptr){ + m_close_requested = true; + m_state.store(State::LISTENING, std::memory_order_release); + sock = m_socket; + m_socket = nullptr; + }else{ + sock = nullptr; + } + } + if (sock != nullptr){ + sock->abort(); + } + + { + std::lock_guard lg(completion->lock); + completion->done = true; + } + completion->cv.notify_all(); +} + +void ServerSocket_Qt::on_send(QSharedPointer send_data){ + QTcpSocket* sock; + { + std::lock_guard lg(m_lock); + m_sends_in_flight.fetch_add(1, std::memory_order_acq_rel); + m_cv.notify_all(); + sock = m_socket; + } + + size_t bytes_sent = 0; + if (sock != nullptr){ + const char* ptr = send_data->packet.data(); + size_t remaining = send_data->packet.size(); + while (remaining > 0 && sock->state() == QAbstractSocket::ConnectedState){ + qint64 current_sent = sock->write(ptr, remaining); + if (current_sent <= 0){ + // The kernel's send buffer is full. Wait for some of it + // to drain, then try again. + sock->waitForBytesWritten(10); + continue; + } + ptr += current_sent; + remaining -= (size_t)current_sent; + } + bytes_sent = send_data->packet.size() - remaining; + } + + { + std::lock_guard lg(m_lock); + m_sends_in_flight.fetch_sub(1, std::memory_order_acq_rel); + m_cv.notify_all(); + } + + { + std::lock_guard lg(send_data->lock); + send_data->bytes_sent = bytes_sent; + send_data->done = true; + } + send_data->cv.notify_all(); +} + +std::unique_ptr ServerSocket_Qt::make_server(const std::string& address, uint16_t port){ + auto server = std::make_unique(); + if (!server->listen(QHostAddress(QString::fromStdString(address)), port)){ + std::lock_guard lg(m_lock); + m_listen_error = server->errorString().toStdString(); + } + + { + std::lock_guard lg(m_lock); + m_event_thread = QThread::currentThread(); + } + + QObject::connect( + this, &ServerSocket_Qt::internal_send, + server.get(), [this](QSharedPointer send_data){ + on_send(send_data); + } + ); + QObject::connect( + this, &ServerSocket_Qt::internal_close_connection, + server.get(), [this](QSharedPointer completion){ + on_close_connection(completion); + } + ); + QObject::connect( + server.get(), &QTcpServer::newConnection, + server.get(), [this]{ + on_new_connection(); + } + ); + + return server; +} + +} diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.h b/Common/Cpp/Sockets/ServerSocket_Qt.h new file mode 100644 index 0000000000..34fb1660dc --- /dev/null +++ b/Common/Cpp/Sockets/ServerSocket_Qt.h @@ -0,0 +1,129 @@ +/* Server Socket (Qt) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#ifndef PokemonAutomation_ServerSocket_Qt_H +#define PokemonAutomation_ServerSocket_Qt_H + +#include +#include +#include +#include +#include "Common/Cpp/Concurrency/Mutex.h" +#include "Common/Cpp/Concurrency/ConditionVariable.h" +#include "AbstractServerSocket.h" + +class QThread; +class QTcpServer; +class QTcpSocket; + +namespace PokemonAutomation{ + + +class ThreadPool; + + +class ServerSocket_Qt final : public QObject, public AbstractServerSocket{ + Q_OBJECT + + // Carries shared ownership (the signals pass it by value) so that a call still + // queued when the caller times out, or when "close()" destroys the server and + // Qt discards the pending call, keeps the payload alive exactly as long as the + // event itself: the handler's copy outlives a timed-out caller, and a + // discarded event leaves the caller's copy as the last reference. + struct SendData{ + std::string packet; + size_t bytes_sent = 0; + bool done = false; + + Mutex lock; + ConditionVariable cv; + }; + + // Synchronizes "close_connection()" with the QTcpServer's event thread when + // called from another thread. Lifetime as SendData. + struct CloseCompletion{ + Mutex lock; + ConditionVariable cv; + bool done = false; + }; + + +public: + ServerSocket_Qt(ThreadPool&){} + + virtual ~ServerSocket_Qt(){ + close(); + } + + // Binds to "address" and listens on "port" on the Qt event thread. This + // blocks until the QTcpServer has been created on that thread. + virtual void start(const std::string& address, uint16_t port) override; + + + virtual void close_connection() override; + + + // Stops the server, destroys the QTcpServer (and its connection + // socket), and returns the socket to State::NOT_RUNNING so "start()" + // can be called again. This blocks until the QTcpServer has been + // destroyed, so any in-flight listener callback is finished by the + // time it returns. It is safe to call multiple times. + // + // This must not be called from within a listener callback, since the + // callbacks run on the QTcpServer's event thread and this destroys the + // server that is running them. + virtual void close() noexcept override; + + + virtual size_t send(const void* data, size_t bytes) override; + + +signals: + void internal_send(QSharedPointer send_data); + void internal_close_connection(QSharedPointer completion); + + +private: + // Runs on the QTcpServer's event thread. + void on_new_connection(); + + // The active connection ended, or was aborted by "close_connection()". + // Only the first of the (errorOccurred, disconnected) signals for a + // given connection fires a callback, and "on_disconnect" is not fired + // for a connection closed by "close_connection()" or "close()". + // Runs on the QTcpServer's event thread. + void connection_lost(QTcpSocket* sock, const std::string& error_message); + + // Runs on the QTcpServer's event thread. + void on_close_connection(QSharedPointer completion); + + // Runs on the QTcpServer's event thread. The context object is the + // server, so the call still runs if the connection socket has already + // been aborted, and it just reports 0 bytes in that case. + void on_send(QSharedPointer send_data); + + std::unique_ptr make_server(const std::string& address, uint16_t port); + +private: + Mutex m_lock; + ConditionVariable m_cv; + + QTcpServer* m_server = nullptr; + QTcpSocket* m_socket = nullptr; + QThread* m_event_thread = nullptr; + bool m_close_requested = false; + bool m_starting = false; + std::atomic m_sends_in_flight{0}; + + // Written on the event thread by "make_server()", read after + // "add_object()" has returned. + std::string m_listen_error; +}; + + + +} +#endif diff --git a/Common/Cpp/Sockets/Sockets_Tests.cpp b/Common/Cpp/Sockets/Sockets_Tests.cpp new file mode 100644 index 0000000000..57a8398628 --- /dev/null +++ b/Common/Cpp/Sockets/Sockets_Tests.cpp @@ -0,0 +1,891 @@ +/* Sockets Tests + * + * From: https://github.com/PokemonAutomation/ + * + * Loopback tests that combine a ClientSocket and a ServerSocket against + * each other on 127.0.0.1. The test logic is written against the abstract + * classes, and is instantiated for every backend that compiles on the + * current platform (Qt: always, POSIX: on non-Windows, WinSocket: on + * Windows; each backend lands in its own PR), so that every platform + * runs 2 of the 3 backends and CI as a whole covers all 3. + * + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "Common/Cpp/Concurrency/AsyncTask.h" +#include "Common/Cpp/Concurrency/ConditionVariable.h" +#include "Common/Cpp/Concurrency/Mutex.h" +#include "Common/Cpp/Concurrency/ThreadPool.h" +#include "Common/Cpp/Sockets/ClientSocket_Qt.h" +#include "Common/Cpp/Sockets/ServerSocket_Qt.h" +#ifndef _WIN32 +#include "Common/Cpp/Sockets/ClientSocket_POSIX.h" +#include "Common/Cpp/Sockets/ServerSocket_POSIX.h" +#endif +#include "Common/Cpp/TestRunners/UnitTestDatabase.h" +#include "Sockets_Tests.h" + +namespace PokemonAutomation{ + +namespace{ + + +struct Waiter{ + Mutex lock; + ConditionVariable cv; + + bool wait_for(std::chrono::milliseconds timeout, const std::function& pred){ + std::unique_lock lg(lock); + return cv.wait_for(lg, timeout, pred); + } + + void notify(){ + std::lock_guard lg(lock); + cv.notify_all(); + } +}; + + +struct ClientListener : AbstractClientSocket::Listener{ + Waiter waiter; + std::atomic connect_finished{0}; + std::string last_connect_error; + std::atomic disconnected{0}; + std::string last_disconnect_error; + std::vector received; + + virtual void on_connect_finished(const std::string& error_message) override{ + connect_finished++; + last_connect_error = error_message; + waiter.notify(); + } + + virtual void on_receive_data(const void* data, size_t bytes) override{ + const uint8_t* ptr = (const uint8_t*)data; + received.insert(received.end(), ptr, ptr + bytes); + waiter.notify(); + } + + virtual void on_disconnect(const std::string& error_message) override{ + disconnected++; + last_disconnect_error = error_message; + waiter.notify(); + } +}; + + +struct ServerListener : AbstractServerSocket::Listener{ + Waiter waiter; + std::atomic listen_finished{0}; + std::string last_listen_error; + std::atomic connected{0}; + std::atomic rejected{0}; + std::atomic disconnected{0}; + std::string last_disconnect_error; + std::atomic bytes_received{0}; + + // When set, received data is echoed back to the client. + AbstractServerSocket* server = nullptr; + + virtual void on_listen_finished(const std::string& error_message) override{ + listen_finished++; + last_listen_error = error_message; + waiter.notify(); + } + + virtual void on_connect() override{ + connected++; + waiter.notify(); + } + + virtual void on_connect_rejected() override{ + rejected++; + waiter.notify(); + } + + virtual void on_receive_data(const void* data, size_t bytes) override{ + bytes_received += bytes; + if (server != nullptr){ + server->send(data, bytes); + } + waiter.notify(); + } + + virtual void on_disconnect(const std::string& error_message) override{ + disconnected++; + last_disconnect_error = error_message; + waiter.notify(); + } +}; + + +template +class TestSockets : public UnitTest{ +public: + TestSockets(ThreadPool& thread_pool, std::string name) + : UnitTest(std::move(name)) + , m_thread_pool(thread_pool) + { + m_threads = 2; + } + + +protected: + bool start_server(ServerSocketType& server, ServerListener& listener, uint16_t& port) const{ + for (int attempt = 0; attempt < 8; attempt++){ + uint16_t candidate; + { + std::lock_guard lg(m_rng_lock); + candidate = (uint16_t)(40000 + m_rng() % 20000); + } + server.start("127.0.0.1", candidate); + if (!listener.waiter.wait_for( + std::chrono::seconds(5), + [&listener]{ return listener.listen_finished.load() > 0; } + )){ + continue; + } + if (listener.last_listen_error.empty()){ + port = candidate; + return true; + } + } + return false; + } + + bool connect_client( + ClientSocketType& client, + ClientListener& listener, + uint16_t port, + int expected_connect + ) const{ + client.connect("127.0.0.1", port); + return listener.waiter.wait_for( + std::chrono::seconds(5), + [&listener, expected_connect]{ + return listener.connect_finished.load() >= expected_connect; + } + ); + } + +protected: + ThreadPool& m_thread_pool; + +private: + mutable Mutex m_rng_lock; + mutable std::mt19937 m_rng{std::random_device{}()}; +}; + + +// Connect, send a packet, verify the echo, disconnect, and check that the +// server observed both the connect and the disconnect. +template +class Test_Sockets_Echo : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + error = "The client did not finish connecting"; + }else if (!client_listener.last_connect_error.empty()){ + error = "The client failed to connect: " + client_listener.last_connect_error; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else if (client.send("hello world", 11) != 11){ + error = "The client did not send all 11 bytes"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.bytes_received.load() >= 11; } + )){ + error = "The server did not receive the data"; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.received.size() >= 11; } + )){ + error = "The client did not receive the echo"; + }else if (std::memcmp(client_listener.received.data(), "hello world", 11) != 0){ + error = "The echo data was corrupted"; + }else{ + ok = true; + } + + client.close(); + } + + if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.disconnected.load() >= 1; } + )){ + error = "The server did not call on_disconnect"; + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Zero-length sends must be a valid no-op that returns 0 without +// touching the connection. +template +class Test_Sockets_ZeroLengthSend : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + error = "The client did not finish connecting"; + }else if (client.send("", 0) != 0){ + error = "A zero-length client send did not return 0"; + }else if (server.send("", 0) != 0){ + error = "A zero-length server send did not return 0"; + }else{ + // The connection must still be usable afterwards. + if (client.send("ping", 4) != 4){ + error = "The client did not send all 4 bytes after a zero-length send"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.bytes_received.load() >= 4; } + )){ + error = "The server did not receive the data"; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.received.size() >= 4; } + )){ + error = "The client did not receive the echo"; + }else if (std::memcmp(client_listener.received.data(), "ping", 4) != 0){ + error = "The echo data was corrupted"; + }else{ + ok = true; + } + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Transfer a megabyte each way (the server echoes it back) and verify the +// data integrity. +template +class Test_Sockets_LargeTransfer : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + constexpr size_t SIZE = 1 << 20; + std::vector payload(SIZE); + for (size_t i = 0; i < SIZE; i++){ + payload[i] = (uint8_t)(i % 251); + } + + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + error = "The client did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else if (client.send(payload.data(), payload.size()) != payload.size()){ + error = "The client did not send all " + std::to_string(payload.size()) + " bytes"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(30), + [&server_listener]{ return server_listener.bytes_received.load() >= SIZE; } + )){ + error = "The server did not receive all the data"; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(30), + [&client_listener]{ return client_listener.received.size() >= SIZE; } + )){ + error = "The client did not receive all the echo data"; + }else if (std::memcmp(client_listener.received.data(), payload.data(), SIZE) != 0){ + error = "The echo data was corrupted"; + }else{ + ok = true; + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Rapidly connect and disconnect many times. Each cycle transfers a +// packet to make sure the new connection actually works. +template +class Test_Sockets_RapidReconnect : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + constexpr int CYCLES = 50; + + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + size_t expected_echo = 0; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + for (int cycle = 0; cycle < CYCLES; cycle++){ + int cycle_no = cycle + 1; + + if (!this->connect_client(client, client_listener, port, cycle_no)){ + error = "The client did not finish connecting (cycle " + std::to_string(cycle_no) + ")"; + break; + } + if (!client_listener.last_connect_error.empty()){ + error = "The client failed to connect (cycle " + std::to_string(cycle_no) + "): " + client_listener.last_connect_error; + break; + } + if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener, cycle_no]{ return server_listener.connected.load() >= cycle_no; } + )){ + error = "The server did not call on_connect (cycle " + std::to_string(cycle_no) + ")"; + break; + } + + std::string packet = "cycle " + std::to_string(cycle_no); + expected_echo += packet.size(); + if (client.send(packet.data(), packet.size()) != packet.size()){ + error = "The client did not send the packet (cycle " + std::to_string(cycle_no) + ")"; + break; + } + if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener, expected_echo]{ return client_listener.received.size() >= expected_echo; } + )){ + error = "The client did not receive the echo (cycle " + std::to_string(cycle_no) + ")"; + break; + } + + client.close(); + if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener, cycle_no]{ return server_listener.disconnected.load() >= cycle_no; } + )){ + error = "The server did not call on_disconnect (cycle " + std::to_string(cycle_no) + ")"; + break; + } + } + + if (error.empty()){ + ok = true; + } + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Send a large packet while the server closes the connection underneath +// it: the send must finish (returning however much was written), the +// server must go back to listening, and the client must observe the +// disconnect. Then the server must accept a new connection. +template +class Test_Sockets_SendCloseRace : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + constexpr size_t SIZE = 1 << 20; + std::vector payload(SIZE, 0xAB); + + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + error = "The client did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else{ + std::atomic bytes_sent{0}; + AsyncTask send_task = this->m_thread_pool.dispatch([&client, &payload, &bytes_sent]{ + bytes_sent = client.send(payload.data(), payload.size()); + }); + + // Give the send a head start, then cut the connection from + // the server side. + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + server.close_connection(); + + send_task.wait_and_ignore_exceptions(); + + if (server.state() != AbstractServerSocket::State::LISTENING){ + error = "The server is not listening after close_connection()"; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.disconnected.load() >= 1; } + )){ + error = "The client did not observe the disconnect"; + }else if (bytes_sent.load() > SIZE){ + error = "The client reported sending more bytes than the packet size"; + }else{ + // The server must accept a new connection. + if (!this->connect_client(client, client_listener, port, 2)){ + error = "The client did not reconnect"; + }else if (!client_listener.last_connect_error.empty()){ + error = "The client failed to reconnect: " + client_listener.last_connect_error; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 2; } + )){ + error = "The server did not accept the new connection"; + }else if (client.send("x", 1) != 1){ + error = "The new connection does not work"; + }else{ + ok = true; + } + } + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// While a client is connected, a second client must be rejected, and the +// first client must keep working. +template +class Test_Sockets_RejectSecondClient : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client_a(this->m_thread_pool); + ClientListener client_a_listener; + client_a.add_listener(client_a_listener); + + ClientSocketType client_b(this->m_thread_pool); + ClientListener client_b_listener; + client_b.add_listener(client_b_listener); + + if (!this->connect_client(client_a, client_a_listener, port, 1)){ + error = "Client A did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else{ + client_b.connect("127.0.0.1", port); + if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.rejected.load() >= 1; } + )){ + error = "The server did not reject the second client"; + }else{ + // The first client must keep working. + if (client_a.send("still here?", 11) != 11){ + error = "Client A did not send all 11 bytes"; + }else if (!client_a_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_a_listener]{ return client_a_listener.received.size() >= 11; } + )){ + error = "Client A did not receive the echo"; + }else if (std::memcmp(client_a_listener.received.data(), "still here?", 11) != 0){ + error = "The echo data was corrupted"; + }else{ + // Client B must have got a final answer to its + // connect attempt, whatever it was. + if (!client_b_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_b_listener]{ return client_b_listener.connect_finished.load() >= 1; } + )){ + error = "Client B did not finish its connect attempt"; + }else{ + ok = true; + } + } + } + } + + client_a.close(); + client_b.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Starting a second server on an already-bound port must fail through +// on_listen_finished and leave the server not running. +template +class Test_Sockets_PortInUse : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + ServerSocketType second_server(this->m_thread_pool); + ServerListener second_listener; + second_server.add_listener(second_listener); + second_server.start("127.0.0.1", port); + if (!second_listener.waiter.wait_for( + std::chrono::seconds(5), + [&second_listener]{ return second_listener.listen_finished.load() > 0; } + )){ + server.close(); + return "The second server did not report its start"; + } + if (second_listener.last_listen_error.empty()){ + server.close(); + second_server.close(); + return "The second server succeeded on a port that is already in use"; + } + if (second_server.state() != AbstractServerSocket::State::NOT_RUNNING){ + server.close(); + second_server.close(); + return "The second server is not NOT_RUNNING after a failed start"; + } + + // The original server must still work. + bool ok = false; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + ok = false; + }else if (client.send("ok", 2) != 2){ + ok = false; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.received.size() >= 2; } + )){ + ok = false; + }else{ + ok = true; + } + + client.close(); + } + + server.close(); + second_server.close(); + + if (!ok){ + return "The original server stopped working"; + } + return true; + } +}; + + +// Qt backend only. Stall the server's event thread for longer than the 5 s +// completion wait (by blocking inside an on_receive_data callback, which +// runs on that thread). close_connection() must then time out and return +// cleanly, and the queued handler, which runs after the caller has gone +// back, must be able to complete: the completion's shared ownership keeps +// it alive until then. Previously this was a stack-use-after-return: the +// handler wrote to the dead stack-local CloseCompletion. +class Test_Sockets_Qt_CompletionLifetime : public UnitTest{ +public: + Test_Sockets_Qt_CompletionLifetime(ThreadPool& thread_pool, std::string name) + : UnitTest(std::move(name)) + , m_thread_pool(thread_pool) + { + m_threads = 2; + } + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + // Blocks the event thread while it receives data. + struct StallListener : AbstractServerSocket::Listener{ + virtual void on_receive_data(const void*, size_t) override{ + std::this_thread::sleep_for(std::chrono::seconds(6)); + } + }; + + ServerSocket_Qt server(m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + StallListener stall_listener; + server.add_listener(stall_listener); + + uint16_t port = 0; + { + std::mt19937 rng{std::random_device{}()}; + for (int attempt = 0; attempt < 8; attempt++){ + uint16_t candidate = (uint16_t)(40000 + rng() % 20000); + server.start("127.0.0.1", candidate); + if (server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.listen_finished.load() > 0; } + ) && server_listener.last_listen_error.empty()){ + port = candidate; + break; + } + } + } + if (port == 0){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocket_Qt client(m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + client.connect("127.0.0.1", port); + if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.connect_finished.load() >= 1; } + )){ + error = "The client did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else{ + // One byte in: the event thread blocks in on_receive_data for 6 s. + char byte = 'x'; + if (client.send(&byte, 1) != 1){ + error = "The client did not send the byte"; + }else{ + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + + auto start = std::chrono::steady_clock::now(); + server.close_connection(); + auto waited = std::chrono::steady_clock::now() - start; + if (waited < std::chrono::seconds(4) || waited >= std::chrono::seconds(8)){ + error = "close_connection() did not time out as expected (" + + std::to_string( + (long long)std::chrono::duration_cast(waited).count() + ) + " ms)"; + }else{ + // Wait for the stall to finish so the late handler actually + // runs, then check that it restored the state. + std::this_thread::sleep_for(std::chrono::seconds(6)); + if (server.state() != AbstractServerSocket::State::LISTENING){ + error = "The late handler did not restore LISTENING"; + }else{ + ok = true; + } + } + } + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } + +private: + ThreadPool& m_thread_pool; +}; + + +template +void add_socket_tests(UnitTestDatabase& database, ThreadPool& thread_pool, const char* backend_name){ + const std::string prefix = "Sockets (" + std::string(backend_name) + ") - "; + + database.add>( + thread_pool, prefix + "Client/server echo round trip" + ); + database.add>( + thread_pool, prefix + "Zero-length send" + ); + database.add>( + thread_pool, prefix + "Large transfer" + ); + database.add>( + thread_pool, prefix + "Rapid reconnect" + ); + database.add>( + thread_pool, prefix + "Send/close race" + ); + database.add>( + thread_pool, prefix + "Reject a second client" + ); + database.add>( + thread_pool, prefix + "Port in use" + ); +} + + + +} // anonymous namespace + + +void add_tests_Sockets(UnitTestDatabase& database, ThreadPool& thread_pool){ + add_socket_tests(database, thread_pool, "Qt"); + + // Qt-specific: the completion-lifetime guarantee is only at risk in the + // Qt backend (a queued call can outlive its caller there). + database.add( + thread_pool, + "Sockets (Qt) - Completion lifetime under event-thread stall" + ); + +#ifndef _WIN32 + add_socket_tests(database, thread_pool, "POSIX"); +#endif +} + + + +} diff --git a/Common/Cpp/Sockets/Sockets_Tests.h b/Common/Cpp/Sockets/Sockets_Tests.h new file mode 100644 index 0000000000..0b651959a4 --- /dev/null +++ b/Common/Cpp/Sockets/Sockets_Tests.h @@ -0,0 +1,28 @@ +/* Sockets Tests + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#ifndef PokemonAutomation_Sockets_Tests_H +#define PokemonAutomation_Sockets_Tests_H + +namespace PokemonAutomation{ + +class ThreadPool; +class UnitTestDatabase; + + +// Loopback tests that combine a ClientSocket and a ServerSocket against +// each other on 127.0.0.1: echo round trip, zero-length sends, a large +// transfer, rapid reconnects, a send/close race, connection rejection, +// and a port-in-use start failure. Each scenario runs for every backend +// that compiles on the current platform (Qt: always, POSIX: non-Windows, +// WinSocket: Windows). `thread_pool` runs the sockets and the test +// workers; it must be a long-lived pool (e.g. +// `GlobalThreadPools::unlimited_normal()`). +void add_tests_Sockets(UnitTestDatabase& database, ThreadPool& thread_pool); + + +} +#endif diff --git a/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp b/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp index 1c19573466..419f0f36a3 100644 --- a/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp +++ b/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp @@ -7,6 +7,7 @@ #include "Common/Cpp/ListenerSet.h" #include "Common/Cpp/ScopeExit.h" #include "Common/Cpp/PrettyPrint.h" +#include "Common/Cpp/Sockets/Sockets_Tests.h" #include "Common/Cpp/TestRunners/TestImageDistortionManager.h" #include "Common/Cpp/TestRunners/UnitTestDatabase.h" #include "Common/Qt/QtThreadPool_Tests.h" @@ -49,6 +50,7 @@ UnitTestDatabase make_UNIT_TESTS_ALL(){ UnitTestDatabase ret; add_tests_QtEventThreadPool(ret, GlobalThreadPools::computation_normal()); + add_tests_Sockets(ret, GlobalThreadPools::unlimited_normal()); add_tests_BlackBorderDetector(ret); add_tests_ListenerSet(ret); add_tests_AudioInfo(ret); diff --git a/SerialPrograms/Source/ML/Inference/ML_PaddleOCRPipeline.cpp b/SerialPrograms/Source/ML/Inference/ML_PaddleOCRPipeline.cpp index ab9ab35fb7..066d79e626 100644 --- a/SerialPrograms/Source/ML/Inference/ML_PaddleOCRPipeline.cpp +++ b/SerialPrograms/Source/ML/Inference/ML_PaddleOCRPipeline.cpp @@ -87,6 +87,7 @@ PaddleOCRPipeline::PaddleOCRPipeline( , m_det_path(std::move(det_path)) { load_dictionary(Filesystem::Path(dict_path)); + std::cerr << "[ORT-DIAG] PaddleOCRPipeline ctor finished: " << rec_path << std::endl; } PaddleOCRDetector& PaddleOCRPipeline::detector(){ @@ -381,6 +382,10 @@ std::string PaddleOCRPipeline::recognize_line(const cv::Mat& cropped_image){ const char* input_names[] = {m_input_name.c_str()}; const char* output_names[] = {m_output_name.c_str()}; + static int diag_run_count = 0; + if (++diag_run_count <= 20){ + std::cerr << "[ORT-DIAG] recognize_line: about to call session->Run() (call " << diag_run_count << ")" << std::endl; + } try{ if (debugging){ m_logger.log("[OCR-DEBUG] Calling m_rec_session.Run() now..."); diff --git a/SerialPrograms/Source/ML/Models/ML_ONNXRuntimeHelpers.cpp b/SerialPrograms/Source/ML/Models/ML_ONNXRuntimeHelpers.cpp index d0dac9d2a8..620173c414 100644 --- a/SerialPrograms/Source/ML/Models/ML_ONNXRuntimeHelpers.cpp +++ b/SerialPrograms/Source/ML/Models/ML_ONNXRuntimeHelpers.cpp @@ -167,9 +167,11 @@ Ort::Session create_session( const std::string& model_cache_path, bool try_gpu ){ + std::cerr << "[ORT-DIAG] create_session(" << model_path << ", try_gpu=" << (try_gpu ? 1 : 0) << ")" << std::endl; bool write_flag_file = true; std::string file_hash; std::tie(write_flag_file, file_hash) = clean_up_old_model_cache(model_cache_path, model_path); + std::cerr << "[ORT-DIAG] clean_up_old_model_cache done, write_flag_file=" << write_flag_file << std::endl; auto onnx_path = str_to_onnx_str(model_path); @@ -205,7 +207,9 @@ Ort::Session create_session( Ort::SessionOptions cpu_options = create_session_options(model_cache_path, false); cpu_options.DisablePerSessionThreads(); + std::cerr << "[ORT-DIAG] calling Ort::Session constructor (CPU, per-session threads disabled)..." << std::endl; Ort::Session session{global_ort_env(), onnx_path.c_str(), cpu_options}; + std::cerr << "[ORT-DIAG] Ort::Session constructor returned: " << model_path << std::endl; logger.log("Ort::Session created"); return session; } diff --git a/SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp b/SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp index c69ace4a39..e16195442e 100644 --- a/SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp +++ b/SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp @@ -4,6 +4,8 @@ * */ +#include +#include #include "Common/Cpp/Exceptions.h" #include "Common/Cpp/PrettyPrint.h" #include "CommonFramework/Logging/Logger.h" @@ -19,11 +21,15 @@ Ort::ThreadingOptions make_ort_threading_options(){ { int threads = (int)PerformanceOptions::instance().ONNX_OPTIONS.MAX_INTRA_OP_THREADS; global_logger_tagged().log("Setting ONNX Intra-Op Threads: " + tostr_u_commas(threads)); + std::cerr << "[ORT-DIAG] ONNX Runtime " << Ort::GetVersionString() + << ", hardware_concurrency() = " << std::thread::hardware_concurrency() + << ", SetGlobalIntraOpNumThreads(" << threads << ")" << std::endl; ret.SetGlobalIntraOpNumThreads(threads); } { int threads = (int)PerformanceOptions::instance().ONNX_OPTIONS.MAX_INTER_OP_THREADS; global_logger_tagged().log("Setting ONNX Inter-Op Threads: " + tostr_u_commas(threads)); + std::cerr << "[ORT-DIAG] SetGlobalInterOpNumThreads(" << threads << ")" << std::endl; ret.SetGlobalInterOpNumThreads(threads); } ret.SetGlobalSpinControl(0); @@ -31,6 +37,7 @@ Ort::ThreadingOptions make_ort_threading_options(){ // them, and they are very slow on Intel CPUs. ONNX Runtime only supports this on x86; // see Common/Cpp/Hardware/FloatingPointMode.h. ret.SetGlobalDenormalAsZero(); + std::cerr << "[ORT-DIAG] SetGlobalSpinControl(0), SetGlobalDenormalAsZero()" << std::endl; return ret; } @@ -42,9 +49,10 @@ Ort::Env& global_ort_env(){ #endif static Ort::Env env( make_ort_threading_options(), - ORT_LOGGING_LEVEL_WARNING, + ORT_LOGGING_LEVEL_INFO, // [ORT-DIAG] was WARNING "Global_ThreadPool_Env" ); + std::cerr << "[ORT-DIAG] Ort::Env created" << std::endl; return env; } diff --git a/SerialPrograms/Source/Tests/CommandLineTests.cpp b/SerialPrograms/Source/Tests/CommandLineTests.cpp index a723fef213..ab6fe321f0 100644 --- a/SerialPrograms/Source/Tests/CommandLineTests.cpp +++ b/SerialPrograms/Source/Tests/CommandLineTests.cpp @@ -7,23 +7,91 @@ #include #include "CommonFramework/Logging/Logger.h" +#include "CommonTools/OCR/OCR_PaddleOCR.h" #include "ComputerPrograms/ComputerPrograms_UnitTestRunner.h" #include "CommandLineTests.h" +#if defined(__APPLE__) || defined(__linux__) +#include +#include +#include +#endif + using std::cout; using std::endl; namespace PokemonAutomation{ +#if defined(__APPLE__) || defined(__linux__) +// On a fatal signal, dump a native backtrace of the crashing thread straight to stderr +// (fd 2, unbuffered) so it survives in the CI log even though stdout's buffer is lost when +// the process segfaults. Then restore the default disposition and re-raise, so the OS still +// writes its crash report (.ips on macOS) and the exit code still reflects the crash. +static void crash_backtrace_handler(int sig){ + // Re-entrancy guard: if we crash while handling a signal, just die. + static volatile sig_atomic_t in_handler = 0; + if (in_handler){ + _exit(128 + sig); + } + in_handler = 1; + + const char msg[] = "\n[PA-CRASH] Fatal signal, backtrace of the crashing thread:\n"; + if (write(2, msg, sizeof(msg) - 1) < 0){ + // Nothing more we can do. + } + + void* frames[128]; + int n = backtrace(frames, 128); + backtrace_symbols_fd(frames, n, 2); // writes one symbolized frame per line to fd 2 + + // Let the OS do its normal crash handling (crash report + exit code). + signal(sig, SIG_DFL); + raise(sig); +} + +static void install_crash_backtrace_handlers(){ + static const int sigs[] = { SIGSEGV, SIGBUS, SIGABRT, SIGILL, SIGFPE }; + for (int sig : sigs){ + signal(sig, crash_backtrace_handler); + } +} +#else +static void install_crash_backtrace_handlers(){} +#endif + int run_command_line_tests(bool augment_color_distortions){ + install_crash_backtrace_handlers(); + // [ORT-DIAG] Keep stdout line-buffered in CI: when piped, stdio defaults to full + // buffering, so the last few KB of test output (and where the crash happened) is lost + // in the buffer when the process dies. + // 8192 bytes: 0 means "implementation-defined" and MSVC's ucrt can end up with a + // zero-size buffer, which corrupts the next stdout write (observed 0xC0000409 on the + // Windows CI runner). + setvbuf(stdout, nullptr, _IOLBF, 8192); cout << "Running parallel unit tests..." << endl; - ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); - if (runner.run(augment_color_distortions)){ - return 1; + + // [ORT-DIAG] The Mac Intel crash is probabilistic (~60% of CI runs) and happens at + // PaddleOCR's ONNX session creation, which normally happens only once per process + // (the sessions are cached). To raise the odds of reproducing it in CI, run the full + // suite several times, clearing the PaddleOCR cache between runs so the sessions are + // re-created each time. + constexpr int k_repeat_runs = 3; + for (int run_index = 1; run_index <= k_repeat_runs; ++run_index){ + std::cerr << "[ORT-DIAG] Test suite run " << run_index << " of " << k_repeat_runs << std::endl; + if (run_index > 1){ + OCR::clear_paddle_ocr_cache(); // Force PaddleOCR to re-create its ONNX sessions. + } + ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); + if (runner.run(augment_color_distortions)){ + std::cerr << "[ORT-DIAG] Test suite run " << run_index << " reported failures; stopping." + << std::endl; + return 1; + } } + cout << "Running parallel unit tests... Done!" << endl; return 0; } diff --git a/SerialPrograms/cmake/SourceFiles.cmake b/SerialPrograms/cmake/SourceFiles.cmake index b0f27c7490..b4725b5616 100644 --- a/SerialPrograms/cmake/SourceFiles.cmake +++ b/SerialPrograms/cmake/SourceFiles.cmake @@ -396,6 +396,7 @@ file(GLOB LIBRARY_SOURCES ../Common/Cpp/Rectangle.tpp ../Common/Cpp/SIMDDebuggers.h ../Common/Cpp/Sockets/AbstractClientSocket.h + ../Common/Cpp/Sockets/AbstractServerSocket.h ../Common/Cpp/Sockets/ClientSocket.cpp ../Common/Cpp/Sockets/ClientSocket.h ../Common/Cpp/Sockets/ClientSocket_POSIX.cpp @@ -404,6 +405,12 @@ file(GLOB LIBRARY_SOURCES ../Common/Cpp/Sockets/ClientSocket_Qt.h ../Common/Cpp/Sockets/ClientSocket_WinSocket.cpp ../Common/Cpp/Sockets/ClientSocket_WinSocket.h + ../Common/Cpp/Sockets/ServerSocket_POSIX.cpp + ../Common/Cpp/Sockets/ServerSocket_Qt.cpp + ../Common/Cpp/Sockets/ServerSocket_POSIX.h + ../Common/Cpp/Sockets/ServerSocket_Qt.h + ../Common/Cpp/Sockets/Sockets_Tests.cpp + ../Common/Cpp/Sockets/Sockets_Tests.h ../Common/Cpp/StreamConnections/MockDevice.cpp ../Common/Cpp/StreamConnections/MockDevice.h ../Common/Cpp/TestRunners/ParallelUnitTestRunner.cpp