Как четко структурировать зависимости между каналами core.async?

Question

Предположим, у меня есть состав вычислений, который я хочу запустить асинхронно с помощью core.async, но, к сожалению, некоторые из функций зависят от вывода других функций. Как мне организовать структурирование этого в моем коде, а также получить лучшую производительность?

Несколько возможных решений я пришел через это

• Prismatic's Graph - кажется разумным, хотя я не проверял его с core.async каналами; тот факт, что он требует использования fnk, немного уступает мне, потому что он требует покупки в DSL для определения функций, но если это лучшее решение, то я не против.
• Javelin cells - только для ClojureScript (в настоящее время) и использует FRP вместо CSP в качестве реализации, но он очень хорошо выполняет моделирование зависимостей между вычислениями через ячейки формулы.
• Onyx - предназначен для распределенных вычислений (в качестве конкурента Apache Storm и т. Д.), Но имеет абстракцию рабочего процесса, которая обрабатывает зависимости между вычислениями и работает с core.async. Это похоже на наиболее подходящий для моего проблемного домена, но я не уверен, что мне нужны накладные расходы на все функции управления кластерами.

Что такое каноническое решение этой проблемы?

Edit: добавлена ​​Onyx

Answer 1

Я не думаю, что существует канонический способ решить, core.async настолько ново, что немногие люди дали ему шанс. Если бы я мог выбирать между тремя вариантами, я бы пошел с Graph, он был развернут и протестирован на производстве некоторое время, и вам не нужен Clojurescript для его запуска. Если вы заинтересованы в решении FRP, взгляните на Java Reactive Extensions, привязки Clojure для него существуют в RxClojure.

Answer 2

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

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

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

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

(let [c (chan)] 
    (thread (>!! c “hello”)) 
    (assert (= “hello” (<!! c))) 
    (close! c) 

Имеются также более сложные механизмы. Функция Alts!! обеспечивает возможность ожидания по многим каналам одновременно. Несколько различных вариантов функции pipeline позволяют моделировать параллелизм в потоке данных как способ.

Существуют ли какие-либо конкретные проблемы, с которыми вы сталкиваетесь, которые не могут быть четко выражены с помощью встроенных функций?