动手写RPC框架 - GeeRPC第一天服务端与消息编码

geektutu commented 3 years ago

https://geektutu.com/post/geerpc-day1.html

7天用 Go语言/golang 从零实现 RPC 框架 GeeRPC 教程(7 days implement golang remote procedure call framework from scratch tutorial)，动手写 RPC 框架，参照 golang 标准库 net/rpc 的实现，实现了服务端(server)、支持异步和并发的客户端(client)、消息编码与解码(message encoding and decoding)、服务注册(service register)、支持 TCP/Unix/HTTP 等多种传输协议。第一天实现了一个简单的服务端和消息的编码与解码。

gaodansoft commented 3 years ago

非常感谢这个教程

geektutu commented 3 years ago

@gaodansoft 笔芯~

JYFiaueng commented 3 years ago

说点什么，那当然是赞了😏

geektutu commented 3 years ago

@JYFiaueng 感谢认可~ 😯

wy-ei commented 3 years ago

请问下面这一行是什么意思呢？

var _ Codec = (*GobCodec)(nil)

geektutu commented 3 years ago

请问下面这一行是什么意思呢？
var _ Codec = (*GobCodec)(nil)

@wy-ei 参考 7days-golang 有价值的问题讨论汇总贴

wangwenjunfromlanzhou commented 3 years ago

写的太好了

geektutu commented 3 years ago

写的太好了

@wangwenjunfromlanzhou 感谢认可~ 😊😊😊

leigexiaohuozi commented 3 years ago

直接这样可以吗defer func() { conn.Close() }() 而不是 defer func() { _ = conn.Close() }()

IAOTW commented 3 years ago

client.sending.Lock() defer client.sending.Unlock() client.mu.Lock() defer client.mu.Unlock() 问题1：不太理解为啥terminateCalls方法，需要sending和mu两把锁，而注册call和删除call，只需要mu锁？问题2：当client的某一个字段比如sending Lock()时，是不是在unlock()之前，这个client对象的sending字段是线程安全的，不会被其他线程或协程访问到吗？

Euraxluo commented 3 years ago

@leigexiaohuozi 直接这样可以吗defer func() { conn.Close() }() 而不是 defer func() { _ = conn.Close() }()

应该是 defer conn.Close()

wilgx0 commented 3 years ago

再一次被大佬精湛的技术按在地板上摩擦

XiaoyeFang commented 3 years ago

刚才照着大佬的代码敲了一遍，发现main函数中的循环发送请求如果不执行readBody或者执行出错，我们的sendResponse就会只打印其中的两三个请求，有时甚至还会报EOF错误：

代码如下：

     for i := 0; i < 5; i++ {
        h := &codec.Header{
            ServiceMethod: "Foo.Sum",
            Seq:           uint64(i),
        }
        _ = cc.Write(h, fmt.Sprintf("geerpc req %d", h.Seq))
        _ = cc.ReadHeader(h)
        //var reply string
        //_ = cc.ReadBody(&reply)
        //log.Println("reply:", reply)
    }

是什么原因阻塞了我们的请求呢，又或者执行什么超时导致主线程退出了

XiaoyeFang commented 3 years ago

@wy-ei 请问下面这一行是什么意思呢？
var _ Codec = (*GobCodec)(nil)

就是检查结构体是否实现了这个接口

wilgx0 commented 3 years ago

@XiaoyeFang 刚才照着大佬的代码敲了一遍，发现main函数中的循环发送请求如果不执行readBody或者执行出错，我们的sendResponse就会只打印其中的两三个请求，有时甚至还会报EOF错误：

代码如下：
     for i := 0; i < 5; i++ {
      h := &codec.Header{
          ServiceMethod: "Foo.Sum",
          Seq:           uint64(i),
      }
      _ = cc.Write(h, fmt.Sprintf("geerpc req %d", h.Seq))
      _ = cc.ReadHeader(h)
      //var reply string
      //_ = cc.ReadBody(&reply)
      //log.Println("reply:", reply)
  }
是什么原因阻塞了我们的请求呢，又或者执行什么超时导致主线程退出了

因为你的主协程发送完5次请求后就退出了此时运行服务器的协程还没有打印完这5次请求也退出了

liyuxuan89 commented 3 years ago

我直接使用的day-1的代码，在windows上可以运行，到我虚拟机的ubuntu上，会阻塞在readRequestHeader，这可能是什么原因呢？

liyuxuan89 commented 3 years ago

运行到箭头就阻塞了，不知道为啥？第一个箭头应该是option，第二个是header和body，client的request应该是发出去了？在windows上没有问题。

liyuxuan89 commented 3 years ago

https://github.com/geektutu/7days-golang/issues/34 不知道是不是这个问题

yangchen97 commented 3 years ago

我想问问如果不在header里指定body长度，会有粘包拆包的问题吗？

wilgx0 commented 3 years ago

@yangchen97 我想问问如果不在header里指定body长度，会有粘包拆包的问题吗？

json 字符串是有数据的边界的即 "{" 和 "}"所以这里并不会出现粘包的问题

tangwy01 commented 3 years ago

问下这行代码 // send options = json.NewEncoder(conn).Encode(geerpc.DefaultOption) 是怎么跟server通讯把option发过去的啊，我看这个conn就是conn, := net.Dial("tcp", <-addr).

XiaoyeFang commented 3 years ago

@wilgx0

@yangchen97 我想问问如果不在header里指定body长度，会有粘包拆包的问题吗？

json 字符串是有数据的边界的即 "{" 和 "}"所以这里并不会出现粘包的问题

学到了

sam-lc commented 3 years ago

func NewGobCodec(conn io.ReadWriteCloser) Codec {
    buf := bufio.NewWriter(conn)
    return &GobCodec{
        conn: conn,
        buf:  buf,
        dec:  gob.NewDecoder(conn),
        enc:  gob.NewEncoder(buf),
    }
}

秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了

4ttenji commented 3 years ago

@sam-lc
func NewGobCodec(conn io.ReadWriteCloser) Codec {
  buf := bufio.NewWriter(conn)
  return &GobCodec{
      conn: conn,
      buf:  buf,
      dec:  gob.NewDecoder(conn),
      enc:  gob.NewEncoder(buf),
  }
}
秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了

请问这里可以解释一下，没太明白buf := bufio.NewWriter(conn)以及enc: gob.NewEncoder(buf),这两句

pieceof commented 3 years ago

感谢这个教程！实现这些项目我一个校招生终于春招找到了大厂工作！确实认认真真的熟悉了golang的编程思想和一些设计模式！十分感谢博主！

gu18168 commented 3 years ago

@4ttenji
@sam-lc
func NewGobCodec(conn io.ReadWriteCloser) Codec {
    buf := bufio.NewWriter(conn)
    return &GobCodec{
        conn: conn,
        buf:  buf,
        dec:  gob.NewDecoder(conn),
        enc:  gob.NewEncoder(buf),
    }
}
秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了
请问这里可以解释一下，没太明白buf := bufio.NewWriter(conn)以及enc: gob.NewEncoder(buf),这两句

encoder 因为是要往 conn 中写入内容, 这里文章中说明了要使用 buffer 来优化写入效率, 所以我们先写入到 buffer 中, 然后我们再调用 buffer.Flush() 来将 buffer 中的全部内容写入到 conn 中, 从而优化效率. 对于读则不需要这方面的考虑, 所以直接在 conn 中读内容即可.

z-learner commented 3 years ago

type GobCodec struct { conn io.ReadWriteCloser buf bufio.Writer dec gob.Decoder enc *gob.Encoder } codec.GobCodec on pkg.go.dev

cannot use (GobCodec)(nil) (value of type GobCodec) as Codec value in variable declaration: missing method

valiner commented 3 years ago

json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？

valiner commented 3 years ago

json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？

即使json有{}的分隔符，但是json.decode里面代码是调用了底层conn.read(),这里可能会把header里面的数据读出来，导致下次读取header数据出现残缺。

valiner commented 3 years ago

@lazzman

@valiner

json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？

即使json有{}的分隔符，但是json.decode里面代码是调用了底层conn.read(),这里可能会把header里面的数据读出来，导致下次读取header数据出现残缺。

个人见解，不对请指正！ 见 /sdk/go1.16.4/src/encoding/json/stream.go:49 中Decode方法的实现从方法注释和代码中的注释能看出来工作机制是从缓冲区中不断读取下一个Json编码内容每次反序列前会从conn中读取所有的数据到缓冲区中，再从缓冲区数据中读取一个完整的Json编码内容，所以消息粘包问题被golang的这种流式编解码机制解决了。
// Decode reads the next JSON-encoded value from its
// input and stores it in the value pointed to by v.
//
// See the documentation for Unmarshal for details about
// the conversion of JSON into a Go value.
func (dec *Decoder) Decode(v interface{}) error {
  if dec.err != nil {
      return dec.err
  }

  if err := dec.tokenPrepareForDecode(); err != nil {
      return err
  }

  if !dec.tokenValueAllowed() {
      return &SyntaxError{msg: "not at beginning of value", Offset: dec.InputOffset()}
  }

  // Read whole value into buffer.
  n, err := dec.readValue()
  if err != nil {
      return err
  }
  dec.d.init(dec.buf[dec.scanp : dec.scanp+n])
  dec.scanp += n

  // Don't save err from unmarshal into dec.err:
  // the connection is still usable since we read a complete JSON
  // object from it before the error happened.
  err = dec.d.unmarshal(v)

  // fixup token streaming state
  dec.tokenValueEnd()

  return err
}

每次反序列前会从conn中读取所有的数据到缓冲区中,这个是时候是不是有可能读到header里面的信息。

happy-v587 commented 3 years ago

如果不加消息编码，本质上是两个tcp的conn直接通信：

w -> conn -> conn -> r；

如果加上消息编码，就变成

w -> bufio -> gob -> conn -> conn -> gob -> r

这个流式处理的很漂亮

walkmiao commented 3 years ago

无意中发现如果GobCodec的Write方法忘记gob.buf.Flush()将数据刷入conn中的话在客户端gob decode的时候会阻塞在conn的read过程中，并且没有任何提示。除非设置net.Conn的读超时时间。

walkmiao commented 3 years ago

@valiner
@lazzman

@valiner

json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？

即使json有{}的分隔符，但是json.decode里面代码是调用了底层conn.read(),这里可能会把header里面的数据读出来，导致下次读取header数据出现残缺。

个人见解，不对请指正！ 见 /sdk/go1.16.4/src/encoding/json/stream.go:49 中Decode方法的实现从方法注释和代码中的注释能看出来工作机制是从缓冲区中不断读取下一个Json编码内容每次反序列前会从conn中读取所有的数据到缓冲区中，再从缓冲区数据中读取一个完整的Json编码内容，所以消息粘包问题被golang的这种流式编解码机制解决了。
// Decode reads the next JSON-encoded value from its
// input and stores it in the value pointed to by v.
//
// See the documentation for Unmarshal for details about
// the conversion of JSON into a Go value.
func (dec *Decoder) Decode(v interface{}) error {
    if dec.err != nil {
        return dec.err
    }

    if err := dec.tokenPrepareForDecode(); err != nil {
        return err
    }

    if !dec.tokenValueAllowed() {
        return &SyntaxError{msg: "not at beginning of value", Offset: dec.InputOffset()}
    }

    // Read whole value into buffer.
    n, err := dec.readValue()
    if err != nil {
        return err
    }
    dec.d.init(dec.buf[dec.scanp : dec.scanp+n])
    dec.scanp += n

    // Don't save err from unmarshal into dec.err:
    // the connection is still usable since we read a complete JSON
    // object from it before the error happened.
    err = dec.d.unmarshal(v)

    // fixup token streaming state
    dec.tokenValueEnd()

    return err
}
每次反序列前会从conn中读取所有的数据到缓冲区中,这个是时候是不是有可能读到header里面的信息。

@lazzman
@valiner
@lazzman

@valiner

json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？

即使json有{}的分隔符，但是json.decode里面代码是调用了底层conn.read(),这里可能会把header里面的数据读出来，导致下次读取header数据出现残缺。

个人见解，不对请指正！ 见 /sdk/go1.16.4/src/encoding/json/stream.go:49 中Decode方法的实现从方法注释和代码中的注释能看出来工作机制是从缓冲区中不断读取下一个Json编码内容每次反序列前会从conn中读取所有的数据到缓冲区中，再从缓冲区数据中读取一个完整的Json编码内容，所以消息粘包问题被golang的这种流式编解码机制解决了。
// Decode reads the next JSON-encoded value from its
// input and stores it in the value pointed to by v.
//
// See the documentation for Unmarshal for details about
// the conversion of JSON into a Go value.
func (dec *Decoder) Decode(v interface{}) error {
  if dec.err != nil {
      return dec.err
  }

  if err := dec.tokenPrepareForDecode(); err != nil {
      return err
  }

  if !dec.tokenValueAllowed() {
      return &SyntaxError{msg: "not at beginning of value", Offset: dec.InputOffset()}
  }

  // Read whole value into buffer.
  n, err := dec.readValue()
  if err != nil {
      return err
  }
  dec.d.init(dec.buf[dec.scanp : dec.scanp+n])
  dec.scanp += n

  // Don't save err from unmarshal into dec.err:
  // the connection is still usable since we read a complete JSON
  // object from it before the error happened.
  err = dec.d.unmarshal(v)

  // fixup token streaming state
  dec.tokenValueEnd()

  return err
}
每次反序列前会从conn中读取所有的数据到缓冲区中,这个是时候是不是有可能读到header里面的信息。
是的，缓冲区中可能缓存了连续多个请求的消息（header|body|header|half body）。debug如下断点可以方便了解工作原理： /sdk/go1.16.4/src/encoding/json/stream.go:90 /sdk/go1.16.4/src/encoding/json/stream.go:149

（header|body|header|half body）这种消息json是如何"拆包"的可以阅读如下源码： /sdk/go1.16.4/src/encoding/json/stream.go:104 /sdk/go1.16.4/src/encoding/json/scanner.go:64

type Codec struct { Name string }

func main (){ r:=strings.NewReader({"Name":"gob"}{"Next":"some more content"}) dec:=json.NewDecoder(r) var cc Codec if err:=dec.Decode(&cc);err!=nil{ log.Println("decode err:",err) return } fmt.Println("Codec Name:",cc.Name) remainData,_:=io.ReadAll(r) fmt.Println("remain data:",string(remainData)) } //Codec Name: gob //remain data: 从实验结果看确实会存在这个问题 json解码的时候可能读到后面的数据

lazzman commented 3 years ago

@valiner

json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？

即使json有{}的分隔符，但是json.decode里面代码是调用了底层conn.read(),这里可能会把header里面的数据读出来，导致下次读取header数据出现残缺。

valiner commented 3 years ago

好的，谢谢解答

---原始邮件--- 发件人: @.> 发送时间: 2021年7月6日(周二) 上午10:37 收件人: @.>; 抄送: @.**@.>; 主题: Re: [geektutu/blog] 动手写RPC框架 - GeeRPC第一天服务端与消息编码 | 极客兔兔 (#91)

@valiner

json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？

即使json有{}的分隔符，但是json.decode里面代码是调用了底层conn.read(),这里可能会把header里面的数据读出来，导致下次读取header数据出现残缺。

— You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub, or unsubscribe.

imtzer commented 3 years ago

非常好的文章，感谢。go小白一只，有些疑惑无法解决，handleRequest的并发控制已经有mutex了，为什么还要加上一个waitgroup

Cherrison commented 2 years ago

请问大佬这个编码这样写的话怎么拓展成protobuf啊， protobuf没有encdoer和decoder啊

running8Man commented 2 years ago

请问这么划分逻辑有什么依据吗？我们在写代码时是否需要画好类图呢，还是一边写一边优化呢。

kysshsy commented 2 years ago

func (server *Server) ServeConn(conn io.ReadWriteCloser) {
    defer func() { _ = conn.Close() }()
    // ....

}
func (server *Server) serveCodec(cc codec.Codec) {
    sending := new(sync.Mutex) // make sure to send a complete response
    wg := new(sync.WaitGroup)  // wait until all request are handled
    for {
        //...
    }
    wg.Wait()
    _ = cc.Close() // 这里
}

gobCodec的close 直接调用了conn.close 这里好像多次调用了conn的close?

stackOverflow的讨论

似乎这样会出问题，我想应该有两种解决方案

把codec的close都实现成幂等的
把ServeConn的Close去掉

hammerZh-Z commented 2 years ago

@HappyUncle 如果不加消息编码，本质上是两个tcp的conn直接通信：
w -> conn -> conn -> r；
如果加上消息编码，就变成
w -> bufio -> gob -> conn -> conn -> gob -> r
这个流式处理的很漂亮

这个总结很赞。然后gob是编解码器

hammerZh-Z commented 2 years ago

@4ttenji
@sam-lc
func NewGobCodec(conn io.ReadWriteCloser) Codec {
    buf := bufio.NewWriter(conn)
    return &GobCodec{
        conn: conn,
        buf:  buf,
        dec:  gob.NewDecoder(conn),
        enc:  gob.NewEncoder(buf),
    }
}
秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了
请问这里可以解释一下，没太明白buf := bufio.NewWriter(conn)以及enc: gob.NewEncoder(buf),这两句

@gu18168
@4ttenji
@sam-lc
func NewGobCodec(conn io.ReadWriteCloser) Codec {
  buf := bufio.NewWriter(conn)
  return &GobCodec{
      conn: conn,
      buf:  buf,
      dec:  gob.NewDecoder(conn),
      enc:  gob.NewEncoder(buf),
  }
}
秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了
请问这里可以解释一下，没太明白buf := bufio.NewWriter(conn)以及enc: gob.NewEncoder(buf),这两句
encoder 因为是要往 conn 中写入内容, 这里文章中说明了要使用 buffer 来优化写入效率, 所以我们先写入到 buffer 中, 然后我们再调用 buffer.Flush() 来将 buffer 中的全部内容写入到 conn 中, 从而优化效率. 对于读则不需要这方面的考虑, 所以直接在 conn 中读内容即可.

所以说这个是针对conn生成了一个带缓存的写入，提升写的效率

iCodeBoxFox commented 2 years ago

@goder-tao 非常好的文章，感谢。go小白一只，有些疑惑无法解决，handleRequest的并发控制已经有mutex了，为什么还要加上一个waitgroup 这个控制的不是并发，是服务处理过程中，防止提前退出

baronwithyou commented 2 years ago

请教个问题：golang里文件描述符(FD)的写入已经是 线程安全 的了，为什么sendResponse的时候还需要加锁？

下面为go/fd_unix.go的源码：

// Write implements io.Writer.
func (fd *FD) Write(p []byte) (int, error) {
    if err := fd.writeLock(); err != nil {
        return 0, err
    }
    defer fd.writeUnlock()
    // 详细代码在此查看 
    // https://github.com/golang/go/tree/master/src/internal/poll
}

----------------分割线-------------------

2021.08.30更新用file IO做了验证，发现这里加锁是为了避免缓冲区 c.buf.Flush() 的时候，其他goroutine也在往同一个缓冲区写入，从而导致 err: short write的错误。（假设不使用缓冲区，就不会有这种问题，但是会牺牲一部分buffer带来的性能优化）

cyj19 commented 2 years ago

@XiaoyeFang 刚才照着大佬的代码敲了一遍，发现main函数中的循环发送请求如果不执行readBody或者执行出错，我们的sendResponse就会只打印其中的两三个请求，有时甚至还会报EOF错误：

代码如下：
     for i := 0; i < 5; i++ {
      h := &codec.Header{
          ServiceMethod: "Foo.Sum",
          Seq:           uint64(i),
      }
      _ = cc.Write(h, fmt.Sprintf("geerpc req %d", h.Seq))
      _ = cc.ReadHeader(h)
      //var reply string
      //_ = cc.ReadBody(&reply)
      //log.Println("reply:", reply)
  }
是什么原因阻塞了我们的请求呢，又或者执行什么超时导致主线程退出了

for循环后加一句time.Sleep(2 * time.Second)，在运行就可以看到5个请求都处理了，所以应该就是主程序提前退出导致的

cyj19 commented 2 years ago

@sam-lc
func NewGobCodec(conn io.ReadWriteCloser) Codec {
  buf := bufio.NewWriter(conn)
  return &GobCodec{
      conn: conn,
      buf:  buf,
      dec:  gob.NewDecoder(conn),
      enc:  gob.NewEncoder(buf),
  }
}
秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了

为什么会导致Option和Header一起传过去啊？使用bufio不是为了提高写入的效率吗？按理说使用gob.NewEncoder(conn)也是可以的呀