Linux Tencent-SNG 5.4.119-1-tlinux4-0010.3 #1 SMP Thu Jan 5 17:31:23 CST 2023 x86_64 x86_64 x86_64 GNU/Linux

func NewThsContext(ctx context.Context, uid int64) *ths.Context {
	nctx := ths.NewContext(ctx)
	nctx.Uin = uint64(uid)
	h := th.NewTmeHeader()
	nauth := h.GetAuthInfo()
	nauth.UID = uint64(uid)
	h.SetAuthInfo(&nauth)
	if h.GetLocalServiceLen() == 0 {
		h.SetLocalServiceName("go_unified_ktv_chorus_game_svr")
	}
	nctx.Head = h
	return nctx
}

func (c *conn) serve(ctx context.Context) {
	ctx = context.WithValue(ctx, ClientAddr, c.remoteAddr)
	ctx, cancelCtx := context.WithCancel(ctx)
	c.cancelCtx = cancelCtx
	defer c.rwc.Close()
c.wg.Add(3)
go c.readRequests(ctx)
go c.handle(ctx)
go c.writeResponses(ctx)
c.wg.Wait()

    
      
      
    
  
}
func (c *conn) handle(ctx context.Context) {

defer func() {

c.cancelCtx()

c.wg.Done()

if err := recover(); err != nil {

c.rwc.SetDeadline(time.Now())

buf := make([]byte, RecoverStackSize)

buf = buf[:runtime.Stack(buf, false)]

log.Output(1, fmt.Sprintf("[PANIC] %v: %v\n%s", c.remoteAddr, err, buf))

}

}()
for {
	select {
	case <-ctx.Done():
		return
	case req := <-c.cin:
		go func() {
			subCtx, cancel := context.WithTimeout(ctx, c.server.msgTimeout)
			rsp, err := c.server.handler.Serve(subCtx, req)
			cancel()
			if err != nil {
				SysLog(err)
				return
			}
			select {
			case <-ctx.Done():
				SysLog("tcp handle business goroutine context done")
				return
			case c.cout <- rsp:
				SysLog("tcp handle business goroutine write rsp to cout channel")
			}
		}()
	}
}

    
      
      
    
  
}
func (sm ServeMutex) Serve(baseCtx context.Context, reqBuf []byte) ([]byte, error) {

conf := cat.GetConfig()

defer func() {

if e := recover(); e != nil {

dataBuf := make([]byte, 41024*1024)

dataBuf = dataBuf[:runtime.Stack(dataBuf, false)]

cat.Logf("[PANIC]%v\n>> %s", e, dataBuf)

}

}()
//parsing the business protocol  head
    head = parseHead(reqBuf)
    //set the context with info from the head
	baseCtx = context.WithValue(baseCtx, RemoteServiceName, head.GetLocalServiceName())
	baseCtx = context.WithValue(baseCtx, LocalServiceName, head.GetServiceName())
	baseCtx = context.WithValue(baseCtx, TraceID, head.GetTraceID())
	baseCtx = context.WithValue(baseCtx, SpanID, head.GetSpanID())
	baseCtx = context.WithValue(baseCtx, ParentSpanID, head.GetParentSpanID())
	baseCtx = context.WithValue(baseCtx, Env, head.GetEnv())

ctx := NewContext(baseCtx)
defer ctx.WriteLog()

// set the protocol head
ctx.Head = head

// found the handler
entry, ok := sm.mapEntries[head.GetCmdPattern()]
if !ok {
	return nil, fmt.Errorf("invalid cmd patrern:%s", head.GetCmdPattern())
}

// set the request and response type
ctx.Req = reflect.New(entry.reqType.Elem()).Interface()
ctx.Rsp = reflect.New(entry.rspType.Elem()).Interface()
    // deocde the request
var err error
    err = BodyUnmarshal(h, reqBuf, ctx.Req)
if err != nil {
	cat.Logf("protocol Decode buf failed, msg:%v", err)
	return nil, err
}

// 
entry.h.Process(ctx)

ctx.ErrCode = ctx.Head.GetResultCode()

// pack header、response
    .....
	return pkgBuf, nil

    
      
      
    
  
}
// NewContext 创建新上下文

func NewContext(ctx context.Context) *Context {

newCtx := Context{

Context:   cat.NewContext(ctx, cat.GetConfig().Log.LogWriter),

startTime: time.Now(),

}

newCtx.LogLevel = int(cat.GetConfig().Log.Level)

newCtx.Head = NewHeader()

return &newCtx

}

func (r *Redo) ExecAsync(obj interface{}, opts ...ExecOptionFunc) {

r.execWithWAL(true, r.ctx, obj, opts...)

}

func (r *Redo) execWithWAL(async bool, ctx context.Context, obj interface{}, opts ...ExecOptionFunc) error {

//.....

f := func() error {

err := r.exec(ctx, key, obj)

if requireCommit {

r.wal.Commit(key)

}

return err

}
if async {
	go f()
	return nil
}
return f()

    
      
      
    
  
}

The default redo context is created in function New.

func New(path string, h Handler, fs ...optionFunc) (*Redo, error) {

	r := Redo{
		h:      h,
		keyGen: newKeyGenerator(),
	}
	...
	r.ctx, r.cancel = context.WithCancel(context.Background())

	if r.opts.Retry.Thread > 1 {
		r.retryCh = make(chan []byte)
		for i := 0; i < int(r.opts.Retry.Thread); i++ {
			r.wg.Go(r.retryLoop)
		}
	}

	r.wg.Go(r.mainLoop)
	return &r, nil
}