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

Wait for I/O to finish after PQputCopyData

pull/1648/head
Marco Slot 2017-09-12 16:04:08 -07:00
parent f5361d52e7
commit d1befa4df9
1 changed files with 20 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
src/backend/distributed/connection/remote_commands.c
View File

@ -497,6 +497,7 @@ PutRemoteCopyData(MultiConnection *connection, const char *buffer, int nbytes)
{ {
PGconn *pgConn = connection->pgConn; PGconn *pgConn = connection->pgConn;
int copyState = 0; int copyState = 0;
bool allowInterrupts = true;
if (PQstatus(pgConn) != CONNECTION_OK) if (PQstatus(pgConn) != CONNECTION_OK)
{ {
@ -506,21 +507,22 @@ PutRemoteCopyData(MultiConnection *connection, const char *buffer, int nbytes)
Assert(PQisnonblocking(pgConn)); Assert(PQisnonblocking(pgConn));
copyState = PQputCopyData(pgConn, buffer, nbytes); copyState = PQputCopyData(pgConn, buffer, nbytes);
if (copyState == -1)
if (copyState == 1)
{
/* successful */
return true;
}
else if (copyState == -1)
{ {
return false; return false;
} }
else
{ /*
bool allowInterrupts = true; * PQputCopyData may have queued up part of the data even if it managed
* to send some of it succesfully. We provide back pressure by waiting
* until the socket is writable to prevent the internal libpq buffers
* from growing excessively.
*
* In the future, we could reduce the frequency of these pushbacks to
* achieve higher throughput.
*/
return FinishConnectionIO(connection, allowInterrupts); return FinishConnectionIO(connection, allowInterrupts);
}
} }
@ -535,6 +537,7 @@ PutRemoteCopyEnd(MultiConnection *connection, const char *errormsg)
{ {
PGconn *pgConn = connection->pgConn; PGconn *pgConn = connection->pgConn;
int copyState = 0; int copyState = 0;
bool allowInterrupts = true;
if (PQstatus(pgConn) != CONNECTION_OK) if (PQstatus(pgConn) != CONNECTION_OK)
{ {
@ -544,21 +547,14 @@ PutRemoteCopyEnd(MultiConnection *connection, const char *errormsg)
Assert(PQisnonblocking(pgConn)); Assert(PQisnonblocking(pgConn));
copyState = PQputCopyEnd(pgConn, errormsg); copyState = PQputCopyEnd(pgConn, errormsg);
if (copyState == -1)
if (copyState == 1)
{
/* successful */
return true;
}
else if (copyState == -1)
{ {
return false; return false;
} }
else
{ /* see PutRemoteCopyData() */
bool allowInterrupts = true;
return FinishConnectionIO(connection, allowInterrupts); return FinishConnectionIO(connection, allowInterrupts);
}
} }