示例:网络 Socket 编程

POSIX 的插口(Socket,另译作「套接字」)API 支持跨网络的进程间通信。插口表示网络通信的端点,它可以是以下几种状态之一:

  • Ready 为初始状态
  • Bound 表示已绑定至某个地址,准备接受传入的连接
  • Listening 表示正在监听传入的连接
  • Open 表示准备发送或接受数据
  • Closed 表示不再活动

下图展示了该 API 提供的操作是如何改变其状态的,其中 Ready 为初始状态:

netstate

如果某个连接为 Open 状态,那么我们可以用 send 发送消息到该连接的另一端,也可以用 recv 从另一端接收消息。

contrib 包提供了一个 Network.Socket 模块,该模块提供了创建插口以及发送和接收消息的原语。其中包含以下函数:

bind : (sock : Socket) -> (addr : Maybe SocketAddress) -> (port : Port) -> IO Int
connect : (sock : Socket) -> (addr : SocketAddress) -> (port : Port) -> IO ResultCode
listen : (sock : Socket) -> IO Int
accept : (sock : Socket) -> IO (Either SocketError (Socket, SocketAddress))
send : (sock : Socket) -> (msg  : String) -> IO (Either SocketError ResultCode)
recv : (sock : Socket) -> (len : ByteLength) -> IO (Either SocketError (String, ResultCode))
close : Socket -> IO ()

这些函数虽然涵盖了上图中所有的状态转移,然而却均未解释这些操作是如何影响其状态的!例如,我们完全有可能在一个尚未就绪的插口上发送消息,或者在插口关闭后从中接收消息。

我们可以用 ST 提供更好的 API,它精确地解释了每个操作是如何影响连接的状态的。在本节中,我们会定义一个插口 API,然后用它来实现一个「回显(echo)」服务器,通过原样返回客户端发送的消息来响应客户端的请求。

定义 Sockets 接口

我们不直接用 IO 进行底层插口编程,而是用 ST 实现一个接口来精确地描述每个操作如何影响插口的状态,以及何时创建或删除插口。我们首先来创建描述插口抽象状态的类型:

data SocketState = Ready | Bound | Listening | Open | Closed

接着定义一个接口,用 Sock 类型表示插口,以其当前状态作为该类型构造子的参数:

interface Sockets (m : Type -> Type) where
  Sock : SocketState -> Type

我们用 socket 方法创建插口。插口库中定义了 SocketType,它描述了插口为 TCP、UDP 还是其它形式。我们后面会一直使用 Stream 来表示 TCP 插口。

socket : SocketType -> ST m (Either () Var) [addIfRight (Sock Ready)]

记住 addIfRight 会在操作结果的形式为 Right val 时添加资源。按照此接口的约定,我们用 Either 来表示可能失败的操作,它可以保存任何关于错误的附加信息。如此一来,我们就能一致地使用 addIfRight 和其它类型级函数了。

现在来定义一个服务器。一旦我们创建了插口,就需要用 bind 方法将它绑定到一个端口上:

bind : (sock : Var) -> (addr : Maybe SocketAddress) -> (port : Port) ->
       ST m (Either () ()) [sock ::: Sock Ready :-> (Sock Closed `or` Sock Bound)]

绑定插口可能会失败,例如如果已经有一个插口被绑定到给定的端口上,因此它同样要返回一个类型为 Either 的值。这里的动作使用了类型级函数 or,它的意思是:

  • bind 失败,插口就会转移到 Sock Closed 状态
  • bind 成功,插口就会转移到 Sock Bound 状态,如前图所示

or 的实现如下:

or : a -> a -> Either b c -> a
or x y = either (const x) (const y)

这样一来,bind 的类型就能写成如下等价形式:

bind : (sock : Var) -> (addr : Maybe SocketAddress) -> (port : Port) ->
       STrans m (Either () ()) [sock ::: Sock Ready]
                    (either [sock ::: Sock Closed] [sock ::: Sock Bound])

然而,使用 or 会更加简洁,它也能尽可能直接地反映出状态转移图,同时仍然刻画了它可能失败的性质。

一旦我们将插口绑定到了端口,就可以开始监听来自客户端的连接了:

listen : (sock : Var) ->
         ST m (Either () ()) [sock ::: Sock Bound :-> (Sock Closed `or` Sock Listening)]

Listening 状态的插口表示准备接受来自独立客户端的连接:

accept : (sock : Var) ->
         ST m (Either () Var)
              [sock ::: Sock Listening, addIfRight (Sock Open)]

如果有客户端传入的连接,accept 会在资源列表的最后添加一个新的资源(按照约定,在列表末尾添加资源可以更好地配合 updateWith 工作,如上一节所述)。 现在,我们有了两个插口:一个继续监听传入的连接,另一个准备与客户端通信。

我们还需要能够在插口上发送和接受数据的方法:

send : (sock : Var) -> String ->
       ST m (Either () ()) [sock ::: Sock Open :-> (Sock Closed `or` Sock Open)]
recv : (sock : Var) ->
       ST m (Either () String) [sock ::: Sock Open :-> (Sock Closed `or` Sock Open)]

一旦我们与另一台机器通过插口进行的通信结束, 就需要用 close 关闭连接并用 remove 移除该插口:

close : (sock : Var) ->
        {auto prf : CloseOK st} -> ST m () [sock ::: Sock st :-> Sock Closed]
remove : (sock : Var) ->
         ST m () [Remove sock (Sock Closed)]

close 使用了断言 CloseOK 作为隐式证明参数,它描述了何时可以关闭插口:

data CloseOK : SocketState -> Type where
     CloseOpen : CloseOK Open
     CloseListening : CloseOK Listening

也就是说,我们可以关闭 Open 状态的插口,告诉另一台机器终止通信。我们也可以关闭 Listening 状态下等待传入连接的插口,这会让服务器停止接受请求。

在本节中,我们实现了一个服务器,不过为了完整性,我们还需要在另一台机器上实现客户端来连接到服务器。这可以通过 connect 来完成:

connect : (sock : Var) -> SocketAddress -> Port ->
          ST m (Either () ()) [sock ::: Sock Ready :-> (Sock Closed `or` Sock Open)]

以下完整的接口作为参考:

interface Sockets (m : Type -> Type) where
  Sock : SocketState -> Type
  socket : SocketType -> ST m (Either () Var) [addIfRight (Sock Ready)]
  bind : (sock : Var) -> (addr : Maybe SocketAddress) -> (port : Port) ->
         ST m (Either () ()) [sock ::: Sock Ready :-> (Sock Closed `or` Sock Bound)]
  listen : (sock : Var) ->
           ST m (Either () ()) [sock ::: Sock Bound :-> (Sock Closed `or` Sock Listening)]
  accept : (sock : Var) ->
           ST m (Either () Var) [sock ::: Sock Listening, addIfRight (Sock Open)]
  connect : (sock : Var) -> SocketAddress -> Port ->
            ST m (Either () ()) [sock ::: Sock Ready :-> (Sock Closed `or` Sock Open)]
  close : (sock : Var) -> {auto prf : CloseOK st} ->
          ST m () [sock ::: Sock st :-> Sock Closed]
  remove : (sock : Var) -> ST m () [Remove sock (Sock Closed)]
  send : (sock : Var) -> String ->
         ST m (Either () ()) [sock ::: Sock Open :-> (Sock Closed `or` Sock Open)]
  recv : (sock : Var) ->
         ST m (Either () String) [sock ::: Sock Open :-> (Sock Closed `or` Sock Open)]

我们稍后会看到如何实现它。这些方法大部分都可以直接通过原始的插口 API 在 IO 中实现。不过首先,我们会看到如何用这些 API 实现一个「回显」服务器。

Sockets 实现「回显」服务器

从顶层来说,我们的回显(echo)服务器在开始和结束时均没有资源可用,它使用了 ConsoleIOSockets 接口:

startServer : (ConsoleIO m, Sockets m) => ST m () []

首先我们需要用 socket 创建一个插口,绑定到一个端口并监听传入的连接。它可能会失败,因此我们需要处理它返回 Right sock 的情况,其中 sock 是新的插口变量,不过也可能返回 Left err

startServer : (ConsoleIO m, Sockets m) => ST m () []
startServer =
  do Right sock <- socket Stream
           | Left err => pure ()
     ?whatNow

交互式地实现这类函数是个不错的想法,我们可以通过挖坑来逐步观察整个系统的状态是如何变化的。在成功调用了 socket 之后,我们就有了 Ready 状态的插口:

  sock : Var
  m : Type -> Type
  constraint : ConsoleIO m
  constraint1 : Sockets m
--------------------------------------
whatNow : STrans m () [sock ::: Sock Ready] (\result1 => [])

接着,我们需要将插口绑定到端口,然后开始监听连接。同样,每一步都可能会失败,此时我们会移除该插口。失败总会导致插口转为 Closed 状态,此时我们能做的就是用 remove 移除它:

startServer : (ConsoleIO m, Sockets m) => ST m () []
startServer =
  do Right sock <- socket Stream        | Left err => pure ()
     Right ok <- bind sock Nothing 9442 | Left err => remove sock
     Right ok <- listen sock            | Left err => remove sock
     ?runServer

最后,我们就有了一个监听传入连接的插口:

  ok : ()
  sock : Var
  ok1 : ()
  m : Type -> Type
  constraint : ConsoleIO m
  constraint1 : Sockets m
--------------------------------------
runServer : STrans m () [sock ::: Sock Listening]
                   (\result1 => [])

我们会在一个独立的函数中实现它。runServer 的类型告诉我们 echoServer 的类型必须是什么(我们无需显式地为 Sock 给出参数 m):

echoServer : (ConsoleIO m, Sockets m) => (sock : Var) ->
             ST m () [remove sock (Sock {m} Listening)]

我们可以完成 startServer 的定义:

startServer : (ConsoleIO m, Sockets m) => ST m () []
startServer =
  do Right sock <- socket Stream        | Left err => pure ()
     Right ok <- bind sock Nothing 9442 | Left err => remove sock
     Right ok <- listen sock            | Left err => remove sock
     echoServer sock

echoServer 中,我们会继续接受并相应请求直到出现失败,此时我们会关闭插口并返回。我们从尝试接受传入的连接开始:

echoServer : (ConsoleIO m, Sockets m) => (sock : Var) ->
             ST m () [remove sock (Sock {m} Listening)]
echoServer sock =
  do Right new <- accept sock | Left err => do close sock; remove sock
     ?whatNow

accept 失败,我们就需要关闭 Listening 状态的插口并在返回前移除它,因为 echoServer 的类型要求如此。

通常,逐步实现 echoServer 意味着我们可以在开发过程中检查当前的状态。如果 accept 成功,那么我们既有的 sock 会继续监听连接,此外一个新的 new 插口会被打开用于通信:

  new : Var
  sock : Var
  m : Type -> Type
  constraint : ConsoleIO m
  constraint1 : Sockets m
--------------------------------------
whatNow : STrans m () [sock ::: Sock Listening, new ::: Sock Open]
                      (\result1 => [])

要完成 echoServer,我们需要从 new 插口上接收一条消息,然后原样返回它。在完成后,我们就关闭 new 插口,然后回到 echoServer 的开始处,准备响应下一次连接:

echoServer : (ConsoleIO m, Sockets m) => (sock : Var) ->
             ST m () [remove sock (Sock {m} Listening)]
echoServer sock =
  do Right new <- accept sock | Left err => do close sock; remove sock
     Right msg <- recv new | Left err => do close sock; remove sock; remove new
     Right ok <- send new ("You said " ++ msg)
           | Left err => do remove new; close sock; remove sock
     close new; remove new; echoServer sock

实现 Sockets

为了在 IO 中实现 Sockets,我们需要从给出具体的 Sock 类型开始。我们可以用原始的插口 API(在 Network.Socket 中实现),将 Socket 存储在 State 中来得到具体的类型,而不必关心该插口所处的抽象状态:

implementation Sockets IO where
  Sock _ = State Socket

大部分方法都可以直接用原始的插口 API 来实现,返回相应的 LeftRight。例如,我们可以实现 socketbindlisten

socket ty = do Right sock <- lift $ Socket.socket AF_INET ty 0
                    | Left err => pure (Left ())
               lbl <- new sock
               pure (Right lbl)
bind sock addr port = do ok <- lift $ bind !(read sock) addr port
                         if ok /= 0
                            then pure (Left ())
                            else pure (Right ())
listen sock = do ok <- lift $ listen !(read sock)
                 if ok /= 0
                    then pure (Left ())
                    else pure (Right ())

然而,这里的 accept 有点不同, 因为我们在用 new 为打开连接创建新资源时,它出现在了资源列表的起始处而非末尾。我们可以通过写出不完整的定义来看到这一点, 使用 returning 来查看返回 Right lbl 需要什么资源:

accept sock = do Right (conn, addr) <- lift $ accept !(read sock)
                       | Left err => pure (Left ())
                 lbl <- new conn
                 returning (Right lbl) ?fixResources

使用 new 将资源添加到列表起始处是很方便的,因为通常来说,这会让 Idris 使用隐式 auto 自动构造证明更加容易。另一方面,当我们用 call 来构造更小的资源集合时,updateWith 会将新创建的资源放到列表的末尾处,因为通常这样会减少需要重新排序的资源的数量。

如果我们查看 fixResources 的类型,就会知道结束 accept 需要做的事情:

  _bindApp0 : Socket
  conn : Socket
  addr : SocketAddress
  sock : Var
  lbl : Var
--------------------------------------
fixResources : STrans IO () [lbl ::: State Socket, sock ::: State Socket]
                      (\value => [sock ::: State Socket, lbl ::: State Socket])

当前资源列表的顺序为 lblsock,而我们需要它们的顺序变成 socklbl。为此,Control.ST 提供了原语 toEnd,它会将一个资源移到列表的末尾。这样我们就能完成 accept 了:

accept sock = do Right (conn, addr) <- lift $ accept !(read sock)
                       | Left err => pure (Left ())
                 lbl <- new conn
                 returning (Right lbl) (toEnd lbl)

Sockets 的完整实现见 Idris 发行版中的 samples/ST/Net/Network.idr 文件。你也可以在同目录下的 EchoServer.idr 文件中找到回显服务器。此外还有一个高级网络协议 RandServer.idr,它基于底层的插口 API,通过状态机的层级实现了一个高级的网络通信协议。它还使用线程来异步地处理传入的请求。你可以在 Edwin Brady 的文稿 State Machines All The Way Down (深入理解状态机)中找到关于线程和随机数服务器的更多详情。