From b11bcd54d7097ab97b691073258dc24e24634479 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Wed, 17 Jun 2026 11:39:57 +0100 Subject: [PATCH 1/4] preparing for the ACCU on Sea presentation --- examples/postgres-talk-c++now.cpp | 217 ++++++++++++++++++++++++++++++ examples/postgres-talk.cpp | 160 +--------------------- 2 files changed, 218 insertions(+), 159 deletions(-) create mode 100644 examples/postgres-talk-c++now.cpp diff --git a/examples/postgres-talk-c++now.cpp b/examples/postgres-talk-c++now.cpp new file mode 100644 index 0000000..eb33a0f --- /dev/null +++ b/examples/postgres-talk-c++now.cpp @@ -0,0 +1,217 @@ +// examples/postgres-talk.cpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include +#include +#include +#include "demo_algorithm.hpp" +#include "demo_http.hpp" +#include +#include +#include +#include +#include +#include +#include + +namespace ex = beman::execution; +namespace net = beman::net; +using namespace std::chrono_literals; + +// PQconnectdb(const char* connstr) -> PGconn* +// PQfinish(PGconn*) - clean-up +// PQexec(const PGconn *conn, const char* query) -> PGresult* - query the database +// PQsendQuery(const PGconn *conn, const char* query) - send a query +// PQconsumeInput(const PGconn *conn) - consume available input, clear socket stat +// PQgetResult(const PGconn *conn) -> PGresult* - get result, return nullptr if no more results, or would block +// PQclear(PGresult) - clean-up +// PQsetnonblocking(const PGconn *conn, int arg) - set non-blocking mode to avoid write blocks +// PQsocket(const PGconn *conn) - get socket +// PQflush(const PGconn *conn) - flush output buffer, return 1 if still pending data +// PQisBusy(const PGconn *conn) -> int - PQgetResult() would block +// PQsetSingleRowMode(PGconn *conn) - set single row mode, return 0 on failure +// PQsetChunkedMode(PGconn *conn, int arg) - set chunked mode, return 0 on failure + +namespace { +const std::string connection_string("user=sruser dbname=sruser"); +const std::string query("select *, pg_sleep(0.5) from messages where 0 < key and key < 5;"); +const std::string query2("select *, pg_sleep(0.5) from messages where 4 < key and key < 8;"); + +inline constexpr auto print_result{[](const PGresult* result) noexcept { + std::cout << "n=" << PQntuples(result) << ", m=" << PQnfields(result) << "\n"; + for (int i = 0, n = PQntuples(result); i < n; ++i) { + for (int j = 0, m = PQnfields(result); j < m; ++j == m || std::cout << ", ") { + std::cout << PQgetvalue(result, i, j); + } + std::cout << "\n"; + } +}}; +} // namespace + +namespace pg { + +struct conn { + std::unique_ptr cn; + net::ip::tcp::socket socket; + conn(net::io_context& io, PGconn* c) : cn(c), socket(io, io.make_socket(PQsocket(c))) {} + operator PGconn*() const { return cn.get(); } +}; + +struct result { + std::unique_ptr res; + result(auto r) : res(r) {} + operator PGresult*() const { return res.get(); } +}; +using results = std::vector; +struct error { + std::string message; + error(std::string m) : message(m) { std::cout << "error:" << m << "\n"; } +}; + +struct env { + using error_types = ex::completion_signatures; +}; + +ex::task exec(auto& conn, std::string query) { + if (!PQsendQuery(conn, query.c_str())) { + co_yield ex::with_error(pg::error(PQerrorMessage(conn))); + } + while (PQflush(conn)) { + co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::out)); + } + pg::results res; + while (true) { + while (PQisBusy(conn)) { + co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in)); + if (!PQconsumeInput(conn)) { + co_yield ex::with_error(pg::error(PQerrorMessage(conn))); + } + } + if (!res.emplace_back(pg::result(PQgetResult(conn)))) { + res.pop_back(); + break; + } + } + co_return std::move(res); +} + +template +class mutex { + struct state_base { + state_base* next{}; + virtual void run() = 0; + }; + Object obj; + bool is_busy{}; + state_base* waiting{}; + + template + using sender_t = decltype(std::declval()(std::declval())); + + template + struct state : state_base { + using operation_state_concept = ex::operation_state_tag; + + std::remove_cvref_t rcvr; + mutex* self; + + struct receiver { + using receiver_concept = ex::receiver_tag; + state* st; + + auto get_env() const noexcept { return ex::get_env(st->rcvr); } + template + void set_value(A&&... a) && noexcept { + st->complete(); + ex::set_value(std::move(st->rcvr), std::forward(a)...); + } + template + void set_error(E&& e) && noexcept { + st->complete(); + ex::set_error(std::move(st->rcvr), std::forward(e)); + } + void set_stopped() && noexcept { + st->complete(); + ex::set_stopped(std::move(st->rcvr)); + } + }; + using inner_state_t = ex::connect_result_t, receiver>; + inner_state_t inner_state; + + state(Rcvr&& r, mutex* s, Fun f) + : rcvr(std::forward(r)), + self(s), + inner_state(ex::connect(std::move(f)(self->obj), receiver(this))) {} + void start() & noexcept { + if (std::exchange(self->is_busy, true)) { + this->next = std::exchange(self->waiting, this); + } else { + run(); + } + } + void run() override { ex::start(inner_state); } + void complete() { + if (self->waiting) { + std::exchange(self->waiting, self->waiting->next)->run(); + } else { + self->is_busy = false; + } + } + }; + template + struct sender { + using sender_concept = ex::sender_tag; + template + static consteval auto get_completion_signatures() { + return ex::get_completion_signatures, Env...>(); + } + mutex* self; + Fun fun; + + template + auto connect(Rcvr&& rcvr) && { + return state(std::forward(rcvr), self, std::move(fun)); + } + }; + + public: + template + mutex(A&&... a) : obj(std::forward(a)...) {} + template + auto run(Fun fun) { + return sender(this, std::move(fun)); + } +}; + +auto exec(pg::mutex& conn, std::string query) { + return conn.run([query = std::move(query)](pg::conn& conn) { return pg::exec(conn, query); }); +} +} // namespace pg + +// ---------------------------------------------------------------------------- + +auto main() -> int { + std::cout << std::unitbuf << "Postgres Example\n"; + net::io_context io; + pg::mutex conn(io, PQconnectdb(connection_string.c_str())); + + auto clock{[](auto sched) -> ex::task<> { + while (true) { + std::cout << "time=" << std::chrono::system_clock::now() << "\n"; + co_await net::resume_after(sched, 1s); + } + }(io.get_scheduler())}; + + auto query1{[](auto sched, auto& conn) -> ex::task<> { + auto res(co_await pg::exec(conn, query)); + std::ranges::for_each(res, print_result); + co_await net::resume_after(sched, 2s); + }(io.get_scheduler(), conn)}; + auto query2(pg::exec(conn, ::query2) | ex::then([](auto res) { std::ranges::for_each(res, print_result); })); + + // ex::sync_wait(demo::when_any()); + // ex::sync_wait(ex::when_all()); + + ex::sync_wait( + demo::when_any(io.async_run(), std::move(clock), ex::when_all(std::move(query1), std::move(query2)))); +} diff --git a/examples/postgres-talk.cpp b/examples/postgres-talk.cpp index eb33a0f..305ed0f 100644 --- a/examples/postgres-talk.cpp +++ b/examples/postgres-talk.cpp @@ -37,7 +37,7 @@ const std::string connection_string("user=sruser dbname=sruser"); const std::string query("select *, pg_sleep(0.5) from messages where 0 < key and key < 5;"); const std::string query2("select *, pg_sleep(0.5) from messages where 4 < key and key < 8;"); -inline constexpr auto print_result{[](const PGresult* result) noexcept { +inline constexpr auto print_result [[maybe_unused]]{[](const PGresult* result) noexcept { std::cout << "n=" << PQntuples(result) << ", m=" << PQnfields(result) << "\n"; for (int i = 0, n = PQntuples(result); i < n; ++i) { for (int j = 0, m = PQnfields(result); j < m; ++j == m || std::cout << ", ") { @@ -50,168 +50,10 @@ inline constexpr auto print_result{[](const PGresult* result) noexcept { namespace pg { -struct conn { - std::unique_ptr cn; - net::ip::tcp::socket socket; - conn(net::io_context& io, PGconn* c) : cn(c), socket(io, io.make_socket(PQsocket(c))) {} - operator PGconn*() const { return cn.get(); } -}; - -struct result { - std::unique_ptr res; - result(auto r) : res(r) {} - operator PGresult*() const { return res.get(); } -}; -using results = std::vector; -struct error { - std::string message; - error(std::string m) : message(m) { std::cout << "error:" << m << "\n"; } -}; - -struct env { - using error_types = ex::completion_signatures; -}; - -ex::task exec(auto& conn, std::string query) { - if (!PQsendQuery(conn, query.c_str())) { - co_yield ex::with_error(pg::error(PQerrorMessage(conn))); - } - while (PQflush(conn)) { - co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::out)); - } - pg::results res; - while (true) { - while (PQisBusy(conn)) { - co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in)); - if (!PQconsumeInput(conn)) { - co_yield ex::with_error(pg::error(PQerrorMessage(conn))); - } - } - if (!res.emplace_back(pg::result(PQgetResult(conn)))) { - res.pop_back(); - break; - } - } - co_return std::move(res); -} - -template -class mutex { - struct state_base { - state_base* next{}; - virtual void run() = 0; - }; - Object obj; - bool is_busy{}; - state_base* waiting{}; - - template - using sender_t = decltype(std::declval()(std::declval())); - - template - struct state : state_base { - using operation_state_concept = ex::operation_state_tag; - - std::remove_cvref_t rcvr; - mutex* self; - - struct receiver { - using receiver_concept = ex::receiver_tag; - state* st; - - auto get_env() const noexcept { return ex::get_env(st->rcvr); } - template - void set_value(A&&... a) && noexcept { - st->complete(); - ex::set_value(std::move(st->rcvr), std::forward(a)...); - } - template - void set_error(E&& e) && noexcept { - st->complete(); - ex::set_error(std::move(st->rcvr), std::forward(e)); - } - void set_stopped() && noexcept { - st->complete(); - ex::set_stopped(std::move(st->rcvr)); - } - }; - using inner_state_t = ex::connect_result_t, receiver>; - inner_state_t inner_state; - - state(Rcvr&& r, mutex* s, Fun f) - : rcvr(std::forward(r)), - self(s), - inner_state(ex::connect(std::move(f)(self->obj), receiver(this))) {} - void start() & noexcept { - if (std::exchange(self->is_busy, true)) { - this->next = std::exchange(self->waiting, this); - } else { - run(); - } - } - void run() override { ex::start(inner_state); } - void complete() { - if (self->waiting) { - std::exchange(self->waiting, self->waiting->next)->run(); - } else { - self->is_busy = false; - } - } - }; - template - struct sender { - using sender_concept = ex::sender_tag; - template - static consteval auto get_completion_signatures() { - return ex::get_completion_signatures, Env...>(); - } - mutex* self; - Fun fun; - - template - auto connect(Rcvr&& rcvr) && { - return state(std::forward(rcvr), self, std::move(fun)); - } - }; - - public: - template - mutex(A&&... a) : obj(std::forward(a)...) {} - template - auto run(Fun fun) { - return sender(this, std::move(fun)); - } -}; - -auto exec(pg::mutex& conn, std::string query) { - return conn.run([query = std::move(query)](pg::conn& conn) { return pg::exec(conn, query); }); -} } // namespace pg // ---------------------------------------------------------------------------- auto main() -> int { std::cout << std::unitbuf << "Postgres Example\n"; - net::io_context io; - pg::mutex conn(io, PQconnectdb(connection_string.c_str())); - - auto clock{[](auto sched) -> ex::task<> { - while (true) { - std::cout << "time=" << std::chrono::system_clock::now() << "\n"; - co_await net::resume_after(sched, 1s); - } - }(io.get_scheduler())}; - - auto query1{[](auto sched, auto& conn) -> ex::task<> { - auto res(co_await pg::exec(conn, query)); - std::ranges::for_each(res, print_result); - co_await net::resume_after(sched, 2s); - }(io.get_scheduler(), conn)}; - auto query2(pg::exec(conn, ::query2) | ex::then([](auto res) { std::ranges::for_each(res, print_result); })); - - // ex::sync_wait(demo::when_any()); - // ex::sync_wait(ex::when_all()); - - ex::sync_wait( - demo::when_any(io.async_run(), std::move(clock), ex::when_all(std::move(query1), std::move(query2)))); } From f0a57b7eb624880778cf9be1e2df657bdffbc6a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Wed, 17 Jun 2026 16:05:02 +0100 Subject: [PATCH 2/4] added the ACCU talk --- CMakeLists.txt | 1 + examples/CMakeLists.txt | 1 + examples/postgres-talk.cpp | 162 +++++++++++++++++++++++++++++++++++++ 3 files changed, 164 insertions(+) diff --git a/CMakeLists.txt b/CMakeLists.txt index 5c59a82..d548db5 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -52,6 +52,7 @@ beman_install_library(${PROJECT_NAME} TARGETS beman.net_headers enable_testing() +set(BEMAN_NET_BUILD_TESTS FALSE) if(BEMAN_NET_BUILD_TESTS) add_subdirectory(tests/beman/net) endif() diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 1c7e81c..b198a3b 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -37,6 +37,7 @@ set(EXAMPLES if(${HAS_POSTGRES}) list(APPEND EXAMPLES postgres postgres-talk populate-postgres) endif() +set(EXAMPLES postgres-talk) foreach(EXAMPLE ${EXAMPLES}) set(EXAMPLE_TARGET beman.net.examples.${EXAMPLE}) diff --git a/examples/postgres-talk.cpp b/examples/postgres-talk.cpp index 305ed0f..84c43fe 100644 --- a/examples/postgres-talk.cpp +++ b/examples/postgres-talk.cpp @@ -49,11 +49,173 @@ inline constexpr auto print_result [[maybe_unused]]{[](const PGresult* result) n } // namespace namespace pg { + struct conn { + std::unique_ptr con; + net::ip::tcp::socket socket; + conn(net::io_context& io, PGconn* c): con(c), socket(io, io.make_socket(PQsocket(con.get()))) {} + operator PGconn*() const { return con.get(); } + }; + struct result { + std::unique_ptr res; + result(PGresult* r): res(r) {} + operator PGresult*() const { return res.get(); } + }; + using results = std::vector; + ex::task exec(auto& conn, std::string query) { + if (!PQsendQuery(conn, query.c_str())) { + std::cout << "ERROR: " << PQerrorMessage(conn) << "\n"; + throw std::runtime_error("ERROR"); + } + while (PQflush(conn)) { + co_await net::async_poll(conn.socket, net::event_type::out); + } + pg::results res; + std::cout << "sent query\n"; + + while (true) { + while (PQisBusy(conn)) { + co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in)); + if (!PQconsumeInput(conn)) { + // error handling + } + } + std::cout << "got a result\n"; + if (!res.emplace_back(PQgetResult(conn))) { + res.pop_back(); + break; + } + } + co_return res; + } + + template + struct mutex { + struct state_base { + state_base* next{}; + virtual void run() = 0; + }; + + Object obj; + bool is_busy{false}; + state_base* waiting{}; + + template + mutex(A&&... a): obj(std::forward(a)...) {} + + template + using sender_t = decltype(std::declval()(std::declval())); + + template + struct state: state_base { + using operation_state_concept = ex::operation_state_tag; + std::remove_cvref_t rcvr; + mutex* mut; + + struct receiver { + using receiver_concept = ex::receiver_tag; + state* s; + auto get_env() const noexcept { + return ex::get_env(s->rcvr); + } + template + void set_value(A&&... a) noexcept { + s->complete(); + ex::set_value(std::move(s->rcvr), std::forward(a)...); + } + template + void set_error(A&& a) noexcept { + s->complete(); + ex::set_error(std::move(s->rcvr), std::forward(a)); + } + void set_stopped() noexcept { + s->complete(); + ex::set_stopped(std::move(s->rcvr)); + } + }; + + using inner_state_t = ex::connect_result_t, receiver>; + + inner_state_t inner_state; + state(Rcvr&& r, Fun f, mutex* m): rcvr(std::forward(r)), mut(m), + inner_state(ex::connect(std::move(f)(mut->obj), receiver{this})) {} + void start() noexcept { + if (!std::exchange(this->mut->is_busy, true)) { + run(); + } + else { + this->next = std::exchange(mut->waiting, this); + } + } + void complete() { + if (mut->waiting) { + std::exchange(mut->waiting, mut->waiting->next)->run(); + } + else { + mut->is_busy = false; + } + } + void run() { + ex::start(inner_state); + } + }; + + template + struct sender { + using sender_concept = ex::sender_tag; + template + static consteval auto get_completion_signatures() { + return ex::get_completion_signatures, Env...>(); + } + + Fun fun; + mutex* mut; + + template + auto connect(Rcvr&& r) && { + return state(std::forward(r), std::move(fun), mut); + } + }; + + template + auto run(Fun fun) { + return sender(std::move(fun), this); + } + }; + + auto exec(pg::mutex& conn, std::string query) { + return conn.run([=](pg::conn& obj) { return exec(obj, query); }); + } } // namespace pg // ---------------------------------------------------------------------------- auto main() -> int { std::cout << std::unitbuf << "Postgres Example\n"; + net::io_context io; + pg::mutex conn(io, PQconnectdb(connection_string.c_str())); + + auto clock = [](auto sched) -> ex::task<> { + while (true) { + std::cout << "time=" << std::chrono::system_clock::now() << "\n"; + // std::this_thread::sleep_for(1s); + co_await net::resume_after(sched, 1s); + } + co_return; + }(io.get_scheduler()); + std::cout << "do more!\n"; + + auto query = [](auto& conn)->ex::task<> { + auto res = co_await pg::exec(conn, ::query); + std::ranges::for_each(res, ::print_result); + co_return; + }(conn); + auto query2 = [](auto& conn)->ex::task<> { + auto res = co_await pg::exec(conn, ::query2); + std::ranges::for_each(res, ::print_result); + co_return; + }(conn); + + ex::sync_wait(demo::when_any(io.async_run(), std::move(clock), + ex::when_all(std::move(query), std::move(query2)))); } From 3450bd6f8a59dd4255ddd798721d2c77d1264cbf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Wed, 17 Jun 2026 16:05:45 +0100 Subject: [PATCH 3/4] added the result into a suitably named file --- examples/postgres-talk-accuonsea.cpp | 221 +++++++++++++++++++++++++++ 1 file changed, 221 insertions(+) create mode 100644 examples/postgres-talk-accuonsea.cpp diff --git a/examples/postgres-talk-accuonsea.cpp b/examples/postgres-talk-accuonsea.cpp new file mode 100644 index 0000000..84c43fe --- /dev/null +++ b/examples/postgres-talk-accuonsea.cpp @@ -0,0 +1,221 @@ +// examples/postgres-talk.cpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include +#include +#include +#include "demo_algorithm.hpp" +#include "demo_http.hpp" +#include +#include +#include +#include +#include +#include +#include + +namespace ex = beman::execution; +namespace net = beman::net; +using namespace std::chrono_literals; + +// PQconnectdb(const char* connstr) -> PGconn* +// PQfinish(PGconn*) - clean-up +// PQexec(const PGconn *conn, const char* query) -> PGresult* - query the database +// PQsendQuery(const PGconn *conn, const char* query) - send a query +// PQconsumeInput(const PGconn *conn) - consume available input, clear socket stat +// PQgetResult(const PGconn *conn) -> PGresult* - get result, return nullptr if no more results, or would block +// PQclear(PGresult) - clean-up +// PQsetnonblocking(const PGconn *conn, int arg) - set non-blocking mode to avoid write blocks +// PQsocket(const PGconn *conn) - get socket +// PQflush(const PGconn *conn) - flush output buffer, return 1 if still pending data +// PQisBusy(const PGconn *conn) -> int - PQgetResult() would block +// PQsetSingleRowMode(PGconn *conn) - set single row mode, return 0 on failure +// PQsetChunkedMode(PGconn *conn, int arg) - set chunked mode, return 0 on failure + +namespace { +const std::string connection_string("user=sruser dbname=sruser"); +const std::string query("select *, pg_sleep(0.5) from messages where 0 < key and key < 5;"); +const std::string query2("select *, pg_sleep(0.5) from messages where 4 < key and key < 8;"); + +inline constexpr auto print_result [[maybe_unused]]{[](const PGresult* result) noexcept { + std::cout << "n=" << PQntuples(result) << ", m=" << PQnfields(result) << "\n"; + for (int i = 0, n = PQntuples(result); i < n; ++i) { + for (int j = 0, m = PQnfields(result); j < m; ++j == m || std::cout << ", ") { + std::cout << PQgetvalue(result, i, j); + } + std::cout << "\n"; + } +}}; +} // namespace + +namespace pg { + struct conn { + std::unique_ptr con; + net::ip::tcp::socket socket; + conn(net::io_context& io, PGconn* c): con(c), socket(io, io.make_socket(PQsocket(con.get()))) {} + operator PGconn*() const { return con.get(); } + }; + + struct result { + std::unique_ptr res; + result(PGresult* r): res(r) {} + operator PGresult*() const { return res.get(); } + }; + using results = std::vector; + ex::task exec(auto& conn, std::string query) { + if (!PQsendQuery(conn, query.c_str())) { + std::cout << "ERROR: " << PQerrorMessage(conn) << "\n"; + throw std::runtime_error("ERROR"); + } + while (PQflush(conn)) { + co_await net::async_poll(conn.socket, net::event_type::out); + } + pg::results res; + std::cout << "sent query\n"; + + while (true) { + while (PQisBusy(conn)) { + co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in)); + if (!PQconsumeInput(conn)) { + // error handling + } + } + std::cout << "got a result\n"; + if (!res.emplace_back(PQgetResult(conn))) { + res.pop_back(); + break; + } + } + co_return res; + } + + template + struct mutex { + struct state_base { + state_base* next{}; + virtual void run() = 0; + }; + + Object obj; + bool is_busy{false}; + state_base* waiting{}; + + template + mutex(A&&... a): obj(std::forward(a)...) {} + + template + using sender_t = decltype(std::declval()(std::declval())); + + template + struct state: state_base { + using operation_state_concept = ex::operation_state_tag; + std::remove_cvref_t rcvr; + mutex* mut; + + struct receiver { + using receiver_concept = ex::receiver_tag; + state* s; + auto get_env() const noexcept { + return ex::get_env(s->rcvr); + } + template + void set_value(A&&... a) noexcept { + s->complete(); + ex::set_value(std::move(s->rcvr), std::forward(a)...); + } + template + void set_error(A&& a) noexcept { + s->complete(); + ex::set_error(std::move(s->rcvr), std::forward(a)); + } + void set_stopped() noexcept { + s->complete(); + ex::set_stopped(std::move(s->rcvr)); + } + }; + + using inner_state_t = ex::connect_result_t, receiver>; + + inner_state_t inner_state; + state(Rcvr&& r, Fun f, mutex* m): rcvr(std::forward(r)), mut(m), + inner_state(ex::connect(std::move(f)(mut->obj), receiver{this})) {} + void start() noexcept { + if (!std::exchange(this->mut->is_busy, true)) { + run(); + } + else { + this->next = std::exchange(mut->waiting, this); + } + } + void complete() { + if (mut->waiting) { + std::exchange(mut->waiting, mut->waiting->next)->run(); + } + else { + mut->is_busy = false; + } + } + void run() { + ex::start(inner_state); + } + }; + + template + struct sender { + using sender_concept = ex::sender_tag; + template + static consteval auto get_completion_signatures() { + return ex::get_completion_signatures, Env...>(); + } + + Fun fun; + mutex* mut; + + template + auto connect(Rcvr&& r) && { + return state(std::forward(r), std::move(fun), mut); + } + }; + + template + auto run(Fun fun) { + return sender(std::move(fun), this); + } + }; + + auto exec(pg::mutex& conn, std::string query) { + return conn.run([=](pg::conn& obj) { return exec(obj, query); }); + } +} // namespace pg + +// ---------------------------------------------------------------------------- + +auto main() -> int { + std::cout << std::unitbuf << "Postgres Example\n"; + net::io_context io; + pg::mutex conn(io, PQconnectdb(connection_string.c_str())); + + auto clock = [](auto sched) -> ex::task<> { + while (true) { + std::cout << "time=" << std::chrono::system_clock::now() << "\n"; + // std::this_thread::sleep_for(1s); + co_await net::resume_after(sched, 1s); + } + co_return; + }(io.get_scheduler()); + std::cout << "do more!\n"; + + auto query = [](auto& conn)->ex::task<> { + auto res = co_await pg::exec(conn, ::query); + std::ranges::for_each(res, ::print_result); + co_return; + }(conn); + auto query2 = [](auto& conn)->ex::task<> { + auto res = co_await pg::exec(conn, ::query2); + std::ranges::for_each(res, ::print_result); + co_return; + }(conn); + + ex::sync_wait(demo::when_any(io.async_run(), std::move(clock), + ex::when_all(std::move(query), std::move(query2)))); +} From 4acc273a064e5aa099aa08a762671c1e09148001 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Wed, 17 Jun 2026 16:10:03 +0100 Subject: [PATCH 4/4] some clean-up to put the talk code inot the right place --- CMakeLists.txt | 1 - examples/CMakeLists.txt | 9 +- examples/postgres-talk-accuonsea.cpp | 241 +++++++++++++-------------- examples/postgres-talk.cpp | 241 +++++++++++++-------------- 4 files changed, 241 insertions(+), 251 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index d548db5..5c59a82 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -52,7 +52,6 @@ beman_install_library(${PROJECT_NAME} TARGETS beman.net_headers enable_testing() -set(BEMAN_NET_BUILD_TESTS FALSE) if(BEMAN_NET_BUILD_TESTS) add_subdirectory(tests/beman/net) endif() diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index b198a3b..d2d865d 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -35,9 +35,14 @@ set(EXAMPLES ) if(${HAS_POSTGRES}) - list(APPEND EXAMPLES postgres postgres-talk populate-postgres) + list(APPEND EXAMPLES + postgres + postgres-talk + postgres-talk-accuonsea + postgres-talk-c++now + populate-postgres + ) endif() -set(EXAMPLES postgres-talk) foreach(EXAMPLE ${EXAMPLES}) set(EXAMPLE_TARGET beman.net.examples.${EXAMPLE}) diff --git a/examples/postgres-talk-accuonsea.cpp b/examples/postgres-talk-accuonsea.cpp index 84c43fe..a7e6b4b 100644 --- a/examples/postgres-talk-accuonsea.cpp +++ b/examples/postgres-talk-accuonsea.cpp @@ -49,150 +49,144 @@ inline constexpr auto print_result [[maybe_unused]]{[](const PGresult* result) n } // namespace namespace pg { - struct conn { - std::unique_ptr con; - net::ip::tcp::socket socket; - conn(net::io_context& io, PGconn* c): con(c), socket(io, io.make_socket(PQsocket(con.get()))) {} - operator PGconn*() const { return con.get(); } - }; - - struct result { - std::unique_ptr res; - result(PGresult* r): res(r) {} - operator PGresult*() const { return res.get(); } - }; - using results = std::vector; - ex::task exec(auto& conn, std::string query) { - if (!PQsendQuery(conn, query.c_str())) { - std::cout << "ERROR: " << PQerrorMessage(conn) << "\n"; - throw std::runtime_error("ERROR"); - } - while (PQflush(conn)) { - co_await net::async_poll(conn.socket, net::event_type::out); - } - pg::results res; - std::cout << "sent query\n"; - - while (true) { - while (PQisBusy(conn)) { - co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in)); - if (!PQconsumeInput(conn)) { - // error handling - } - } - std::cout << "got a result\n"; - if (!res.emplace_back(PQgetResult(conn))) { - res.pop_back(); - break; +struct conn { + std::unique_ptr con; + net::ip::tcp::socket socket; + conn(net::io_context& io, PGconn* c) : con(c), socket(io, io.make_socket(PQsocket(con.get()))) {} + operator PGconn*() const { return con.get(); } +}; + +struct result { + std::unique_ptr res; + result(PGresult* r) : res(r) {} + operator PGresult*() const { return res.get(); } +}; +using results = std::vector; +ex::task exec(auto& conn, std::string query) { + if (!PQsendQuery(conn, query.c_str())) { + std::cout << "ERROR: " << PQerrorMessage(conn) << "\n"; + throw std::runtime_error("ERROR"); + } + while (PQflush(conn)) { + co_await net::async_poll(conn.socket, net::event_type::out); + } + pg::results res; + std::cout << "sent query\n"; + + while (true) { + while (PQisBusy(conn)) { + co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in)); + if (!PQconsumeInput(conn)) { + // error handling } } - co_return res; + std::cout << "got a result\n"; + if (!res.emplace_back(PQgetResult(conn))) { + res.pop_back(); + break; + } } + co_return res; +} - template - struct mutex { - struct state_base { - state_base* next{}; - virtual void run() = 0; - }; +template +struct mutex { + struct state_base { + state_base* next{}; + virtual void run() = 0; + }; - Object obj; - bool is_busy{false}; - state_base* waiting{}; - - template - mutex(A&&... a): obj(std::forward(a)...) {} - - template - using sender_t = decltype(std::declval()(std::declval())); - - template - struct state: state_base { - using operation_state_concept = ex::operation_state_tag; - std::remove_cvref_t rcvr; - mutex* mut; - - struct receiver { - using receiver_concept = ex::receiver_tag; - state* s; - auto get_env() const noexcept { - return ex::get_env(s->rcvr); - } - template - void set_value(A&&... a) noexcept { - s->complete(); - ex::set_value(std::move(s->rcvr), std::forward(a)...); - } - template - void set_error(A&& a) noexcept { - s->complete(); - ex::set_error(std::move(s->rcvr), std::forward(a)); - } - void set_stopped() noexcept { - s->complete(); - ex::set_stopped(std::move(s->rcvr)); - } - }; - - using inner_state_t = ex::connect_result_t, receiver>; - - inner_state_t inner_state; - state(Rcvr&& r, Fun f, mutex* m): rcvr(std::forward(r)), mut(m), - inner_state(ex::connect(std::move(f)(mut->obj), receiver{this})) {} - void start() noexcept { - if (!std::exchange(this->mut->is_busy, true)) { - run(); - } - else { - this->next = std::exchange(mut->waiting, this); - } + Object obj; + bool is_busy{false}; + state_base* waiting{}; + + template + mutex(A&&... a) : obj(std::forward(a)...) {} + + template + using sender_t = decltype(std::declval()(std::declval())); + + template + struct state : state_base { + using operation_state_concept = ex::operation_state_tag; + std::remove_cvref_t rcvr; + mutex* mut; + + struct receiver { + using receiver_concept = ex::receiver_tag; + state* s; + auto get_env() const noexcept { return ex::get_env(s->rcvr); } + template + void set_value(A&&... a) noexcept { + s->complete(); + ex::set_value(std::move(s->rcvr), std::forward(a)...); } - void complete() { - if (mut->waiting) { - std::exchange(mut->waiting, mut->waiting->next)->run(); - } - else { - mut->is_busy = false; - } + template + void set_error(A&& a) noexcept { + s->complete(); + ex::set_error(std::move(s->rcvr), std::forward(a)); } - void run() { - ex::start(inner_state); + void set_stopped() noexcept { + s->complete(); + ex::set_stopped(std::move(s->rcvr)); } }; - template - struct sender { - using sender_concept = ex::sender_tag; - template - static consteval auto get_completion_signatures() { - return ex::get_completion_signatures, Env...>(); + using inner_state_t = ex::connect_result_t, receiver>; + + inner_state_t inner_state; + state(Rcvr&& r, Fun f, mutex* m) + : rcvr(std::forward(r)), mut(m), inner_state(ex::connect(std::move(f)(mut->obj), receiver{this})) {} + void start() noexcept { + if (!std::exchange(this->mut->is_busy, true)) { + run(); + } else { + this->next = std::exchange(mut->waiting, this); + } + } + void complete() { + if (mut->waiting) { + std::exchange(mut->waiting, mut->waiting->next)->run(); + } else { + mut->is_busy = false; } + } + void run() { ex::start(inner_state); } + }; - Fun fun; - mutex* mut; + template + struct sender { + using sender_concept = ex::sender_tag; + template + static consteval auto get_completion_signatures() { + return ex::get_completion_signatures, Env...>(); + } - template - auto connect(Rcvr&& r) && { - return state(std::forward(r), std::move(fun), mut); - } - }; + Fun fun; + mutex* mut; - template - auto run(Fun fun) { - return sender(std::move(fun), this); + template + auto connect(Rcvr&& r) && { + return state(std::forward(r), std::move(fun), mut); } }; - auto exec(pg::mutex& conn, std::string query) { - return conn.run([=](pg::conn& obj) { return exec(obj, query); }); + template + auto run(Fun fun) { + return sender(std::move(fun), this); } +}; + +auto exec(pg::mutex& conn, std::string query) { + return conn.run([=](pg::conn& obj) { return exec(obj, query); }); +} } // namespace pg // ---------------------------------------------------------------------------- auto main() -> int { std::cout << std::unitbuf << "Postgres Example\n"; - net::io_context io; + net::io_context io; pg::mutex conn(io, PQconnectdb(connection_string.c_str())); auto clock = [](auto sched) -> ex::task<> { @@ -205,17 +199,16 @@ auto main() -> int { }(io.get_scheduler()); std::cout << "do more!\n"; - auto query = [](auto& conn)->ex::task<> { + auto query = [](auto& conn) -> ex::task<> { auto res = co_await pg::exec(conn, ::query); std::ranges::for_each(res, ::print_result); co_return; }(conn); - auto query2 = [](auto& conn)->ex::task<> { + auto query2 = [](auto& conn) -> ex::task<> { auto res = co_await pg::exec(conn, ::query2); std::ranges::for_each(res, ::print_result); co_return; }(conn); - ex::sync_wait(demo::when_any(io.async_run(), std::move(clock), - ex::when_all(std::move(query), std::move(query2)))); + ex::sync_wait(demo::when_any(io.async_run(), std::move(clock), ex::when_all(std::move(query), std::move(query2)))); } diff --git a/examples/postgres-talk.cpp b/examples/postgres-talk.cpp index 84c43fe..a7e6b4b 100644 --- a/examples/postgres-talk.cpp +++ b/examples/postgres-talk.cpp @@ -49,150 +49,144 @@ inline constexpr auto print_result [[maybe_unused]]{[](const PGresult* result) n } // namespace namespace pg { - struct conn { - std::unique_ptr con; - net::ip::tcp::socket socket; - conn(net::io_context& io, PGconn* c): con(c), socket(io, io.make_socket(PQsocket(con.get()))) {} - operator PGconn*() const { return con.get(); } - }; - - struct result { - std::unique_ptr res; - result(PGresult* r): res(r) {} - operator PGresult*() const { return res.get(); } - }; - using results = std::vector; - ex::task exec(auto& conn, std::string query) { - if (!PQsendQuery(conn, query.c_str())) { - std::cout << "ERROR: " << PQerrorMessage(conn) << "\n"; - throw std::runtime_error("ERROR"); - } - while (PQflush(conn)) { - co_await net::async_poll(conn.socket, net::event_type::out); - } - pg::results res; - std::cout << "sent query\n"; - - while (true) { - while (PQisBusy(conn)) { - co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in)); - if (!PQconsumeInput(conn)) { - // error handling - } - } - std::cout << "got a result\n"; - if (!res.emplace_back(PQgetResult(conn))) { - res.pop_back(); - break; +struct conn { + std::unique_ptr con; + net::ip::tcp::socket socket; + conn(net::io_context& io, PGconn* c) : con(c), socket(io, io.make_socket(PQsocket(con.get()))) {} + operator PGconn*() const { return con.get(); } +}; + +struct result { + std::unique_ptr res; + result(PGresult* r) : res(r) {} + operator PGresult*() const { return res.get(); } +}; +using results = std::vector; +ex::task exec(auto& conn, std::string query) { + if (!PQsendQuery(conn, query.c_str())) { + std::cout << "ERROR: " << PQerrorMessage(conn) << "\n"; + throw std::runtime_error("ERROR"); + } + while (PQflush(conn)) { + co_await net::async_poll(conn.socket, net::event_type::out); + } + pg::results res; + std::cout << "sent query\n"; + + while (true) { + while (PQisBusy(conn)) { + co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in)); + if (!PQconsumeInput(conn)) { + // error handling } } - co_return res; + std::cout << "got a result\n"; + if (!res.emplace_back(PQgetResult(conn))) { + res.pop_back(); + break; + } } + co_return res; +} - template - struct mutex { - struct state_base { - state_base* next{}; - virtual void run() = 0; - }; +template +struct mutex { + struct state_base { + state_base* next{}; + virtual void run() = 0; + }; - Object obj; - bool is_busy{false}; - state_base* waiting{}; - - template - mutex(A&&... a): obj(std::forward(a)...) {} - - template - using sender_t = decltype(std::declval()(std::declval())); - - template - struct state: state_base { - using operation_state_concept = ex::operation_state_tag; - std::remove_cvref_t rcvr; - mutex* mut; - - struct receiver { - using receiver_concept = ex::receiver_tag; - state* s; - auto get_env() const noexcept { - return ex::get_env(s->rcvr); - } - template - void set_value(A&&... a) noexcept { - s->complete(); - ex::set_value(std::move(s->rcvr), std::forward(a)...); - } - template - void set_error(A&& a) noexcept { - s->complete(); - ex::set_error(std::move(s->rcvr), std::forward(a)); - } - void set_stopped() noexcept { - s->complete(); - ex::set_stopped(std::move(s->rcvr)); - } - }; - - using inner_state_t = ex::connect_result_t, receiver>; - - inner_state_t inner_state; - state(Rcvr&& r, Fun f, mutex* m): rcvr(std::forward(r)), mut(m), - inner_state(ex::connect(std::move(f)(mut->obj), receiver{this})) {} - void start() noexcept { - if (!std::exchange(this->mut->is_busy, true)) { - run(); - } - else { - this->next = std::exchange(mut->waiting, this); - } + Object obj; + bool is_busy{false}; + state_base* waiting{}; + + template + mutex(A&&... a) : obj(std::forward(a)...) {} + + template + using sender_t = decltype(std::declval()(std::declval())); + + template + struct state : state_base { + using operation_state_concept = ex::operation_state_tag; + std::remove_cvref_t rcvr; + mutex* mut; + + struct receiver { + using receiver_concept = ex::receiver_tag; + state* s; + auto get_env() const noexcept { return ex::get_env(s->rcvr); } + template + void set_value(A&&... a) noexcept { + s->complete(); + ex::set_value(std::move(s->rcvr), std::forward(a)...); } - void complete() { - if (mut->waiting) { - std::exchange(mut->waiting, mut->waiting->next)->run(); - } - else { - mut->is_busy = false; - } + template + void set_error(A&& a) noexcept { + s->complete(); + ex::set_error(std::move(s->rcvr), std::forward(a)); } - void run() { - ex::start(inner_state); + void set_stopped() noexcept { + s->complete(); + ex::set_stopped(std::move(s->rcvr)); } }; - template - struct sender { - using sender_concept = ex::sender_tag; - template - static consteval auto get_completion_signatures() { - return ex::get_completion_signatures, Env...>(); + using inner_state_t = ex::connect_result_t, receiver>; + + inner_state_t inner_state; + state(Rcvr&& r, Fun f, mutex* m) + : rcvr(std::forward(r)), mut(m), inner_state(ex::connect(std::move(f)(mut->obj), receiver{this})) {} + void start() noexcept { + if (!std::exchange(this->mut->is_busy, true)) { + run(); + } else { + this->next = std::exchange(mut->waiting, this); + } + } + void complete() { + if (mut->waiting) { + std::exchange(mut->waiting, mut->waiting->next)->run(); + } else { + mut->is_busy = false; } + } + void run() { ex::start(inner_state); } + }; - Fun fun; - mutex* mut; + template + struct sender { + using sender_concept = ex::sender_tag; + template + static consteval auto get_completion_signatures() { + return ex::get_completion_signatures, Env...>(); + } - template - auto connect(Rcvr&& r) && { - return state(std::forward(r), std::move(fun), mut); - } - }; + Fun fun; + mutex* mut; - template - auto run(Fun fun) { - return sender(std::move(fun), this); + template + auto connect(Rcvr&& r) && { + return state(std::forward(r), std::move(fun), mut); } }; - auto exec(pg::mutex& conn, std::string query) { - return conn.run([=](pg::conn& obj) { return exec(obj, query); }); + template + auto run(Fun fun) { + return sender(std::move(fun), this); } +}; + +auto exec(pg::mutex& conn, std::string query) { + return conn.run([=](pg::conn& obj) { return exec(obj, query); }); +} } // namespace pg // ---------------------------------------------------------------------------- auto main() -> int { std::cout << std::unitbuf << "Postgres Example\n"; - net::io_context io; + net::io_context io; pg::mutex conn(io, PQconnectdb(connection_string.c_str())); auto clock = [](auto sched) -> ex::task<> { @@ -205,17 +199,16 @@ auto main() -> int { }(io.get_scheduler()); std::cout << "do more!\n"; - auto query = [](auto& conn)->ex::task<> { + auto query = [](auto& conn) -> ex::task<> { auto res = co_await pg::exec(conn, ::query); std::ranges::for_each(res, ::print_result); co_return; }(conn); - auto query2 = [](auto& conn)->ex::task<> { + auto query2 = [](auto& conn) -> ex::task<> { auto res = co_await pg::exec(conn, ::query2); std::ranges::for_each(res, ::print_result); co_return; }(conn); - ex::sync_wait(demo::when_any(io.async_run(), std::move(clock), - ex::when_all(std::move(query), std::move(query2)))); + ex::sync_wait(demo::when_any(io.async_run(), std::move(clock), ex::when_all(std::move(query), std::move(query2)))); }