Неблокирующие очереди: обмен сообщениями между потоками

[@halyavin]

Пробовали смотреть на code.google.com/p/disruptor/? Вроде он решает эту же задачу.

[@B08AH]

Если не ошибаюсь, он как раз использует кольцевой буфер. Следовательно возможны потери сообщений.

[@Yakud]

Отформатируйте, пожалуйста, код.
Неудобно читать!

[@B08AH]

тэг code почему то все снес влево вне зависимости от отступов,
каким тэгом делают, чтоб код прилично выглядел?

[@akzhan]

source lang=

[@gribozavr]

Все ошибки, которые можно допустить в многопоточной программе, вы здесь допустили, прикрываясь термином «lockless», изобретённым тоже вами.

[@romik]

А вы тестировали этот код? А на многоядерной машине?
Может изначальная реализация на C и работала, но этот код работать не будет: в нём нет ни одного memory barrier и видимость данных в разных потоках ничем не обеспечивается.
В статье, на которую вы дали ссылку, неспроста упоминается ключевое слово volatile.

[@B08AH]

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

[@romik]

Ну значит вам везло. Впрочем, тестирование всё равно не докажет что код корректен. Оно может лишь доказать что он некорректен.
В данном случае в его некорректности нет никаких сомнений.

[@GrossHo]

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

[@romik]

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

[@romik]

Ну, в С, где стоит вопрос освобождения памяти, нет сборщика мусора. А где есть сборщик мусора — её не надо освобождать :)
Другое дело что если выделение/освобождение памяти делать в разных потоках, то либл сами malloc/free будут блокирующими и называть алгоритм lock-free уже нельзя, либо malloc/free неблокирующие, но тогда внутри них самих реализована lock-free queue :)
В жизни, насколько я понимаю, возможны оба варианта в зависимости от системы и размера запрашиваемой области памяти.

[@B08AH]

Не вижу причин почему нельзя malloc делать в одном треде, а free в другом. А вот с тем, что он может быть блокирующим возможно и связано то, что уменьшив количество блоков в 8 раз я получил двукратный рост скорости. 32х уже дает единицы процентов. Полностью lock-free конечно уже нельзя назвать, однако на блокирующей очереди блокировка идет на каждый вызов и количество блокировок никак не уменьшить, а количество маллоков урезается на раз.

[@akalend]

malloc делать в одном треде, а free в другом делать нельзя.

[@korisk]

Почему?

[@akalend]

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

[@GrossHo]

Как в Си будет работать malloc/free реализация зависит от используемого аллокатора. Умные выступают за повторное использование ресурсов. И если один поток выделяет память, передает другому, где тот складывает неиспользуемые ресурсы у себя — повторное использование не столь эффективно, как если б выделение и высвобождение производилось бы в одном потоке.

[@zelyony]

public bool consume(out object message)
{
if (first == last || first.next == null)
{
message = null;
return false;
}
message = first.next.message;
first = first.next;
return true;
}
а здесь правда нет рейс-кондишенов? )

[@zvulon]

MSMQ?
Async CTP?
Go?
Erlang ?!

вам поиграться или в продакшн?
ну или вы в R&D?

вобщем странно все это.

[@adontz]

msdn.microsoft.com/en-us/library/dd267265.aspx

[@Ryadovoy]

Сдается мне что мьютекс здесь нужно использовать не только при добавлении/удалении threadNode, но и при поиске threadNode в момент отправки.
Могу ошибаться из-за недопонимания C#, но разве безопасно в одном потоке вызывать метод объекта byTID.TryGetValue(), а в другом подменять ссылку на объект byTID = newbyTID?

На С++ данный момент описан мной в статье (см. последний пример кода)

[@B08AH]

Мютекс используется не для добавления элементов в словарь, а для его подмены. Тут неважно получил ли поток этот поинтер до апдейта или после, он в обоих случаях валидный.
Но в случае разрегистрации сообщение может добавиться в очередь той ноды которая только что разрегистрировалась.
В случае шарпа указатель на ноду «потеряется» и вместе с содержанием очереди уйдет к GC. А вот в случае с неплюсовым си, пришлось делать т.н. отложенную очистку. Там, в частности из-за подобных вещей, код выглядит более громоздко, на шарпе все минимализировалось. Очень понравился язык.

[@Ryadovoy]

Что означает выражение «получил поток поинтер»?

Можно ликбез по C#?
Обязан ли вызов метода класса, увеличивать время жизни ссылки-члена класса при обращении, если эта ссылка-член класса будет подменяться в другом потоке?
Метод класса считывает ссылку-член класса каждый раз при обращении, или кеширует ее?

Если Ваш алгоритм рассчитан на какие-то особенности языка, это должно быть указано в явном виде, лучше прямо в коде в виде комментария.

[@B08AH]

К C# не относится, поинтер есть поинтер. Чтобы получить адрес метода TryGetValue резолвится поинтер byTID, там в момент апдейта фактически два объекта, и поток может выполнить эту операцию как со старым объектом, так и с новым. Операция замены поинтера атомарна по определению, поэтому вникуда попасть нельзя, в любом случае там валидный объект.

[@Ryadovoy]

>>К C# не относится, поинтер есть поинтер
Относится, т.к. здесь очень важен вопрос времени жизни объекта byTID.

Если без лока разрезолвить поинтер и вызвать у него метод, то на C++ объект может удалится вторым потоком в любой момент времени, и синхронизация просто необходима для корректной работы.

На C# есть зборщик мусора, и все объекты, как я понимаю, являются ссылками.
Как оно работает я в точности не знаю, но подозреваю что здесь есть свои ньюансы, и что все не так просто и нельзя так бездумно пользоваться такими вещами в многопоточной среде.

Что будет, к примеру, если сборщик мусора будет производить очистку в отдельном потоке в произвольный момент времени?

[@B08AH]

->на C++ объект может удалится вторым потоком в любой момент времени
А не надо его сразу удалять. Перекладываешь его в другое место и удаляешь позже, по таймауту.
GC эту задачу решает за разработчика и удаляет объект когда он гарантировано не используется. Если один тред успел схватить поинтер на одну процессорную операцию раньше, чем второй его подменил, GC его не удалит.

А вообще в моем понимании никаких новых тредов при работе быть не должно. Они все должны стартовать при начале и завершаться в конце.

[@Ryadovoy]

>>А не надо его сразу удалять. Перекладываешь его в другое место и удаляешь позже, по таймауту.

Никакого там таймаута не нужно, нужно лишь правильно пользоваться синхронизацией и умными указателями.
В моей реализации схожего алгоритма на C++ был один неочевидный нюанс — умный указатель на список, перед использованием, должен быть скопирован под локом, и нельзя напрямую обращаться к умному указателю, хранящемуся мембером, т.к. он в любой момент времени может быть подменен и удален в другом потоке и есть большая вероятность обращения к удаленному объекту.

Мне кажется схожие проблемы могут быть и в Вашей реализации на C#.
По крайней мере стоит упомянуть о проблемных моментах и что использование именно C# их решает.

>>А вообще в моем понимании никаких новых тредов при работе быть не должно. Они все должны стартовать при начале и завершаться в конце.

Зачем тогда усложнять код использованием синхронизации и копированием при записи?

[@B08AH]

Ежели боязно делать непосредственно, можно сделать сначала tmp=byTID; а потом tmp.TryGetValue();

[@DSL88]

Помнится совсем недавно писал систему сообщений между потоками без блокировки на Delphi.
Была только одна проблема — опаздывание сообщений, поскольку в список оно могло добавиться в тот же момент, как в нем была проверка.
Пришлось писать систему «звонка», когда очередь сама сообщала о новых сообщениях потоку.

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

[@B08AH]

Поэтому надо делать разрегистрацию.

 register(...);
 while(!terminated)
 {
   //process messages
 }
 unregister();

[@khaale]

А вот и исходники ConcurrentQueue под Mono
без Interlocked.CompareExchange и там не обошлось.

[@B08AH]

Хочется обратить внимание на метод TryPeek
Интерлока при чтении нет, только при записи.
Мораль: один пишет, многие читают = потокобезопасность.

    public bool TryPeek (out T value)
    {
      if (IsEmpty) {
        value = default (T);
        return false;
      }
      
      Node first = head.Next;
      value = first.Value;
      return true;
    }

[@romik]

Только вот Peek и Take разные операции. Take (достать элемент из очереди) меняет состояние.
А Peek в ConcurrentQueue.cs работает правильно без «интерлока» только потому, что об этом позаботились в методе Enqueue, правильно расставив Interlocked операции. В общем же случае «один пишет, многие читают» совсем не означает потокобезопасность.

Так что B08AH, озаботились бы вы хотя бы чтением теории перед тем как лезть в отнюдь не простую область lock-free алгоритмов.

[@B08AH]

->Только вот Peek и Take разные операции.
Да неужели. А что будет если во время Peek, после проверки isEmpty, операция TryDequeue из другого потока считает последний элемент?
Опять жу будет резолвиться указатель head, и в это время другой поток его подменяет.
Ryadovoy вот возмущается именно по этому поводу.
->В общем же случае «один пишет, многие читают» совсем не означает потокобезопасность.
В общем нет, а в случае атомарных операций(подмена одного указателя), где данные валидны и до и после апдейта — все безопасно. Об этом кстати можно прочитать если пройти по ссылке, которую я дал.

[@romik]

> Да неужели. А что будет если во время Peek, после проверки isEmpty, операция TryDequeue из другого потока считает последний элемент?

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

> В общем нет, а в случае атомарных операций(подмена одного указателя), где данные валидны и до и после апдейта — все безопасно. Об этом кстати можно прочитать если пройти по ссылке, которую я дал.

Только если подмена указателя делается через Interlocked, иначе публикация ссылки вполне может произойти раньше чем обновление данных, несмотря на порядок выражений в коде. Почитайте уже наконец теорию, сколько можно объяснять базовые вещи.

[@B08AH]

->публикация ссылки вполне может произойти раньше чем обновление данных, несмотря на порядок выражений в коде
Теория говорит, что операции записи, в т.ч. если один из них внутри вызова функции, не могут меняться местами. habrahabr.ru/blogs/net/130318/

[@romik]

Согласен, оказывается в .NET модель памяти в этом месте более жёсткая чем в привычной мне Java. Впрочем, от этого число проблем синхронизации в вашем коде сильно меньше не стало.

[@B08AH]

С синхронизацией проблем никаких нет. Траффик вертится сколько угодно без каких либо потерь сообщений и ексепшенов.
Но есть проблемы с производительностью. Простая конструкция на ConcurrentDictionary<string,ConcurrentQueue

[@B08AH]

Потерялась часть сообщения.… работает существенно быстрее.

[@akalend]

>точнее профайлер, выявил, что malloc это относительно дорогая операция
по этому все используют пулы