fix error result cache

This commit is contained in:
Zheng Kai
2023-06-25 18:49:24 +08:00
parent 05fe88de8b
commit 645873a6d1
4 changed files with 50 additions and 32 deletions

View File

@@ -4,8 +4,9 @@ import "sync"
// Core ... // Core ...
type Core struct { type Core struct {
pool map[[16]byte]*row serial int
mux sync.Mutex pool map[[16]byte]*row
mux sync.Mutex
} }
// NewCore ... // NewCore ...

View File

@@ -11,14 +11,14 @@ import (
"time" "time"
) )
func (pr *row) fetchRemote() (ab []byte, ok bool, err error) { func (pr *row) fetchRemote() (ab []byte, err error) {
r := pr.req r := pr.req
b := pr.log b := pr.log
u, err := url.Parse(config.OpenAIBase + r.Url) u, err := url.Parse(config.OpenAIBase + r.Url)
if err != nil { if err != nil {
return nil, false, err return nil, err
} }
// zj.J(`real url`, u.String()) // zj.J(`real url`, u.String())
@@ -39,9 +39,7 @@ func (pr *row) fetchRemote() (ab []byte, ok bool, err error) {
return return
} }
if rsp.StatusCode >= 200 || rsp.StatusCode < 300 { if rsp.StatusCode < 200 || rsp.StatusCode >= 300 {
ok = true
} else {
err = fmt.Errorf(`status code fail: %d`, rsp.StatusCode) err = fmt.Errorf(`status code fail: %d`, rsp.StatusCode)
b.WriteString(err.Error()) b.WriteString(err.Error())
} }

View File

@@ -19,11 +19,14 @@ func (c *Core) add(req *pb.Req, hr *http.Request) (pr *row, cached bool) {
return return
} }
c.serial++
pr = &row{ pr = &row{
hr: hr, serial: c.serial,
hash: hash, hr: hr,
req: req, hash: hash,
t: time.Now(), req: req,
t: time.Now(),
core: c,
} }
pr.mux.Lock() pr.mux.Lock()
go pr.run() go pr.run()
@@ -31,3 +34,12 @@ func (c *Core) add(req *pb.Req, hr *http.Request) (pr *row, cached bool) {
c.mux.Unlock() c.mux.Unlock()
return return
} }
func (c *Core) delete(r *row) {
c.mux.Lock()
pr, ok := c.pool[r.hash]
if ok && pr == r {
delete(c.pool, r.hash)
}
c.mux.Unlock()
}

View File

@@ -11,15 +11,22 @@ import (
) )
type row struct { type row struct {
hash [16]byte serial int
hr *http.Request hash [16]byte
req *pb.Req hr *http.Request
rsp []byte req *pb.Req
err error rsp []byte
done bool err error
t time.Time done bool
mux sync.RWMutex t time.Time
log *bytes.Buffer mux sync.RWMutex
log *bytes.Buffer
core *Core
}
func (pr *row) suicide() {
pr.core.delete(pr)
pr.core = nil
} }
func (pr *row) run() { func (pr *row) run() {
@@ -33,20 +40,20 @@ func (pr *row) run() {
// return // return
} }
var ok bool pr.rsp, pr.err = pr.fetchRemote()
pr.rsp, ok, pr.err = pr.fetchRemote()
if pr.err == nil && ok {
// pr.failLog.Reset()
go writeFailLog(pr.hash, pr.log.Bytes())
} else {
go writeFailLog(pr.hash, pr.log.Bytes())
}
go pr.saveFile()
// go pr.metrics()
pr.done = true pr.done = true
pr.mux.Unlock() pr.mux.Unlock()
// go pr.metrics()
if pr.err == nil {
// pr.failLog.Reset()
// go writeFailLog(pr.hash, pr.log.Bytes())
go pr.saveFile()
} else {
go writeFailLog(pr.hash, pr.log.Bytes())
go pr.suicide()
}
} }
func (pr *row) wait() { func (pr *row) wait() {