为什么要有超时处理？

[devhg]

超时处理是 RPC 框架一个比较基本的能力，如果缺少超时处理机制，无论是服务端还是客户端都容易因为网络或其他错误导致挂死，资源耗尽，这些问题的出现大大地降低了服务的可用性。因此，我们需要在 RPC 框架中加入超时处理的能力。

纵观整个远程调用的过程，需要客户端处理超时的地方有：

• 与服务端建立连接，导致的超时
• 发送请求到服务端，写报文导致的超时
• 等待服务端处理时，等待处理导致的超时（比如服务端已挂死，迟迟不响应）
• 从服务端接收响应时，读报文导致的超时

需要服务端处理超时的地方有：

• 读取客户端请求报文时，读报文导致的超时
• 发送响应报文时，写报文导致的超时
• 调用映射服务的方法时，处理报文导致的超时

[devhg]

drpc如何去做的

1. 在客户端建立链接时 Dial()

// 在这里实现了一个超时处理的外壳 dialTimeout，
// 这个壳将 NewClient 作为入参，在 2 个地方添加了超时处理的机制。
// 1)将 net.Dial 替换为 net.DialTimeout，如果连接创建超时，将返回错误。
// 2)使用子协程执行 NewClient，执行完成后则通过信道 ch 发送结果，
// 如果 time.After() 信道先接收到消息，则说明 NewClient 执行超时，返回错误。
func dialTimeout(f newClientFunc, network, addr string, opts ...*Option) (client *Client, err error) {
        //...
        conn, err := net.DialTimeout(network, addr, option.ConnectTimeout)
        //...
	ch := make(chan clientResult)
	go func() {
		client, err := f(conn, option)
		ch <- clientResult{client, err}
	}()

	// block if ConnectTimout is equal with zero
	if option.ConnectTimeout == 0 {
		result := <-ch
		return result.client, result.err
	}

	// no block else
	select {
	case <-time.After(option.ConnectTimeout):
		return nil, fmt.Errorf("rpc client: connect timeout")
	case result := <-ch:
		return result.client, result.err
	}
}

2. 在客户端Client.Call()整个过程导致的超时（包含发送报文，等待处理，接收报文所有阶段）

// Call 是客户端暴露给用户的RPC服务调用接口，它是对 Go 的封装。
// 阻塞等待call.Done()，等待响应返回，是一个同步接口
// Client.Call 的超时处理机制，使用 context 包实现，控制权交给用户，控制更为灵活。
func (c *Client) Call(ctx context.Context, serviceMethod string, args, reply interface{}) error {
	call := c.Go(serviceMethod, args, reply, make(chan *Call, 1))
	select {
	case <-ctx.Done():
		c.removeCall(call.Seq)
		return errors.New("rpc client: call failed: " + ctx.Err().Error())
	case call := <-call.Done:
		return call.Error
	}
}

3. 服务端处理报文，即 Server.handleRequest 超时

// 这里需要确保 sendResponse 仅调用一次，因此将整个过程拆分为 called 和 sent 两个阶段，
// 在这段代码中只会发生如下两种情况：
// 1) called 信道接收到消息，代表处理没有超时，继续执行 sendResponse。
// 2) time.After() 先于 called 接收到消息，说明处理已经超时，called 和 sent 都将被阻塞。
//    在 case <-time.After(timeout) 处调用 sendResponse。
func (server *Server) handleRequest(cc codec.Codec, req *request,
	sending *sync.Mutex, wg *sync.WaitGroup, timeout time.Duration) {
	defer wg.Done()

	called, sent := make(chan struct{}), make(chan struct{})
	go func() {
		err := req.servci.call(req.mTyp, req.argv, req.replyv)
		called <- struct{}{}
		if err != nil {
			req.h.Error = err.Error()
			server.sendResponse(cc, req.h, invalidRequest, sending)
			sent <- struct{}{}
			return
		}
		server.sendResponse(cc, req.h, req.replyv.Interface(), sending)
		sent <- struct{}{}
	}()

	// block if HandleTimeout is equal with zero
	if timeout == 0 {
		<-called
		<-sent
	}

	select {
	case <-time.After(timeout):
		req.h.Error = fmt.Sprintf("rpc server: request handle timeout: expect within %s", timeout)
		server.sendResponse(cc, req.h, invalidRequest, sending)
	case <-called:
		<-sent
	}
}