Параллельная обработка в потоке трубопровода

Question

Мне очень нравится концепция трубопровода/труб для применения операций с потоковым источником ввода-вывода. Мне интересно создавать инструменты, которые работают с очень большими файлами журнала. Одна из достопримечательностей перехода на Haskell из Python/Ruby - это более простой способ написания параллельного кода, но я не могу найти документацию об этом. Как я мог настроить поток канала, который считывает строки из файла и работает на них параллельно (т. Е. С 8 ядрами, он должен читать восемь строк и передавать их на восемь различных потоков, которые нужно обработать, а затем снова собирать и т. д.), в идеале с минимальной «церемонией», как это возможно ...

Возможно, можно указать, должны ли линии соединяться по порядку или нет, если это может повлиять на скорость процесса?

Я уверен, что можно было бы что-то соблазнить собой, используя идеи из книги Parallel Haskell, но мне кажется, что выполнение чистой функции параллельно (parmap и т. Д.) В середине рабочего процесса Conduit должно быть очень легко?

Answer 1

В качестве примера «внутреннего параллелизма» упоминается Петр Pudlák в своем комментарии, считают эту функцию (я использую pipes, но может быть реализован с conduit так же легко):

import Control.Monad 
import Control.Lens (view) 
import Control.Concurrent.Async (mapConcurrently) 
import Pipes 
import qualified Pipes.Group as G 
import qualified Control.Foldl as L 

concProd :: Int -> (a -> IO b) -> Producer a IO r -> Producer b IO r 
concProd groupsize action producer = 
     L.purely G.folds L.list (view (G.chunksOf groupsize) producer) 
     >-> 
     forever (await >>= liftIO . mapConcurrently action >>= mapM G.yield) 

Этот функция принимает в качестве параметров размер группы, действие, которое мы хотим запустить для каждого значения типа a, и значение Producer значений a.

Он возвращает новый Producer. Внутренне производитель считывает значения a партиями groupsize, обрабатывает их одновременно и дает результаты один за другим.

Код использует Pipes.Group, чтобы «разделить» оригинального производителя на суб-производителей размером groupsize, а затем Control.Foldl, чтобы «свернуть» каждого субпродюсера в список.

Для более сложных задач вы можете обратиться к асинхронным каналам, предоставленным pipes-concurrency или stm-conduit. Но они вытаскивают вас из «единого трубопровода» мировоззрения ванильных труб/трубопроводов.