add comments

Author: seeflood

components/rpc/callback/callback.go

@@ -57,7 +57,7 @@ var (
 	// to storage BeforeFactory
 	beforeInvokeRegistry = map[string]BeforeFactory{}
 	// to storage AfterFactory
-	afterInvokeRegistry  = map[string]AfterFactory{}
+	afterInvokeRegistry = map[string]AfterFactory{}
 )
 
 // NewCallback is created Callback
@@ -84,7 +84,7 @@ func (c *callback) AddBeforeInvoke(conf rpc.CallbackFunc) {
 	c.beforeInvoke = append(c.beforeInvoke, f.Create())
 }
 
-// AddAfterInvoke is add beforeInvoke into callback.afterInvoke
+// AddAfterInvoke is used to add beforeInvoke into callback.afterInvoke
 func (c *callback) AddAfterInvoke(conf rpc.CallbackFunc) {
 	f, ok := afterInvokeRegistry[conf.Name]
 	if !ok {
@@ -98,7 +98,7 @@ func (c *callback) AddAfterInvoke(conf rpc.CallbackFunc) {
 	c.afterInvoke = append(c.afterInvoke, f.Create())
 }
 
-// BeforeInvoke is get RPCRequest in callback.beforeInvoke
+// BeforeInvoke is used to invoke beforeInvoke callbacks
 func (c *callback) BeforeInvoke(request *rpc.RPCRequest) (*rpc.RPCRequest, error) {
 	var err error
 	for _, cb := range c.beforeInvoke {
@@ -109,7 +109,7 @@ func (c *callback) BeforeInvoke(request *rpc.RPCRequest) (*rpc.RPCRequest, error
 	return request, err
 }
 
-// AfterInvoke is get RPCResponse in callback.afterInvoke
+// AfterInvoke is used to invoke afterInvoke callbacks
 func (c *callback) AfterInvoke(response *rpc.RPCResponse) (*rpc.RPCResponse, error) {
 	var err error
 	for _, cb := range c.afterInvoke {

components/rpc/invoker/mosn/channel/connpool.go

@@ -88,7 +88,6 @@ func newConnPool(
 	return p
 }
 
-
 // connPool is connected pool
 type connPool struct {
 	maxActive   int
@@ -131,6 +130,7 @@ func (p *connPool) Get(ctx context.Context) (*wrapConn, error) {
 	if p.stateFunc != nil {
 		wc.state = p.stateFunc()
 	}
+	// start a readloop gorountine to read and handle data
 	if p.onDataFunc != nil {
 		utils.GoWithRecover(func() {
 			p.readloop(wc)
@@ -171,6 +171,7 @@ func (p *connPool) readloop(c *wrapConn) {
 
 	c.buf = buffer.NewIoBuffer(defaultBufSize)
 	for {
+		// read data from connection
 		n, readErr := c.buf.ReadOnce(c)
 		if readErr != nil {
 			err = readErr
@@ -182,6 +183,8 @@ func (p *connPool) readloop(c *wrapConn) {
 		}
 
 		if n > 0 {
+			// handle data.
+			// it will delegate to hstate if it's constructed by httpchannel
 			if onDataErr := p.onDataFunc(c); onDataErr != nil {
 				err = onDataErr
 				log.DefaultLogger.Errorf("[runtime][rpc]connpool onData err: %s", onDataErr.Error())

components/rpc/invoker/mosn/channel/httpchannel.go

@@ -36,8 +36,11 @@ func init() {
 	RegistChannel("http", newHttpChannel)
 }
 
+// hstate is a pipe for readloop goroutine to communicate with request goroutine
 type hstate struct {
+	// request goroutine will read data from it
 	reader net.Conn
+	// readloop goroutine will write data to it
 	writer net.Conn
 }
 
@@ -60,20 +63,30 @@ type httpChannel struct {
 	pool *connPool
 }
 
-// newHttpChannel is create rpc.Channel by ChannelConfig
+// newHttpChannel is used to create rpc.Channel according to ChannelConfig
 func newHttpChannel(config ChannelConfig) (rpc.Channel, error) {
 	hc := &httpChannel{}
 	hc.pool = newConnPool(
 		config.Size,
+		// dialFunc
 		func() (net.Conn, error) {
 			local, remote := net.Pipe()
 			localTcpConn := &fakeTcpConn{c: local}
 			remoteTcpConn := &fakeTcpConn{c: remote}
 			if err := acceptFunc(remoteTcpConn, config.Listener); err != nil {
 				return nil, err
 			}
+			// the goroutine model is:
+			// request goroutine --->  localTcpConn ---> mosn
+			//		^										|
+			//		|										|
+			//		|										|
+			// 		hstate <-- readloop goroutine 	<------
 			return localTcpConn, nil
-		}, func() interface{} {
+		},
+		// stateFunc
+		func() interface{} {
+			// hstate is a pipe for readloop goroutine to communicate with request goroutine
 			s := &hstate{}
 			s.reader, s.writer = net.Pipe()
 			return s
@@ -84,25 +97,29 @@ func newHttpChannel(config ChannelConfig) (rpc.Channel, error) {
 	return hc, nil
 }
 
-// Do is handle RPCRequest to RPCResponse
+// Do is used to handle RPCRequest and return RPCResponse
 func (h *httpChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
+	// 1. context.WithTimeout
 	timeout := time.Duration(req.Timeout) * time.Millisecond
 	ctx, cancel := context.WithTimeout(req.Ctx, timeout)
 	defer cancel()
 
+	// 2. get a fake connection with mosn
+	// The pool will start a readloop gorountine,
+	// which aims to read data from mosn and then write data to the hstate.writer
 	conn, err := h.pool.Get(ctx)
 	if err != nil {
 		return nil, err
 	}
-
+	// 3. set deadline before write data to this connection
 	hstate := conn.state.(*hstate)
 	deadline, _ := ctx.Deadline()
 	if err = conn.SetWriteDeadline(deadline); err != nil {
 		hstate.close()
 		h.pool.Put(conn, true)
 		return nil, common.Error(common.UnavailebleCode, err.Error())
 	}
-
+	// 4. write data to this fake connection
 	httpReq := h.constructReq(req)
 	defer fasthttp.ReleaseRequest(httpReq)
 
@@ -112,6 +129,7 @@ func (h *httpChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
 		return nil, common.Error(common.UnavailebleCode, err.Error())
 	}
 
+	// 5. read response data and parse it into fasthttp.Response
 	httpResp := &fasthttp.Response{}
 	hstate.reader.SetReadDeadline(deadline)
 
@@ -121,6 +139,8 @@ func (h *httpChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
 		return nil, common.Error(common.UnavailebleCode, err.Error())
 	}
 	h.pool.Put(conn, false)
+
+	// 6. convert result to rpc.RPCResponse,which is the response of rpc invoker
 	body := httpResp.Body()
 	if httpResp.StatusCode() != http.StatusOK {
 		return nil, common.Errorf(common.UnavailebleCode, "http response code %d, body: %s", httpResp.StatusCode(), string(body))

components/rpc/invoker/mosn/channel/xchannel.go

@@ -51,6 +51,7 @@ func newXChannel(config ChannelConfig) (rpc.Channel, error) {
 	m := &xChannel{proto: proto}
 	m.pool = newConnPool(
 		config.Size,
+		// dialFunc
 		func() (net.Conn, error) {
 			local, remote := net.Pipe()
 			localTcpConn := &fakeTcpConn{c: local}
@@ -60,6 +61,7 @@ func newXChannel(config ChannelConfig) (rpc.Channel, error) {
 			}
 			return localTcpConn, nil
 		},
+		// stateFunc
 		func() interface{} {
 			return &xstate{calls: map[uint32]chan call{}}
 		},
@@ -90,18 +92,20 @@ type xChannel struct {
 
 // Do is handle RPCRequest to RPCResponse
 func (m *xChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
+	// 1. context.WithTimeout
 	timeout := time.Duration(req.Timeout) * time.Millisecond
 	ctx, cancel := context.WithTimeout(req.Ctx, timeout)
 	defer cancel()
 
+	// 2. get fake connection with mosn
 	conn, err := m.pool.Get(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	xstate := conn.state.(*xstate)
 
-	// encode request
+	// 3. encode request
 	frame := m.proto.ToFrame(req)
 	id := atomic.AddUint32(&xstate.reqid, 1)
 	frame.SetRequestId(uint64(id))
@@ -131,6 +135,7 @@ func (m *xChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
 	}
 	m.pool.Put(conn, false)
 
+	// read response and decode it
 	select {
 	case res := <-callChan:
 		if res.err != nil {

components/rpc/invoker/mosn/mosninvoker.go

@@ -89,24 +89,26 @@ func (m *mosnInvoker) Invoke(ctx context.Context, req *rpc.RPCRequest) (resp *rp
 		}
 	}()
 
+	// 1. validate request
 	if req.Timeout == 0 {
 		req.Timeout = 3000
 	}
 	req.Ctx = ctx
 	log.DefaultLogger.Debugf("[runtime][rpc]request %+v", req)
+	// 2. beforeInvoke callback
 	req, err = m.cb.BeforeInvoke(req)
 	if err != nil {
 		log.DefaultLogger.Errorf("[runtime][rpc]before filter error %s", err.Error())
 		return nil, err
 	}
-
+	// 3. do invocation
 	resp, err = m.channel.Do(req)
 	if err != nil {
 		log.DefaultLogger.Errorf("[runtime][rpc]error %s", err.Error())
 		return nil, err
 	}
-
 	resp.Ctx = req.Ctx
+	// 4. afterInvoke callback
 	resp, err = m.cb.AfterInvoke(resp)
 	if err != nil {
 		log.DefaultLogger.Errorf("[runtime][rpc]after filter error %s", err.Error())

components/rpc/types.go

@@ -53,7 +53,7 @@ func (r RPCHeader) Get(key string) string {
 // RPCRequest is request info
 type RPCRequest struct {
 	// context
-	Ctx         context.Context
+	Ctx context.Context
 	// request id
 	Id          string
 	Timeout     int32
@@ -88,9 +88,9 @@ type Callback interface {
 	// AddAfterInvoke is add AfterInvoke func
 	AddAfterInvoke(CallbackFunc)
 
-	// BeforeInvoke is get BeforeInvoke by RPCRequest
+	// BeforeInvoke is used to invoke beforeInvoke callbacks
 	BeforeInvoke(*RPCRequest) (*RPCRequest, error)
-	// AfterInvoke is get AfterInvoke by RPCRequest
+	// AfterInvoke is used to invoke afterInvoke callbacks
 	AfterInvoke(*RPCResponse) (*RPCResponse, error)
 }