하스켈 : 스폰을 사용하지 않고 파이프 (브로드 캐스트) 나누기

Question

이 질문은 bit codegolf 및 newb입니다.

안녕하세요,라이브러리를 하스켈에서 사용하고 있습니다. 동일한 데이터를 여러 채널 (브로드 캐스트)을 통해 보내도록 파이프를 분할하고 싶습니다. Pipes.Concurrent tutorial은 Output의 모노oid 상태를 활용하여 spawn을 사용하여 사서함을 만들 것을 제안합니다.

main = do 
(output1, input1) <- spawn Unbounded 
(output2, input2) <- spawn Unbounded 
let effect1 = fromInput input1 >-> pipe1 
let effect2 = fromInput input2 >-> pipe2 
let effect3 = P.stdinLn >-> toOutput (output1 <> output2) 
... 

정말 필요한 사서함을 통해이 간접인가 : 예를 들어 , 우리는 같은 것을 할 수 있는가? 대신 다음과 같이 작성할 수 있습니까?

main = do 
let effect3 = P.stdinLn >-> (pipe1 <> pipe2) 
... 

Pipe는 Monoid 인스턴스를 가지고 있지 않기 때문에 위의 컴파일되지 않습니다. 여기에는 적절한 이유가 있습니까? 첫 번째 방법이 실제로 파이프를 분할하는 가장 깨끗한 방법입니까?

Answer 1

동시성을 사용하지 않고이 작업을 수행하는 두 가지 방법이 있는데, 둘 다주의해야합니다. 이것은 단지 작성하는 것입니다 해결하기

p1 = for cat f -- i.e. p1 = forever $ await >>= f 
p2 = for cat g -- i.e. p2 = forever $ await >>= g 

... 다음 쉬운 방법 :

첫 번째 방법은 pipe1 및 pipe2이 단순한 경우 Consumer의 그 루프가 영원히 좋아한다는 것입니다

for P.stdinLn $ \str -> do 
    f str 
    g str 

예를 들어

, p1 단지 print 모든 값 보내고 인 경우

p1 = for cat (lift . print) 

을 012,351,

... 그리고 p2 그냥 핸들에 그 값을 쓰고있다 :

p2 = for cat (lift . hPutStrLn h) 

을 ... 당신은 지금처럼 그들을 결합 할 것이다 :

for P.stdinLn $ \str -> do 
    lift $ print str 
    lift $ hPutStrLn h str 

그러나,이 단순화는 Consumer 작동 그 고리는 사소한 것입니다. 더 일반적인 솔루션은 파이프 용 ArrowChoice 인스턴스를 정의하는 것입니다. 나는 풀 기반 Pipe s는 올바른 법을 준수하는 인스턴스를 허용하지만, 푸시 기반하지 않는 것이 Pipe의 할 생각 :

newtype Edge m r a b = Edge { unEdge :: a -> Pipe a b m r } 

instance (Monad m) => Category (Edge m r) where 
    id = Edge push 
    (Edge p2) . (Edge p1) = Edge (p1 >~> p2) 

instance (Monad m) => Arrow (Edge m r) where 
    arr f = Edge (push />/ respond . f) 
    first (Edge p) = Edge $ \(b, d) -> 
     evalStateP d $ (up \>\ unsafeHoist lift . p />/ dn) b 
     where 
     up() = do 
      (b, d) <- request() 
      lift $ put d 
      return b 
     dn c = do 
      d <- lift get 
      respond (c, d) 

instance (Monad m) => ArrowChoice (Edge m r) where 
    left (Edge k) = Edge (bef >=> (up \>\ (k />/ dn))) 
     where 
      bef x = case x of 
       Left b -> return b 
       Right d -> do 
        _ <- respond (Right d) 
        x2 <- request() 
        bef x2 
      up() = do 
       x <- request() 
       bef x 
      dn c = respond (Left c) 

이이 유형의 매개 변수 순서가 ArrowChoice이 기대에 있도록 newtype은 필요합니다. 당신이 용어 Pipe 푸시 기반 익숙하지 않다면

, 그것은 기본적으로 대신 최 하류 파이프의 가장 상류 파이프에서 시작하는 Pipe, 그리고 그들 모두는 다음과 같은 모양이 있습니다

a -> Pipe a b m r 

을 업스트림에서 적어도 하나의 값을 수신 할 때까지 "갈"수없는 Pipe으로 생각하십시오.

이 푸시 기반 Pipe의 자신의 구성 사업자와 정체성 완료, 기존의 풀 기반 Pipe의에 "이중"입니다

(>~>) :: (Monad m) 
     => (a -> Pipe a b m r) 
     -> (b -> Pipe b c m r) 
     -> (a -> Pipe a c m r) 

push :: (Monad m) 
     -> a -> Pipe a a m r 

...하지만 단방향 Pipes API는 내 보내지 않습니다 이것은 기본적으로. 이 연산자는 Pipes.Core에서만 얻을 수 있습니다. (더 자세히 공부하여 어떻게 작동하는지 직감을 얻을 수도 있습니다.) 이 모듈은 푸시 기반 Pipe과 끌어 오기 기반 Pipe이 둘 다 일반적인 양방향 버전의 특수한 경우이며 양방향 대소 문자를 이해하는 것이 왜 서로 쌍방인지를 어떻게 알 수 있는지 보여줍니다. 그런 다음이 완료되면 풀 (pull) 기반의 파이프에 그 변환 runEdge을 사용

p >>> bifurcate >>> (p1 +++ p2) 
    where 
    bifurcate = Edge $ pull ~> \a -> do 
     yield (Left a) -- First give `p1` the value 
     yield (Right a) -- Then give `p2` the value 

:

당신은 푸시 기반 파이프에 대한 Arrow 인스턴스가되면, 당신은 뭔가를 쓸 수 있습니다.

이 접근법에는 끌어 오기 기반 파이프를 밀어 넣기 기반 파이프로 자동 업그레이드 할 수 없다는 단점이 있습니다 (하지만 일반적으로 수동으로 수행하는 방법을 찾는 것은 어렵지 않습니다). 예를 들어, 수 Pipes.Prelude.map를 업그레이드 푸시 기반 Pipe, 당신은 작성합니다 물론

mapEdge :: (Monad m) => (a -> b) -> Edge m r a b 
mapEdge f = Edge (mapPush f) 

, 다음 Arrow에 싸서받을 권리 유형이 다음

mapPush :: (Monad m) => (a -> b) -> (a -> Pipe a b m r) 
mapPush f a = do 
    yield (f a) 
    Pipes.Prelude.map f 

더 간단한 방법은 처음부터 작성하는 것입니다 :

mapEdge f = Edge $ push ~> yield . f 

사용 중 접근 방식은 가장 적합한.

실제로 정확히 같은 질문에 대답하려고했기 때문에 Arrow 및 ArrowChoice 인스턴스를 만들었습니다. 동시성을 사용하지 않고 이러한 문제를 어떻게 해결할 수 있습니까? 나는이 더 일반적인 주제에 대해 다른 스택 오버 플로우 응답 here에 대해 썼습니다. 여기에서이 Arrow 및 ArrowChoice 인스턴스를 사용하여 동시 시스템을 동등한 순수 시스템으로 증분하는 방법을 설명합니다.