thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-12-12 17:32:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix bug with buffered read

Browse Source
This commit is contained in:
alesapin 2020-11-26 15:26:41 +03:00
parent 0b079cdeb8
commit fc0bca1ff2
1 changed files with 15 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
src/Server/TestKeeperTCPHandler.cpp
View File

@ -89,6 +89,7 @@ struct SocketInterruptablePollWrapper
#if defined(POCO_HAVE_FD_EPOLL) #if defined(POCO_HAVE_FD_EPOLL)
int rc; int rc;
epoll_event evout[2]; epoll_event evout[2];
memset(evout, 0, sizeof(evout));
do do
{ {
Poco::Timestamp start; Poco::Timestamp start;
@ -284,16 +285,20 @@ void TestKeeperTCPHandler::runImpl()
auto state = poll_wrapper->poll(session_timeout); auto state = poll_wrapper->poll(session_timeout);
if (state & SocketInterruptablePollWrapper::HAS_REQUEST) if (state & SocketInterruptablePollWrapper::HAS_REQUEST)
{ {
auto received_op = receiveRequest(); do
if (received_op == Coordination::OpNum::Close)
{ {
LOG_DEBUG(log, "Received close request for session #{}", session_id); auto received_op = receiveRequest();
break; if (received_op == Coordination::OpNum::Close)
} {
else if (received_op == Coordination::OpNum::Heartbeat) LOG_DEBUG(log, "Received close request for session #{}", session_id);
{ break;
session_stopwatch.restart(); }
} else if (received_op == Coordination::OpNum::Heartbeat)
{
LOG_TRACE(log, "Received heartbeat for session #{}", session_id);
session_stopwatch.restart();
}
} while (in->available());
} }
if (state & SocketInterruptablePollWrapper::HAS_RESPONSE) if (state & SocketInterruptablePollWrapper::HAS_RESPONSE)
@ -360,7 +365,7 @@ Coordination::OpNum TestKeeperTCPHandler::receiveRequest()
Coordination::ZooKeeperRequestPtr request = Coordination::ZooKeeperRequestFactory::instance().get(opnum); Coordination::ZooKeeperRequestPtr request = Coordination::ZooKeeperRequestFactory::instance().get(opnum);
request->xid = xid; request->xid = xid;
request->readImpl(*in); request->readImpl(*in);
int response_fd = poll_wrapper->getResponseFD(); int response_fd = poll_wrapper->getResponseFD();
if (opnum != Coordination::OpNum::Close) if (opnum != Coordination::OpNum::Close)
{ {
auto promise = std::make_shared<std::promise<Coordination::ZooKeeperResponsePtr>>(); auto promise = std::make_shared<std::promise<Coordination::ZooKeeperResponsePtr>>();