惰性状态转换器在 2D 递归中急切地消耗惰性列表

Question

我正在使用状态转换器在 2D 递归游走的每个点对数据集进行随机采样，从而输出一起满足条件的 2D 样本网格列表。我想懒洋洋地从结果中提取，但我的方法在提取第一个结果之前在每个点都耗尽了整个数据集。

具体来说，考虑这个程序：

import Control.Monad ( sequence, liftM2 )
import Data.Functor.Identity
import Control.Monad.State.Lazy ( StateT(..), State(..), runState )

walk :: Int -> Int -> [State Int [Int]]
walk _ 0 = [return [0]]
walk 0 _ = [return [0]]
walk x y =
  let st :: [State Int Int]
      st = [StateT (\s -> Identity (s, s + 1)), undefined]
      unst :: [State Int Int] -- degenerate state tf
      unst = [return 1, undefined]
  in map (\m_z -> do
      z <- m_z
      fmap concat $ sequence [
          liftM2 (zipWith (\x y -> x + y + z)) a b -- for 1D: map (+z) <$> a
          | a <- walk x (y - 1) -- depth
          , b <- walk (x - 1) y -- breadth -- comment out for 1D
        ]
    ) st -- vs. unst

main :: IO ()
main = do
  std <- getStdGen
  putStrLn $ show $ head $ fst $ (`runState` 0) $ head $ walk 2 2

程序从(x, y) 到(0, 0) 遍历矩形网格并对所有结果求和，包括状态单子列表之一的值：读取并推进其状态的非平凡转换器st ，或琐碎的变压器unst。有趣的是该算法是否探索了st 和unst 的头部。

在呈现的代码中，它会抛出 undefined。我将此归结为我对链接转换顺序的错误设计，特别是状态处理的问题，因为使用unst 代替（即从状态转换中分离结果）确实会产生结果。但是，我随后发现，即使使用状态转换器，一维递归也能保持惰性（删除宽度步长 b <- walk... 并将 liftM2 块交换为 fmap）。

如果我们trace (show (x, y))，我们还会看到它确实在触发之前遍历了整个网格：

$ cabal run
Build profile: -w ghc-8.6.5 -O1
...
(2,2)
(2,1)
(1,2)
(1,1)
(1,1)
sandbox: Prelude.undefined

我怀疑我对 sequence 的使用在这里有问题，但是由于 monad 的选择和 walk 的维度会影响其成功，我不能笼统地说 sequenceing 转换是严格本身。

是什么导致这里一维和二维递归的严格性不同，我怎样才能达到我想要的惰性？

Answer 1

考虑以下简化示例：

import Control.Monad.State.Lazy

st :: [State Int Int]
st = [state (\s -> (s, s + 1)), undefined]

action1d = do
  a <- sequence st
  return $ map (2*) a

action2d = do
  a <- sequence st
  b <- sequence st
  return $ zipWith (+) a b

main :: IO ()
main = do
  print $ head $ evalState action1d 0
  print $ head $ evalState action2d 0

在这里，在 1D 和 2D 计算中，结果的头显式仅取决于输入的头（对于 1D 动作，head a 和 head a 和 head b 对于 2D 动作） .但是，在 2D 计算中，b（甚至只是它的头部）对当前状态存在隐式依赖，并且该状态取决于对整体的评估a，不仅仅是它的头。

您的示例中有类似的依赖关系，尽管它被状态操作列表的使用所掩盖。

假设我们想要手动运行操作walk22_head = head $ walk 2 2 并检查结果列表中的第一个整数：

main = print $ head $ evalState walk22_head

明确写出状态动作列表st的元素：

st1, st2 :: State Int Int
st1 = state (\s -> (s, s+1))
st2 = undefined

我们可以把walk22_head写成：

walk22_head = do
  z <- st1
  a <- walk21_head
  b <- walk12_head
  return $ zipWith (\x y -> x + y + z) a b

请注意，这仅取决于定义的状态操作st1 以及walk 2 1 和walk 1 2 的头部。反过来，这些头可以写成：

walk21_head = do
  z <- st1
  a <- return [0] -- walk20_head
  b <- walk11_head
  return $ zipWith (\x y -> x + y + z) a b

walk12_head = do
  z <- st1
  a <- walk11_head
  b <- return [0] -- walk02_head
  return $ zipWith (\x y -> x + y + z) a b

同样，这些仅取决于定义的状态操作st1 和walk 1 1 的头部。

现在，让我们试着写下walk11_head的定义：

walk11_head = do
  z <- st1
  a <- return [0]
  b <- return [0]
  return $ zipWith (\x y -> x + y + z) a b

这仅取决于定义的状态操作st1，因此有了这些定义，如果我们运行main，我们会得到一个明确的答案：

> main
10

但这些定义并不准确！在walk 1 2 和walk 2 1 中的每一个中，头部动作都是动作的序列，从调用walk11_head 的动作开始，然后继续基于walk11_tail 的动作。因此，更准确的定义是：

walk21_head = do
  z <- st1
  a <- return [0] -- walk20_head
  b <- walk11_head
  _ <- walk11_tail  -- side effect of the sequennce
  return $ zipWith (\x y -> x + y + z) a b

walk12_head = do
  z <- st1
  a <- walk11_head
  b <- return [0] -- walk02_head
  _ <- walk11_tail  -- side effect of the sequence
  return $ zipWith (\x y -> x + y + z) a b

与：

walk11_tail = do
  z <- undefined
  a <- return [0]
  b <- return [0]
  return [zipWith (\x y -> x + y + z) a b]

有了这些定义，单独运行walk12_head 和walk21_head 就没有问题了：

> head $ evalState walk12_head 0
1
> head $ evalState walk21_head 0
1

这里的状态副作用不需要计算答案，因此从不调用。但是，不可能同时运行它们：

> head $ evalState (walk12_head >> walk21_head) 0
*** Exception: Prelude.undefined
CallStack (from HasCallStack):
  error, called at libraries/base/GHC/Err.hs:78:14 in base:GHC.Err
  undefined, called at Lazy2D_2.hs:41:8 in main:Main

因此，尝试运行 main 失败的原因相同：

> main
*** Exception: Prelude.undefined
CallStack (from HasCallStack):
  error, called at libraries/base/GHC/Err.hs:78:14 in base:GHC.Err
  undefined, called at Lazy2D_2.hs:41:8 in main:Main

因为在计算walk22_head时，即使是最开始walk21_head的计算也依赖于walk11_tail发起的状态副作用walk12_head。

您原来的 walk 定义与这些模型的行为方式相同：

> head $ evalState (head $ walk 1 2) 0
1
> head $ evalState (head $ walk 2 1) 0
1
> head $ evalState (head (walk 1 2) >> head (walk 2 1)) 0
*** Exception: Prelude.undefined
CallStack (from HasCallStack):
  error, called at libraries/base/GHC/Err.hs:78:14 in base:GHC.Err
  undefined, called at Lazy2D_0.hs:15:49 in main:Main
> head $ evalState (head (walk 2 2)) 0
*** Exception: Prelude.undefined
CallStack (from HasCallStack):
  error, called at libraries/base/GHC/Err.hs:78:14 in base:GHC.Err
  undefined, called at Lazy2D_0.hs:15:49 in main:Main

很难说如何解决这个问题。您的玩具示例非常适合说明问题，但尚不清楚在您的“真实”问题中如何使用状态，以及 head $ walk 2 1 是否真的对 walk 1 1 的 walk 1 1 动作有状态依赖性head $ walk 1 2.

Answer 2

The accepted answer by K.A. Buhr 是对的：虽然在每个方向上领先一步都很好（尝试 walk 与 x < 2 或 y < 2）在 liftM2 中隐式 >>= 的组合， a 值中的sequence 和b 值中的状态依赖使得b 依赖于a 的所有副作用。正如他还指出的那样，有效的解决方案取决于实际需要的依赖项。

我将针对我的特殊情况分享一个解决方案：每个walk 调用至少取决于调用者的状态，也许还有其他一些状态，基于网格的预订遍历和@987654335 中的替代方案@。此外，正如问题所暗示的那样，我想在测试st 中任何不需要的替代方案之前尝试做出完整的结果。这在视觉上有点难以解释，但这是我能做的最好的事情：左边显示每个坐标处 st 备选方案的可变数量（这是我在实际用例中所拥有的），右边显示 [相当messy] 状态所需依赖顺序的映射：我们看到它首先在 3D DFS 中遍历 x-y，“x”作为深度（最快轴），“y”作为广度（中轴），最后选择最慢轴（以虚线和空心圆圈显示）。

原始实现的核心问题来自于状态转换的排序列表以适应非递归返回类型。让我们将 list 类型完全替换为 monad 参数中的递归类型，以便调用者可以更好地控制依赖顺序：

data ML m a = MCons a (MML m a) | MNil -- recursive monadic list
newtype MML m a = MML (m (ML m a)) -- base case wrapper

[1, 2] 的一个例子：

MCons 1 (MML (return (MCons 2 (MML (return MNil)))))

Functor 和 Monoid 行为经常被使用，下面是相关的实现：

instance Functor m => Functor (ML m) where
  fmap f (MCons a m) = MCons (f a) (MML $ (fmap f) <$> coerce m)
  fmap _ MNil = MNil

instance Monad m => Semigroup (MML m a) where
  (MML l) <> (MML r) = MML $ l >>= mapper where
    mapper (MCons la lm) = return $ MCons la (lm <> (MML r))
    mapper MNil = r

instance Monad m => Monoid (MML m a) where
  mempty = MML (pure MNil)

有两个关键操作：在两个不同的轴上组合步骤，以及在同一坐标上组合来自不同备选方案的列表。分别：

1. 根据图表，我们希望首先从 x 步骤获得单个完整结果，然后从 y 步骤获得完整结果。每一步都返回来自内部坐标的可行备选方案的所有组合的结果列表，因此我们对两个列表进行笛卡尔积，也偏向一个方向（在本例中为 y 最快）。首先，我们定义一个“连接”，它在裸列表 ML 的末尾应用基本案例包装器 MML：

   nest :: Functor m => MML m a -> ML m a -> ML m a
   nest ma (MCons a mb) = MCons a (MML $ nest ma <$> coerce mb)
   

   然后是笛卡尔积：

   prodML :: Monad m => (a -> a -> a) -> ML m a -> ML m a -> ML m a
   prodML f x (MCons ya ym) = (MML $ prodML f x <$> coerce ym) `nest` ((f ya) <$> x)
   prodML _ MNil _ = MNil
   

2. 我们希望将来自不同备选方案的列表拆分为一个列表，我们不在乎这会在备选方案之间引入依赖关系。这是我们在 Monoid 实例中使用 mconcat 的地方。

总而言之，它看起来像这样：

walk :: Int -> Int -> MML (State Int) Int
-- base cases
walk _ 0 = MML $ return $ MCons 1 (MML $ return MNil)
walk 0 _ = walk 0 0

walk x y =
  let st :: [State Int Int]
      st = [StateT (\s -> Identity (s, s + 1)), undefined]
      xstep = coerce $ walk (x-1) y
      ystep = coerce $ walk x (y-1)
     -- point 2: smash lists with mconcat
  in mconcat $ map (\mz -> MML $ do
      z <- mz
                              -- point 1: product over results
      liftM2 ((fmap (z+) .) . prodML (+)) xstep ystep
    ) st

headML (MCons a _) = a
headML _ = undefined

main :: IO ()
main = putStrLn $ show $ headML $ fst $ (`runState` 0) $ (\(MML m) -> m) $ walk 2 2

请注意，结果已随语义发生变化。这对我来说并不重要，因为我的目标只需要从状态中提取随机数，并且可以通过将列表元素正确引导到最终结果中来控制所需的任何依赖顺序。

（我还要警告说，如果没有记忆或注意严格性，这种实现对于大 x 和 y 来说效率非常低。）