eelke/pgLab
1
0
Fork 0
Code Issues 29 Pull requests Projects Releases 1 Packages Wiki Activity Actions

Switch to cmake + vcpkg, builds on linux, windows needs verification.

Browse source 
Code has been modified to be portable.
This commit is contained in:
eelke 2026-10-17 16:21:22 +02:00
parent c1a5ac1aad
commit 49f6619a4e
34 changed files with 678 additions and 327 deletions
Download patch file Download diff file Expand all files Collapse all files

200
pglablib/ASyncDBConnection.cpp
View file

@ -1,13 +1,14 @@
#include "ASyncDBConnection.h"
#include "ASyncDBConnection.h"
#include "ScopeGuard.h"
#include "util.h"
#include "Pgsql_PgException.h"
#include "Win32Event.h"
#include "WaitHandleList.h"
#include <atomic>
#include <chrono>
#include <queue>
#include <WinSock2.h>
#include <QDebug>
#include <QEventLoop>
#include <QSocketNotifier>
#include <QTimer>
namespace {
@ -45,10 +46,6 @@ public:
struct t_Command {
std::mutex m_mutex;
t_CommandQueue m_queue;
Win32Event m_newEvent;
t_Command()
: m_newEvent(Win32Event::Reset::Auto, Win32Event::Initial::Clear)
{}
} m_commandQueue;
ConnectionConfig m_config;
@ -64,18 +61,29 @@ public:
void stop();
/// Wakes the worker thread when it is waiting in its event loop
void wake();
private:
enum class WaitResult {
Socket, ///< the socket became ready
Wakeup, ///< woken up by stop() or by a new command
Timeout ///< the timeout expired
};
ASyncDBConnection *asyncConnObject = nullptr;
/// \todo Implement new method to stop the thread
Win32Event m_stopEvent;
std::atomic<bool> terminateRequested{false}; ///< is set when the thread should stop
std::mutex m_wakeMutex;
QEventLoop *m_loop = nullptr; ///< loop the worker blocks in, guarded by m_wakeMutex
Pgsql::Connection m_connection;
bool terminateRequested = false; ///< is set when the thread should stop
bool m_terminated = true;
Pgsql::Canceller m_canceller;
QElapsedTimer m_timer;
bool hasQueuedCommand();
WaitResult waitForSocket(QSocketNotifier::Type type, int timeout_ms, bool wakeOnNewCommand);
bool makeConnection();
void communicate();
@ -97,7 +105,6 @@ private:
ASyncDBConnectionThread::ASyncDBConnectionThread(ASyncDBConnection *asco)
: asyncConnObject(asco)
, m_stopEvent(Win32Event::Reset::Manual, Win32Event::Initial::Clear)
{}
void ASyncDBConnectionThread::run()
@ -154,6 +161,59 @@ bool ASyncDBConnectionThread::cancel()
return m_canceller.cancel(nullptr);
}
bool ASyncDBConnectionThread::hasQueuedCommand()
{
std::lock_guard<std::mutex> lg(m_commandQueue.m_mutex);
return !m_commandQueue.m_queue.empty();
}
ASyncDBConnectionThread::WaitResult ASyncDBConnectionThread::waitForSocket(QSocketNotifier::Type type, int timeout_ms, bool wakeOnNewCommand)
{
QEventLoop loop;
WaitResult result = WaitResult::Wakeup;
QSocketNotifier notifier(m_connection.socket(), type);
QObject::connect(&notifier, &QSocketNotifier::activated, &loop, [&result, &loop] {
result = WaitResult::Socket;
loop.quit();
});
QTimer timer;
timer.setSingleShot(true);
if (timeout_ms >= 0) {
QObject::connect(&timer, &QTimer::timeout, &loop, [&result, &loop] {
result = WaitResult::Timeout;
loop.quit();
});
timer.start(timeout_ms);
}
{
std::lock_guard<std::mutex> lg(m_wakeMutex);
m_loop = &loop;
}
// Checked after publishing the loop, so a wake-up that happened before is not lost
bool pending = terminateRequested || (wakeOnNewCommand && hasQueuedCommand());
if (!pending) {
loop.exec();
}
{
std::lock_guard<std::mutex> lg(m_wakeMutex);
m_loop = nullptr;
}
return result;
}
void ASyncDBConnectionThread::wake()
{
std::lock_guard<std::mutex> lg(m_wakeMutex);
if (m_loop) {
QMetaObject::invokeMethod(m_loop, [loop = m_loop] { loop->quit(); }, Qt::QueuedConnection);
}
}
bool ASyncDBConnectionThread::makeConnection()
{
using namespace std::literals::chrono_literals;
@ -165,30 +225,14 @@ bool ASyncDBConnectionThread::makeConnection()
bool ok = m_connection.connectStart(conn_string); //keywords, values);
auto start = std::chrono::steady_clock::now();
if (ok && m_connection.status() != CONNECTION_BAD) {
Win32Event socket_event(Win32Event::Reset::Auto, Win32Event::Initial::Clear);
long fd = FD_WRITE;
QSocketNotifier::Type type = QSocketNotifier::Write;
while (true) {
// poll till complete or failed (we can get an abort command)
int sock = m_connection.socket();
WSAEventSelect(sock, socket_event.handle(), fd);
WaitHandleList whl;
auto wait_result_socket_event = whl.add(socket_event);
auto wait_result_stop = whl.add(m_stopEvent);
auto elapsed = std::chrono::steady_clock::now() - start;
auto remaining = std::chrono::duration_cast<std::chrono::milliseconds>(30s - elapsed).count();
WaitResult res = waitForSocket(type, remaining > 0 ? static_cast<int>(remaining) : 0, false);
auto nu = std::chrono::steady_clock::now();
std::chrono::duration<float, std::milli> diff = -(nu-start);
diff += 30s;
DWORD timeout = diff.count();
DWORD res = MsgWaitForMultipleObjectsEx(
whl.count(), // _In_ DWORD nCount,
whl, // _In_ const HANDLE *pHandles,
timeout, // _In_ DWORD dwMilliseconds,
0, // _In_ DWORD dwWakeMask,
0 // _In_ DWORD dwFlags
);
if (res == wait_result_socket_event) {
if (res == WaitResult::Socket) {
auto poll_state = m_connection.connectPoll();
if (poll_state == PGRES_POLLING_OK) {
qDebug() << "ASyncDBConnection Connected";
@ -202,17 +246,21 @@ bool ASyncDBConnectionThread::makeConnection()
}
else if (poll_state == PGRES_POLLING_READING) {
doStateCallback({ ASyncDBConnection::State::Connecting, "Negotiating" });
fd = FD_READ;
type = QSocketNotifier::Read;
}
else if (poll_state == PGRES_POLLING_WRITING) {
doStateCallback({ ASyncDBConnection::State::Connecting, "Negotiating" });
fd = FD_WRITE;
type = QSocketNotifier::Write;
}
}
else if (res == wait_result_stop) {
else if (terminateRequested) {
m_connection.close();
return false;
}
else if (res == WaitResult::Timeout) {
doStateCallback({ ASyncDBConnection::State::NotConnected, "Connection timed out" });
m_connection.close();
return false;
}
} // end while (true)
}
@ -252,7 +300,7 @@ void ASyncDBConnectionThread::communicate()
void ASyncDBConnectionThread::stop()
{
terminateRequested = true;
m_stopEvent.set();
wake();
}
void ASyncDBConnectionThread::doStateCallback(ASyncDBConnection::StateData state)
@ -264,34 +312,18 @@ void ASyncDBConnectionThread::doStateCallback(ASyncDBConnection::StateData state
bool ASyncDBConnectionThread::waitForAndSendCommand()
{
SOCKET sock = static_cast<SOCKET>(m_connection.socket());
Win32Event socket_event(Win32Event::Reset::Manual, Win32Event::Initial::Clear);
long fd = FD_READ | FD_CLOSE;
WSAEventSelect(sock, socket_event.handle(), fd);
WaitHandleList whl;
auto wait_result_new_command = whl.add(m_commandQueue.m_newEvent);
auto wait_result_socket = whl.add(socket_event);
whl.add(m_stopEvent);
DWORD res = MsgWaitForMultipleObjectsEx(
whl.count(), // _In_ DWORD nCount,
whl, // _In_ const HANDLE *pHandles,
INFINITE, // _In_ DWORD dwMilliseconds,
0, // _In_ DWORD dwWakeMask,
0 // _In_ DWORD dwFlags
);
if (res == wait_result_new_command) {
if (hasQueuedCommand()) {
doNewCommand();
return true;
}
// Idle connection: a readable socket carries notifies or notices, or signals a lost connection
if (waitForSocket(QSocketNotifier::Read, -1, true) == WaitResult::Socket) {
if (!m_connection.consumeInput()) {
doStateCallback({ ASyncDBConnection::State::NotConnected, "Connection lost" });
return false;
}
}
else if (res == wait_result_socket) {
WSANETWORKEVENTS net_events;
WSAEnumNetworkEvents(sock, socket_event.handle(), &net_events);
if (net_events.lNetworkEvents & FD_CLOSE) {
doStateCallback({ ASyncDBConnection::State::NotConnected, "Connection lost" });
return false;
}
}
// Note if it was stop we can just return and function
// above will stop looping because terminateRequested has been set too by stop
return true;
@ -347,42 +379,16 @@ bool ASyncDBConnectionThread::consumeResultInput()
void ASyncDBConnectionThread::waitForResult()
{
SOCKET sock = static_cast<SOCKET>(m_connection.socket());
Win32Event socket_event(Win32Event::Reset::Manual, Win32Event::Initial::Clear);
long fd = FD_READ | FD_CLOSE;
bool finished = false;
while ( ! finished) {
WSAEventSelect(sock, socket_event.handle(), fd);
WaitResult res = waitForSocket(QSocketNotifier::Read, -1, false);
WaitHandleList whl;
auto wait_result_socket = whl.add(socket_event);
auto wait_result_stop = whl.add(m_stopEvent);
DWORD res = MsgWaitForMultipleObjectsEx(
whl.count(), // _In_ DWORD nCount,
whl, // _In_ const HANDLE *pHandles,
INFINITE, // _In_ DWORD dwMilliseconds,
0, // _In_ DWORD dwWakeMask,
0 // _In_ DWORD dwFlags
);
if (res == wait_result_socket) {
WSANETWORKEVENTS net_events;
WSAEnumNetworkEvents(sock, socket_event.handle(), &net_events);
if (net_events.lNetworkEvents & FD_READ) {
if (consumeResultInput()) {
finished = true;
}
if (res == WaitResult::Socket) {
if (consumeResultInput()) {
finished = true;
}
else if (net_events.lNetworkEvents & FD_CLOSE) {
doStateCallback({ ASyncDBConnection::State::NotConnected, "Close while waiting for result" });
finished = true;
stop();
}
}
if (res == wait_result_stop) {
else if (terminateRequested) {
// Send cancel, close connection and terminate thread
cancel();
doStateCallback(ASyncDBConnection::State::Terminating);
@ -455,8 +461,8 @@ bool ASyncDBConnection::send(const std::string &command, on_result_callback on_r
{
std::lock_guard<std::mutex> lg(m_threadData->m_commandQueue.m_mutex);
m_threadData->m_commandQueue.m_queue.emplace(command, on_result);
m_threadData->m_commandQueue.m_newEvent.set();
}
m_threadData->wake();
return true;
}
@ -468,8 +474,8 @@ bool ASyncDBConnection::send(const std::string &command, Pgsql::Params params, o
{
std::lock_guard<std::mutex> lg(m_threadData->m_commandQueue.m_mutex);
m_threadData->m_commandQueue.m_queue.emplace(command, std::move(params), on_result);
m_threadData->m_commandQueue.m_newEvent.set();
}
m_threadData->wake();
return true;
}