Как правильно создать >=1000 потоков?

Question

[Макс Максимов]

Дано: есть база данных на PostgreSQL, в который есть 2 хранимые процедуры. Есть консолька на C# с драйвером Npgsql. В консольке надо создать 1000 потоков, каждый из которых имеет своё соединение и вызывает эти самые хранимки. Вызываю потоки и запрос следующим образом:

new Thread(() =>
                {
                    Thread.CurrentThread.IsBackground = true;
                    for (var i = 0; i < 500; i++) {
                        try {
                            var thread = new Thread(() =>
                            {
                                Console.WriteLine($"Running thread: {Thread.CurrentThread.ManagedThreadId}");
                                SaveByteA(connectionString);
                                var a = 2 + 4;
                            }) {IsBackground = true,Priority = ThreadPriority.AboveNormal};
                            thread.Start();
                        }
                        catch (Exception e) {
                            Console.WriteLine(e);
                            throw;
                        }
                    }

                    Console.WriteLine("Finished Thread #1.");

                })
                { IsBackground = true, Priority = ThreadPriority.Highest }.Start();


                new Thread(() =>
                {
                    Thread.CurrentThread.IsBackground = true;
                    for (var i = 0; i < 500; i++)
                    {
                        try {
                            var thread = new Thread(() =>
                                {
                                    Console.WriteLine($"Running  2 thread: {Thread.CurrentThread.ManagedThreadId}");
                                    SaveByteB(connectionString);

                                })
                                { IsBackground = true, Priority = ThreadPriority.AboveNormal };
                            thread.Start();
                        }
                        catch (Exception e) {
                            Console.WriteLine(e);
                            throw;
                        }
                    }
                    Console.WriteLine("Finished Thread #2.");
                }) { IsBackground = true, Priority = ThreadPriority.Highest}.Start();

Надо: Сделать столько запросов (то бишь потоков), чтобы одновременное количество соединённых клиентов с выполняющимися транзакциями было примерно ~1000.
Проблема :когда запускаю консольку с малым количеством потоков, то всё обрабатывается быстро -- БД очень мощная, поэтому результат в ~1000 соединении и обрабатываемых транзакции пропадает. А когда ставлю большое количество потоков, то получаю SocketException -- соединение было прервано насильно удалённым хостом. (Приём консколька жалуется на базу, а база на клиент. кто прав -- не пойму).

В чём моя проблема? Куда копать? Есть лучше варианты работы с потоками или варианты нагрузки базы так, как написано в условии?

Comments

[Владимир Дубровин]

А какой у вас max_connections на стороне сервера?

[devalone]

Правильно не создавать 1000 потоков, если конечно твой процессор не поддерживает столько аппаратно.

[Макс Максимов]

Владимир Дубровин, 1500 стоит, shared_buffers = 1024MB.

Answer 1

[alexs0ff]

Если у Вас компьютер не 1000 ядерный, вам нужны не CPU bound потоки, а IO bound потоки. Поищите асинхронные методы в вашем PostgesQl провайдере (скорее всего они с приставкой Sync)

Comments

[Макс Максимов]

Разве Thread, Task не одно и тоже с async/await?

[alexs0ff]

Task (async/await) - это асинхронное программирование, которое может быть многопоточным, но не обязано. Когда мы создаем таски, типа TaskFactory.Create, они управляются планировщиком, который как правило использует ThreadPool (правда, иногда можно создать и полноценный поток через LongRunning флаг).

Плюс у Вас проблемы с сокетами, это скорее всего проблема с максимальным количеством возможных подключений по сети, поищите есть флаг в AppConfig (погуглите "c# connectionManagement" или address = "*" maxconnection = "2" ), который может также влиять на данную проблему.

[eRKa]

Макс Максимов, согласен с alexs0ff. Нет смысла создавать столько потоков, в реальности они не будет работать одновременно и будет тратится куча ресурсов на переключения контекста. Для того, чтобы создать поток, уже нужно время и ресурсы, а на 8 ядрах 1000 потоков будут больше времени тратить, чтобы поделить между собой ресурсы процессора.
Во первых не стоит использовать Thread в чистом виде, лучше использовать пул потоков. А во вторых посмотрите в сторону TPL и его метода Parallel.ForEach. В его настройках можно указать и количество параллельных потоков.

[Макс Максимов]

alexs0ff, поменял всё в вид Task и async/await. Конфиг добавил 65к лимит. Так же получаю ошибки на сокет:
System.Net.Sockets.SocketException: 'Попытка установить соединение была безуспешной, т.к. от другого компьютера за требуемое время не получен нужный отклик, или было разорвано уже установленное соединение из-за неверного отклика уже подключенного компьютера 10.1.0.16:5432'

for (var i = 0; i < 500; i++) {
                        try {
                            var task = new Task(async () =>
                            {
                                await SaveByteA(connectionString);
                            });
                            task.Start();
                        }
                        catch (Exception e) {
                            Console.WriteLine(e);
                            throw;
                        }
                    }

Код метода:

private static async Task SaveByteA(NpgsqlConnectionStringBuilder connectionString) {
            using (var connection = new NpgsqlConnection(connectionString.ConnectionString)) {
                await connection.OpenAsync();

                using (var transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted)) {
                    var rawSql =
                        @"select data.save_bytea_inh(@i_paydate) from generate_series(1, 1, 1);";
                    using (var command = new NpgsqlCommand(rawSql, connection, transaction))
                    {
                        command.Parameters.AddWithValue("i_paydate", NpgsqlDbType.Timestamp, RandomDay());
                        command.Prepare();


                        await command.ExecuteScalarAsync();
                    };

                    
                    await transaction.CommitAsync();
                }

                Console.WriteLine("Finished working: A");
            }
        }

[alexs0ff]

Макс Максимов, Попробуйте установить локально postrgress и применить этот же код, если все будет хорошо - значит копайте в сторону сети или настроек сервера. Потому, как в данном коде я не вижу явных проблем с многопоточностью.

[Макс Максимов]

alexs0ff, хз каким образом, но получилось сделать так:
Main:

new Thread(() =>
                {
                    foreach (var fiscal in fiscalList)
                        new Thread(() => { DisplayResultAsync(fiscal); })
                            {IsBackground = true, Priority = ThreadPriority.AboveNormal}.Start();
                    ;
                }) {IsBackground = true, Priority = ThreadPriority.AboveNormal}.Start();

                new Thread(() =>
                {
                    foreach (var fiscal in fiscalList2)
                        new Thread(() => { DisplayResultAsync(fiscal); })
                            {IsBackground = true, Priority = ThreadPriority.AboveNormal}.Start();
                    ;
                }) {IsBackground = true, Priority = ThreadPriority.AboveNormal}.Start();

static async void DisplayResultAsync(Fiscal fiscal)
        {
            await fiscal.SaveByteAAsync();
        }

Вспомогательный класс:

public class Fiscal {
        private readonly string connectionString;

        public Fiscal(NpgsqlConnectionStringBuilder connectionString) {
            this.connectionString = connectionString.ConnectionString;
        }


        public async Task SaveByteAAsync() {
            using (var connection = new NpgsqlConnection(connectionString)) {
                var connectionTask = connection.OpenAsync();
                Task.WaitAll(connectionTask);
                if (connectionTask.IsFaulted) throw new Exception("Connection failure", connectionTask.Exception);

                Console.WriteLine("Connected! Delaying...");
                var delay = Task.Delay(400);
                Task.WaitAll(delay);

                using (var transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted)) {
                    var rawSql =
                        @"select data.save_bytea_inh(@i_paydate) from generate_series(1, 1, 1);";
                    using (var command = new NpgsqlCommand(rawSql, connection, transaction)) {
                        command.Parameters.AddWithValue("i_paydate", NpgsqlDbType.Timestamp, DateTime.Today);
                        command.Prepare();

                        await command.ExecuteScalarAsync();
                    }

                    await transaction.CommitAsync();
                }

                Console.WriteLine("Finished working: A");
            }
        }
    }

Спасибо за помощь! Плюс ещё помог ответ: https://stackoverflow.com/a/23140651

Answer 2

[profesor08]

Похоже на первый опыт ддоса своей бд. Распредели нагрузку так, чтоб число соединений в секунду не превышало лимита, на который настроена бд. И все получится.

Answer 3

[Дмитрий Донской]

Макс Максимов, посмотри сколько одновременных соединений разрешено у БД. Если мало, то увеличь.
https://stackoverflow.com/questions/30778015/how-t...

Так же почитай документацию на драйвер: www.npgsql.org/doc/connection-string-parameters.html
Тебя должно интересовать Pooling. Попробуй так же увеличь количество зарезервированных подключений.

Pooling=true;Min Pool Size=0;Max Pool Size=1000;

Помониторь на каком потоке происходит исключение, узнай его индекс. Добавь временную переменную или как угодно отследи его. Если вылетает на одном и том же или +-, то скорее всего смотри начало ответа).

Comments

[Макс Максимов]

С конфигурацией БД всё нормально. Максимальное количество соединении 1500, шардинг буфера 1 гб.
Мне нужно именно 1000 соединении, которые будут работать параллельно. Чтобы протестировать БД. В итоге получаю Socket.Exception. Смотрите коммент : Как правильно создать >=1000 потоков?