Может ли общий источник работать с несколькими блоками крана одновременно?

Рассмотрим этот код, в котором кратковременное нажатие завершается. Все блоки работают одновременно (сразу же выводятся), затем спят. Большинство из них не заканчиваются, потому что программа заканчивается раньше, чем они:

my $supply = Supply.interval(0.2);
my $tap = $supply.tap: { say "1 $^a"; sleep 5;  };
sleep 5;

Выход (имеет выход) имеет 25 строк (по одному для каждого тика 0,2 в 5 секунд):

1. 0
1. 1
...
1. 24

Затем я меняю это предложение на .share:

my $supply = Supply.interval(0.2).share;
my $tap = $supply.tap: { say "1. $^a"; sleep 5 };
sleep 5;

Я вижу только одну строку ввода, но я ожидал того же выхода:

1. 1

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

my $supply = Supply.interval(0.2).share;
my $tap  = $supply.tap: { say "1. $^a"; sleep 5 };
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

Тем не менее выход имеет выход только для первого крана и по-прежнему имеет только одну линию. Я ожидал 25 строк для каждого:

1. 1

Ответ 1

Основными правилами Supply являются:

Без введения параллелизма без явного запроса
Противодавление через модель отправителя-платит
Сообщение обрабатывается полностью до следующего (так что .map({...something with state... }) можно доверять, чтобы не вызвать конфликты над состоянием)

Правило 3 на самом деле не применяется к share поскольку после этого происходит отдельная цепочка операций после операции, но правила 1 и 2 действительно выполняются. Целью share является публикация/подписка, а также предоставление повторного использования части обработки несколькими обработчиками сообщений ниже по потоку. Внедрение параллельной обработки сообщений - это отдельная проблема.

Существуют различные варианты. Один из них - иметь сообщения для параллельной обработки, вставленные в Channel. Это явно вводит место для буферизации сообщений (ну, пока у вас не закончится память... именно поэтому Supply поставляется с моделью обратного давления с отправителем). Принуждение Channel обратно в Supply получает значения, вытащенные из Channel и испускаемые этим источником Supply в потоке бассейна. Это выглядит так:

my $supply = Supply.interval(0.2).share;
my $tap  = $supply.Channel.Supply.tap: { say "1. $^a"; sleep 5 };
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

Обратите внимание, что поскольку whenever автоматически применяется то, на что он просил реагировать на Supply, тогда это будет выглядеть whenever $supply.Channel { }, что делает его довольно кратким решением, - но в то же время красиво явным в том whenever $supply.Channel { }, что оно указывает как работает механизм обратного давления. Другим свойством этого решения является то, что он сохраняет порядок сообщений и по-прежнему дает обработку в один момент времени после Channel.

Альтернативой является реагировать на каждое сообщение, вместо этого запуская какую-то асинхронную часть работы для ее обработки. start операция в Supply рассылает блок, который передается для запуска в пуле потоков для каждого принятого сообщения, что не блокирует приход следующего сообщения. Результатом является Supply Supply. Это заставляет нажимать на каждый внутренний Supply чтобы на самом деле что-то произошло, что вначале кажется слегка противоречащим интуиции, но на самом деле это полезно для программиста: он дает понять, что есть дополнительный бит асинхронной работы для отслеживания. Я настоятельно рекомендую использовать это в сочетании с синтаксисом " react/whenever, который автоматически выполняет управление подпиской и распространение ошибок. Наиболее прямое преобразование кода в вопрос:

my $supply = Supply.interval(0.2).share;
my $tap  = supply { whenever $supply.start({ say "1. $^a"; sleep 5 }) { whenever $_ {} } }.tap;
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

Хотя также можно написать вместо этого:

my $supply = Supply.interval(0.2).share;
my $tap  = supply { whenever $supply -> $a { whenever start { say "1. $a"; sleep 5 } {} } }.tap;
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

Что указывает на возможность написания parallelize Supply:

my $supply = Supply.interval(0.2).share;
my $tap  = parallelize($supply, { say "1. $^a"; sleep 5 }).tap;
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

sub parallelize(Supply $messages, &operation) {
    supply {
        whenever $messages -> $value {
            whenever start operation($value) {
                emit $_;
            }
        }
     }
}

Выход этого подхода сильно отличается от первого Channel, поскольку все операции запускаются, как только приходит сообщение. Также он не сохраняет порядок сообщений. Там все еще есть неявная очередь (в отличие от явной с подходом " Channel), это просто то, что теперь очередь работы с планировщиком пула потоков и планировщик ОС, который должен отслеживать незавершенную работу. И опять же, нет противодавления, но обратите внимание, что было бы вполне возможно реализовать это, отслеживая выдающиеся Promises и блокируя дальнейшие входящие сообщения с await Promise.anyof(@outstanding).

Наконец, я хотел бы отметить, что race whenever построении race whenever либо рассматривался вопрос о hyper whenever чтобы обеспечить некоторый механизм языкового уровня для параллельной обработки сообщений о Supply. Однако семантика таких, и как они играют в supply -block цели проектирования и свойство безопасности, представляют собой значительные проблемы проектирования.

Ответ 2

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

my $supply = Supply.interval(0.2).share;
my $tap  = $supply.tap: { say "1. $^a in #{+$*THREAD}" };
my $tap2 = $supply.tap: { say "2. $^a in #{+$*THREAD}" };
sleep 0.5;
===================
1. 1 in #4
2. 1 in #4
1. 2 in #4
2. 2 in #4

Так что ответ в настоящее время: no