Отслеживание изменений массива данных из второго потока

@SuprSonic · Регистрация: 07.08.2016

Author24 — интернет-сервис помощи студентам

Хочу узнать мнения гуру касаемо подхода в реализации следующего. Есть массив данных и два потока. Один поток производит операции с элементами этого массива (добавление и обновление рандомных элементов). Другой поток должен отслеживать изменения в этом массиве и строго дублировать его. Первый поток должен быть полностью автономным и не зависеть от работы второго. Т.е работа второго никаким образом не должна тормозить работу первого, обратное же допустимо. Т.к. сам массив может быть очень большим(а в теории бесконечным), то проходить циклом во втором потоке весь массив и искать изменения не совсем разумно. Получается первый поток должен сообщить второму о том, какие элементы он изменил (добавил/обновил).

У меня получилось вот это:

C#

public class _MultiThread
{
        public ManualResetEventSlim EVENT;
        private bool isRead;
        private int index;
        private int[] data;
 
        public _MultiThread()
        {
            this.data = new int[100];
            this.index = -1;
            this.isRead = false;
            this.EVENT = new ManualResetEventSlim(true);
        }
 
        private void _clear()
        {
            this.EVENT.Reset();
            this.index = -1;
            this.isRead = false;
            this.EVENT.Set();
        }
 
        // Добавляем только уникальные значения
        public unsafe void AddUnique(int _value)
        {
            if (isRead)
            {
                _clear();
                this.data[++this.index] = _value;
                return;
            }
            fixed (int* ptr = &this.data[0])
            {
                for (int i = 0; i <= this.index; i++)
                {
                    if (ptr[i] == _value) return;
                }
                this.data[++this.index] = _value;
            }
        }
 
        public void Reader(Action<int> func)
        {
            this.EVENT.Wait();
            for(int i = 0; i <= this.index; i++)
            {
                func(this.data[i]);
            }
            this.isRead = true;
        }
}
 
public class WORK{
        public _MultiThread CHANGES = new _MultiThread();
        public int[] _ARRAY;
 
        public WORK(){
                this._ARRAY = new int[100];
        }
 
        public void Working(){
                // Какая-то полезная работа с записью результата в случайный элемент массива
                int index = new Random().Next( 0, 99 );  // Выбираем случайный элемент
                this._ARRAY[index] = new Random().Next( 0, 9999999 ); // Генерируем и записываем в него случайное число
                this.CHANGES.AddUnique(index); // Сообщаем какой элемент был изменен
        }
}
 
.....
//В основном цикле
WORK worker = new WORK();
 
new Thread(() =>
{
       // Раз в 1 сек "слушаем" изменения
       while(true){
            Thread.Sleep(1000);
            worker.CHANGES.Reader((index) => {
                 Console.WriteLine ("Был изменен элемент массива под индексом: " + index);
            });
       }
}).Start();
 
new Thread(() =>
{
        while(true){
             Thread.Sleep(25);
             worker.Working();
        }
}).Start();

Для простоты основной массив будет содержат только INT. Для поддержки связи между потоками использую собственный класс _MultiThread. Т.к. первый поток может работать быстрее, то к тому моменту пока второй поток начнет "прослушивать изменения", один и тот же элемент массива может быть обновлен несколько раз, а поскольку второй поток не интересует сколько раз элементы подвергались изменению, то использую функцию AddUnique.
После того как Reader прочитал все изменения, выставляем isRead в true. Первый поток увидев, что изменения были считаны перед добавлением в _MultiThread.data затирает все с помощью _clear().

И вроде все работает как надо. Но, так как я в многопоточности новичок, хотел бы знать как можно иначе реализовать алгоритм? Возможно ли использовать lock и обойтись без ManualResetEventSlim ? И есть ли в моей реализации ошибки?

@TopLayer · 21.01.2019, 21:40

Сообщение от SuprSonic

Прочтите пожалуйста еще раз мои посты повнимательнее. Потому как попахивает каким-то стебом уже.
Чего там понимать? В байте 8 бит. 2^8 = 256 в byte же можно запихнуть значение от 0 до 255. Что тут сверхнового?

Ну в было элементе битового массива (который состоит из байт в конкретном случае) значение, например, 128. Первый поток поменял массив данных и теперь должен в массиве бит изменить число 128 на число 130. Для этого он сначала считывает число 128 из этого массива в регистр. Сразу после этого второй поток тоже считывает оттуда же число 128, просматривает все его биты и уже собирается записать в массив 0 вместо 128, как его опережает первый поток, который записывает значение 130. После этого второй поток сразу же записывает значение 0 вместо 130. То есть бит, установленный первым потоком никогда не будет прочитан вторым. Вот я об этом говорил. И оперирование общей для потоков памяти идёт байтами, а не битами. А вы мне пишете:

Сообщение от SuprSonic

Почему? Ведь второй поток пробегается от начала и до конца по каждому биту и меняет 1 на 0;

Сообщение от SuprSonic

Почему X и Y то? Там либо 0 либо 1.

@SuprSonic · 21.01.2019, 22:03 **[ТС]**

Сообщение от TopLayer

Для этого он сначала считывает число 128 из этого массива в регистр.

Ага, дошло. Спасибо.

Тогда можно ли как-то блокировать определенный байт ? Вместо блокировки всего массива байтов?

Добавлено через 12 минут
volatile здесь как то нужен?

@TopLayer · 21.01.2019, 22:08

Сообщение от SuprSonic

Тогда можно ли как-то блокировать определенный байт ? Вместо блокировки всего массива байтов?

Нет. И наверное лучше не в байтах хранить, а в long-ах. Вот такая конструкция часто используется для обновления.

C#

    int[] x = { 1, 2, 3 };
    
    int oldValue;
    int newValue;
    do
    {
        oldValue = x[1];
        newValue = oldValue + 155;
    } while (Interlocked.CompareExchange(ref x[1], newValue, oldValue) != oldValue);
    
    Console.WriteLine(x[1]);

Добавлено через 3 минуты

Сообщение от SuprSonic

volatile здесь как то нужен?

volatile не позволяет компилятору/процессору переупорядочивать инструкции определённым образом. Например тут

C#

        public void Reader(Action<int> func)
        {
            this.EVENT.Wait();
            for(int i = 0; i <= this.index; i++)
            {
                func(this.data[i]);
            }
            this.isRead = true;
        }

this.isRead = true могло выполнится до цикла. А вот если бы была волатильная запись, то уже были бы соответствующие гарантии.

@SuprSonic · 21.01.2019, 22:40 **[ТС]**

Сообщение от TopLayer

И наверное лучше не в байтах хранить, а в long-ах. Вот такая конструкция часто используется для обновления.

Как я понял, там сравниваются значения до и после?

А если оставить байты и просто сообщать второму потоку в каком именно байте в данный момент времени копошится первый поток, а перед тем как обнулить бит проверять не находится ли первый поток в том же байте? Если находится, то прочитать текущий байт побитно заново.

@TopLayer · 21.01.2019, 22:43

Сообщение от SuprSonic

Как я понял, там сравниваются значения до и после?

Ну да, там проверяется, что в массиве значение не изменилось, пока поток вычислял новое.

Сообщение от SuprSonic

а перед тем как обнулить бит проверять не находится ли первый поток в том же байте?

Между проверкой и записью может пройти сколько угодно времени, и к этому моменту результаты проверки могут стать не актуальными.

@SuprSonic · 22.01.2019, 00:27 **[ТС]**

Сообщение от TopLayer

Между проверкой и записью может пройти сколько угодно времени, и к этому моменту результаты проверки могут стать не актуальными.

Тогда остается вариант считывания байта во временную переменную, его побитовое изменение и перед обратной записью в массив сверка предыдущего значения с текущим в массиве через Interlocked.CompareExchange ?

Interlocked.CompareExchange получается блокирует элемент массива или чаво?

Добавлено через 1 час 5 минут

Сообщение от TopLayer

И наверное лучше не в байтах хранить, а в long-ах.

Почему то byte быстрее работают

@Элд Хасп · 22.01.2019, 00:41

Сообщение от SuprSonic

Почему то byte быстрее работают

Это зависит от процессора в большинстве случаев целочисленные операции (в том числе логические) быстрее всего работаю с int и uint. Но надо проверять на конкретном процессоре.

Добавлено через 1 минуту
Если Вам нужны именно биты, то лучше всего их хранить как bool. Иначе компилятор C# может вставлять ненужные приведения типов.

@SuprSonic · 22.01.2019, 00:44 **[ТС]**

C#

            Random random = new Random();
 
            int GetRandom()
            {
                return random.Next(500_000, 1_000_000);
            }
 
            using (new Diagnostics("bit int[]"))
            {
                int read_sum = 0, length = 0, from = 1024000;
                int bit = 0;
                int[] array = new int[32000];
 
                for (int i = 0; i < count; i++)
                {
                    var value = GetRandom();
                    bit = (value % 32);
                    if (((array[value / 32] >> bit) & 1) == 0)
                    {
                        array[value / 32] ^= 1<<bit;
                        length = Math.Max(length, value + 1);
                        from = Math.Min(from, value);
                    }
 
                }
 
                // Считываем каждый бит по отдельности и высчитываем общую сумму
                // по завершении побитового чтения каждого элемента array[i], затираем его
                for (int i = from / 32; i < length / 32 + (length % 32 > 0 ? 1 : 0); i++)
                {
                    int old_value = array[i];
                    do
                    {
                        if (((array[i] >> 0) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 0));
                        }
                        if (((array[i] >> 1) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 1));
                        }
                        if (((array[i] >> 2) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 2));
                        }
                        if (((array[i] >> 3) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 3));
                        }
                        if (((array[i] >> 4) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 4));
                        }
                        if (((array[i] >> 5) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 5));
                        }
                        if (((array[i] >> 6) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 6));
                        }
                        if (((array[i] >> 7) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 7));
                        }
                        if (((array[i] >> 8) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 8));
                        }
                        if (((array[i] >> 9) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 9));
                        }
                        if (((array[i] >> 10) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 10));
                        }
                        if (((array[i] >> 11) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 11));
                        }
                        if (((array[i] >> 12) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 12));
                        }
                        if (((array[i] >> 13) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 13));
                        }
                        if (((array[i] >> 14) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 14));
                        }
                        if (((array[i] >> 15) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 15));
                        }
                        if (((array[i] >> 16) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 16));
                        }
                        if (((array[i] >> 17) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 17));
                        }
                        if (((array[i] >> 18) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 18));
                        }
                        if (((array[i] >> 19) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 19));
                        }
                        if (((array[i] >> 20) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 20));
                        }
                        if (((array[i] >> 21) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 21));
                        }
                        if (((array[i] >> 22) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 22));
                        }
                        if (((array[i] >> 23) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 23));
                        }
                        if (((array[i] >> 24) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 24));
                        }
                        if (((array[i] >> 25) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 25));
                        }
                        if (((array[i] >> 26) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 26));
                        }
                        if (((array[i] >> 27) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 27));
                        }
                        if (((array[i] >> 28) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 28));
                        }
                        if (((array[i] >> 29) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 29));
                        }
                        if (((array[i] >> 30) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 30));
                        }
                        if (((array[i] >> 31) & 1) > 0)
                        {
                            read_sum += ((i * 32 + 31));
                        }
                    } while (Interlocked.CompareExchange(ref array[i], 0, old_value) != old_value);
                }
 
                Console.Write("Read_sum: " + read_sum + " | Length:" + length + " || From: " + from + " || ");
            }

Добавлено через 3 минуты

Сообщение от Элд Хасп

Это зависит от процессора в большинстве случаев целочисленные операции (в том числе логические) быстрее всего работаю с int и uint. Но надо проверять на конкретном процессоре.

Ну на int быстрее чем на byte, но на long что-то как-то медленно. Win 64. Хотя может это из-за того что из под виртуалки сижу...

@Элд Хасп · 22.01.2019, 01:15

Сообщение от SuprSonic

Ну на int быстрее чем на byte, но на long что-то как-то медленно. Win 64. Хотя может это из-за того что из под виртуалки сижу...

Это не от системы зависит, а от процессора.
Основной тип данных для АЛУ процессора - 32 разряда. Может только у новых перешли на 64 (как основной тип) - точно не знаю.
Все операции с байтами делаются через приведение к int. Поэтому потеря тактов на этом вполне возможна. Может быть можно как-то указать в настройках оптимизации или небезопасный код, чтобы этого избежать. Но не возился с этим, не было такой потребности.

@SuprSonic · 22.01.2019, 03:24 **[ТС]**

Протестировал еще один способ, попробовав BitArray из System.Collections;

C#

            using (new Diagnostics("BitArray"))
            {
                int read_sum = 0, index = 0;
                BitArray array = new BitArray(10240000, false);
 
                for (int i = 0; i < count; i++)
                {
                    int value = GetRandom();
                    array.Set(value, true);
                }
 
                // Считываем каждое значение по отдельности и высчитываем общую сумму
                lock (array.SyncRoot)
                {
                    foreach (bool item in array)
                    {
                        if (item) read_sum += index;
                        index++;
                    }
                    array.SetAll(false);
                }
 
                Console.Write("Read_sum: " + read_sum + " | Length:" + array.Length + " || From: " + "foreach" + " || ");
            }

При большой длине основного массива и при условии что второй поток работает медленнее первого (в тесте это кол-во итераций), то BitArray отрабатывает быстрее на 10%. Но, прочесть его вторым потоком без lock увы невозможно. И еще слишком много памяти отъедает — в раз 10 больше в сравнении с предыдущим.

@SuprSonic 7 / 4 / 3 Регистрация: 07.08.2016 Сообщений: 65
	21.01.2019, 22:03 [ТС]	42
	Сообщение от TopLayer Для этого он сначала считывает число 128 из этого массива в регистр. Ага, дошло. Спасибо. Тогда можно ли как-то блокировать определенный байт ? Вместо блокировки всего массива байтов? Добавлено через 12 минут volatile здесь как то нужен? 0

@TopLayer 907 / 664 / 318 Регистрация: 23.10.2016 Сообщений: 1,543
	21.01.2019, 22:43	45
	Сообщение от SuprSonic Как я понял, там сравниваются значения до и после? Ну да, там проверяется, что в массиве значение не изменилось, пока поток вычислял новое. Сообщение от SuprSonic а перед тем как обнулить бит проверять не находится ли первый поток в том же байте? Между проверкой и записью может пройти сколько угодно времени, и к этому моменту результаты проверки могут стать не актуальными. 1

@SuprSonic 7 / 4 / 3 Регистрация: 07.08.2016 Сообщений: 65
	22.01.2019, 00:27 [ТС]	46
	Сообщение от TopLayer Между проверкой и записью может пройти сколько угодно времени, и к этому моменту результаты проверки могут стать не актуальными. Тогда остается вариант считывания байта во временную переменную, его побитовое изменение и перед обратной записью в массив сверка предыдущего значения с текущим в массиве через Interlocked.CompareExchange ? Interlocked.CompareExchange получается блокирует элемент массива или чаво? Добавлено через 1 час 5 минут Сообщение от TopLayer И наверное лучше не в байтах хранить, а в long-ах. Почему то byte быстрее работают 0

@Элд Хасп Модератор 15466 / 10712 / 2786 Регистрация: 21.04.2018 Сообщений: 31,532 Записей в блоге: 2
	22.01.2019, 00:41	47
	Сообщение от SuprSonic Почему то byte быстрее работают Это зависит от процессора в большинстве случаев целочисленные операции (в том числе логические) быстрее всего работаю с int и uint. Но надо проверять на конкретном процессоре. Добавлено через 1 минуту Если Вам нужны именно биты, то лучше всего их хранить как bool. Иначе компилятор C# может вставлять ненужные приведения типов. 0

@Элд Хасп Модератор 15466 / 10712 / 2786 Регистрация: 21.04.2018 Сообщений: 31,532 Записей в блоге: 2
	22.01.2019, 01:15	49
	Сообщение от SuprSonic Ну на int быстрее чем на byte, но на long что-то как-то медленно. Win 64. Хотя может это из-за того что из под виртуалки сижу... Это не от системы зависит, а от процессора. Основной тип данных для АЛУ процессора - 32 разряда. Может только у новых перешли на 64 (как основной тип) - точно не знаю. Все операции с байтами делаются через приведение к int. Поэтому потеря тактов на этом вполне возможна. Может быть можно как-то указать в настройках оптимизации или небезопасный код, чтобы этого избежать. Но не возился с этим, не было такой потребности. 0

@TopLayer 907 / 664 / 318 Регистрация: 23.10.2016 Сообщений: 1,543
	21.01.2019, 21:40	41
	Сообщение от SuprSonic Прочтите пожалуйста еще раз мои посты повнимательнее. Потому как попахивает каким-то стебом уже. Чего там понимать? В байте 8 бит. 2^8 = 256 в byte же можно запихнуть значение от 0 до 255. Что тут сверхнового? Ну в было элементе битового массива (который состоит из байт в конкретном случае) значение, например, 128. Первый поток поменял массив данных и теперь должен в массиве бит изменить число 128 на число 130. Для этого он сначала считывает число 128 из этого массива в регистр. Сразу после этого второй поток тоже считывает оттуда же число 128, просматривает все его биты и уже собирается записать в массив 0 вместо 128, как его опережает первый поток, который записывает значение 130. После этого второй поток сразу же записывает значение 0 вместо 130. То есть бит, установленный первым потоком никогда не будет прочитан вторым. Вот я об этом говорил. И оперирование общей для потоков памяти идёт байтами, а не битами. А вы мне пишете: Сообщение от SuprSonic Почему? Ведь второй поток пробегается от начала и до конца по каждому биту и меняет 1 на 0; Сообщение от SuprSonic Почему X и Y то? Там либо 0 либо 1. 1

@SuprSonic 7 / 4 / 3 Регистрация: 07.08.2016 Сообщений: 65
	21.01.2019, 22:40 [ТС]	44
	Сообщение от TopLayer И наверное лучше не в байтах хранить, а в long-ах. Вот такая конструкция часто используется для обновления. Как я понял, там сравниваются значения до и после? А если оставить байты и просто сообщать второму потоку в каком именно байте в данный момент времени копошится первый поток, а перед тем как обнулить бит проверять не находится ли первый поток в том же байте? Если находится, то прочитать текущий байт побитно заново. 0

	22.01.2019, 03:24