Go语言P2P文件传输并行请求异常排查与修复求助

云上老周

2026-6-20

Go P2P文件传输系统并行请求异常排查与修复

问题背景

我用Go构建P2P文件传输系统，需要从对等节点并行获取数据块。顺序请求完全正常，但用goroutines发起并行请求时出现异常：偶发sync.WaitGroup计数器负数panic、部分数据块传输不完整。

代码结构概述

main.go：初始化并启动两个对等节点服务器，用goroutines模拟节点间文件请求
peer.go：定义PeerServer结构体，处理节点连接、文件请求逻辑
tcp_transport.go：实现TCP传输层，负责连接建立与数据传输
transport.go：定义传输接口规范
encoding.go：提供消息编解码功能

核心异常

sync.WaitGroup计数器负数：并行请求时偶发该panic
数据传输不完整：goroutines模式下部分数据块未完成接收或写入

关键代码片段

main.go

func main() {
    // Initialize and start peer servers
    peerServer1 := NewPeerServer(opts1, "localhost:5000")
    peerServer2 := NewPeerServer(opts2, "localhost:5001")

    var wg sync.WaitGroup
    wg.Add(2)
    go func() { defer wg.Done(); peerServer1.Start() }()
    go func() { defer wg.Done(); peerServer2.Start() }()

    time.Sleep(2 * time.Second) // Wait for servers to start

    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            peerServer1.TestFileRequest(i, "127.0.0.1:5001")
        }(i)
    }

    wg.Wait()
    log.Println("Servers shut down successfully.")
}

peer.go

type PeerServer struct {
    peerLock  sync.Mutex
    peers     map[string]Peer
    Transport Transport
}

func (p *PeerServer) TestFileRequest(i int, ipAddr string) error {
    err := p.Transport.Dial(ipAddr, "peer-server")
    if err != nil {
        return err
    }

    dataMessage := DataMessage{Payload: TestSend{ChunkId: i, ChunkName: fmt.Sprintf("chunk_%d.chunk", i)}}
    var buf bytes.Buffer
    encoder := gob.NewEncoder(&buf)
    if err := encoder.Encode(dataMessage); err != nil {
        return fmt.Errorf("Failed to encode data: %v", err)
    }

    p.peerLock.Lock()
    peer, exists := p.peers[ipAddr]
    p.peerLock.Unlock()
    if !exists {
        return fmt.Errorf("peer with address %s not found", ipAddr)
    }

    if err := peer.Send(buf.Bytes()); err != nil {
        return fmt.Errorf("Failed to send data: %v", err)
    }

    time.Sleep(time.Millisecond * 500)

    var fileSize int64
    if err := binary.Read(peer, binary.LittleEndian, &fileSize); err != nil {
        return fmt.Errorf("Failed to read file size: %v", err)
    }

    file, err := os.Create(fmt.Sprintf("chunk_%d.chunk", i))
    if err != nil {
        return fmt.Errorf("Failed to create file: %v", err)
    }
    defer file.Close()

    if _, err := io.CopyN(file, peer, fileSize); err != nil {
        return fmt.Errorf("Failed to copy file: %v", err)
    }

    fmt.Printf("Successfully received and saved chunk_%d.chunk\n", i)
    peer.CloseStream(i)

    return nil
}

tcp_transport.go

type TCPPeer struct {
    net.Conn
    outbound bool
    wg       *sync.WaitGroup
}

func (p *TCPPeer) CloseStream(i int) {
    p.wg.Done()
}

func (p *TCPPeer) Send(b []byte) error {
    _, err := p.Write(b)
    return err
}

type TCPTransport struct {
    TransportOpts
    listener net.Listener
    msgch    chan msg
}

func (t *TCPTransport) Dial(addr string, connType string) error {
    conn, err := net.Dial("tcp", addr)
    if err != nil {
        return err
    }
    go t.handleConn(conn, true, connType)
    return nil
}

func (t *TCPTransport) handleConn(conn net.Conn, outbound bool, connType string) {
    peer := NewTCPPeer(conn, outbound)
    if t.OnPeer != nil {
        if err := t.OnPeer(peer, connType); err != nil {
            return
        }
    }
    for {
        msg := msg{}
        err := t.Decoder.Reader(conn, &msg)//reads the message if stream is sent makes msg.stream==true
        if err != nil {
            return
        }
        msg.From = conn.RemoteAddr().String()
        if msg.Stream {
            peer.wg.Add(1)
            peer.wg.Wait()
            continue
        }
        t.msgch <- msg
    }
}

疑似问题点

WaitGroup竞态：TCPPeer中的wg被并发操作时未加锁，Add和Done的执行顺序混乱导致计数器异常
时序错误：TestFileRequest中调用peer.CloseStream(i)可能早于handleConn里的peer.wg.Add(1)，直接造成计数器负数
TCP连接复用问题：并行请求复用同一个Peer连接，数据传输时出现交叉，导致数据不完整
硬编码休眠不可靠：time.Sleep(500ms)无法保证远端已准备好响应，可能导致读文件大小操作提前执行失败

修复方案

1. 修正WaitGroup的并发安全问题

为每个数据流维护独立的WaitGroup，避免共享实例引发的竞态：

// 修改TCPPeer结构体
type TCPPeer struct {
    net.Conn
    outbound bool
    streamWg map[int]*sync.WaitGroup // 按ChunkId绑定独立WaitGroup
    wgLock   sync.Mutex              // 保护streamWg的并发访问
}

// 初始化TCPPeer
func NewTCPPeer(conn net.Conn, outbound bool) *TCPPeer {
    return &TCPPeer{
        Conn:     conn,
        outbound: outbound,
        streamWg: make(map[int]*sync.WaitGroup),
    }
}

// 初始化对应ChunkId的WaitGroup
func (p *TCPPeer) InitStream(chunkId int) {
    p.wgLock.Lock()
    defer p.wgLock.Unlock()
    p.streamWg[chunkId] = &sync.WaitGroup{}
    p.streamWg[chunkId].Add(1)
}

// CloseStream仅操作对应ChunkId的WaitGroup
func (p *TCPPeer) CloseStream(chunkId int) {
    p.wgLock.Lock()
    wg, exists := p.streamWg[chunkId]
    p.wgLock.Unlock()
    if exists {
        wg.Done()
    }
}

// 修改handleConn中的流处理逻辑
func (t *TCPTransport) handleConn(conn net.Conn, outbound bool, connType string) {
    peer := NewTCPPeer(conn, outbound)
    if t.OnPeer != nil {
        if err := t.OnPeer(peer, connType); err != nil {
            return
        }
    }
    for {
        msg := msg{}
        err := t.Decoder.Reader(conn, &msg)
        if err != nil {
            return
        }
        msg.From = conn.RemoteAddr().String()
        if msg.Stream {
            // 假设msg中携带ChunkId，需调整解码逻辑传递该信息
            chunkId := msg.Payload.(TestSend).ChunkId
            p.wgLock.Lock()
            wg, exists := peer.streamWg[chunkId]
            p.wgLock.Unlock()
            if exists {
                wg.Wait()
            }
            continue
        }
        t.msgch <- msg
    }
}

2. 避免TCP连接复用的数据交叉

每个并行请求建立独立TCP连接，或在连接上实现数据流多路复用：

// 修改TestFileRequest，使用独立连接通信
func (p *PeerServer) TestFileRequest(i int, ipAddr string) error {
    // 建立独立TCP连接
    conn, err := net.Dial("tcp", ipAddr)
    if err != nil {
        return err
    }
    defer conn.Close()

    dataMessage := DataMessage{Payload: TestSend{ChunkId: i, ChunkName: fmt.Sprintf("chunk_%d.chunk", i)}}
    var buf bytes.Buffer
    encoder := gob.NewEncoder(&buf)
    if err := encoder.Encode(dataMessage); err != nil {
        return fmt.Errorf("Failed to encode data: %v", err)
    }

    if _, err := conn.Write(buf.Bytes()); err != nil {
        return fmt.Errorf("Failed to send data: %v", err)
    }

    // 用就绪信号替代硬编码休眠
    var readySignal bool
    if err := binary.Read(conn, binary.LittleEndian, &readySignal); err != nil {
        return fmt.Errorf("Failed to read ready signal: %v", err)
    }
    if !readySignal {
        return fmt.Errorf("Remote peer not ready for chunk %d", i)
    }

    var fileSize int64
    if err := binary.Read(conn, binary.LittleEndian, &fileSize); err != nil {
        return fmt.Errorf("Failed to read file size: %v", err)
    }

    file, err := os.Create(fmt.Sprintf("chunk_%d.chunk", i))
    if err != nil {
        return fmt.Errorf("Failed to create file: %v", err)
    }
    defer file.Close()

    if _, err := io.CopyN(file, conn, fileSize); err != nil {
        return fmt.Errorf("Failed to copy file: %v", err)
    }

    fmt.Printf("Successfully received and saved chunk_%d.chunk\n", i)
    return nil
}

3. 移除硬编码休眠，改用同步信号

用明确的就绪信号替代time.Sleep，确保远端准备好后再执行数据读取操作。

并发调试技巧

启用Go竞态检测器：编译时添加-race参数，直接检测竞态条件：go run -race main.go
日志追踪：在关键操作（WaitGroup的Add/Done、数据读写）处添加日志，标记goroutine ID和ChunkId，便于追踪时序问题
单元测试：编写并发场景测试用例，模拟多个goroutine同时操作的情况，复现问题
隔离变量：确保每个goroutine操作独立的变量实例，减少共享状态引发的竞态

内容的提问来源于stack exchange，提问作者Tarun Kavipurapu

Go语言P2P文件传输并行请求异常排查与修复求助

问题背景 #

代码结构概述 #

核心异常 #

关键代码片段 #

main.go #

peer.go #

tcp_transport.go #

疑似问题点 #

修复方案 #

1. 修正WaitGroup的并发安全问题 #

2. 避免TCP连接复用的数据交叉 #

3. 移除硬编码休眠，改用同步信号 #

并发调试技巧 #