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

Set initial pool size to cached connection count

pull/2848/head
Marco Slot 2019-07-21 14:50:43 +02:00
parent 4c0c33365e
commit 4444d92dbc
3 changed files with 1 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -333,7 +333,6 @@ StartNodeUserDatabaseConnection(uint32 flags, const char *hostname, int32 port,
connection = StartConnectionEstablishment(&key); connection = StartConnectionEstablishment(&key);
dlist_push_tail(entry->connections, &connection->connectionNode); dlist_push_tail(entry->connections, &connection->connectionNode);
entry->connectionCount++;
ResetShardPlacementAssociation(connection); ResetShardPlacementAssociation(connection);
@ -1065,8 +1064,6 @@ AfterXactHostConnectionHandling(ConnectionHashEntry *entry, bool isCommit)
cachedConnectionCount++; cachedConnectionCount++;
} }
} }
entry->connectionCount = cachedConnectionCount;
} }
@ -1172,29 +1169,3 @@ TrimLogLevel(const char *message)
return chompedMessage + n; return chompedMessage + n;
} }
/*
* NodeConnectionCount gets the number of connections to the given node
* for the current username and database.
*/
int
NodeConnectionCount(char *hostname, int port)
{
ConnectionHashKey key;
ConnectionHashEntry *entry = NULL;
bool found = false;
strlcpy(key.hostname, hostname, MAX_NODE_LENGTH);
key.port = port;
strlcpy(key.user, CurrentUserName(), NAMEDATALEN);
strlcpy(key.database, CurrentDatabaseName(), NAMEDATALEN);
entry = (ConnectionHashEntry *) hash_search(ConnectionHash, &key, HASH_FIND, &found);
if (!found)
{
return 0;
}
return entry->connectionCount;
}

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

@ -1472,8 +1472,7 @@ FindOrCreateWorkerPool(DistributedExecution *execution, WorkerNode *workerNode)
workerPool->distributedExecution = execution; workerPool->distributedExecution = execution;
/* "open" connections aggressively when there are cached connections */ /* "open" connections aggressively when there are cached connections */
nodeConnectionCount = NodeConnectionCount(workerNode->workerName, nodeConnectionCount = MaxCachedConnectionsPerWorker;
workerNode->workerPort);
workerPool->maxNewConnectionsPerCycle = Max(1, nodeConnectionCount); workerPool->maxNewConnectionsPerCycle = Max(1, nodeConnectionCount);
dlist_init(&workerPool->pendingTaskQueue); dlist_init(&workerPool->pendingTaskQueue);

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

@ -128,7 +128,6 @@ typedef struct ConnectionHashEntry
{ {
ConnectionHashKey key; ConnectionHashKey key;
dlist_head *connections; dlist_head *connections;
int connectionCount;
} ConnectionHashEntry; } ConnectionHashEntry;
/* hash entry for cached connection parameters */ /* hash entry for cached connection parameters */
@ -191,7 +190,6 @@ extern MultiConnection * StartNodeUserDatabaseConnection(uint32 flags,
extern void CloseNodeConnectionsAfterTransaction(char *nodeName, int nodePort); extern void CloseNodeConnectionsAfterTransaction(char *nodeName, int nodePort);
extern void CloseConnection(MultiConnection *connection); extern void CloseConnection(MultiConnection *connection);
extern void ShutdownConnection(MultiConnection *connection); extern void ShutdownConnection(MultiConnection *connection);
extern int NodeConnectionCount(char *nodeName, int nodePort);
/* dealing with a connection */ /* dealing with a connection */
extern void FinishConnectionListEstablishment(List *multiConnectionList); extern void FinishConnectionListEstablishment(List *multiConnectionList);