1 | #include <Poco/Net/NetException.h> |
2 | |
3 | #include <IO/ReadBufferFromPocoSocket.h> |
4 | #include <Common/Exception.h> |
5 | #include <Common/NetException.h> |
6 | #include <Common/Stopwatch.h> |
7 | |
8 | |
9 | namespace ProfileEvents |
10 | { |
11 | extern const Event NetworkReceiveElapsedMicroseconds; |
12 | } |
13 | |
14 | |
15 | namespace DB |
16 | { |
17 | |
18 | namespace ErrorCodes |
19 | { |
20 | extern const int NETWORK_ERROR; |
21 | extern const int SOCKET_TIMEOUT; |
22 | extern const int CANNOT_READ_FROM_SOCKET; |
23 | } |
24 | |
25 | |
26 | bool ReadBufferFromPocoSocket::nextImpl() |
27 | { |
28 | ssize_t bytes_read = 0; |
29 | Stopwatch watch; |
30 | |
31 | /// Add more details to exceptions. |
32 | try |
33 | { |
34 | bytes_read = socket.impl()->receiveBytes(internal_buffer.begin(), internal_buffer.size()); |
35 | } |
36 | catch (const Poco::Net::NetException & e) |
37 | { |
38 | throw NetException(e.displayText() + ", while reading from socket (" + peer_address.toString() + ")" , ErrorCodes::NETWORK_ERROR); |
39 | } |
40 | catch (const Poco::TimeoutException &) |
41 | { |
42 | throw NetException("Timeout exceeded while reading from socket (" + peer_address.toString() + ")" , ErrorCodes::SOCKET_TIMEOUT); |
43 | } |
44 | catch (const Poco::IOException & e) |
45 | { |
46 | throw NetException(e.displayText() + ", while reading from socket (" + peer_address.toString() + ")" , ErrorCodes::NETWORK_ERROR); |
47 | } |
48 | |
49 | if (bytes_read < 0) |
50 | throw NetException("Cannot read from socket (" + peer_address.toString() + ")" , ErrorCodes::CANNOT_READ_FROM_SOCKET); |
51 | |
52 | /// NOTE: it is quite inaccurate on high loads since the thread could be replaced by another one |
53 | ProfileEvents::increment(ProfileEvents::NetworkReceiveElapsedMicroseconds, watch.elapsedMicroseconds()); |
54 | |
55 | if (bytes_read) |
56 | working_buffer.resize(bytes_read); |
57 | else |
58 | return false; |
59 | |
60 | return true; |
61 | } |
62 | |
63 | ReadBufferFromPocoSocket::ReadBufferFromPocoSocket(Poco::Net::Socket & socket_, size_t buf_size) |
64 | : BufferWithOwnMemory<ReadBuffer>(buf_size), socket(socket_), peer_address(socket.peerAddress()) |
65 | { |
66 | } |
67 | |
68 | bool ReadBufferFromPocoSocket::poll(size_t timeout_microseconds) |
69 | { |
70 | return offset() != buffer().size() || socket.poll(timeout_microseconds, Poco::Net::Socket::SELECT_READ | Poco::Net::Socket::SELECT_ERROR); |
71 | } |
72 | |
73 | } |
74 | |