Thread-safe specialized queue - .NET

Здравствуйте, Аноним, Вы писали:

А>По поводу этого я уже высказался. С таким подходом придётся периодически будить поток только для того, чтобы проверить состояние очереди. Меня это не устраивает.

Не нужно периодически будить поток, код в другой подветке.

А>Любопытно, кто это Вам сказал? К сожалению, Вам неправильно объяснили. Кэши процессоров имеют специальные механизмы, поддерживающие их в актуальном состоянии, в том числе — в многопроцессорных машишах. И какие-либо блокировки на это никак не влияют. А иначе критические секции, да и другие средства синхронизации, оказались бы в большинстве случаев бесполезным хламом.

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

А>То есть как что?! Она завершится, что же ещё? А иначе поток просто останется висеть на одном из этих объектов до самого вызова ExitProcess, после чего будет просто жёстко терминирован. Если до вызова ExitProcess дело вообще дойдёт — кто его знает, этот NET. А то ведь может статься, что только TerminateProcess остановит это безобразие

Да, торможу. Про то что в MonitorRoutine terminate проверяется я забыл.

Здравствуйте, Serginio1, Вы писали:

Так. Если я правильно понял, то может статься, что поле нестатическое поле класса может оказаться расположенным в регистре процессора? Забавно

А под кэшем Вы что подразумевали? Я не совсем уловил.

Здравствуйте, Аноним, Вы писали:

А>Или вот у меня объявлено private volatile int timeout. В нативе я ы не задумываясь использовал бы эту переменную без всяких блокировок

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

Здравствуйте, Undying, Вы писали:

U>Здравствуйте, Аноним, Вы писали:

А>>То есть, если я правильно понял, Вы предлагаете периодически будить поток по таймауту, безотносительно наличия сообщений в очереди, так? Вообще-то именно этого я и пытался избежать. А если Вы что-то другое имели в виду, продемонстрируйте пожалуйста этот момент кодом.

U>Примерно так (локи опущены):

U>

U>DateTime lastSendingTime;
U>void MonitorRoutine()
U>{
U>  while (!terminate)
U>  {
U>    DateTime curTime = DateTime.UtcNow;
U>    TimeSpan interval = curTime - lastSendingTime;
U>    if (interval < timeout)
U>    {
U>      waitHandle.WaitOne(timeout - interval, false);
U>      continue;
U>    }
    
U>    if (msgQueue.Count == 0)
U>    {
U>      waitHandle.WaitOne();
U>    }
U>    else
U>    {
U>      //Отправляем сообщение

U>      lastSendingTime = DateTime.UtcNow;
U>      waitHandle.WaitOne(timeout, false);
U>    }
U>  }
U>}

U>        public void EnqueueMessage(string message)
U>        {
U>            lock (this.queueSync)
U>            {
U>                msgQueue.Enqueue(message);
U>            }
U>            if (!isPaused)
U>              waitHandle.Set();
              
U>        }

U>        volatile bool isPaused = false;
U>        public void Pause()
U>        {
U>          isPaused = true;
U>          waitHandle.WaitOne();
U>        }

U>        // Возобновление отправок
U>        public void Resume()
U>        {
U>          isPaused = false;
U>          waitHandle.Set();
U>        }

U>

К сожалению, приведённый Вами код не соответствует сформулированной автором теме задаче. В частности, у Вас добавление сообщения в очередь автоматически приводит к снятию с паузы, а этого происходить не должно. Попытка же этот исправить без добавления эвента неизбежно привёдет либо к возникновению гонок, либо к необходимости будить поток, даже в состоянии паузы. Да собственно, гонки ми в этом коде имеют место — вызов EnqueueMessage может легко вклиниться if (msgQueue.Count == 0) и waitHandle.WaitOne(), а залочить это не получится.

Вообще, Вам не кажется, что Вы несколько непоследовательны?

С одной стороны предлагаете всё лочить одним объектом, дабы на корню убить гипотетические деадлоки, хотя один объект блокировки таких гарантий всё равно дать не может. С другой — готовы идти на усложнение логики, лишь бы избавиться от одного-двух объектов, а ведь усложнение логики — прямой путь к появлению логических ошибок, к тому-же трудноуловимых в многопоточном коде.

А>>Вообще-то лаконичность кода для меня — дело третье

По моей шкале эффективность и читабельность кода имеют куда больший приоритет. А вот что Вы подразумеваете под "лучше по безопасности"? В каком смысле?

U>Безопаснее в том смысле, что код с volatile гораздо сложнее сломать изменениями, чем код с Interlocked.Exchange.

Вероятно, Вы правы. Хотя в нативе мы как-то обходимся

Впрочем, по поводу преимуществ volatile убедительно сказано здесь

Давайте всё-таки сделаем допущение, что с логикой всё в порядке и сосредоточимся на особенностях её реализации на C#, мне именно это более всего интересно

PS Надеюсь, автор темы простит меня за то, что я так беспардонно в неё вторгся. Я нечаянно

Здравствуйте, Аноним, Вы писали:

А>Здравствуйте, Serginio1, Вы писали:

А>Так. Если я правильно понял, то может статься, что поле нестатическое поле класса может оказаться расположенным в регистре процессора? Забавно

А что мешает? Это просто кусок памяти, RAMы.

А>А под кэшем Вы что подразумевали? Я не совсем уловил.
Кэш ядра процессора.

Здравствуйте, Undying, Вы писали:

U>Здравствуйте, Аноним, Вы писали:

А>>По поводу этого я уже высказался. С таким подходом придётся периодически будить поток только для того, чтобы проверить состояние очереди. Меня это не устраивает.

U>Не нужно периодически будить поток, код в другой подветке.

Мне как-то удобнее смотреть в плоском виде

Но я видел и там отписался.

А>>Любопытно, кто это Вам сказал? К сожалению, Вам неправильно объяснили. Кэши процессоров имеют специальные механизмы, поддерживающие их в актуальном состоянии, в том числе — в многопроцессорных машишах. И какие-либо блокировки на это никак не влияют. А иначе критические секции, да и другие средства синхронизации, оказались бы в большинстве случаев бесполезным хламом.

U>Volatile принудительно сбрасывает кэши процессоров, если переменная таким образом не помечена и используется не из под лока, то устаревшее значение переменной может жить в кэшах процессоров неопределенное время (т.е. до тех пор пока кто-нибудь другой кэш не сбросит).

Ещё раз, вот смотрите. Имеем переменную, расположенную в памяти, допустим int32 Value = 10. Допустим, два потока на разных процессорах уже обращались к ней для чтения и она оказалась в кэшах обоих процессоров. Теперь наступает момент, когда эти два потока на тех-же процессорах пытаются выполнить следующий псевдокод

  EnterCriticalSection(CS);
  Value = Value + 5;
  LeaveCriticalSection(CS);

Если бы не существовало механизма поддержания когерентности кэшей процессоров, то каждый из потоков имел бы дело с собственной переменной Value в своём кэше,. и переменная Value в каждом из кэшей оказалась бы равна 15. То есть от применения критической секции не было бы никакого проку. Но так как такой механизм таки существует, и существует он на "железном" уровне, то ничего подобного не происходит, критические секции работают, а переменная Value окажется равна 20, как ей и полагается. Давайте Вы хорошенько это обдумаете, и мы к этому вопросу больше возвращаться не будем, договорились? Скучно, ей-Богу

При оптимизации даже при обноядерном процессоре.
А>А под кэшем Вы что подразумевали? Я не совсем уловил.
Кэш ядра. Он у каждого ядра свой, а так как потоки могут выполняться на разных ядрах, то и дашшые одного кэша могут быть невидимы для другого.
Но тут нужно смотреть архитектуру.

Здравствуйте, Кондраций, Вы писали:

К>Здравствуйте, Аноним, Вы писали:

А>>Здравствуйте, Serginio1, Вы писали:

А>>Так. Если я правильно понял, то может статься, что поле нестатическое поле класса может оказаться расположенным в регистре процессора? Забавно

К>А что мешает? Это просто кусок памяти, RAMы.

А>>А под кэшем Вы что подразумевали? Я не совсем уловил.
К>Кэш ядра процессора.

Да ничто вобщем-то не мешает, просто как-то нелогично. А скорее просто не привычно

PS RAM всё-таки принято применять к чисто ОЗУ, к регистрам процессоров оно ка-бы не совсем подходит.

Здравствуйте, Аноним, Вы писали:

А>Здравствуйте, Аноним, Вы писали:

А>>Или вот у меня объявлено private volatile int timeout. В нативе я ы не задумываясь использовал бы эту переменную без всяких блокировок

А>При запуске на машине с числом физических ядер более одного, нативный volitile становится злой шуткой вместо гарантированной синхронизации.

Честно говоря, совсем Вас не понял

Не моглибы Вы повторить это в несколько более развёрнутом виде? Спасибо.

Здравствуйте, Аноним, Вы писали:

U>>Volatile принудительно сбрасывает кэши процессоров, если переменная таким образом не помечена и используется не из под лока, то устаревшее значение переменной может жить в кэшах процессоров неопределенное время (т.е. до тех пор пока кто-нибудь другой кэш не сбросит).

А>Ещё раз, вот смотрите. Имеем переменную, расположенную в памяти, допустим int32 Value = 10. Допустим, два потока на разных процессорах уже обращались к ней для чтения и она оказалась в кэшах обоих процессоров. Теперь наступает момент, когда эти два потока на тех-же процессорах пытаются выполнить следующий псевдокод
А>

А>  EnterCriticalSection(CS);
А>  Value = Value + 5;
А>  LeaveCriticalSection(CS);
А>

Смотри выделенное. лок соответствует критической секцией

Здравствуйте, Serginio1, Вы писали:

S>Кэш ядра. Он у каждого ядра свой,

Вообще-то это не для всех многоядерников справедливо, есть и с, грубо говоря, совмещённым...Ну там довольно сложный механизм. По поводу когерентности, я не хочу повторяться, вот здесь посмотрите.

Здравствуйте, Serginio1, Вы писали:

S>Здравствуйте, Аноним, Вы писали:

U>>>Volatile принудительно сбрасывает кэши процессоров, если переменная таким образом не помечена и используется не из под лока, то устаревшее значение переменной может жить в кэшах процессоров неопределенное время (т.е. до тех пор пока кто-нибудь другой кэш не сбросит).

А>>Ещё раз, вот смотрите. Имеем переменную, расположенную в памяти, допустим int32 Value = 10. Допустим, два потока на разных процессорах уже обращались к ней для чтения и она оказалась в кэшах обоих процессоров. Теперь наступает момент, когда эти два потока на тех-же процессорах пытаются выполнить следующий псевдокод
А>>

А>>  EnterCriticalSection(CS);
А>>  Value = Value + 5;
А>>  LeaveCriticalSection(CS);
А>>

S>Смотри выделенное.

Признаться, я не понял, что нужно смотреть, и что значит "лок соответствует критической секцией" В WinApi мы просто вызываем EnterCriticalSection, при чём тут лок? Вы можете с успехом заменить EnterCriticalSection на WaitForSingleObject(Semaphore), ничего не поменяется. Здесь нет никаких иницируемых системой сбросов кэша, этого просто не нужно. Когерентность обеспечивается механизмом самих кэшей, на уровне железа. Вообще принудительный сброс кэша нужен только в очень редких, весьма экзотических, случааях.

В общем, я не знаю, как это ещё нужно объяснять, и более к этому возвращаться не намерен. Если не понятно — более ничем не могу помочь, извините

Могу только посоветовать ознакомиться в общих чертах с архитектурой современных процессоров.

Здравствуйте, Аноним, Вы писали:

http://rsdn.ru/article/dotnet/CSThreading2.xml#ENUAE
Имелось ввиду Барьеры в памяти.
В свое время шел разговор и об кэшах процессоров.

Здравствуйте, Undying, Вы писали:
U>Примерно так (локи опущены):

А если вот так?

public sealed class MessageSender
{
    public event Action<string> MessageReceived;

    private readonly ManualResetEvent sendEvent;
    private readonly Queue<string> msgQueue;
    private readonly object syncRoot;

    private volatile bool isPaused;
    private int interval;

    public MessageSender()
    {
        this.sendEvent = new ManualResetEvent(false);
        this.msgQueue = new Queue<string>(32);
        this.syncRoot = new object();
        this.interval = 1000;

        ThreadPool.QueueUserWorkItem(SenderJob);
    }

    // Интервал между посылками события
    public int Interval
    {
        get { return this.interval; }
        set
        {
            if(value < 0) value = 100;
            this.interval = value;
        }
    }

    // Метод потока-отправителя
    private void SenderJob(object state)
    {
        while(true)
        {
            this.sendEvent.WaitOne();

            string message = null;
            lock(this.syncRoot)
            {
                if (this.msgQueue.Count > 0)
                {
                    message = this.msgQueue.Dequeue();
                }
                else this.sendEvent.Reset();
            }

            if (message != null)
            {
                // Возбуждение события
                Action<string> receiver = this.MessageReceived;
                if (receiver != null) receiver(message);
            }

            Thread.Sleep(this.interval);
        }
    }

    // Добавление сообщений в очередь
    public void EnqueueMessage(string message)
    {
        lock(this.syncRoot)
        {
            this.msgQueue.Enqueue(message);

            if (!this.isPaused)
            {
                this.sendEvent.Set();
            }
        }
    }

    // Приостановка отправок
    public void Pause()
    {
        lock (this.syncRoot)
        {
            this.isPaused = true;
            this.sendEvent.Reset();
        }
    }

    // Возобновление отправок
    public void Resume()
    {
        if   (this.isPaused)
        lock (this.syncRoot)
        {
            this.isPaused = false;
            if (this.msgQueue.Count > 0)
            {
                this.sendEvent.Set();
            }
        }
    }

    // Очистка очереди
    public void Abort()
    {
        lock(this.syncRoot)
        {
            this.msgQueue.Clear();
        }
    }
}

Работа с WaitHandle в локах осознанно, так как поток отправителя сам сбрасывает ивент... То есть может выйти так, что:
1. Поток-отправитель проверит очередь, убедится что в ней нету сообщений.
2. Другой поток добавит в очередь сообщение и включит ивент в сигнальное состояние.
3. Убедившись, что сообщений нет, поток-отправитель, сбросит событие.
4. В итоге отправитель ждёт события, а в очереди 1 сообщение.

Работа устраивает, проблем не наблюдается, но хочется довести до ума чтобы "уж наверняка"...

Здравствуйте, Serginio1, Вы писали:

S>Здравствуйте, Аноним, Вы писали:

S>http://rsdn.ru/article/dotnet/CSThreading2.xml#ENUAE
S>Имелось ввиду Барьеры в памяти.
S>В свое время шел разговор и об кэшах процессоров.

Я просмотрел раздел "Барьеры в памяти и асинхронная изменчивость (volatility)", Вы ведь его имели в виду? Но при чём тут кэш процессора? Ключевая фраза могут кэшироваться в регистрах CPU, ну всё правильно, видимо. Я не знаю, может Вы просто не в курсе, что "регистры процессора" и "L1, L2 кэши процессора" зто абсолютно разные, никак не связанные вещи? Не обижайтесь, если это не так, и Вы в курсе этой разницы, но у меня просто не осталось других идей

А то, что там названо "барьерами", это действительно, самые обычные критические секции, к но NET, судя по всему, добавляет внутрь защищённого блока ещё и код, копирующий значения переменных из регистров в ОЗУ. Но дело в том, что на самом деле этот код копирует из регистров в кэш. Понимаете, современные процессы вообще не умеют работать напрямую с ОЗУ, но сами они об этом не знают. Пытаясь записать что-то в ОЗУ, они на самом деле пишут в кэш, и при чтении читают из кэша. Но само ядро об этом даже не догадывается, да и не нужно ему это. Контролеер кэша делает кэш прозрачным для ядра, а следовательно и для программы. Именно контроллер следит, чтобы значение по адресу в озу, к которму обращается ядро, совпадало с тем, что лежит в кэше. И он-же обеспечивает непротиворечивость данных в кэшах процессоров на многопроцессорной системе.

В общем, это меня утомило, честное слово

Здравствуйте, Пельмешко, Вы писали:

П>Здравствуйте, Undying, Вы писали:
U>>Примерно так (локи опущены):

П>Работа устраивает, проблем не наблюдается, но хочется довести до ума чтобы "уж наверняка"...

А зачем вам while(true), лишняя логика по проверке очереди на ноль для Reset-а и флаг isPaused?

можно, например, так:

private void SenderJob(object state)
{
            sendEvent.WaitOne();

            string message;
            lock (msgQueue)
            {
                message = _queues.Dequeue();
            }
            
            if (message != null)
            {
                //непосредственно перед
                lock(syncRoot)
                {
                    Thread.Sleep(interval);
                }

                Action<string> receiver = this.MessageReceived;
                if (receiver != null) receiver(message);
            }
}

public void EnqueueMessage(string message)
{
        lock(msgQueue)
        {
            msgQueue.Enqueue(message);
        }
        ThreadPool.QueueUserWorkItem(SenderJob, null);
}

Здравствуйте, Аноним, Вы писали:

А>В общем, это меня утомило, честное слово

Все, кто предпочитает думать, что volatile и Lock обладают каким-то волшебством, недоступным нативному, даже ассемблерному, коду, тот, разумеется волен в своих желаниях. А кто хочет разобраться, советую хотя-бы в общих чертах ознакомиться с устройством процессоров. А здесь об этом говорить как-то не очень подходяще.
Я же оговорился, что свое время речь шла о кэшах процессоров, и от том, что зависит от архитектуры.
Буду рад если дашь толковую ссылку на Контролеер кэшей. Особенно когда ядра будут исчисляться сотнямию
volatility так же как и в нативе просто запрещает оптимизацию для кэширования значений в CPU.
А Lock это обычная критическая секция. Так, что Net в этом плане недалеко от натива ушел

Здравствуйте, <Аноним>, Вы писали:

А>Здравствуйте, Serginio1, Вы писали:

А>Признаться, я не понял, что нужно смотреть, и что значит "лок соответствует критической секцией" В WinApi мы просто вызываем EnterCriticalSection, при чём тут лок? Вы можете с успехом заменить EnterCriticalSection на WaitForSingleObject(Semaphore), ничего не поменяется. Здесь нет никаких иницируемых системой сбросов кэша, этого просто не нужно. Когерентность обеспечивается механизмом самих кэшей, на уровне железа. Вообще принудительный сброс кэша нужен только в очень редких, весьма экзотических, случааях.

АААА.... Я хочу такое железо как у него. Почему мне не такое продали???

... << RSDN@Home 1.2.0 alpha 4 rev. 1228>>

Здравствуйте, Аноним, Вы писали:
А>В общем, это меня утомило, честное слово

Все, кто предпочитает думать, что volatile и Lock обладают каким-то волшебством, недоступным нативному, даже ассемблерному, коду, тот, разумеется волен в своих желаниях. А кто хочет разобраться, советую хотя-бы в общих чертах ознакомиться с устройством процессоров. А здесь об этом говорить как-то не очень подходяще.

Ща перечитал топик и никак не пойму что и кому Вас утомило доказывать.
Зато я понял:
1. Чем чаще употребляешь слово "нативный" в managed-ветке, тем ты авторитетнее.
2. Советовать всем изучить архитектуру CPU — это просто высший пилотаж!

Здравствуйте, Arboz, Вы писали:
П>>Работа устраивает, проблем не наблюдается, но хочется довести до ума чтобы "уж наверняка"...

A>А зачем вам while(true), лишняя логика по проверке очереди на ноль для Reset-а и флаг isPaused?
Действительно, и зачем только такое понадобилось?

А перечитайте начало топика, посмотрите какие требования к логике представлены нужны и что подсказал Undying по поводу постоянного создания потоков.

                lock(syncRoot)
                {
                    Thread.Sleep(interval);
                }

Даже не знаю что Вам на ЭТО ответить...

Здравствуйте, Serginio1, Вы писали:

S>Здравствуйте, Аноним, Вы писали:

А>>В общем, это меня утомило, честное слово

Все, кто предпочитает думать, что volatile и Lock обладают каким-то волшебством, недоступным нативному, даже ассемблерному, коду, тот, разумеется волен в своих желаниях. А кто хочет разобраться, советую хотя-бы в общих чертах ознакомиться с устройством процессоров. А здесь об этом говорить как-то не очень подходяще.
S> Я же оговорился, что свое время речь шла о кэшах процессоров, и от том, что зависит от архитектуры.
S>Буду рад если дашь толковую ссылку на Контролеер кэшей. Особенно когда ядра будут исчисляться сотнямию
S> volatility так же как и в нативе просто запрещает оптимизацию для кэширования значений в CPU.
S>А Lock это обычная критическая секция. Так, что Net в этом плане недалеко от натива ушел

Извините, если всё-таки обидел, не правильно поняв, я не хотел. Вы имели в виду, что кэши ранее обсуждались где-то в другом месте, а не в статье? И к каким выводам пришло обсуждение? Случайно не помните, хотя-бы где и когда, это было, любопытно было-бы взглянуть.

По поводу ссылок — с удовольствием-бы, но я их не сохранил. Всё-таки это не вчера было, и даже не позавчера

Если не ошибаюсь, на IXBT было несколько очень неплохих статей. Да я думаю, найти будет не сложно, во всяком случае не помню, чтобы это вызвало какие-то затруднения. Надо только не забывать, что Intel и AMD строит свои кэши по-разному. Одноко для обеспечения когерентности существует достаточно старый протокол, принятый ещё в 90-х годах., и все производители его поддерживают. Название протокола — извините, тоже забыл

	От:	Undying
	Дата:	02.07.09 11:29
	Оценка:

	От:	Аноним
	Дата:	02.07.09 11:32
	Оценка:

	От:	Аноним
	Дата:	02.07.09 11:42
	Оценка:

	От:	Аноним
	Дата:	02.07.09 12:07
	Оценка:

	От:	Кондраций
	Дата:	02.07.09 12:20
	Оценка:	+1

От:	Serginio1	https://habrahabr.ru/users/serginio1/topics/
Дата:	02.07.09 12:37
Оценка:

От:	Пельмешко	blog
Дата:	02.07.09 13:55
Оценка: