主页 / 问题 / 1318106
Accepted
Inventor
Asked:2022-08-16 19:19:25 +0000 UTC2022-08-16 19:19:25 +0000 UTC

从 syscall.RawConn golang 读取数据

  • 772

有一个tcp服务器。我需要从 syscall.RawConn 获取数据。我正在尝试这样做：

for {
    for _, conn := range tcpConns {
        rawConn, err := conn.SyscallConn()
        if err != nil {
            fmt.Printf("get raw conn error: %s\n", err)
        }
        rawConn.Read(readConn)
    }
}

func readConn(fd uintptr) bool {
    f := os.NewFile(fd, "file")
    b, err := ioutil.ReadAll(f)
    if err != nil {
        fmt.Printf("read file error: %s\n", err)
        return false
    } else {
        fmt.Printf("New data: %s\n", string(b))
        return true
    }
}

执行时，在终端中我观察到：

read file error: read file: resource temporarily unavailable

那些。返回此错误ioutil.ReadAll(f)。此外，该函数readConn返回 false 并且程序执行被阻止。那些。rawConn.Read(readConn)即使我的函数已执行并返回结果，代码也不会继续。文档说，在 readConn 返回 false 的情况下，将在可能时再尝试读取数据。但是什么都没有发生，也不清楚“何时成为可能”是什么意思。

我已经尝试在函数中返回 true，readConn但随后在该行的无限循环的第二个循环上执行块ioutil.ReadAll(f)。我应该怎么做才能获取数据？

我知道你不能syscall.RawConn从 net.TCPConn 接收和读取，但我不喜欢你需要为那里的每个连接创建一个 goroutine 并且它会占用大量内存，所以我想创建 1 个无限循环来读取来自没有 goroutines 的所有连接的数据

golang

1 个回答

  1. Best Answer

    Go 以非阻塞方式创建套接字。如果没有数据，则系统调用read返回一个错误EAGAIN，Go 提供一个测试消息resource temporarily unavailable

    如果忽略此错误，则它rawConn可以工作。

    注：如果你使用RawConn.Read，那么你不需要调用ReadAll--RawConn.Read块，这意味着它会调用你的函数readConn，直到函数返回true。

    Read 在底层连接的文件描述符或句柄上调用 f；f 应尝试从文件描述符中读取。如果 f 返回 true，则 Read 返回。否则，Read 会阻塞等待连接准备好读取并重复尝试。文件描述符保证在 f 执行时保持有效，但在 f 返回后不会。

    用法示例RawConn.Read。该程序使用系统调用从套接字文件描述符中读取syscall.Read。

    package main

import (
    "fmt"
    "net"
    "os"
    "syscall"
)

func main() {
    addr, err := net.ResolveTCPAddr("tcp", "localhost:54321")
    if err != nil {
        fmt.Println("Address resolution failed:", err.Error())
        os.Exit(1)
    }

    l, err := net.ListenTCP("tcp", addr)
    if err != nil {
        fmt.Println("Error listening:", err.Error())
        os.Exit(1)
    }
    // Close the listener when the application closes.
    defer l.Close()
    fmt.Println("Listening on ", addr)

    conn, err := l.AcceptTCP()
    if err != nil {
        fmt.Println("Accept failed: ", err)
    }
    defer conn.Close()

    rawConn, err := conn.SyscallConn()
    if err != nil {
        fmt.Printf("get raw conn error: %s\n", err)
    }

    rawConn.Read(readConnWithSyscall)
}

func readConnWithSyscall(fd uintptr) (done bool) {
    fmt.Println("readConn: file descriptor:", fd)
    byteBuf := make([]byte, 1024)

    n, err := syscall.Read(int(fd), byteBuf)
    if err != nil {
        if err == syscall.EAGAIN {
            println("readconn: EAGAIN")
            // Try once more
            return false
        } else {
            println("read: error ", err)
            return true
        }
    }
    if n > 0 {
        println("readconn: read", n, "bytes: ", string(byteBuf[:n]))
        // More left to read
        return false
    } else {
        // read returned 0 == EOF
        println("readconn: finished")
        return true
    }
}

    测试运行echo "Test" | nc -N -w 1 -v localhost 54321 服务器程序输出：

    Listening on  127.0.0.1:54321
readConn: file descriptor: 7
readconn: EAGAIN
readConn: file descriptor: 7
readconn: read 5 bytes:  Test

readConn: file descriptor: 7
readconn: finished

    更新

    异步套接字读取器。此选项不会阻止RawConn.Read.

    package main

import (
    "fmt"
    "net"
    "os"
    "syscall"
)

type asyncReader struct {
    data []byte
    done bool
}

func (r *asyncReader) Read(fd uintptr) (done bool) {
    byteBuf := make([]byte, 1024)
    done = true

    n, err := syscall.Read(int(fd), byteBuf)
    if err != nil {
        if err == syscall.EAGAIN {
            println("asyncReader: EAGAIN")
        } else {
            println("asyncReader: error ", err)
            r.done = true
        }
    } else if n > 0 {
        println("asyncReader: read", n, "bytes: ", string(byteBuf[:n]))
        r.data = append(r.data, byteBuf[:n]...)
    } else {
        // read returned 0 == EOF
        println("asyncReader: finished")
        r.done = true
    }
    return
}

func main() {
    addr, err := net.ResolveTCPAddr("tcp", "localhost:54321")
    if err != nil {
        fmt.Println("Address resolution failed:", err.Error())
        os.Exit(1)
    }

    l, err := net.ListenTCP("tcp", addr)
    if err != nil {
        fmt.Println("Error listening:", err.Error())
        os.Exit(1)
    }
    // Close the listener when the application closes.
    defer l.Close()
    fmt.Println("Listening on ", addr)

    conn, err := l.AcceptTCP()
    if err != nil {
        fmt.Println("Accept failed: ", err)
    }
    defer conn.Close()

    rawConn, err := conn.SyscallConn()
    if err != nil {
        fmt.Printf("get raw conn error: %s\n", err)
    }

    // rawConn.Read(readConnWithSyscall)
    r := &asyncReader{done: false, data: make([]byte, 0)}
    for !r.done {
        rawConn.Read(r.Read)
        println("main: data size:", len(r.data))
    }

    println("main: received:", string(r.data))
}

    示例 - 客户端发送一行，然后等待一秒钟并发送另一行：

    { echo "Test 1"; sleep 1; echo "Test 2"; } | nc -N -w 1 -v localhost 54321

    服务器：

    Listening on  127.0.0.1:54321
asyncReader: EAGAIN
main: data size: 0
asyncReader: EAGAIN
main: data size: 0
asyncReader: EAGAIN
main: data size: 0
asyncReader: EAGAIN
main: data size: 0
asyncReader: EAGAIN
main: data size: 0
asyncReader: read 7 bytes:  Test 1

main: data size: 7
asyncReader: EAGAIN
main: data size: 7
.
.
.
asyncReader: read 7 bytes:  Test 2

main: data size: 14
.
.
.
asyncReader: finished
main: data size: 14
main: received: Test 1
Test 2

    省略号中隐藏了 9 万个（！）异步读取器调用。

    主要缺点是此选项会 100% 加载 CPU，因为它会不断拉动套接字。

    对于没有冗余请求的多个套接字的真正异步处理，您需要syscall.Select使用syscall.EpollWait

    • 2

相关问题