go - 需要帮助了解 libchan 的工作原理

Question

我正在尝试使用 libchan使用类似 go channel 的传输在机器之间发送消息的库。

根据我收集到的信息，大致的想法是这样的:

1. 您有一个 SPDY 客户端，它通过 tcp 将一个序列化的命令对象发送到一个地址。此命令对象包含一个名为 Pipe 的 libchan channel ，通过它发送响应。
2. 当服务器接收到传入连接时，它会等待一个命令对象。当它得到一个时，它通过对象中包含的 Pipe 发送响应。

这是我的困惑点。对于在两台机器之间持续存在的 channel ，它们必须共享内存或至少共享连接它们两者的抽象。根据我对 libchan 代码库的了解，我不知道这怎么可能。

这里是 repo 中示例的一个片段:

// client

    receiver, remoteSender := libchan.Pipe()
    command := &RemoteCommand{
        Cmd:        os.Args[1],
        Args:       os.Args[2:],
        Stdin:      os.Stdin,
        Stdout:     os.Stdout,
        Stderr:     os.Stderr,
        StatusChan: remoteSender,
    }

    err = sender.Send(command)
    if err != nil {
        log.Fatal(err)
    }
    err = receiver.Receive(response)
    if err != nil {
        log.Fatal(err)
    }

    os.Exit(response.Status)

和服务器:

// server
t := spdy.NewTransport(p)

        go func() {
            for {
                receiver, err := t.WaitReceiveChannel()
                if err != nil {
                    log.Print("receiver error")
                    log.Print(err)
                    break
                }
                log.Print("about to spawn receive proc")
                go func() {
                    for {
                        command := &RemoteReceivedCommand{}
                        err := receiver.Receive(command)
                        returnResult := &CommandResponse{}
                        if res != nil {
                            if exiterr, ok := res.(*exec.ExitError); ok {
                                returnResult.Status = exiterr.Sys(). 
                              (syscall.WaitStatus).ExitStatus()
                            } else {
                                log.Print("res")
                                log.Print(res)
                                returnResult.Status = 10
                            }
                        }
                        err = command.StatusChan.Send(returnResult)

我要磨练的重点是:

libchan.Pipe()

根据消息来源，这将返回一个 channel 。一个引用保存在客户端，另一个发送到服务器。然后使用此 channel 将值从后者传递到前者。这在实践中如何运作？

client 的完整代码和 server

Answer 1

首先，很高兴知道 Pipe() 所做的只是创建一个 channel 并返回内存中的发送方/接收方对。

来自 inmem.go :

// Pipe returns an inmemory Sender/Receiver pair.
func Pipe() (Receiver, Sender) {
    c := make(chan interface{})
    return pReceiver(c), pSender(c)
}

那你可以在inmem_test.go里面找一个简单的端到端示例。

This struct等同于演示中的 RemoteCommand。

type InMemMessage struct {
    Data   string
    Stream io.ReadWriteCloser
    Ret    Sender
}

在 TestInmemRetPipe() 中，创建了一个简单的客户端和服务器。

The client使用 Pipe() 创建本地发送方/接收方对，同时 the server只需使用 libchan.Sender interface在 InMemMessage 结构中。

请注意，客户端和服务器是分别接收 Sender 或 Receiver 作为参数的函数。在下一个代码片段中有更多相关内容。

func TestInmemRetPipe(t *testing.T) {
    client := func(t *testing.T, w Sender) {
        ret, retPipe := Pipe()
        message := &InMemMessage{Data: "hello", Ret: retPipe}

        err := w.Send(message)
        if err != nil {
            t.Fatal(err)
        }
        msg := &InMemMessage{}
        err = ret.Receive(msg)
        if err != nil {
            t.Fatal(err)
        }

        if msg.Data != "this better not crash" {
            t.Fatalf("%#v", msg)
        }

    }
    server := func(t *testing.T, r Receiver) {
        msg := &InMemMessage{}
        err := r.Receive(msg)
        if err != nil {
            t.Fatal(err)
        }

        if msg.Data != "hello" {
            t.Fatalf("Wrong message:\n\tExpected: %s\n\tActual: %s", "hello", msg.Data)
        }
        if msg.Ret == nil {
            t.Fatal("Message Ret is nil")
        }

        message := &InMemMessage{Data: "this better not crash"}
        if err := msg.Ret.Send(message); err != nil {
            t.Fatal(err)
        }
    }
    SpawnPipeTestRoutines(t, client, server)

}

SpawnPipeTestRoutines() 执行客户端和服务器函数。在此函数中，另一个发送器/接收器对通过 Pipe() 实例化。

在演示应用程序中，此处由 Pipe() 执行的功能(即促进客户端和服务器实例之间的通信)改为通过网络通信处理。

func SpawnPipeTestRoutines(t *testing.T, s SendTestRoutine, r ReceiveTestRoutine) {
    end1 := make(chan bool)
    end2 := make(chan bool)

    receiver, sender := Pipe()

    go func() {
        defer close(end1)
        s(t, sender)
        err := sender.Close()
        if err != nil {
            t.Fatalf("Error closing sender: %s", err)
        }
    }()

    go func() {
        defer close(end2)
        r(t, receiver)
    }()
    ...

在演示应用程序中，通过调用 Transport.NewSendChannel() 来促进通信在客户端和Transport.WaitReceiveChannel() ，分别返回一个 libchan.Sender 和 libchan.Receiver。这些 libchan 实例通过网络处理促进“管道”。

来自 client.go:

sender, err := transport.NewSendChannel()
...
err = sender.Send(command)

来自 server.go:

receiver, err := t.WaitReceiveChannel()
...
err := receiver.Receive(command)

在这两种情况下，先决条件传输配置都是预先完成的(即绑定(bind)到套接字、使用 TLS 等)。

可能还值得注意的是 the spdy library being used is part of the libchan distribution ，因此它提供了 libchan 原语。