ClickHouse/dbms/include/DB/Interpreters/InterserverIOHandler.h

157 lines
4.2 KiB
C++
Raw Normal View History

2014-03-21 13:42:14 +00:00
#pragma once
#include <DB/IO/ReadBuffer.h>
#include <DB/IO/WriteBuffer.h>
2016-01-28 01:00:27 +00:00
#include <DB/IO/ReadBufferFromString.h>
#include <DB/IO/ReadHelpers.h>
#include <DB/IO/WriteBufferFromString.h>
#include <DB/IO/WriteHelpers.h>
2014-03-21 13:42:14 +00:00
#include <DB/Core/Types.h>
#include <map>
#include <atomic>
2014-03-21 13:42:14 +00:00
#include <Poco/Net/HTMLForm.h>
namespace DB
{
namespace ErrorCodes
{
extern const int DUPLICATE_INTERSERVER_IO_ENDPOINT;
extern const int NO_SUCH_INTERSERVER_IO_ENDPOINT;
}
2016-01-28 01:00:27 +00:00
/** Местонахождение сервиса.
*/
struct InterserverIOEndpointLocation
{
public:
InterserverIOEndpointLocation(const std::string & name_, const std::string & host_, UInt16 port_)
: name(name_), host(host_), port(port_)
{
}
/// Создаёт местонахождение на основе его сериализованного представления.
InterserverIOEndpointLocation(const std::string & serialized_location)
{
ReadBufferFromString buf(serialized_location);
readBinary(name, buf);
readBinary(host, buf);
readBinary(port, buf);
assertEOF(buf);
}
/// Сериализует местонахождение.
std::string toString() const
{
std::string serialized_location;
WriteBufferFromString buf(serialized_location);
writeBinary(name, buf);
writeBinary(host, buf);
writeBinary(port, buf);
buf.next();
return serialized_location;
}
public:
std::string name;
std::string host;
UInt16 port;
};
2014-03-21 13:42:14 +00:00
/** Обработчик запросов от других серверов.
*/
class InterserverIOEndpoint
{
public:
2016-01-28 01:00:27 +00:00
virtual std::string getId(const std::string & path) const = 0;
2016-03-01 17:47:53 +00:00
virtual void processQuery(const Poco::Net::HTMLForm & params, ReadBuffer & body, WriteBuffer & out) = 0;
2014-03-21 13:42:14 +00:00
virtual ~InterserverIOEndpoint() {}
void cancel() { is_cancelled = true; }
protected:
/// Нужно остановить передачу данных.
std::atomic<bool> is_cancelled {false};
2014-03-21 13:42:14 +00:00
};
using InterserverIOEndpointPtr = std::shared_ptr<InterserverIOEndpoint>;
2014-03-21 13:42:14 +00:00
/** Сюда можно зарегистрировать сервис, обрататывающий запросы от других серверов.
2014-03-22 14:44:44 +00:00
* Используется для передачи кусков в ReplicatedMergeTree.
2014-03-21 13:42:14 +00:00
*/
class InterserverIOHandler
{
public:
void addEndpoint(const String & name, InterserverIOEndpointPtr endpoint)
{
std::lock_guard<std::mutex> lock(mutex);
2014-03-21 13:42:14 +00:00
if (endpoint_map.count(name))
throw Exception("Duplicate interserver IO endpoint: " + name, ErrorCodes::DUPLICATE_INTERSERVER_IO_ENDPOINT);
endpoint_map[name] = endpoint;
}
void removeEndpoint(const String & name)
{
std::lock_guard<std::mutex> lock(mutex);
2014-03-21 13:42:14 +00:00
if (!endpoint_map.count(name))
throw Exception("No interserver IO endpoint named " + name, ErrorCodes::NO_SUCH_INTERSERVER_IO_ENDPOINT);
endpoint_map.erase(name);
}
InterserverIOEndpointPtr getEndpoint(const String & name)
{
std::lock_guard<std::mutex> lock(mutex);
2014-03-21 13:42:14 +00:00
if (!endpoint_map.count(name))
throw Exception("No interserver IO endpoint named " + name, ErrorCodes::NO_SUCH_INTERSERVER_IO_ENDPOINT);
return endpoint_map[name];
}
private:
using EndpointMap = std::map<String, InterserverIOEndpointPtr>;
2014-03-21 13:42:14 +00:00
EndpointMap endpoint_map;
std::mutex mutex;
2014-03-21 13:42:14 +00:00
};
2014-03-22 14:44:44 +00:00
/// В конструкторе вызывает addEndpoint, в деструкторе - removeEndpoint.
class InterserverIOEndpointHolder
{
public:
2014-04-02 07:59:43 +00:00
InterserverIOEndpointHolder(const String & name_, InterserverIOEndpointPtr endpoint_, InterserverIOHandler & handler_)
: name(name_), endpoint(endpoint_), handler(handler_)
2014-03-22 14:44:44 +00:00
{
handler.addEndpoint(name, endpoint);
}
2014-04-02 07:59:43 +00:00
InterserverIOEndpointPtr getEndpoint()
{
return endpoint;
}
2014-03-22 14:44:44 +00:00
~InterserverIOEndpointHolder()
{
try
{
handler.removeEndpoint(name);
/// После уничтожения объекта, endpoint ещё может жить, так как владение им захватывается на время обработки запроса,
/// см. InterserverIOHTTPHandler.cpp
2014-03-22 14:44:44 +00:00
}
catch (...)
{
tryLogCurrentException("~InterserverIOEndpointHolder");
}
}
void cancel() { endpoint->cancel(); }
2014-03-22 14:44:44 +00:00
private:
String name;
2014-04-02 07:59:43 +00:00
InterserverIOEndpointPtr endpoint;
2014-03-22 14:44:44 +00:00
InterserverIOHandler & handler;
};
using InterserverIOEndpointHolderPtr = std::shared_ptr<InterserverIOEndpointHolder>;
2014-03-22 14:44:44 +00:00
2014-03-21 13:42:14 +00:00
}