Loading...
Searching...
No Matches
pipeline.hxx
/* Definition of the pqxx::pipeline class.
*
* Throughput-optimized mechanism for executing queries.
*
* DO NOT INCLUDE THIS FILE DIRECTLY; include pqxx/pipeline instead.
*
* Copyright (c) 2000-2024, Jeroen T. Vermeulen.
*
* See COPYING for copyright license. If you did not receive a file called
* COPYING with this source code, please notify the distributor of this
* mistake, or contact the author.
*/
#ifndef PQXX_H_PIPELINE
#define PQXX_H_PIPELINE
#if !defined(PQXX_HEADER_PRE)
# error "Include libpqxx headers as <pqxx/header>, not <pqxx/header.hxx>."
#endif
#include <limits>
#include <map>
#include <string>
#include "pqxx/transaction_base.hxx"
namespace pqxx
{
// TODO: libpq 14 introduced a similar "pipeline mode." Can we use that?
class PQXX_LIBEXPORT pipeline : public transaction_focus
{
public:
using query_id = long;
pipeline(pipeline const &) = delete;
pipeline &operator=(pipeline const &) = delete;
pipeline(pipeline &&) = delete;
pipeline &operator=(pipeline &&) = delete;
explicit pipeline(transaction_base &t) : transaction_focus{t, s_classname}
{
init();
}
pipeline(transaction_base &t, std::string_view tname) :
transaction_focus{t, s_classname, tname}
{
init();
}
~pipeline() noexcept;
query_id insert(std::string_view) &;
void complete();
void flush();
void cancel();
[[nodiscard]] bool is_finished(query_id) const;
result retrieve(query_id qid)
{
return retrieve(m_queries.find(qid)).second;
}
std::pair<query_id, result> retrieve();
[[nodiscard]] bool empty() const noexcept { return std::empty(m_queries); }
int retain(int retain_max = 2) &;
void resume() &;
private:
struct PQXX_PRIVATE Query
{
explicit Query(std::string_view q) :
query{std::make_shared<std::string>(q)}
{}
std::shared_ptr<std::string> query;
result res;
};
using QueryMap = std::map<query_id, Query>;
void init();
void attach();
void detach();
static constexpr query_id qid_limit() noexcept
{
// Parenthesise this to work around an eternal Visual C++ problem:
// Without the extra parentheses, unless NOMINMAX is defined, the
// preprocessor will mistake this "max" for its annoying built-in macro
// of the same name.
return (std::numeric_limits<query_id>::max)();
}
PQXX_PRIVATE query_id generate_id();
bool have_pending() const noexcept
{
return m_issuedrange.second != m_issuedrange.first;
}
PQXX_PRIVATE void issue();
void set_error_at(query_id qid) noexcept
{
PQXX_UNLIKELY
if (qid < m_error)
m_error = qid;
}
[[noreturn]] PQXX_PRIVATE void internal_error(std::string const &err);
PQXX_PRIVATE bool obtain_result(bool expect_none = false);
PQXX_PRIVATE void obtain_dummy();
PQXX_PRIVATE void get_further_available_results();
PQXX_PRIVATE void check_end_results();
PQXX_PRIVATE void receive_if_available();
PQXX_PRIVATE void receive(pipeline::QueryMap::const_iterator stop);
std::pair<pipeline::query_id, result> retrieve(pipeline::QueryMap::iterator);
QueryMap m_queries;
std::pair<QueryMap::iterator, QueryMap::iterator> m_issuedrange;
int m_retain = 0;
int m_num_waiting = 0;
query_id m_q_id = 0;
bool m_dummy_pending = false;
query_id m_error = qid_limit();
internal::encoding_group m_encoding;
static constexpr std::string_view s_classname{"pipeline"};
};
} // namespace pqxx
#endif
pqxx::pipeline
Processes several queries in FIFO manner, optimized for high throughput.
Definition pipeline.hxx:51
pqxx::pipeline::pipeline
pipeline(transaction_base &t)
Start a pipeline.
Definition pipeline.hxx:63
pqxx::pipeline::pipeline
pipeline(transaction_base &t, std::string_view tname)
Start a pipeline. Assign it a name, for more helpful error messages.
Definition pipeline.hxx:68
pqxx::pipeline::query_id
long query_id
Identifying numbers for queries.
Definition pipeline.hxx:54
pqxx::result
Result set containing data returned by a query or command.
Definition result.hxx:92
pqxx::transaction_focus
Base class for things that monopolise a transaction's attention.
Definition transaction_focus.hxx:29
pqxx::transaction_base
Interface definition (and common code) for "transaction" classes.
Definition transaction_base.hxx:151
pqxx
The home of all libpqxx classes, functions, templates, etc.
Definition array.cxx:27