external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Avoid unnecessary calls to PQconsumeInput

release-8.3
Marco Slot 2019-07-21 01:52:44 +02:00 committed by Hanefi Onaldi
parent 0de8807aee
commit 03f7e7479d
No known key found for this signature in database
GPG Key ID: 95177DABDC09D1F5
1 changed files with 13 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/backend/distributed/executor/adaptive_executor.c
View File

@ -389,6 +389,9 @@ typedef struct WorkerSession
/* index in the wait event set */
int waitEventSetIndex;
/* events reported by the latest call to WaitEventSetWait */
int latestUnconsumedWaitEvents;
} WorkerSession;
@ -1713,6 +1716,7 @@ RunDistributedExecution(DistributedExecution *execution)
}
session = (WorkerSession *) event->user_data;
session->latestUnconsumedWaitEvents = event->events;
ConnectionStateMachine(session);
}
@ -2564,11 +2568,13 @@ CheckConnectionReady(WorkerSession *session)
waitFlags = waitFlags | WL_SOCKET_WRITEABLE;
}
/* if reading fails, there's not much we can do */
if (PQconsumeInput(connection->pgConn) == 0)
if ((session->latestUnconsumedWaitEvents & WL_SOCKET_READABLE) != 0)
{
connection->connectionState = MULTI_CONNECTION_LOST;
return false;
if (PQconsumeInput(connection->pgConn) == 0)
{
connection->connectionState = MULTI_CONNECTION_LOST;
return false;
}
}
if (!PQisBusy(connection->pgConn))
@ -2578,6 +2584,9 @@ CheckConnectionReady(WorkerSession *session)
UpdateConnectionWaitFlags(session, waitFlags);
/* don't consume input redundantly if we cycle back into CheckConnectionReady */
session->latestUnconsumedWaitEvents = 0;
return connectionReady;
}