《游戏服务端编程实践》3.4.1 客户端请求与响应模型

// file: ch03/rpc/call.go
// tier: L1
// verify: go test ./ch03/rpc/

package rpc

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

var (
	ErrTimeout      = errors.New("rpc: call timeout")
	ErrClosed       = errors.New("rpc: caller closed")
	ErrSeqExhausted = errors.New("rpc: no free request id")
)

// Response is the reply the transport hands back to the Caller.
type Response struct {
	Seq  uint32
	Code int32
	Body []byte
}

// Sender writes one request to the connection. It must not block on the reply:
// the reply arrives later through Dispatch.
type Sender func(seq uint32, cmd uint16, body []byte) error

type result struct {
	resp Response
	err  error
}

type pendingCall struct {
	ch chan result
}

// Caller correlates responses with requests through a sequence number. One
// Caller owns one connection; multiplexing several connections through one
// Caller would make sequence reuse ambiguous.
type Caller struct {
	seq     atomic.Uint32
	mu      sync.Mutex
	pending map[uint32]*pendingCall
	send    Sender
	timeout time.Duration
	closed  atomic.Bool
	late    atomic.Uint64
}

func NewCaller(send Sender, timeout time.Duration) *Caller {
	if timeout <= 0 {
		timeout = 3 * time.Second
	}
	return &Caller{
		pending: make(map[uint32]*pendingCall),
		send:    send,
		timeout: timeout,
	}
}

// register allocates a sequence number and inserts the pending entry under one
// lock, so two concurrent calls can never observe or share the same id.
func (c *Caller) register() (uint32, *pendingCall, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed.Load() {
		return 0, nil, ErrClosed
	}
	for i := 0; i < 1<<16; i++ {
		seq := c.seq.Add(1)
		if seq == 0 {
			continue // 0 is reserved for unsolicited pushes
		}
		if _, busy := c.pending[seq]; busy {
			continue // wrap-around landed on a live request
		}
		pc := &pendingCall{ch: make(chan result, 1)}
		c.pending[seq] = pc
		return seq, pc, nil
	}
	return 0, nil, ErrSeqExhausted
}

func (c *Caller) remove(seq uint32) {
	c.mu.Lock()
	delete(c.pending, seq)
	c.mu.Unlock()
}

// Call sends one request and waits for the response carrying the same seq.
// A timeout removes the pending entry before returning, so a response that
// arrives later has nothing to attach to and is counted as late.
func (c *Caller) Call(ctx context.Context, cmd uint16, body []byte) (Response, error) {
	seq, pc, err := c.register()
	if err != nil {
		return Response{}, err
	}
	defer c.remove(seq)

	if err := c.send(seq, cmd, body); err != nil {
		return Response{}, fmt.Errorf("rpc: send cmd=%d seq=%d: %w", cmd, seq, err)
	}

	timer := time.NewTimer(c.timeout)
	defer timer.Stop()

	select {
	case r := <-pc.ch:
		return r.resp, r.err
	case <-timer.C:
		return Response{}, fmt.Errorf("%w: cmd=%d seq=%d", ErrTimeout, cmd, seq)
	case <-ctx.Done():
		return Response{}, ctx.Err()
	}
}

// Dispatch delivers a response to the waiting call. It reports false when no
// call is waiting, which is the normal fate of a response that misses its
// deadline. Counting these events is how you detect a latent timeout that is
// too aggressive for the current network.
func (c *Caller) Dispatch(resp Response) bool {
	if resp.Seq == 0 {
		c.late.Add(1)
		return false
	}
	c.mu.Lock()
	pc, ok := c.pending[resp.Seq]
	if ok {
		delete(c.pending, resp.Seq)
	}
	c.mu.Unlock()
	if !ok {
		c.late.Add(1)
		return false
	}
	select {
	case pc.ch <- result{resp: resp}:
		return true
	default:
		// The waiter left between delete and send; do not block the reader.
		c.late.Add(1)
		return false
	}
}

// Close fails every in-flight call and rejects new ones. It does not close the
// underlying connection; the owner of the connection does that after Close.
func (c *Caller) Close() {
	if !c.closed.CompareAndSwap(false, true) {
		return
	}
	c.mu.Lock()
	pending := c.pending
	c.pending = make(map[uint32]*pendingCall)
	c.mu.Unlock()
	for _, pc := range pending {
		select {
		case pc.ch <- result{err: ErrClosed}:
		default:
		}
	}
}

// Pending reports how many calls are currently waiting for a response.
func (c *Caller) Pending() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return len(c.pending)
}

// LateDiscarded reports how many responses arrived after their call had
// already given up.
func (c *Caller) LateDiscarded() uint64 {
	return c.late.Load()
}

go test -race ./ch03/rpc/