Haskell 中的半依赖动作

Question

我正在寻找一种 Haskell 设计，以某种方式组合一元操作链（通常是 IO），使后面的操作依赖于之前的操作，但在某些情况下可以在它们完成之前执行。

目前我想出的解决方案是：

type Future m a = m (m a)

读取：一个单子动作，它启动某个过程并返回一个动作，该动作将返回该过程的结果（可能通过等待该过程完成）。

所以在某个链中a >>= b >>= c b 得到一个返回 a 的结果的动作。如果 b 评估此操作，它会等待 a 完成，否则它将并行运行。这也意味着，如果某个动作不需要前一个动作的结果作为参数，则根据定义它不依赖于它，因此依赖关系是显式的。

一些示例代码：

date :: Future IO String   -- long process to find out the date
date = do
    print "attempting to get date"  -- will usually start some thread or process to compute the date
    return (print "today")  -- will wait for this thread or process and return the computed date

main = do
    d <- date   -- starts recieving the date
    print "foo" -- some other process
    d >>= print -- waits until the date has been computed and prints it out

输出：

"attempting to get date"
"foo"
"today"

存在一个问题：如果一个动作决定等待前一个动作，它将始终依赖于之前的所有其他动作（在我的情况下）。但是在上面的例子中，如果 c 决定等待 b 但 b 没有决定等待 a，c 可能会在 a 完成之前开始，这不应该发生。

作为解决方案，我编写了另一个组合运算符：

(>=>) :: Monad m => Future m a -> (m a -> Future m b) -> Future m b
a >=> f = do
    r1 <- a
    r2 <- f r1
    return (r1 >> r2)

所以这将结合“等待动作”和a >=> b >=> c 将正常工作，如果 c 等待 b 这个等待动作也将等待 a。 然而，这种方法还有另一个问题（除此之外，您需要记住使用 >=> 而不是 >>=）：可能会多次评估等待操作。 如果 b 等待 a 并且 c 等待 b 则等待 b 将连接到等待 a，因此等待 a 将被执行两次。

实际问题出在 >=> 中：f r1 可能会评估 r1，在这种情况下，它不需要在 return 语句中使用 r2 进行排序（因为它已经被执行，因此 a 已经完成）。但它也可能不会，我不知道。

所以我基本上想要的正是这个，但不可能多次运行等待操作。不幸的是，我在功能设计方面不是很有经验。

所以我希望你能以某种方式启发我如何增强或改变我的设计，或者为我指出一种不同的、更灵活的方法。

编辑根据到目前为止的答案，我想对我真正想要的内容进行更多说明：

我不想推迟（甚至跳过）动作的执行，我也不需要线程或类似的并行功能。实际上我正在调用外部进程。一个例子是

backup :: Future IO ExitCode
backup = do
    pid <- startProcess "backup"
    return (waitForProcessAndGetExitCode pid)

当我现在链接 backup >=> otherAction 之类的操作时，otherAction 可以在备份运行的同时运行（总体上节省了很多时间）。但是 otherAction 可能需要完成备份，这种情况下它可以使用它的参数等待备份并检查是否成功。无论哪种方式，都必须执行备份。

我现在正在寻找一个不错的通用解决方案，最好不要绑定到 IO monad。

更新我找到了适合我的解决方案。我在下面的单独答案中对其进行了描述。

Answer 1

我很确定你真的想要这个签名：

(>>=) :: Future m a -> (a -> Future m b) -> Future m b

以下是您实现所需内容的方式：

import Control.Concurrent
import Control.Monad
import Control.Monad.Trans

newtype Future m a = Future { runFuture :: m (m a) }

instance (Monad m) => Monad (Future m) where
    return = Future . return . return
    m >>= f = Future $ do
        fut1 <- runFuture m
        return $ join $ join $ liftM (runFuture . f) fut1

instance MonadTrans Future where
    lift = Future . liftM return

换句话说，Future 是一个 monad 转换器，它的实现并没有专门针对 IO monad。但是，以下示例将展示如何将它与 IO monad 结合使用来链接期货：

parallel :: IO a -> Future IO a
parallel m = Future $ do
    v <- newEmptyMVar
    forkIO $ m >>= putMVar v
    return $ takeMVar v

future1 = parallel $ do
    threadDelay 1000000
    putStrLn "Hello, World" 
    return 1
future2 n = parallel $ do
    threadDelay 1000000
    print n
    return 2
future3 = future1 >>= future2

main = do
    f <- runFuture future3
    putStrLn "I'm waiting..."
    r <- f
    print r

我还没有证明它满足 monad 定律或 monad 转换器定律，但我会尝试这样做，我会通知你它是否检查出来。在那之前，可能会有一个错位的join。

编辑：不！差远了。它绝对不满足单子定律。我不知道我是否接近，但现在假设这个答案是不正确的。不过，我现在有点好奇，想知道这是否可能。

Answer 2

也许一种可能性是拒绝运行f，直到它的输出被要求：

mma >=> fab = return $ do
    ma <- mma
    b  <- fab ma
    b

根据您的需要，首先运行mma 可能很重要：

mma >=> fab = do
    ma <- mma
    return $ do
        b <- fab ma
        b

Answer 3

如果您添加了对m 具有MonadIO 实例的限制，则可以执行以下操作（根据内存，未经测试）：

share :: IO a -> IO (IO a)
share m = do
    ref <- newIORef Nothing
    let reader = do
          cached <- readIORef ref
          case cached of
            Just a -> return a
            Nothing -> m >>= \a -> writeIORef ref (Just a) >> return a
    return reader

您可以通过将 IORef 创建包装在 unsafePerformIO 中来将其更改为 share2 :: IO a -> IO a，并且很容易推广到任何 MonadIO 实例。

但是，根据您的问题，使用threads 或IVar 之类的内容可能会更好。

Answer 4

对于某些情况，当您想激发一些线程并在某个时刻收集结果时，请检查 http://hackage.haskell.org/packages/archive/base/4.5.1.0/doc/html/Control-Concurrent-SampleVar.html 和 http://hackage.haskell.org/packages/archive/base/4.5.1.0/doc/html/Control-Concurrent.html#g:2 ，因为它们似乎相关

对于某些情况，当您需要按需执行操作时，您可能会发现此代码很有用 未在 GHC 中检查，但在修正错别字后应该可以工作

module Promise (SuspendedAction, createSuspendedAction, getValueFromSuspendedAction)
import Data.IORef

data Promise a = Suspended (IO a) | Done a

data SuspendedAction = SA (IORef (Promise a))

createSuspendedAction :: m a -> m (SuspendedAction a)
createSuspendedAction act = newIORef (Suspended act)

readSuspendedAction :: SuspendedAction a -> m a
readSuspendedAction (SA ref) = readIORef ref >>= \suspended -> case suspended of
  Done a -> return a
  Suspended sact -> sact >>= \rv -> writeIORef ref (Done rv) >> return rv

顺便说一句，仔细检查 hackage，有一个包允许在尊重其顺序的同时懒惰地执行 IO 操作。

Answer 5

我自己找到了解决方案，尽管不完全是针对我所遇到的问题。

我意识到我必须事先以某种方式知道一个动作是否依赖于之前的动作。我尝试了各种方法，最终想出了我现在要描述的东西。 我的解决方案允许编写类似的代码

a :: Process IO x ()
a = independant $ do
    print "start a"
    return $ print "end a"

b :: Process IO x Int
b = independant $ do
    print "start b"
    return $ print "end b" >> return 0

c :: Process IO Int ()
c = dependant $ \x -> do
    print $ "start c with " ++ show x
    return $  print ("end c, started with " ++ show x)

chain = a >~ b >~ c
main = exec chain

-- outputs:
"start a" "start b" "end a" "end b" "start c with 0" "end c, started with 0"

（下面有更多示例）

我使用了以下类型

type Future m a = m (m a)
type Action m a b = a -> Future m b
type Process m a b = forall c. Action m c a -> Action m c b  -- will need -XRank2Types

使用以下原语：

-- sequences f after g, f is dependant of g and gets its result
-- dependant :: Monad m => Action m a b -> Action m c a -> Action c b
dependant :: Monad m => Action m a b -> Process m a b
dependant f g a = join (g a) >>= f

-- sequences f after g, f is independant of g
independant :: Monad m => Future m a -> Process m b a
independant f g a = do
    w1 <- g a
    w2 <- f
    return (w1 >> w2)

-- concenation of processes
(>~) = flip (.)

这种方法也允许其他原语更容易处理，例如：

-- lifts a pure function into an action
pureA :: Monad m => (a -> b) -> Action m a b
pureA f a = return . return $ f a

-- makes an action wich always returns the same result
constA :: Monad m => b -> Action m a b
constA = pureA . const

-- no operation action
nop :: Monad m => Action m a ()
nop = constA ()

-- puts a sequence point
wait :: Monad m => Process m a a
wait = dependant $ pureA id

-- modify its result with a pure function
modify :: (Monad m, Functor m) => (a -> b) -> Process m a b
modify f act a = do
    x <- act a
    return (fmap f x)

-- makes a process, wich always returns the same result
constP :: (Monad m, Functor m) => b -> Process m a b
constP = modify . const

最后是一个运行进程的函数：

-- executes a process
exec :: Monad m => Process m () b -> m b
exec p = join $ p nop undefined

再举几个稍微复杂一点的例子：

simleI :: String -> a -> Process IO b a
simpleI name r = independant $ do
    print ("start " ++ name)
    return $ print ("end " ++ name) >> return r

simpleD :: (Show a, Show b) => String -> (a -> b) -> Process IO a b
simpleD name f = dependant $ \a -> do
    print ("start " ++ name ++ " with " ++ show a)
    let r = f a
    return $ print ("end " ++ name ++ " with " ++ show r ++ " (started with " ++ show a ++ ")") >> return r

a = simpleI "a" ()
b = simpleI "b" 42
c = simpleD "c" (+1)
d = simpleI "d" ()

chain1 = a >~ b >~ c >~ d       -- == d . c . b . a
chain2 = a >~ wait >~ b >~ c >~ d
chain3 = a >~ b >~ modify (+1) >~ c >~ d

main = do
    exec chain1
    print "---"
    exec chain2
    print "---"
    exec chain3

输出：

"start a"
"start b"
"end a"
"end b"
"start c with 42"
"start d"
"end c with 43 (started with 42)"
"end d"
"---"
"start a"
"end a"
"start b"
"end b"
"start c with 42"
"start d"
"end c with 43 (started with 42)"
"end d"
"---"
"start a"
"start b"
"end a"
"end b"
"start c with 43"
"start d"
"end c with 44 (started with 43)"
"end d"

这几乎正是我想要的。

我有点好奇如何对 Action 和 Process 进行分类。它们不是单子。他们可能是箭，但我对箭太陌生，无法分辨。进程可能是一个带有 fmap = modify 和 pure = const 的 Applicative 。 constA 或类似的东西。

如果您想到我的方法，请随时发表评论，尤其是如何扩展或简化它。