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

Refactor CheckShardPlacements() and improve support for node removal 

This commit refactors CheckShardPlacements() so that it only considers
modifyingConnection. Also, it skips nodes which are removed from the
cluster.
pull/1122/head
Onder Kalaci 2017-01-12 03:16:56 +02:00
parent dbaf7f0e7e
commit aed5f817fa
4 changed files with 21 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/backend/distributed/connection/connection_management.c
View File

@ -307,11 +307,11 @@ GetConnectionFromPGconn(struct pg_conn *pqConn)
/*
* CloseNodeConnections closes all the connections to a particular node.
* This is mainly used when a worker leaves the cluster
* CloseNodeConnectionsAfterTransaction sets the sessionLifespan flag of the connections
* to a particular node as false. This is mainly used when a worker leaves the cluster.
*/
void
CloseNodeConnections(char *nodeName, int nodePort)
CloseNodeConnectionsAfterTransaction(char *nodeName, int nodePort)
{
HASH_SEQ_STATUS status;
ConnectionHashEntry *entry;
@ -319,21 +319,21 @@ CloseNodeConnections(char *nodeName, int nodePort)
hash_seq_init(&status, ConnectionHash);
while ((entry = (ConnectionHashEntry *) hash_seq_search(&status)) != 0)
{
dlist_head *connections = entry->connections;
dlist_iter iter;
dlist_head *connections = NULL;
if (strcmp(entry->key.hostname, nodeName) != 0 || entry->key.port != nodePort)
{
continue;
}
while (!dlist_is_empty(connections))
connections = entry->connections;
dlist_foreach(iter, connections)
{
dlist_node *currentNode = dlist_pop_head_node(connections);
MultiConnection *connection =
dlist_container(MultiConnection, connectionNode, currentNode);
dlist_container(MultiConnection, connectionNode, iter.cur);
CloseConnection(connection);
connection->sessionLifespan = false;
}
}
}

27
src/backend/distributed/connection/placement_connection.c
View File

@ -546,35 +546,20 @@ CheckShardPlacements(ConnectionShardHashEntry *shardEntry,
{
ConnectionPlacementHashEntry *placementEntry =
dlist_container(ConnectionPlacementHashEntry, shardNode, placementIter.cur);
dlist_iter referenceIter;
ConnectionReference *modifyingConnection = placementEntry->modifyingConnection;
MultiConnection *connection = NULL;
dlist_foreach(referenceIter, &placementEntry->connectionReferences)
{
ConnectionReference *reference =
dlist_container(ConnectionReference, placementNode, referenceIter.cur);
MultiConnection *connection = reference->connection;
/*
* If neither DDL nor DML were executed, there's no need for
* invalidation.
*/
if (!reference->hadDDL && !reference->hadDML)
/* we only consider shards that are modified */
if (modifyingConnection == NULL)
{
continue;
}
/*
* Failed if connection was closed, or remote transaction was
* unsuccessful.
*/
connection = modifyingConnection->connection;
if (!connection || connection->remoteTransaction.transactionFailed)
{
placementEntry->failed = true;
}
}
if (placementEntry->failed)
{
failures++;
}
else

4
src/backend/distributed/utils/node_metadata.c
View File

@ -381,8 +381,8 @@ RemoveNodeFromCluster(char *nodeName, int32 nodePort, bool forceRemove)
nodeDeleteCommand = NodeDeleteCommand(workerNode->nodeId);
/* make sure we don't have any open connections */
CloseNodeConnections(nodeName, nodePort);
/* make sure we don't have any lingering session lifespan connections */
CloseNodeConnectionsAfterTransaction(nodeName, nodePort);
SendCommandToWorkers(WORKERS_WITH_METADATA, nodeDeleteCommand);
}

2
src/include/distributed/connection_management.h
View File

@ -131,7 +131,7 @@ extern MultiConnection * StartNodeUserDatabaseConnection(uint32 flags,
const char *user,
const char *database);
extern MultiConnection * GetConnectionFromPGconn(struct pg_conn *pqConn);
extern void CloseNodeConnections(char *nodeName, int nodePort);
extern void CloseNodeConnectionsAfterTransaction(char *nodeName, int nodePort);
extern void CloseConnection(MultiConnection *connection);
extern void CloseConnectionByPGconn(struct pg_conn *pqConn);