package main
import (
"context"
"fmt"
"io"
"strconv"
"sync/atomic"
"github.com/chenjie199234/Corelib/cerror"
"github.com/chenjie199234/Corelib/container/list"
"github.com/chenjie199234/Corelib/monitor"
)
type Msg struct {
H *MsgHeader
B *MsgBody
WithB bool
}
type MsgBody struct {
Body []byte
Error *cerror.Error
}
type MsgHeader struct {
Callid uint64
Path string
Type int
Deadline int64
Metadata map[string]string
Tracedata map[string]string
Traildata map[string]string
}
type rw struct {
callid uint64
path string
metadata map[string]string
traceddata map[string]string
deadline int64
reader *list.BlockList[*MsgBody]
sender func(context.Context, *Msg) error
status atomic.Int32 //use right 4 bit,the bit from left to right:peer_read_status,peer_send_status,self_read_status,self_send_status
e error
}
func newrw(callid uint64, path string, deadline int64, md, td map[string]string, sender func(context.Context, *Msg) error) *rw {
tmp := &rw{
callid: callid,
path: path,
metadata: md,
traceddata: td,
deadline: deadline,
reader: list.NewBlockList[*MsgBody](),
sender: sender,
}
tmp.status.Store(0b1111)
return tmp
}
func (this *rw) init(body *MsgBody) error {
return this.sender(context.Background(), &Msg{
H: &MsgHeader{
Callid: this.callid,
Path: this.path,
Type: 1,
Deadline: this.deadline,
Metadata: this.metadata,
Tracedata: this.traceddata,
},
B: body,
WithB: body != nil,
})
}
func (this *rw) send(body *MsgBody) error {
if this.status.Load()&0b0001 == 0 {
if this.e != nil {
return this.e
}
return cerror.ErrCanceled
}
if this.status.Load()&0b1000 == 0 {
return io.EOF
}
if e := this.sender(context.Background(), &Msg{
H: &MsgHeader{
Callid: this.callid,
Path: this.path,
Type: 2,
},
B: body,
WithB: body != nil,
}); e != nil {
return e
}
if body.Error != nil {
//if we send error to peer,means we stop send
this.status.And(0b1110)
}
return nil
}
func (this *rw) closesend() error {
if old := this.status.And(0b1110); old&0b0001 == 0 {
return nil
}
return this.sender(context.Background(), &Msg{
H: &MsgHeader{
Callid: this.callid,
Path: this.path,
Type: 3,
},
})
}
func (this *rw) closerecv() error {
if old := this.status.And(0b1101); old&0b0010 == 0 {
return nil
}
this.reader.Close()
return this.sender(context.Background(), &Msg{
H: &MsgHeader{
Callid: this.callid,
Path: this.path,
Type: 4,
},
})
}
func (this *rw) closerecvsend(trail bool, err error) error {
if old := this.status.And(0b1100); old&0b0011 == 0 {
return nil
}
this.e = err
this.reader.Close()
m := &Msg{
H: &MsgHeader{
Callid: this.callid,
Path: this.path,
Type: 5,
},
}
if trail {
m.H.Traildata = map[string]string{"Cpu-Usage": strconv.FormatFloat(monitor.LastUsageCPU, 'g', 10, 64)}
}
return this.sender(context.Background(), m)
}
func (this *rw) recv() ([]byte, error) {
if this.status.Load()&0b0010 == 0 {
if this.e != nil {
return nil, this.e
}
return nil, cerror.ErrCanceled
}
m, e := this.reader.Pop(context.Background())
if e != nil {
if e == list.ErrClosed {
if this.e != nil {
return nil, this.e
}
if this.status.Load()&0b0100 == 0 {
return nil, io.EOF
}
if this.status.Load()&0b0010 == 0 {
return nil, cerror.ErrCanceled
}
return nil, cerror.ErrClosed
} else {
//this is impossible
return nil, cerror.Convert(e)
}
}
//fix interface nil problem
if m.Error == nil || m.Error.Code == 0 {
return m.Body, nil
}
//if we read error from peer,means peer stop send
this.status.And(0b1011)
this.reader.Close()
return m.Body, m.Error
}
func (this *rw) cache(m *MsgBody) error {
_, e := this.reader.Push(m)
if e == list.ErrClosed {
e = cerror.ErrClosed
}
return e
}
func main() {
rw := newrw(101, "/test", 0, nil, nil, func(ctx context.Context, m *Msg) error {
return nil
})
rw.closerecvsend(true, nil)
fmt.Println("finish")
select {}
}
module demo
go 1.23.0
require github.com/chenjie199234/Corelib v0.0.133-0.20250221101755-b7a3a9d25afa
Go version
1.24.0
go env
What did you do?
in my project,there has a very strange panic when update from 1.23.6 to 1.24.0
What did you see happen?
same code
works fine in 1.23.6
panic in 1.24.0
What did you expect to see?
works fine