ClickHouse/dbms/include/DB/Interpreters/ProcessList.h
Vitaliy Lyudvichenko f8eb9028fd Refined query_log, processes and merges metrics.
Also, dbms/CMakeLists.txt was rewritten.
2016-12-09 13:11:25 +03:00

236 lines
6.0 KiB
C++

#pragma once
#include <map>
#include <list>
#include <memory>
#include <mutex>
#include <Poco/Condition.h>
#include <Poco/Net/IPAddress.h>
#include <DB/Common/Stopwatch.h>
#include <DB/Core/Defines.h>
#include <DB/Core/Progress.h>
#include <DB/Common/Exception.h>
#include <DB/Common/MemoryTracker.h>
#include <DB/IO/WriteHelpers.h>
#include <DB/Interpreters/QueryPriorities.h>
#include <DB/Interpreters/ClientInfo.h>
#include <DB/Storages/IStorage.h>
#include <DB/Common/CurrentMetrics.h>
namespace CurrentMetrics
{
extern const Metric Query;
}
namespace DB
{
/** List of currently executing queries.
* Also implements limit on their number.
*/
/** Information of process list element.
* To output in SHOW PROCESSLIST query. Does not contain any complex objects, that do something on copy or destructor.
*/
struct ProcessInfo
{
String query;
double elapsed_seconds;
size_t read_rows;
size_t read_bytes;
size_t total_rows;
size_t written_rows;
size_t written_bytes;
Int64 memory_usage;
ClientInfo client_info;
};
/// Query and information about its execution.
struct ProcessListElement
{
String query;
ClientInfo client_info;
Stopwatch watch;
/// Progress of input stream
Progress progress_in;
/// Progress of output stream
Progress progress_out;
MemoryTracker memory_tracker;
QueryPriorities::Handle priority_handle;
CurrentMetrics::Increment num_queries {CurrentMetrics::Query};
bool is_cancelled = false;
/// Temporary tables could be registered here. Modify under mutex.
Tables temporary_tables;
ProcessListElement(
const String & query_,
const ClientInfo & client_info_,
size_t max_memory_usage,
double memory_tracker_fault_probability,
QueryPriorities::Handle && priority_handle_)
: query(query_), client_info(client_info_), memory_tracker(max_memory_usage),
priority_handle(std::move(priority_handle_))
{
memory_tracker.setDescription("(for query)");
current_memory_tracker = &memory_tracker;
if (memory_tracker_fault_probability)
memory_tracker.setFaultProbability(memory_tracker_fault_probability);
}
~ProcessListElement()
{
current_memory_tracker = nullptr;
}
bool updateProgressIn(const Progress & value)
{
progress_in.incrementPiecewiseAtomically(value);
if (priority_handle)
priority_handle->waitIfNeed(std::chrono::seconds(1)); /// NOTE Could make timeout customizable.
return !is_cancelled;
}
bool updateProgressOut(const Progress & value)
{
progress_out.incrementPiecewiseAtomically(value);
return !is_cancelled;
}
ProcessInfo getInfo() const
{
ProcessInfo res;
res.query = query;
res.client_info = client_info;
res.elapsed_seconds = watch.elapsedSeconds();
res.read_rows = progress_in.rows;
res.read_bytes = progress_in.bytes;
res.total_rows = progress_in.total_rows;
res.written_rows = progress_out.rows;
res.written_bytes = progress_out.bytes;
res.memory_usage = memory_tracker.get();
return res;
}
};
/// Data about queries for one user.
struct ProcessListForUser
{
/// Query_id -> ProcessListElement *
using QueryToElement = std::unordered_map<String, ProcessListElement *>;
QueryToElement queries;
/// Limit and counter for memory of all simultaneously running queries of single user.
MemoryTracker user_memory_tracker;
};
class ProcessList;
/// Keeps iterator to process list and removes element in destructor.
class ProcessListEntry
{
private:
using Container = std::list<ProcessListElement>;
ProcessList & parent;
Container::iterator it;
public:
ProcessListEntry(ProcessList & parent_, Container::iterator it_)
: parent(parent_), it(it_) {}
~ProcessListEntry();
ProcessListElement * operator->() { return &*it; }
const ProcessListElement * operator->() const { return &*it; }
ProcessListElement & get() { return *it; }
const ProcessListElement & get() const { return *it; }
};
class ProcessList
{
friend class ProcessListEntry;
public:
using Element = ProcessListElement;
using Entry = ProcessListEntry;
/// list, for iterators not to invalidate. NOTE: could replace with cyclic buffer, but not worth.
using Container = std::list<Element>;
using Info = std::vector<ProcessInfo>;
/// User -> queries
using UserToQueries = std::unordered_map<String, ProcessListForUser>;
private:
mutable std::mutex mutex;
mutable Poco::Condition have_space; /// Number of currently running queries has become less than maximum.
Container cont;
size_t cur_size; /// In C++03 or C++11 and old ABI, std::list::size is not O(1).
size_t max_size; /// 0 means no limit. Otherwise, when limit exceeded, an exception is thrown.
UserToQueries user_to_queries;
QueryPriorities priorities;
/// Limit and counter for memory of all simultaneously running queries.
MemoryTracker total_memory_tracker;
public:
ProcessList(size_t max_size_ = 0) : cur_size(0), max_size(max_size_) {}
using EntryPtr = std::shared_ptr<ProcessListEntry>;
/** Register running query. Returns refcounted object, that will remove element from list in destructor.
* If too much running queries - wait for not more than specified (see settings) amount of time.
* If timeout is passed - throw an exception.
*/
EntryPtr insert(const String & query_, const ClientInfo & client_info, const Settings & settings);
/// Number of currently executing queries.
size_t size() const { return cur_size; }
/// Get current state of process list.
Info getInfo() const
{
std::lock_guard<std::mutex> lock(mutex);
Info res;
res.reserve(cur_size);
for (const auto & elem : cont)
res.emplace_back(elem.getInfo());
return res;
}
void setMaxSize(size_t max_size_)
{
std::lock_guard<std::mutex> lock(mutex);
max_size = max_size_;
}
/// Register temporary table. Then it is accessible by query_id and name.
void addTemporaryTable(ProcessListElement & elem, const String & table_name, StoragePtr storage);
/// Find temporary table by query_id and name. NOTE: doesn't work fine if there are many queries with same query_id.
StoragePtr tryGetTemporaryTable(const String & query_id, const String & table_name) const;
};
}