pipeline.hxx

Go to the documentation of this file.

/* 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-2026, 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_PIPELINE_HXX
#define PQXX_PIPELINE_HXX
 
#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 final : 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, sl loc = sl::current()) :
          transaction_focus{t, s_classname}, m_created_loc{loc}
  {
    init(loc);
  }
  pipeline(
    transaction_base &t, std::string_view tname, sl loc = sl::current()) :
          transaction_focus{t, s_classname, tname}, m_created_loc{loc}
  {
    init(loc);
  }
 
  ~pipeline() noexcept;
 
 
  query_id insert(std::string_view, sl = sl::current()) &;
 
 
  void complete(sl = sl::current());
 
 
  void flush(sl = sl::current());
 
 
  void cancel(sl = sl::current());
 
  [[nodiscard]] bool is_finished(query_id) const;
 
 
  result retrieve(query_id qid, sl loc = sl::current())
  {
    return retrieve(m_queries.find(qid), loc).second;
  }
 
 
  std::pair<query_id, result> retrieve(sl = sl::current());
 
  [[nodiscard]] bool empty() const noexcept { return std::empty(m_queries); }
 
 
  int retain(int retain_max = 2) &;
 
  void resume(sl loc = sl::current()) &;
 
private:
  struct PQXX_PRIVATE Query final
  {
    explicit Query(std::string_view q) :
            query{std::make_shared<std::string>(q)}
    {}
 
    // NOLINTNEXTLINE(misc-non-private-member-variables-in-classes)
    std::shared_ptr<std::string> query;
    // NOLINTNEXTLINE(misc-non-private-member-variables-in-classes)
    result res;
  };
 
  using QueryMap = std::map<query_id, Query>;
 
  void init(sl);
  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();
 
  [[nodiscard]] bool have_pending() const noexcept
  {
    return m_issuedrange.second != m_issuedrange.first;
  }
 
  PQXX_PRIVATE void issue(sl);
 
  void set_error_at(query_id qid) noexcept
  {
    if (qid < m_error) [[unlikely]]
      m_error = qid;
  }
 
  [[noreturn]] PQXX_PRIVATE void internal_error(std::string const &err, sl);
 
  PQXX_PRIVATE bool obtain_result(bool expect_none, sl);
 
  PQXX_PRIVATE void obtain_dummy(sl);
  PQXX_PRIVATE void get_further_available_results(sl);
  PQXX_PRIVATE void check_end_results();
 
  PQXX_PRIVATE void receive_if_available(sl loc);
 
  PQXX_PRIVATE void receive(pipeline::QueryMap::const_iterator stop, sl);
  std::pair<pipeline::query_id, result>
    retrieve(pipeline::QueryMap::iterator, sl);
 
  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();
 
 
  encoding_group m_encoding{encoding_group::unknown};
 
  sl m_created_loc;
 
  static constexpr std::string_view s_classname{"pipeline"};
};
} // namespace pqxx
#endif