Modelling deadlock in cyclic pipelines in Haskell with coroutine libraries

Question

I'm modelling two processes which have been put into a cyclic pipeline - the output of each feeds into the input of the other - so as to work out when they've deadlocked (i.e., neither can progress since they're each waiting on output from the other).

I've used the conduit package for this, as I couldn't see an easy way to do it using pipes, and streaming doesn't really look suited to this sort of task. It looks like monad-coroutine would be another possibility for this, but I didn't investigate it further.

Define the problem as follows:

We have 2 processes executing a program made up of two sorts of instruction: "send" and "receive". Each process has an incoming mailbox and can send to an outgoing mailbox, each of which is the end of a FIFO queue of unbounded length. Let's assume the items of data they're sending and receiving are Ints.

The instruction send n will send the int n to the process's outgoing mailbox (from which it can be retrieved by whatever program is monitoring the other end of the queue). receive will try and retrieve an int from the process's incoming mailbox. If there's nothing in the mailbox, the process will block until there is.

Processes execute each instruction in turn, and if there are no more instructions, they exit.

Now, assume we join 2 processes A and B "head to toe" in a cycle: process A receives data from process B's outbox, and sends data to process B's inbox, and vice versa.

Given a particular program P, we wish to simulate their execution to find out (a) whether the 2 processes deadlock, when given program P, and (b) how many items of data each process sends before the 2 of them either deadlock or come to the end of their instructions.

I think this problem is of interest because pre-emptive concurrency doesn't seem like a good way of solving it -- most concurrency libraries do their best to help you avoid deadlock, not model processes that have got into deadlock. (But perhaps I'm wrong, and this is easily modelled with a standard concurrency library - I'd be keen to hear.) Also it gave me a good reason to look at some of the streaming data packages (conduit, pipes and streaming), all of which I believe are modelled around the idea of processes that can "yield" data "downstream", or "await" it from "upstream", which is exactly what this problem requires.

Here's my code:

(NB: contains possible spoilers for the Advent of Code 2017, day 18 problem, Part 2 - but this is not relevant to my question, which is about modelling deadlock with coroutines.)

-- requires packages:
--    microlens-platform
--    mtl
--    conduit
--    conduit-combinators

{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE FlexibleContexts #-}

import Conduit                    ( (.|), ConduitM, yield, await, fuseBoth,
                                    yieldMany, runConduitPure ) 
import Data.Conduit.Lift          (execStateC)
import Data.Conduit.List          (consume) 

import Lens.Micro.Platform

import Control.Monad.State        (MonadState)
import Control.Monad              (when, unless)

data Instr =
    Send Int 
  | Recv
  -- ... stub: imagine further operations here, e.g.
  --  acting on a store, conditional jumps to other instructions, etc.
  deriving Show

-- | state of a program
data ProgState = ProgState {
      _program   :: [Instr] -- ^ remaining instructions to run 
    , _outputCnt :: Int     -- ^ how many times we've done a "Send"
  }
  deriving Show

-- programs initially haven't sent anything
mkProgState :: [Instr] -> ProgState
mkProgState instrs = ProgState instrs 0

makeLenses ''ProgState

-- | perform one operation, using 'yield' and 'await'
-- to "send" and "receive" values.
-- return a Bool - whether we can continue, or are 
-- blocked on a receive and should abort.
applyOp
  :: MonadState ProgState m => 
      Instr -> ConduitM Int Int m Bool
applyOp instr = case instr of
  Send n    -> do yield n
                  outputCnt += 1
                  return True
  Recv      -> do valM <- await
                  case valM of
                     Nothing  -> return False
                     Just _val -> -- stub: ..do something with received vals
                                 return True

-- Given initial state:
--  Execute instructions in sequence until either
--  (a) there are no more left, or
--  (b) we're blocked while receiving
--  and return the new state.
runLoop :: Monad m => ProgState -> ConduitM Int Int m ProgState
runLoop state =
  execStateC state loop
  where
  loop :: MonadState ProgState m => ConduitM Int Int m ()
  loop = do
    prog <- use program
    unless (null prog) $ do  -- still more instructions
      let instr = head prog
      canContinue <- applyOp instr
      when canContinue $ do
        program %= tail -- step forward 1 instruction
        loop

-- | put 2 program processes in sequence, one feeding the other.
-- In addition to program states, takes input to program A,
-- and returns output from program B.
pipe
  :: [Int]
     -> (ProgState, ProgState) -> ((ProgState, ProgState), [Int])
pipe  input (stateA, stateB) =  
  let
      (=.|=) = fuseBoth  -- join 2 conduits in sequence,
                         -- and return results from both as a tuple

      -- get the side effect result of both programs A and B,
      -- also what B emits, as a list (using 'consume')
      conduit = 
          yieldMany input .| 
            runLoop stateA   =.|=
                  runLoop stateB   =.|=
                     consume
  in runConduitPure conduit

-- simulate the effect of joining our pipeline to its own
-- start, creating a cycle - and keep running until
-- the processes finish or are deadlocked (i.e.,
-- produce no output because neither can continue)
runCycle :: ProgState -> ProgState -> (ProgState, ProgState)
runCycle  =
  loop [] 
  where
    loop input stateA stateB = do
      let ((stateA', stateB'), output) = pipe input (stateA, stateB)
      if null output
        then (stateA', stateB')
        else loop output stateA' stateB'

-- Give 2 processes a program to run that is guaranteed
-- to result in them deadlocking, when joined in a cyclic
-- pipeline. 
-- count how many items each outputs before deadlock happens.
test :: (Int, Int)
test = 
  let instrs = [
          Send 1
        , Send 2
        , Recv
        , Recv
        , Recv
        ]
      (stateA, stateB) = runCycle (mkProgState instrs) (mkProgState instrs)
  in (stateA ^. outputCnt, stateB ^. outputCnt)

main :: IO ()
main = do
  let (aCount, bCount) = test
  putStrLn $ "program A emitted " ++ show aCount ++ " items"
  putStrLn $ "program B emitted " ++ show bCount ++ " items"

My questions are:

• Can you see any opportunities for improvement here, especially simpler ways of modelling the problem?
• Could pipes or streaming be used instead? I couldn't see an obvious way to do so. Or would some other package be better - should I have tried monad-coroutine? Or machines, perhaps? (I saw from its description that it might be relevant, but haven't investigated further -- ...there's only so many hours in the day.)
• I assume modelling this with any sort of pre-emptive multitasking library, like Control.Concurrent, would be tricky and pointless. (Since concurrency libraries tend to be about avoiding deadlock, rather than letting it happen and letting you inspect threads' current state.)

Answer 1

Here's your simpler way of modelling the problem: Each Send Int eliminates the other program's next Recv until neither sends.

data Instr = Send Int | Recv deriving Eq

execute :: [Instr] -> [Instr] -> (Bool, [Int], [Int]) -- Success, outputs
execute (Send i:x) y = (\(a,b,c) -> (a,i:b,c)) $ execute x (delete Recv y)
execute x (Send i:y) = (\(a,b,c) -> (a,b,i:c)) $ execute (delete Recv x) y
execute [] [] = (True, [], [])
execute _ _ = (False, [], [])