haskell - 惰性状态转换器在 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 转换本身就是严格性的来源。

是什么导致了 1D 和 2D 递归之间的严格性差异，我怎样才能实现我想要的惰性？

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 计算中，结果的头部显式地仅取决于输入的头部（仅head a适用于 1D 动作以及两者head a和head b2D 动作）。但是，在 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 1walk 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

这仅取决于定义的状态 action 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_tailwalk12_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真的sequence对walk 1 1由head $ walk 1 2.

Answer 2

KA Buhr 接受的答案是正确的：虽然在每个方向上领先一步都很好（尝试walk使用x < 2或y < 2）隐式>>=in liftM2、sequencein 值的组合和 make 值中a的状态依赖关系取决于的所有副作用。正如他还指出的那样，一个可行的解决方案取决于实际需要什么依赖项。bba

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

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

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 最快）。首先，我们定义一个“连接”，它MML在裸列表的末尾应用一个基本案例包装器ML：

   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. 我们希望将来自不同备选方案的列表粉碎成一个列表，我们不在乎这会在备选方案之间引入依赖关系。这是我们mconcat从 Monoid 实例中使用的地方。

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

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 是非常低效的。）