【问题标题】:Clean and type-safe state machine implementation in a statically typed language?以静态类型语言实现干净且类型安全的状态机?
【发布时间】:2011-12-03 17:39:19
【问题描述】:

我用 Python 实现了一个简单的状态机:

import time

def a():
    print "a()"
    return b

def b():
    print "b()"
    return c

def c():
    print "c()"
    return a


if __name__ == "__main__":
    state = a
    while True:
        state = state()
        time.sleep(1)

我想将它移植到 C,因为它不够快。但是 C 不允许我创建一个返回相同类型函数的函数。我尝试制作这种类型的函数:typedef *fn(fn)(),但它不起作用,所以我不得不使用结构来代替。现在代码很丑!

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

typedef struct fn {
    struct fn (*f)(void);
} fn_t;

fn_t a(void);
fn_t b(void);
fn_t c(void);

fn_t a(void)
{
    fn_t f = {b};

    (void)printf("a()\n");

    return f;
}

fn_t b(void)
{
    fn_t f = {c};

    (void)printf("b()\n");

    return f;
}

fn_t c(void)
{
    fn_t f = {a};

    (void)printf("c()\n");

    return f;
}

int main(void)
{
    fn_t state = {a};

    for(;; (void)sleep(1)) state = state.f();

    return EXIT_SUCCESS;
}

所以我认为这是 C 的类型系统损坏的问题。所以我使用了一种具有真实类型系统的语言(Haskell),但同样的问题发生了。我不能只做这样的事情:

type Fn = IO Fn
a :: Fn
a = print "a()" >> return b
b :: Fn
b = print "b()" >> return c
c :: Fn
c = print "c()" >> return a

我收到错误消息,Cycle in type synonym declarations

所以我必须像对 C 代码那样做一些包装器:

import Control.Monad
import System.Posix

data Fn = Fn (IO Fn)

a :: IO Fn
a = print "a()" >> return (Fn b)

b :: IO Fn
b = print "b()" >> return (Fn c)

c :: IO Fn
c = print "c()" >> return (Fn a)

run = foldM (\(Fn f) () -> sleep 1 >> f) (Fn a) (repeat ())

为什么用静态类型语言制作状态机这么难?我还必须在静态类型语言中产生不必要的开销。动态类型语言没有这个问题。有没有更简单的方法来使用静态类型语言?

【问题讨论】:

  • 如果速度不够快...取消睡眠?
  • 这是一个简化版本,用于说明我用来实现我的状态机的习语。
  • 只是想确定和/或巨魔。
  • 您可以改用newtype Fn = Fn (IO Fn)。与data 定义不同,这具有零开销。循环被报告为错误的原因是(1)它们通常错误,(2)它们难以打印好的错误消息,(3)它们使类型检查更加困难(4 ) 我敢肯定还有更多原因。
  • 顺便说一句,您不需要将函数调用显式转换为void 来丢弃它们的返回值;这是自动发生的。

标签: python c language-agnostic haskell typing


【解决方案1】:

在 Haskell 中,这个习惯用法就是继续执行下一个状态:

type StateMachine = IO ()
a, b, c :: StateMachine
a = print "a()" >> b
b = print "b()" >> c
c = print "c()" >> a

您不必担心这会溢出堆栈或类似的东西。如果你坚持有状态,那么你应该使数据类型更明确:

data PossibleStates = A | B | C
type StateMachine = PossibleStates -> IO PossibleStates
machine A = print "a()" >> return B
machine B = print "b()" >> return C
machine C = print "c()" >> return A

然后,您可以获得关于任何忘记某些状态的StateMachine 的编译器警告。

【讨论】:

    【解决方案2】:

    如果您使用newtype 而不是data,则不会产生任何开销。此外,您可以在定义点包装每个状态的函数,因此使用它们的表达式不必:

    import Control.Monad
    
    newtype State = State { runState :: IO State }
    
    a :: State
    a = State $ print "a()" >> return b
    
    b :: State
    b = State $ print "b()" >> return c
    
    c :: State
    c = State $ print "c()" >> return a
    
    runMachine :: State -> IO ()
    runMachine s = runMachine =<< runState s
    
    main = runMachine a
    

    编辑runMachine 有一个更通用的形式让我印象深刻; iterate 的一元版本:

    iterateM :: Monad m => (a -> m a) -> a -> m [a]
    iterateM f a = do { b <- f a
                      ; as <- iterateM f b
                      ; return (a:as)
                      }
    
    main = iterateM runState a
    

    编辑:嗯,iterateM 导致空间泄漏。也许iterateM_ 会更好。

    iterateM_ :: Monad m => (a -> m a) -> a -> m ()
    iterateM_ f a = f a >>= iterateM_ f
    
    main = iterateM_ runState a
    

    编辑:如果你想通过状态机线程化一些状态,你可以对State使用相同的定义,但是将状态函数更改为:

    a :: Int -> State
    a i = State $ do{ print $ "a(" ++ show i ++ ")"
                    ; return $ b (i+1)
                    }
    
    b :: Int -> State
    b i = State $ do{ print $ "b(" ++ show i ++ ")"
                    ; return $ c (i+1)
                    }
    
    c :: Int -> State
    c i = State $ do{ print $ "c(" ++ show i ++ ")"
                    ; return $ a (i+1)
                    }
    
    main = iterateM_ runState $ a 1
    

    【讨论】:

    • 包装是为了让类型检查器满意,但没有运行时开销。一般来说,状态之间可能存在多对多的关系。在定义点包裹是O(n),在使用点包裹可能是O(N^2)。
    • "iterateM 不是尾递归的,因此会导致空间泄漏"。不,尾递归与它无关。如果我没记错的话,它会导致空间泄漏,因为您正在收集一个充满() 的巨大列表。
    • @Dan 它正在收集大量访问过的States 列表,但我希望能够识别出结果未被使用,并且在创建时会被垃圾收集。我想我对Monad 中的尾递归感到困惑。 IO 不是对iterateM 的递归调用进行排序,并且仅在产生结果后才执行(:)(这永远不会)?
    • 如果每个后续状态的状态都依赖于前一个状态会是什么样子?例如,如果您希望它打印 a(1) b(2) c(3) 等...
    • @zmanian 查看更新以了解如何通过状态机线程化附加状态
    【解决方案3】:

    您的 Haskell 代码的问题是,type 只引入了同义词,这与 C 中的 typedef 非常相似。一个重要的限制是,类型的扩展必须是有限的,你不能给你的状态机一个有限的扩展。一个解决方案是使用newtypenewtype 是一个只存在于类型检查器的包装器;绝对零开销(由于无法删除的泛化而发生的排除的东西)。这是您的签名;它类型检查:

    newtype FN = FN { unFM :: (IO FN) }
    

    请注意,每当您想使用FN 时,您必须先使用unFN 解压它。每当你返回一个新函数时,使用FN 来打包它。

    【讨论】:

    • 我知道 newtype(虽然我没想过用它来消除这里的开销)。但是我仍然必须在每个状态中进行包装/展开,这与动态类型语言不同。
    • @megazord:你可以定义一个助手next = return . Fn来消除重复。
    • @megazord:因为 Haskell 不是 Python。类型安全是以一些为代价的,但是如果你不小心返回一个整数或字符串作为下一个状态,或者根本忘记返回下一个状态,这又是一个编译时错误。你的 Python 代码直到运行时才会检测到这一点。
    【解决方案4】:

    在类 C 类型系统中,函数不是一阶公民。处理它们有一定的限制。这是为了实现/执行的简单性和速度而做出的决定。为了让函数表现得像对象,通常需要支持闭包。然而,大多数处理器的指令集并不自然地支持这些。由于 C 被设计为靠近金属,因此没有支撑它们。

    在 C 中声明递归结构时,类型必须是完全可扩展的。这样做的结果是,您只能在结构声明中将指针作为自引用:

    struct rec;
    struct rec {
        struct rec *next;
    };
    

    此外,我们使用的每个标识符都必须声明。函数类型的限制之一是不能转发声明。

    C 中的状态机通常通过在 switch 语句或跳转表中进行从整数到函数的映射来工作:

    typedef int (*func_t)();
    
    void run() {
        func_t table[] = {a, b, c};
    
        int state = 0;
    
        while(True) {
            state = table[state]();
        }
    }
    

    您也可以profile your Python code 并尝试找出您的代码运行缓慢的原因。您可以将关键部分移植到 C/C++ 并继续使用 Python 作为状态机。

    【讨论】:

    • 我认为这就是问题所在,试图以一种语言惯用的方式做某事,并以完全相同的方式将其应用到另一种语言中。他们通常不是 1-1。
    • 这比我的解决方案还要糟糕。函数是 Haskell 中的“一等”公民,我的 Haskell 实现与 C 实现存在确切的问题。所以我不认为你的第一句话是相关的。
    • 不能为 Haskell 说话。但这就是为什么它在 Python 和 C 中的工作方式不同。
    • 与函数是否“一流”无关。这是类型系统限制我可以编写的代码类型的问题。
    • 您正试图在静态类型语言上强制使用鸭子类型解决方案,这显然不起作用?由于历史原因,C 不允许这样做。我举了一个在 C 中实现精益快速状态机的标准方法示例?
    【解决方案5】:

    像往常一样,尽管已经给出了很好的答案,但我还是忍不住自己尝试了一下。关于所呈现的内容困扰我的一件事是它忽略了输入。状态机——我熟悉的那些——根据输入在各种可能的转换之间进行选择。

    data State vocab = State { stateId :: String
                             , possibleInputs :: [vocab]
                             , _runTrans :: (vocab -> State vocab)
                             }
                          | GoalState { stateId :: String }
    
    instance Show (State a) where
      show = stateId
    
    runTransition :: Eq vocab => State vocab -> vocab -> Maybe (State vocab)
    runTransition (GoalState id) _                   = Nothing
    runTransition s x | x `notElem` possibleInputs s = Nothing
                      | otherwise                    = Just (_runTrans s x)
    

    这里我定义了一个类型State,它由一个词汇类型vocab参数化。现在让我们定义一种方法,我们可以通过输入输入来跟踪状态机的执行。

    traceMachine :: (Show vocab, Eq vocab) => State vocab -> [vocab] -> IO ()
    traceMachine _ [] = putStrLn "End of input"
    traceMachine s (x:xs) = do
      putStrLn "Current state: "
      print s
      putStrLn "Current input: "
      print x
      putStrLn "-----------------------"
      case runTransition s x of
        Nothing -> putStrLn "Invalid transition"
        Just s' -> case s' of
          goal@(GoalState _) -> do
            putStrLn "Goal state reached:"
            print s'
            putStrLn "Input remaining:"
            print xs
          _ -> traceMachine s' xs
    

    现在让我们在一个忽略其输入的简单机器上试一试。请注意:我选择的格式相当冗长。但是,后面的每个函数都可以被视为状态机图中的一个节点,我认为您会发现详细程度与描述这样一个节点完全相关。我使用stateId 以字符串格式编码一些关于该状态如何表现的视觉信息。

    data SimpleVocab = A | B | C deriving (Eq, Ord, Show, Enum)
    
    simpleMachine :: State SimpleVocab
    simpleMachine = stateA
    
    stateA :: State SimpleVocab
    stateA = State { stateId = "A state. * -> B"
                   , possibleInputs = [A,B,C]
                   , _runTrans = \_ -> stateB
                   }
    
    stateB :: State SimpleVocab
    stateB = State { stateId = "B state. * -> C"
                   , possibleInputs = [A,B,C]
                   , _runTrans = \_ -> stateC
                   }
    
    stateC :: State SimpleVocab
    stateC = State { stateId = "C state. * -> A"
                   , possibleInputs = [A,B,C]
                   , _runTrans = \_ -> stateA
                   }
    

    由于该状态机的输入无关紧要,因此您可以输入任何内容。

    ghci> traceMachine simpleMachine [A,A,A,A]
    

    我不会包含输出,这也很冗长,但您可以清楚地看到它从stateAstateBstateC 又回到stateA。现在让我们做一个稍微复杂一点的机器:

    lessSimpleMachine :: State SimpleVocab
    lessSimpleMachine = startNode
    
    startNode :: State SimpleVocab
    startNode = State { stateId = "Start node. A -> 1, C -> 2"
                      , possibleInputs = [A,C]
                      , _runTrans = startNodeTrans
                      }
      where startNodeTrans C = node2
            startNodeTrans A = node1
    
    node1 :: State SimpleVocab
    node1 = State { stateId = "node1. B -> start, A -> goal"
                  , possibleInputs = [B, A]
                  , _runTrans = node1trans
                  }
      where node1trans B = startNode
            node1trans A = goalNode
    
    node2 :: State SimpleVocab
    node2 = State { stateId = "node2. C -> goal, A -> 1, B -> 2"
                  , possibleInputs = [A,B,C]
                  , _runTrans = node2trans
                  }
      where node2trans A = node1
            node2trans B = node2
            node2trans C = goalNode
    
    goalNode :: State SimpleVocab
    goalNode = GoalState "Goal. :)"
    

    每个节点的可能输入和转换不需要进一步解释,因为它们在代码中已详细描述。我会让你自己玩traceMachine lessSipmleMachine inputs。看看inputs 无效(不遵守“可能的输入”限制)或在输入结束之前命中目标节点时会发生什么。

    我想我的解决方案的冗长有点无法满足您的基本要求,即减少杂乱无章的内容。但我认为它也说明了描述性 Haskell 代码的能力。尽管它非常冗长,但它在表示状态机图的节点方面也非常简单。

    【讨论】:

    • 我不记得为什么我对我的状态机进行了不同的建模,我现在有点醉了(感恩节),我会回复你的。对于具有更多功能的东西来说看起来不错
    • “我认为它也说明了 Haskell 代码的描述性”确实,我可以通过查看类型 State 来判断它的作用。我相信这可以表示为Arrow
    【解决方案6】:

    在 Haskell 中制作状态机并不难,一旦你意识到它们不是单子!像您想要的状态机是一个箭头,准确地说是一个自动机箭头:

    newtype State a b = State (a -> (b, State a b))
    

    这是一个函数,它接受一个输入值并产生一个输出值以及一个新版本的自身。这不是 monad,因为你不能为它写 join(&gt;&gt;=)。等效地,一旦你把它变成了一个箭头,你就会意识到不可能为它写一个ArrowApply 实例。

    以下是实例:

    import Control.Arrow
    import Control.Category
    import Prelude hiding ((.), id)
    
    instance Category State where
        id = State $ \x -> (x, id)
    
        State f . State g =
            State $ \x ->
                let (y, s2) = g x
                    (z, s1) = f y
                in (z, s1 . s2)
    
    instance Arrow State where
        arr f = let s = State $ \x -> (f x, s) in s
        first (State f) =
            State $ \(x1, x2) ->
                let (y1, s) = f x1
                in ((y1, x2), first s)
    

    玩得开心。

    【讨论】:

      【解决方案7】:

      你想要的是递归类型。不同的语言有不同的方法。

      例如,在 OCaml(一种静态类型语言)中,有一个可选的编译器/解释器标志 -rectypes 支持递归类型,允许您定义如下内容:

      let rec a () = print_endline "a()"; b
      and b () = print_endline "b()"; c
      and c () = print_endline "c()"; a
      ;;
      

      虽然这并不像您在 C 示例中所抱怨的那样“丑陋”,但 下面 发生的事情仍然是一样的。编译器只是替你担心,而不是强迫你写出来。

      正如其他人指出的那样,在 Haskell 中您可以使用 newtype 并且不会有任何“开销”。但是您抱怨必须显式包装和解包递归类型,这是“丑陋的”。 (与您的 C 示例类似;没有“开销”,因为在机器级别,1 成员结构与其成员相同,但它“丑陋”。)

      我想提的另一个例子是 Go(另一种静态类型语言)。在 Go 中,type 构造定义了一个新类型。它不是一个简单的别名(如 C 中的 typedef 或 Haskell 中的 type),而是创建了一个成熟的新类型(如 Haskell 中的 newtype),因为这样的类型具有独立的“方法集”方法你可以定义它。因此,类型定义可以是递归的:

      type Fn func () Fn
      

      【讨论】:

      • IMO 这是对“为什么使用静态类型语言中的函数实现状态机如此困难?”的最佳答案:实际上并非如此!类型推断为您提供了动态类型的大部分优点在编译时对您的代码的信心。
      【解决方案8】:

      您可以在 C 中获得与在 Python 代码中相同的效果,只需声明函数返回 (void*)

      #include "stdio.h"
      
      typedef void* (*myFunc)(void);
      
      void* a(void);
      void* b(void);
      void* c(void);
      
      void* a(void) {
          printf("a()\n");
          return b;
      }
      
      void* b(void) {
          printf("b()\n");
          return c;
      }
      
      void* c(void) {
          printf("c()\n");
          return a;
      }
      
      void main() {
          void* state = a;
          while (1) {
              state = ((myFunc)state)();
              sleep(1);
          }
      }
      

      【讨论】:

      • 我想出了同样的事情,但我认为重点是不要通过使用强制转换来破坏类型检查器!
      • @pat 这也是我的推理。
      • 在我看来,你@megazord 根本不知道你想要什么......如果你真的想要一些与你的 Python 代码最相似的解决方案 - 那么void* 是唯一的出路.否则 - 更改您的问题的表述以反映您的实际问题。
      • @0x69,megazord 有几个 C 解决方案,但问题的关键是是否有一个干净的类型安全解决方案。 struct wrapper 解决方案是类型安全的,但不是干净的。您的解决方案是干净的,但不是类型安全的。
      • Python 版本不是类型安全的。这里的要求不是速度,不一定是类型安全吗?
      【解决方案9】:

      你的问题之前有过:Recursive declaration of function pointer in C

      C++ 运算符重载可用于隐藏与您的 C 和 Haskell 解决方案基本相同的机制,正如 Herb Sutter 在 GotW #57: Recursive Declarations 中所描述的那样。

      struct FuncPtr_;
      typedef FuncPtr_ (*FuncPtr)();
      
      struct FuncPtr_
      {
        FuncPtr_( FuncPtr pp ) : p( pp ) { }
        operator FuncPtr() { return p; }
        FuncPtr p;
      };
      
      FuncPtr_ f() { return f; } // natural return syntax
      
      int main()
      {
        FuncPtr p = f();  // natural usage syntax
        p();
      }
      

      但是这种具有函数的业务很可能比具有数字状态的业务表现更差。您应该使用switch 语句或状态表,因为在这种情况下您真正想要的是与goto 等效的结构化语义。

      【讨论】:

      • 这与滥用语法是一回事。几乎和铸造一样糟糕
      • 它就像是邪恶的超浓缩液体形式。但它有效。
      【解决方案10】:

      F# 中的一个例子:

      类型 Cont = Cont of (unit -> Cont) 让记录 a() = printfn "a()" 续 (fun () -> b 42) 和 b n = printfn "b(%d)" n 续 和 c() = printfn "c()" 续 让REC运行(续f)= 让 f = f() 运行 f 运行(续)

      关于“为什么使用静态类型语言中的函数实现状态机这么难?”的问题:那是因为a 和朋友的类型有点奇怪:一个函数在返回一个函数时返回一个函数返回一个函数...

      如果我从示例中删除 Cont,F# 编译器会抱怨并说:

      期待 'a 但给定的单位 -> 'a。当统一 'a 和 unit -> 'a 时,结果类型将是无限的。

      另一个答案显示了 OCaml 中的一个解决方案,其类型推断足够强大,无需声明 Cont,这表明静态类型不是罪魁祸首,而是许多静态类型语言缺乏强大的类型推断。

      我不知道为什么 F# 不这样做,我猜这可能会使类型推断算法更复杂、更慢或“太强大”(它可以设法推断错误类型表达式的类型,稍后失败,给出难以理解的错误消息)。

      请注意,您提供的 Python 示例并不安全。在我的示例中,b 表示由整数参数化的状态族。在无类型语言中,很容易出错并返回 bb 42 而不是正确的 lambda,并在代码执行之前错过该错误。

      【讨论】:

      • 这与我的 C 代码相同。当然它更短,因为它具有类型推断并且是功能性的。但我的 Haskell 代码大小几乎相同,但由于 monadic IO 更安全。
      • 与您的 C(或 Python)代码不同。请参阅b 和我关于国家家族的评论。如果您使用函数对状态机进行编码,那么尝试利用参数来限制状态数量是很自然的。
      【解决方案11】:

      您发布的Python代码将被转换为递归函数,但不会进行尾调用优化,因为Python没有尾调用优化,所以它会在某些时候堆栈溢出。所以 Python 代码实际上是有问题的,并且需要更多的工作才能使其与 Haskell 或 C 版本一样好。

      这是我的意思的一个例子:

      so.py:

      import threading
      stack_size_bytes = 10**5
      threading.stack_size(10**5)
      machine_word_size = 4
      
      def t1():
          print "start t1"
          n = stack_size_bytes/machine_word_size
          while n:
              n -= 1
          print "done t1"
      
      def t2():
          print "start t2"
          n = stack_size_bytes/machine_word_size+1
          while n:
              n -= 1
          print "done t2"
      
      if __name__ == "__main__":
          t = threading.Thread(target=t1)
          t.start()
          t.join()
          t = threading.Thread(target=t2)
          t.start()
          t.join()
      

      外壳:

      $ python so.py
      start t1
      done t1
      start t2
      Exception in thread Thread-2:
      Traceback (most recent call last):
        File "/usr/lib/python2.7/threading.py", line 530, in __bootstrap_inner
          self.run()
        File "/usr/lib/python2.7/threading.py", line 483, in run
          self.__target(*self.__args, **self.__kwargs)
        File "so.py", line 18, in t2
          print "done t2"
      RuntimeError: maximum recursion depth exceeded
      

      【讨论】:

      • a 返回b 不是 b(),差别很大!
      • @djenga49 我在 OSX 上运行 Python 2.7.2,它不喜欢 105 的堆栈大小。我改成215,甚至把machine_word_size改成1,还是可以正常工作:$ python so.py start t1 done t1 start t2 done t2
      • @djenga49 如果您所说的大体上属实,那么编写无限期运行的 Python 程序将是不可能,因为所有无限迭代都不可避免地会导致无限递归。显然不是这样的!
      • @djenga49 在 Python 中不是这样。 Python 循环不是通过将它们转换为递归函数来实现的,然后通过将不纯操作视为可以返回然后由 IO monad 驱动程序执行的单子计算以某种方式进行记忆(你到底从哪里得到这个想法从?)。它们是通过直接依次执行每个字节码操作来实现的,其中一些是向后条件分支。
      • @djenga49 所以向我们展示一个导致这种行为的while True 循环。没有必要让一个循环巧妙地执行太多次,无限循环会更优雅地证明你的观点。当您使用它时,请在您奇怪的系统上尝试from dis import dis; dis(t2);在我的身上,你可以清楚地看到用于实现循环的向后跳转操作。
      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2012-10-31
      • 2016-09-26
      • 2010-09-28
      • 1970-01-01
      相关资源
      最近更新 更多