thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-21 09:10:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

dbms: probably better [#METR-10240].

Browse Source
This commit is contained in:
Alexey Milovidov 2014-03-07 08:04:02 +04:00
parent cd2e6663a5
commit 3d0b31dfed
4 changed files with 124 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
dbms/include/DB/IO/ReadBufferFromPocoSocket.h
View File

@ -1,10 +1,6 @@
#pragma once
#include <Poco/Net/Socket.h>
#include <Poco/Net/NetException.h>
#include <DB/Core/Exception.h>
#include <DB/Core/ErrorCodes.h>
#include <DB/IO/ReadBuffer.h>
#include <DB/IO/BufferWithOwnMemory.h>
@ -26,50 +22,12 @@ protected:
*/
Poco::Net::SocketAddress peer_address;
bool nextImpl()
{
ssize_t bytes_read = 0;
/// Добавляем в эксепшены более подробную информацию.
try
{
bytes_read = socket.impl()->receiveBytes(internal_buffer.begin(), internal_buffer.size());
}
catch (const Poco::Net::NetException & e)
{
throw Exception(e.displayText(), "while reading from socket (" + peer_address.toString() + ")", ErrorCodes::NETWORK_ERROR);
}
catch (const Poco::TimeoutException & e)
{
throw Exception("Timeout exceeded while reading from socket (" + peer_address.toString() + ")", ErrorCodes::SOCKET_TIMEOUT);
}
catch (const Poco::IOException & e)
{
throw Exception(e.displayText(), "while reading from socket (" + peer_address.toString() + ")", ErrorCodes::NETWORK_ERROR);
}
if (bytes_read < 0)
throw Exception("Cannot read from socket (" + peer_address.toString() + ")", ErrorCodes::CANNOT_READ_FROM_SOCKET);
if (bytes_read)
working_buffer.resize(bytes_read);
else
return false;
return true;
}
bool nextImpl();
public:
ReadBufferFromPocoSocket(Poco::Net::Socket & socket_, size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE)
: BufferWithOwnMemory<ReadBuffer>(buf_size), socket(socket_), peer_address(socket.peerAddress())
{
}
ReadBufferFromPocoSocket(Poco::Net::Socket & socket_, size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE);
bool poll(size_t timeout_microseconds)
{
return offset() != buffer().size() || socket.poll(timeout_microseconds, Poco::Net::Socket::SELECT_READ | Poco::Net::Socket::SELECT_ERROR);
}
bool poll(size_t timeout_microseconds);
};
}

55
dbms/include/DB/IO/WriteBufferFromPocoSocket.h
View File

@ -1,10 +1,6 @@
#pragma once
#include <Poco/Net/Socket.h>
#include <Poco/Net/NetException.h>
#include <DB/Core/Exception.h>
#include <DB/Core/ErrorCodes.h>
#include <DB/IO/WriteBuffer.h>
#include <DB/IO/BufferWithOwnMemory.h>
@ -27,57 +23,12 @@ protected:
Poco::Net::SocketAddress peer_address;
void nextImpl()
{
if (!offset())
return;
size_t bytes_written = 0;
while (bytes_written < offset())
{
ssize_t res = 0;
/// Добавляем в эксепшены более подробную информацию.
try
{
res = socket.impl()->sendBytes(working_buffer.begin() + bytes_written, offset() - bytes_written);
}
catch (const Poco::Net::NetException & e)
{
throw Exception(e.displayText() + "while writing to socket (" + peer_address.toString() + ")", ErrorCodes::NETWORK_ERROR);
}
catch (const Poco::TimeoutException & e)
{
throw Exception("Timeout exceeded while writing to socket (" + peer_address.toString() + ")", ErrorCodes::SOCKET_TIMEOUT);
}
catch (const Poco::IOException & e)
{
throw Exception(e.displayText(), "while reading from socket (" + peer_address.toString() + ")", ErrorCodes::NETWORK_ERROR);
}
if (res < 0)
throw Exception("Cannot write to socket (" + peer_address.toString() + ")", ErrorCodes::CANNOT_WRITE_TO_SOCKET);
bytes_written += res;
}
}
void nextImpl();
public:
WriteBufferFromPocoSocket(Poco::Net::Socket & socket_, size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE)
: BufferWithOwnMemory<WriteBuffer>(buf_size), socket(socket_), peer_address(socket.peerAddress())
{
}
WriteBufferFromPocoSocket(Poco::Net::Socket & socket_, size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE);
~WriteBufferFromPocoSocket()
{
try
{
next();
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
~WriteBufferFromPocoSocket();
};
}

55
dbms/src/IO/ReadBufferFromPocoSocket.cpp Normal file
View File

@ -0,0 +1,55 @@
#include <Poco/Net/NetException.h>
#include <DB/Core/Exception.h>
#include <DB/Core/ErrorCodes.h>
#include <DB/IO/ReadBufferFromPocoSocket.h>
namespace DB
{
bool ReadBufferFromPocoSocket::nextImpl()
{
ssize_t bytes_read = 0;
/// Добавляем в эксепшены более подробную информацию.
try
{
bytes_read = socket.impl()->receiveBytes(internal_buffer.begin(), internal_buffer.size());
}
catch (const Poco::Net::NetException & e)
{
throw Exception(e.displayText(), "while reading from socket (" + peer_address.toString() + ")", ErrorCodes::NETWORK_ERROR);
}
catch (const Poco::TimeoutException & e)
{
throw Exception("Timeout exceeded while reading from socket (" + peer_address.toString() + ")", ErrorCodes::SOCKET_TIMEOUT);
}
catch (const Poco::IOException & e)
{
throw Exception(e.displayText(), "while reading from socket (" + peer_address.toString() + ")", ErrorCodes::NETWORK_ERROR);
}
if (bytes_read < 0)
throw Exception("Cannot read from socket (" + peer_address.toString() + ")", ErrorCodes::CANNOT_READ_FROM_SOCKET);
if (bytes_read)
working_buffer.resize(bytes_read);
else
return false;
return true;
}
ReadBufferFromPocoSocket::ReadBufferFromPocoSocket(Poco::Net::Socket & socket_, size_t buf_size)
: BufferWithOwnMemory<ReadBuffer>(buf_size), socket(socket_), peer_address(socket.peerAddress())
{
}
bool ReadBufferFromPocoSocket::poll(size_t timeout_microseconds)
{
return offset() != buffer().size() || socket.poll(timeout_microseconds, Poco::Net::Socket::SELECT_READ | Poco::Net::Socket::SELECT_ERROR);
}
}

63
dbms/src/IO/WriteBufferFromPocoSocket.cpp Normal file
View File

@ -0,0 +1,63 @@
#include <Poco/Net/NetException.h>
#include <DB/Core/Exception.h>
#include <DB/Core/ErrorCodes.h>
#include <DB/IO/WriteBufferFromPocoSocket.h>
namespace DB
{
void WriteBufferFromPocoSocket::nextImpl()
{
if (!offset())
return;
size_t bytes_written = 0;
while (bytes_written < offset())
{
ssize_t res = 0;
/// Добавляем в эксепшены более подробную информацию.
try
{
res = socket.impl()->sendBytes(working_buffer.begin() + bytes_written, offset() - bytes_written);
}
catch (const Poco::Net::NetException & e)
{
throw Exception(e.displayText() + "while writing to socket (" + peer_address.toString() + ")", ErrorCodes::NETWORK_ERROR);
}
catch (const Poco::TimeoutException & e)
{
throw Exception("Timeout exceeded while writing to socket (" + peer_address.toString() + ")", ErrorCodes::SOCKET_TIMEOUT);
}
catch (const Poco::IOException & e)
{
throw Exception(e.displayText(), "while reading from socket (" + peer_address.toString() + ")", ErrorCodes::NETWORK_ERROR);
}
if (res < 0)
throw Exception("Cannot write to socket (" + peer_address.toString() + ")", ErrorCodes::CANNOT_WRITE_TO_SOCKET);
bytes_written += res;
}
}
WriteBufferFromPocoSocket::WriteBufferFromPocoSocket(Poco::Net::Socket & socket_, size_t buf_size)
: BufferWithOwnMemory<WriteBuffer>(buf_size), socket(socket_), peer_address(socket.peerAddress())
{
}
WriteBufferFromPocoSocket::~WriteBufferFromPocoSocket()
{
try
{
next();
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
}