Concurrency: 6 способов жить с shared state

[@NCrashed]

Замечание из личного опыта: Isolated mutability гораздо сложнее покрыть тестами, нужно учитывать, что сообщения могут скапливаться в очередях и собранные с акторов мгновенные значения (например, количество обслуживаемых подключений) могут не совпадать с ожидаемыми (сумма активных и неактивных соединений может быть не равна общему числу соединений, как пример).

[@pdima]

А как покрывать тестами без isolated mutability? Тестировать многопоточные приложения с shared state куда сложнее, тут isolated mutability часто как раз помогает.

[@NCrashed]

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

[@pdima]

спасибо, интересно было бы посмотреть на примеры таких тестов. Как они работают понятно, но не очень хочется добавлять зависимость от таких мютексов в API.

А проблем тестирования самих акторов в случае с isolated mutability я не вижу, нужно тестировать реакцию на различные сообщения независимо от наличия очереди, точно так же как в однопоточном варианте.

[@NCrashed]

Пример взят отсюда, тестирование read-write mutex (сорри, что код на D, но в данном случае от джавы почти не отличается):

    //...
    void readerFn()
    {
        synchronized (mutex.reader)
        {
            atomicOp!"+="(numReaders, 1);
            rdSemA.notify();
            rdSemB.wait();
            atomicOp!"-="(numReaders, 1);
        }
    }

    void writerFn()
    {
        synchronized (mutex.writer)
        {
            atomicOp!"+="(numWriters, 1);
            wrSemA.notify();
            wrSemB.wait();
            atomicOp!"-="(numWriters, 1);
        }
    }

    // ...

    scope group = new ThreadGroup;

    // 2 simultaneous readers
    group.create(&readerFn); group.create(&readerFn);
    rdSemA.wait(); rdSemA.wait();
    assert(numReaders == 2);
    rdSemB.notify(); rdSemB.notify();
    group.joinAll();
    assert(numReaders == 0);
    foreach (t; group) group.remove(t);

А проблем тестирования самих акторов в случае с isolated mutability я не вижу, нужно тестировать реакцию на различные сообщения независимо от наличия очереди, точно так же как в однопоточном варианте.

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

[@seregamorph]

Стоит заметить, что в варианте неблокирующего оптимистичного алгоритма есть существенный недостаток: при высокой конкуренции вероятность получить false на compare-and-swap становится достаточно высока, особенно с большими значениями BigInteger (что приведет к большому числу переповторов впустую), поэтому есть смысл подумать о других вариантах, например synchronized/ReentrantLock.

Кроме того, пожалуй, метод val() излишний в api, потому что класс в таком виде становится не Thread safe (условно, конечно, но его контракт несколько странный).

[@f0y]

в варианте неблокирующего оптимистичного алгоритма есть существенный недостаток: при высокой конкуренции вероятность получить false на compare-and-swap становится достаточно высока, особенно с большими значениями BigInteger

Да, вы правы. Я сделал перформанс тесты (спасибо огромное TheShade) и до определенного момента неблокирующий алгоритм выигрывает в 2 раза у блокирующих. Затем деградириует и в конце концов становится медленнее. Здесь он просто приведен для демонстрации подхода.

Кроме того, пожалуй, метод val() излишний в api, потому что класс в таком виде становится не Thread safe

В какой реализации? BigInteger же immutable.

[@seregamorph]

Я не об этом, между next и val нет синхронизации, но если рассматривать как аналогию AtomicInteger get/incrementAndGet, то для примера пойдет, а в боевой реализации я бы без нужды метод val добавлять не стал. Это все условности, поэтому не стоит придавать этому большое значение.

[@Throwable]

Одну важную вещь всегда опускают, когда говорят про транзакции — это принцип «let it fault». Транзакция не обязана завершиться успешно. В случае, когда возникает коллизия на write, одна из них всегда откатывается. Поэтому приложение должно уметь отлавливать такие случаи и повторять попытку изменений с самого начала. Практически нигде в примерах это даже не упоминается. К сожалению, во всех системах, с которыми я имел дело, подобная проверка отсутствует (если это не делается самой платформой).

Другое свойство, которое не часто упоминается, это изолированность транзакций. Большинство STM сделано на схеме MVCC, а она не отвечает «full-serialized» уровню, поскольку отсутствуют read locks. Работа на таком уровне чревата неприятным эффектом write skew.

На самом деле я не вижу сильно ограниченным использование STM, разве что в экзотических решениях. Для несложных структур можно использовать стандартные классы библиотеки java.concurrent, а сложные графы хранить в памяти без persistence как правило нет нужды. В большинстве архитектур shared state реализуется как раз на уровне persistence в базе данных, а все, что находится в памяти — это временные данные: либо immutable объекты, либо «isolated mutability». STM не масштабируются: для этого есть множество других решений типа distributed cache. Обход большого графа в STM достаточно накладен, требуются дополнительные индексирующие структуры для быстрого доступа. Вобщем, in-memory движки баз данных уже имеют весь необходимый функционал и являются более стандартным решением для хранения shared state.

В одно время в проекте встал вопрос использовать STM или in-memory database, и выбор именно на последний (JPA+H2).

[@f0y]

Спасибо за комментарий!

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

В комментах упоминается

Если да, то нас опередили, но мы оптимистичны и повторяем транзакцию еще раз.

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

private final TransactionFactory txFactory = new TransactionFactoryBuilder().setMaxRetries(1_000_000) .build();

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

[@Colwin]

Поэтому операции внутри транзакции должны быть идемпотентны.

Этого часто ой как сложно добиться :-)

[@rfq]

6 этих способов можно разбить на группы:

a) универсальные:
— пессимитические: synchronized, locks
— оптимистические: lock-free, stm
b) частные:
— иммутабельные: обеспечивают только чтение общего состояния
— isolated mutability: инициирующий поток не нуждается в результате (по крайней мере, немедленно), поэтому может выдать задание (или послать сообщение) и продолжить свою работу. Замечу, выдача задания также требует работы с общим состянием (планом вычислений) и, соответственно, обращения к универсальным методам.

[@Lol4t0]

Что-то мне кажется, что никакой выгоды от применения хитрой синхронизации вместо обычных блокировок вы тут не получите, особенно, если слегка модифицировать первый пример, использовав Atomic<BigInteger> curr (или как там в Java) и отказаться от synchronized в get.

В таком случае Fine-Grained становится просто не нужен.

В 'Lock free' вы точно так же блокируетесь на записи, только еще и загружаете все время при этом процессор.

Транзакции внутри, наверное, тоже используют блокировки.

Immutable — совершенно другой интерфейс, его просто нельзя ставить в один ряд со всеми остальными.

Передача сообщений опять же внутри использует блокировки.

Это я все к тому, что иногда нет смысла придумывать что-то сложное, когда решение лежит на поверхности.

Как обзор технологий — норм, только пример надо было поудачней подобрать.

[@vanxant]

Сообщения можно и без блокировок отправлять, если у каждого потока своя личная очередь исходящих сообщений, в которую пишет только он (хотя это и усложняет чтение входящих)

[@Colwin]

Тогда блокировки будут на чтении ) за все надо платить.

[@nehaev]

Раз уж в посте упомянута Akka, не могу не привести ссылку на разбор авторами этого проекта основных моделей многопоточного программирования Concurrency – The good, the bad, the ugly.

[@orionll]

Есть еще один способ: использовать агенты. Они есть в Clojure и Akka.

[@kemsky]

LockFree вариант делает лишнюю работу и чем больше нагрузка и сложнее операция, тем больше. В пределе это практически livelocking.
Мне кажется, Immutable пример неполный, смысл ведь не в том, чтобы генерировать числа (тогда можно просто брать новый генератор и все), а в том, чтобы не было повторов, в этом случае придется где-то еще хранить ссылку на последний Immutable и иметь все связанные с этим проблемы. Поправьте, если я ошибаюсь.

[@f0y]

О каких проблемах идет речь? Если ссылку на immutable объект расшарить между двумя потоками, то потоки, одновременно работающие с ним, не смогут его «испортить», так как объект неизменяемый. В этом суть данного подхода. Конечно он не полностью решает все проблемы и не всегда подходит.

[@kemsky]

Расшарить не проблема, но как гарантировать последовательность чисел? Новое число будет одинаковым для потоков между которыми расшарен immutable, что отличается от всех остальных примеров в статье.