C # Задачи, которые работают на разных независимых объектах, ошибки синхронизации все еще возникают, почему?

Question

Моя программа выполняет задачи в группах из n задач одновременно.Каждая задача записывает данные в собственный объект Queue<string>, предоставленный индексом в Queue<string> в List<Queue<string>> очередях.задачи не разделяют данные или очереди, но я все еще получаю ошибки синхронизации.Я знаю, что структуры данных не являются потокобезопасными, я не понимаю, почему они должны быть и почему я получаю ошибки, поскольку каждый Task имеет свою собственную структуру данных, что может вызвать ошибки?

Вот простой код для демонстрации:

class Program
{
    static int j = 0;
    List<Queue<string>> queueList = new List<Queue<string>>();

    public void StartTasts(int n)
    {
        for (int i = 0; i < n; i++)
            queueList.Add(new Queue<string>());

        List<Task> tsk = new List<Task>();
        for (int TaskGroup = 0; TaskGroup < 10; TaskGroup++)
        {   //10 groups of task
            //each group has 'n' tasks working in parallel
            for (int i = 0; i < n; i++)
            {
                //each task gets its own and independent queue from the list
                tsk.Add(Task.Factory.StartNew(() =>
                {
                    DoWork(j % n);
                }));
                j++;
            }
            //waiting for each task group to finish
            foreach (Task t in tsk)
                t.Wait();
            //after they all finished working with the queues, clear queues
            //making them ready for the nest task group
            foreach (Queue<string> q in queueList)
                q.Clear();
        }
    }

    public void DoWork(int queue)
    {
        //demonstration of generating strings 
        //and put them in the correct queue
        for (int k = 0; k < 10000; k++)
            queueList[queue].Enqueue(k + "");
    }


    static void Main(string[] args)
    {
        new Program().StartTasts(10);
    }

}

эта программа генерирует некоторые ошибки, такие как:

System.ArgumentException: 'Массив назначения был недостаточно длинным.Проверьте destIndex и длину, а также нижние границы массива. '

System.IndexOutOfRangeException:' Индекс находился за пределами массива. '(в очереди)

System.AggregateException: произошла одна или несколько ошибок.---> System.ArgumentException: исходный массив был недостаточно длинным.Проверьте srcIndex и длину, а также нижние границы массива.

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

Answer 1

Проблема в нормальных проблемах закрытия переменных.Поскольку все задачи совместно используют один и тот же экземпляр переменной j, все они будут иметь одинаковое значение, скорее всего, происходит то, что ваш цикл запускает 10 задач очень быстро, но прежде чем любая из них сможет добраться до j % n, значение j имеетуже становится 10.

Создайте локальную копию k, объявленную в рамках цикла for, и это должно решить вашу проблему.

public void StartTasts(int n)
{
    for (int i = 0; i < n; i++)
        queueList.Add(new Queue<string>());

    List<Task> tsk = new List<Task>();
    for (int TaskGroup = 0; TaskGroup < 10; TaskGroup++)
    {   //10 groups of task
        //each group has 'n' tasks working in parallel
        for (int i = 0; i < n; i++)
        {
            int k = j; // `int k = i;` would work here too and give you the same results.

            tsk.Add(Task.Factory.StartNew(() =>
            {
                DoWork(k % n);
            }));
            j++;
        }
        //waiting for each task group to finish
        foreach (Task t in tsk)
            t.Wait();
        //after they all finished working with the queues, clear queues
        //making them ready for the nest task group
        foreach (Queue<string> q in queueList)
            q.Clear();
    }
}

Если вы хотите увидеть проблему вдействие с более простым воссозданием, попробуйте этот простой код.

public static void Main(string[] args)
{

    for (int i = 0; i < 10; i++)
    {
        int j = i;
        Task.TaskFactory.StartNew(() =>
        {
            Thread.Sleep(10); //Give a little time for the for loop to complete.
            Console.WriteLine("i: " + i + " j: " + j);
        }
    });
    Console.ReadLine();
}

Answer 2

Вы вычислили идентификатор задачи внутри задачи и изменили базу для расчета вне задачи.Я изменил логику лишь немного.У меня не было ошибок.

namespace Project1
{
    using System.Collections.Generic;
    using System.Threading.Tasks;

    internal class Program
    {
        private static int j = 0;
        private readonly List<Queue<string>> queueList = new List<Queue<string>>();

        public void StartTasts(int n)
        {
            for (var i = 0; i < n; i++)
            {
                this.queueList.Add(new Queue<string>());
            }

            var taskList = new List<Task>();
            for (var taskGroup = 0; taskGroup < 10; taskGroup++)
            {
                // 10 groups of task
                // each group has 'n' tasks working in parallel
                for (var i = 0; i < n; i++)
                {
                    // each task gets its own and independent queue from the list
                    var taskId = j % n;
                    taskList.Add(
                        Task.Factory.StartNew(
                            () =>
                            {
                                this.DoWork(taskId);
                            }));
                    j++;
                }

                // waiting for each task group to finish
                foreach (var t in taskList)
                {
                    t.Wait();
                }

                // after they all finished working with the queues, clear queues
                // making them ready for the nest task group
                foreach (var q in this.queueList)
                {
                    q.Clear();
                }
            }
        }

        public void DoWork(int queue)
        {
            // demonstration of generating strings 
            // and put them in the correct queue
            for (var k = 0; k < 10000; k++)
            {
                this.queueList[queue].Enqueue(k + string.Empty);
            }
        }

        private static void Main(string[] args)
        {
            new Program().StartTasts(10);
        }      
    }
}

Answer 3

Я не думаю, что ваша проблема в очереди, похоже, что проблема в самом списке может быть проблемой.

Поскольку правило использования параллельных или синхронных процессов, список не является потоком, кроме DS.

Попробуйте использовать поток сохранения DS Как Класс ConcurrentBag