sockets - binary.Write to socket in go blocks

Question

我正在尝试编写一个简单的客户端-服务器应用程序。它在同一台计算机上的双方都可以正常工作，但在两台不同的计算机上运行时会阻塞。这些计算机是 RaspBerry Pi 2 和 MacBook Pro。

问题是客户端在二进制中阻塞。在几次良好运行后写入。通常，接收数据的服务器打印它已收到 63，而发送数据的客户端打印它将发送 64。

如果我将截止日期通知服务器，它会超时退出。客户端也有截止日期但不会超时。

我尝试用简单的字节数组或单个 int 替换数据。它仍然阻塞。但数据越小，正确的次数就越多。

如果我删除客户端中的 Printf 函数，它也会运行更长的时间。

我还尝试通过从套接字到 STDIO 的简单副本来替换服务器的功能(我在某处找到的提示)，但它仍然无法从 binary.Write 中出来。

有什么想法吗？

这是完整的源代码。那里有一些伪造的东西(例如计数循环)，但它们真的很简单。还是我弄错了？

package main

import (
    //  "bytes"
    "encoding/binary"
    "fmt"
    "io"
    "net"
    "os"
    "os/exec"
    "runtime/trace"
    "time"
    "unsafe"
)

type TheMessageType struct {
    X             int32
    RestOfMessage [1000]byte
}

func listener(theCounter int, active bool, f *os.File) {
    var message TheMessageType
    var err error
    var conn net.Conn

    count := theCounter

    for {
        // Dial the server
        fmt.Printf("LISTENER %d: connecting to socket\n", count)
        conn, err = net.Dial("tcp", "192.168.1.113:49152")
        if err != nil {
            fmt.Printf("LISTENER %d: not connected, retrying in 3 seconds\n", count)
            time.Sleep(3 * time.Second)
        } else {
            fmt.Printf("LISTENER %d: connected\n", count)
            break
        }
    }
    defer conn.Close()

    // Loop writing to socket
    for {
        // Set deadline
        // conn.SetDeadline(time.Now().Add(2 * time.Second))

        // Print some info
        fmt.Printf("LISTENER %d: writing from %v to %v\n", count, conn.LocalAddr(), conn.RemoteAddr())

        // Prepare message to write to socket
        message.X = int32(count)
        theString := fmt.Sprintf("%d", count)
        copy(message.RestOfMessage[:], theString)

        // binary write directly to socket
        fmt.Printf("LISTENER %d: binary.Write this: %d, %s\n", count, message.X, message.RestOfMessage)

        f.Sync()
        err = binary.Write(conn, binary.LittleEndian, message)

        f.Sync()
        fmt.Printf("LISTENER %d: written\n", count)
        // time.Sleep(50 * time.Millisecond)
        checkError(err, "LISTENER")

        count = count + 1
    }
}

func main() {
    var myMessage TheMessageType
    var ln net.Listener
    var conn net.Conn
    var err error
    var theCount int
    var avgspeed, speed float64
    var speedlist [10]float64

    curspeed := 0

    const listenerActive = true
    const mainActive = false

    f, err := os.Create("theFile.out")
    trace.Start(f)

    t0 := time.Now()
    t1 := time.Now()

    transferSize := unsafe.Sizeof(myMessage)

    fmt.Printf("MAIN: timestamp %s ; size of transfers is %d\n", t0.String(), transferSize)

    if mainActive {
        fmt.Println("MAIN: listen")
        ln, err = net.Listen("tcp", ":49152")
        fmt.Println("MAIN: defer")
        defer ln.Close()
        fmt.Println("MAIN: checkError")
        checkError(err, "MAIN")
    }

    // launch listener
    if listenerActive {
        go listener(theCount, listenerActive, f)
    }

    for theCount = 1; ; theCount++ {

        if mainActive {
            fmt.Println("MAIN: accept")
            conn, err = ln.Accept()
            checkError(err, "MAIN")

            exit := false

            for !exit {
                // Set deadline
                // conn.SetDeadline(time.Now().Add(2 * time.Second))

                // Print some info
                fmt.Printf("MAIN: reading on %v from %v\n", conn.LocalAddr(), conn.RemoteAddr())

                // move time
                t0 = t1

                // read from socket
                fmt.Println("MAIN: reader")

                f.Sync()
                err = binary.Read(conn, binary.LittleEndian, &myMessage)

                f.Sync()
                if err == io.EOF {
                    fmt.Println("EOF!")
                    exit = true
                    conn.Close()
                } else {
                    checkError(err, "MAIN")
                }

                // clear screen
                c := exec.Command("clear")
                c.Stdout = os.Stdout
                c.Run()

                // print received data
                fmt.Printf("MAIN: socket message received: %d, %s\n", myMessage.X, myMessage.RestOfMessage)

                // calculate speed & average
                t1 = time.Now()
                tdif := t1.Sub(t0)
                speed = float64(transferSize) / tdif.Seconds() / 1000000.0 * 8.0
                speedlist[curspeed] = speed
                curspeed = curspeed + 1
                if curspeed == 10 {
                    curspeed = 0
                }
                totalspeed := 0.0
                for i := 0; i < 10; i++ {
                    totalspeed += speedlist[i]
                }
                avgspeed = totalspeed / 10

                // print speed
                fmt.Printf("MAIN: avg mbit/s %9.2f ; actual mbit/s: %9.2f\n", avgspeed, speed)
                // time.Sleep(1 * time.Second)
            }
        }
    }
}

func checkError(err error, name string) {
    if err != nil {
        fmt.Fprintf(os.Stderr, "%s: Fatal error: %s\n", name, err.Error())
        //      panic(fmt.Sprintf("%v", err.Error()))
        os.Exit(1)
    }
}

这是客户端被阻止写入时的堆栈跟踪:

goroutine 1 [running]:
runtime.throw(0x1927b8, 0x5)
    /usr/local/go/src/runtime/panic.go:527 +0x90 fp=0xc82004bd78 sp=0xc82004bd60
runtime.sigpanic()
    /usr/local/go/src/runtime/sigpanic_unix.go:27 +0x2ba fp=0xc82004bdc8 sp=0xc82004bd78
main.main()
    /Users/ple/dev/go/src/Peter-test/main.go:108 +0x33c fp=0xc82004bf50 sp=0xc82004bdc8
runtime.main()
    /usr/local/go/src/runtime/proc.go:111 +0x2b0 fp=0xc82004bfa0 sp=0xc82004bf50
runtime.goexit()
    /usr/local/go/src/runtime/asm_amd64.s:1696 +0x1 fp=0xc82004bfa8 sp=0xc82004bfa0

goroutine 17 [syscall, locked to thread]:
runtime.goexit()
    /usr/local/go/src/runtime/asm_amd64.s:1696 +0x1

goroutine 5 [trace reader (blocked)]:
runtime.ReadTrace(0x0, 0x0, 0x0)
    /usr/local/go/src/runtime/trace.go:339 +0x207
runtime/trace.Start.func1(0x5a42e0, 0xc82002a020)
    /usr/local/go/src/runtime/trace/trace.go:28 +0x18
created by runtime/trace.Start
    /usr/local/go/src/runtime/trace/trace.go:34 +0x78

goroutine 6 [runnable]:
reflect.Value.Index(0x1234c0, 0xc8204ac004, 0x51, 0x75, 0x12ac00, 0xc8204ac078, 0x48)
    /usr/local/go/src/reflect/value.go:823
encoding/binary.(*encoder).value(0xc8200475a0, 0x1234c0, 0xc8204ac004, 0x51)
    /usr/local/go/src/encoding/binary/binary.go:509 +0x808
encoding/binary.(*encoder).value(0xc8200475a0, 0x162ba0, 0xc8204ac000, 0x59)
    /usr/local/go/src/encoding/binary/binary.go:518 +0xb4f
encoding/binary.Write(0x621138, 0xc8200a2000, 0x621160, 0x289ef0, 0x162ba0, 0xc8204ac000, 0x0, 0x0)
    /usr/local/go/src/encoding/binary/binary.go:316 +0x1792
main.listener(0x0, 0x1, 0xc82002a020)
    /Users/ple/dev/go/src/Peter-test/main.go:59 +0xac8
created by main.main
    /Users/ple/dev/go/src/Peter-test/main.go:105 +0x332

Answer 1

吉姆的回答是正确的。 Main 进入忙循环。我通过将一个 channel 传递给监听器并在 main 的末尾等待它来解决它。客户端在 for 循环之后“写入”它，这是一个无限循环，所以 main 永远不会结束。从那时起，程序就一直在传递数据。为了确定，我会让它运行一整夜。

我只得到大约 12.5Mbit/s，这有点滑稽。连接是 1Gbit Wifi AC 到 100mbit 以太网交换机到 Raspberry Pi。但这是另一个问题。

谢谢!