CODEWITHSUNDEEP
c#multithreadingpostgresql
Stefan Orr
Stefan Orr

Reputation: 311

Connection failures to PostgreSQL when multi-threading. C#

I have a process that streams results from hundreds of queries on 5 different databases to a file on disk using a producer-consumer pattern. The queries and connections all work when run sequentially (or if I continuously retry them concurrently), however when threading them, some of the connections fail.

I am currently just adding failed attempts to a failed collection and then rerunning them recursively so they eventually all finish, but a lot of time is wasted with these failed connections.

Here is the relevant code I am using:

public void executeNodes(List<Node> nodeList, BlockingCollection<Dictionary<string, object>> producer, int retryAttempts)
{
    BlockingCollection<Node> retryNodes = new BlockingCollection<Node>();
    Parallel.ForEach(nodeList, new ParallelOptions() { MaxDegreeOfParallelism = 10 }, node =>
    {
        NpgsqlConnection conn = new NpgsqlConnection(node.ConnectionString);
        try
        {
            conn.Open();
            NpgsqlCommand npgQuery = new NpgsqlCommand(node.Query, conn);
            NpgsqlDataReader reader = npgQuery.ExecuteReader();
            while (reader.Read())
            {
                Dictionary<string, object> row = new Dictionary<string, object>();
                for (int i = 0; i < reader.FieldCount; i++)
                {
                    row[reader.GetName(i)] = reader.GetValue(i);
                }
                producer.Add(row);
            }
            conn.Close();
            //Console.WriteLine("Success");
        }
        catch (Exception e)
        {
            //Console.WriteLine("Failed");
            retryNodes.Add(node);
        }
    });
    if (retryNodes.Count == 0 || retryAttempts == 0) producer.CompleteAdding();
    else executeNodes(retryNodes.ToList<Node>(), producer, retryAttempts - 1);
}

Where a node is simply:

public class Node
{
    public string ConnectionString;
    public string Query;

    public Node(string connectionString, string query)
    {
        ConnectionString = connectionString;
        Query = query;
    }
}

This process will run successfully and correctly, but it is annoying that I am getting connection failures. Any help would be greatly appreciated.

Notes:

I have tried adding:

ServicePointManager.DefaultConnectionLimit = 20;

I have tried adding registry entries specified in this page

The connection failures are not due to the connection pool being reached.

The errors I am seeing are:

Unable to read data from the transport connection: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond.

Upvotes: 1

Views: 1067

Answers (1)

Stefan Orr
Stefan Orr

Reputation: 311

So my problem was with connection pooling. Once I set it to false in the connection string, everything worked fine.

Upvotes: 1

Related Questions