gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(client): close conn when aysnc write failed

Browse Source 
with: #1000293406

Signed-off-by: clinx <chenlin1@oppo.com>
(cherry picked from commit 497c4cb4c2)
This commit is contained in:
clinx 2025-08-19 20:05:04 +08:00 committed by chihe
parent ff1f157f20
commit 16dd2c30e0
3 changed files with 74 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
proto/distributed_cache.go
View File

@ -12,7 +12,7 @@ import (
const (
PageSize = 4 * 1024
CACHE_BLOCK_PACKET_SIZE = 8 * PageSize
CACHE_BLOCK_PACKET_SIZE = 32 * PageSize
CACHE_BLOCK_SIZE = 1 << 20
CACHE_OBJECT_BLOCK_SIZE = 4 * 1024 * 1024
ReadCacheTimeout = 1 // second

22
sdk/remotecache/client.go
View File

@ -114,6 +114,10 @@ func NewRemoteCacheClient(config *ClientConfig) (rc *RemoteCacheClient, err erro
if err != nil {
return nil, errors.New("failed to init log")
}
_, err = stat.NewStatistic(config.LogDir, "remoteCacheClient", int64(stat.DefaultStatLogSize), stat.DefaultTimeOutUs, true)
if err != nil {
return nil, errors.New("failed to stat log")
}
}
if proto.Buffers == nil {
@ -175,6 +179,7 @@ func (rc *RemoteCacheClient) Stop() {
close(rc.stopC)
rc.conns.Close()
log.LogFlush()
stat.WriteStat()
rc.wg.Wait()
}
@ -564,8 +569,7 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re
}
bgTime := stat.BeginStat()
defer func() {
forceClose := err != nil && !proto.IsFlashNodeLimitError(err)
rc.conns.PutConnect(conn, forceClose)
rc.conns.PutConnect(conn, err != nil)
parts := strings.Split(addr, ":")
if len(parts) > 0 && addr != "" {
stat.EndStat(fmt.Sprintf("flashPutBlock:%v", parts[0]), err, bgTime, 1)
@ -1137,7 +1141,13 @@ func (rc *RemoteCacheClient) NewRemoteCacheReader(ctx context.Context, conn *net
func (reader *RemoteCacheReader) read(p []byte) (n int, err error) {
bg := stat.BeginStat()
defer stat.EndStat(fmt.Sprintf("readCache:%v", reader.flashIp), err, bg, 1)
defer func() {
err1 := err
if err1 == io.EOF {
err1 = nil
}
stat.EndStat(fmt.Sprintf("readCache:%v", reader.flashIp), err1, bg, 1)
}()
if reader.ctx.Err() != nil {
log.LogErrorf("RemoteCacheReader:reqID(%v) err(%v)", reader.reqID, reader.ctx.Err())
return 0, reader.ctx.Err()
@ -1175,7 +1185,9 @@ func (reader *RemoteCacheReader) read(p []byte) (n int, err error) {
}
extentOffset := reply.ExtentOffset
reader.localData = p[extentOffset : extentOffset+expectLen]
log.LogDebugf("RemoteCacheReader:Read offset(%v) extentOffset(%v) expectLen(%v) p.len(%v)", reply.KernelOffset, reply.ExtentOffset, expectLen, len(p))
if log.EnableDebug() {
log.LogDebugf("RemoteCacheReader:Read reqID(%v) offset(%v) extentOffset(%v) expectLen(%v) p.len(%v) cost(%v)", reader.reqID, reply.KernelOffset, reply.ExtentOffset, expectLen, len(p), time.Since(*bg))
}
atomic.StoreInt64(&reader.alreadyReadLen, reader.alreadyReadLen+expectLen)
return int(expectLen), nil
}
@ -1246,7 +1258,7 @@ func (reader *RemoteCacheReader) Close() error {
log.LogDebugf("RemoteCacheReader Close, reqId(%v)", reader.reqID)
reader.closed = true
bytespool.Free(reader.buffer)
reader.rc.conns.PutConnect(reader.conn, false)
reader.rc.conns.PutConnect(reader.conn, reader.currOffset < reader.endOffset)
}
return nil
}

114
tool/remotecache-benchmark/main.go
View File

@ -25,7 +25,6 @@ import (
"github.com/cubefs/cubefs/tool/remotecache-benchmark/storage"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/bytespool"
"github.com/cubefs/cubefs/util/stat"
"github.com/google/uuid"
)
@ -93,10 +92,6 @@ func NewBenchmarkTester(dataDir, hddBase, nvmeBase string, verify bool, master s
if err != nil {
fmt.Printf("\nCreate a remote cache Tester failed:%v\n", err)
}
_, err = stat.NewStatistic("/tmp/cfs", "remoteCacheClient", int64(stat.DefaultStatLogSize), stat.DefaultTimeOutUs, true)
if err != nil {
fmt.Printf("\nCreate a remote cache stat log failed:%v\n", err)
}
}
return tester
@ -215,6 +210,7 @@ func main() {
clear := flag.Bool("clear-cache", true, "Clear page cache")
removeKey := flag.Bool("remove-key", false, "Remove specific key")
blockKey := flag.String("block-key", "", "Block unique key")
readRepeat := flag.Int("read-repeat", 1, "Read the repeat count of a key press")
flag.Parse()
fmt.Printf("Parsed command-line arguments:\n")
@ -232,6 +228,7 @@ func main() {
fmt.Printf(" -clear-cache: %v\n", *clear)
fmt.Printf(" -remove-key: %v\n", *removeKey)
fmt.Printf(" -block-key: %v\n", *blockKey)
fmt.Printf(" -read-repeat: %v\n", *readRepeat)
tester := NewBenchmarkTester(ensureAbsolutePath(*dataDir), *hddBase, *nvmeBase, *verify, *master, *clear)
if *needGenerate {
if err := tester.GenerateTestData(*totalSizeGB, *genConcurrency); err != nil {
@ -256,7 +253,7 @@ func main() {
if *runGetTest {
// Ensure all data has been preheated into the storage system.
tester.RunGetBenchmark(ctx, *concurrency)
tester.RunGetBenchmark(ctx, *concurrency, *readRepeat)
}
if *removeKey && *blockKey != "" {
@ -276,7 +273,6 @@ func (t *BenchmarkTester) Stop() {
if t.cacheStorage != nil {
t.cacheStorage.Stop()
}
stat.WriteStat()
}
func (t *BenchmarkTester) RunPutBenchmark(ctx context.Context, concurrency int) {
@ -510,23 +506,23 @@ func (fh *FileHandler) Close() error {
return err
}
func (t *BenchmarkTester) RunGetBenchmark(ctx context.Context, concurrency int) {
func (t *BenchmarkTester) RunGetBenchmark(ctx context.Context, concurrency int, repeats int) {
fmt.Printf("\n=== Starting Get performance test (concurrency: %d) ===\n", concurrency)
if t.hddStorage != nil {
hddResult := t.runStorageGetBenchmark(ctx, t.hddStorage, concurrency)
hddResult := t.runStorageGetBenchmark(ctx, t.hddStorage, concurrency, repeats)
t.printBenchmarkResult(hddResult)
}
if t.nvmeStorage != nil {
nvmeResult := t.runStorageGetBenchmark(ctx, t.nvmeStorage, concurrency)
nvmeResult := t.runStorageGetBenchmark(ctx, t.nvmeStorage, concurrency, repeats)
t.printBenchmarkResult(nvmeResult)
}
if t.cacheStorage != nil {
remoteResult := t.runStorageGetBenchmark(ctx, t.cacheStorage, concurrency)
remoteResult := t.runStorageGetBenchmark(ctx, t.cacheStorage, concurrency, repeats)
t.printBenchmarkResult(remoteResult)
}
}
func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage storage.Storage, concurrency int) *BenchmarkResult {
func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage storage.Storage, concurrency int, repeats int) *BenchmarkResult {
fmt.Printf("Testing %s Get performance...\n", storage.Name())
// if t.clearPageCache {
// clearPageCache()
@ -561,58 +557,60 @@ func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage st
continue
}
fh := value.(*FileHandler)
startTime := time.Now()
from := int64(0)
to := fh.Size
reqId := uuid.New().String()
r, len1, _, err := storage.Get(ctx, reqId, fh.FileName, from, to)
var latency time.Duration
var dataBuf, tmpBuf []byte
for i := 0; i < repeats; i++ {
startTime := time.Now()
from := int64(0)
to := fh.Size
reqId := uuid.New().String()
r, len1, _, err := storage.Get(ctx, reqId, fh.FileName, from, to)
var latency time.Duration
var dataBuf, tmpBuf []byte
if err == nil && r != nil {
dataBuf = bytespool.Alloc(proto.CACHE_BLOCK_PACKET_SIZE)
tmpBuf = bytespool.Alloc(int(to - from))
reads := 0
for {
readBytes, readErr := r.Read(dataBuf)
if readErr != nil && readErr != io.EOF {
fmt.Printf("storage %v read data %v failed reqID %v: readErr %v\n", storage.Name(), fh.FileName, reqId, readErr)
err = readErr
break
}
copy(tmpBuf[reads:], dataBuf[:readBytes])
reads += readBytes
if readErr == io.EOF {
break
}
}
if err == nil && reads != int(len1) {
err = fmt.Errorf("wrong len %v:expected[%v]", reads, len1)
} else if err == nil {
latency = time.Since(startTime)
if t.verify {
readCrc := crc32.ChecksumIEEE(tmpBuf[:reads])
actualCrc := crc32.ChecksumIEEE(fh.bytes[from:to])
if actualCrc != readCrc {
err = fmt.Errorf("wrong crc %v:expected[%v]", actualCrc, readCrc)
if err == nil && r != nil {
dataBuf = bytespool.Alloc(proto.CACHE_BLOCK_PACKET_SIZE)
tmpBuf = bytespool.Alloc(int(to - from))
reads := 0
for {
readBytes, readErr := r.Read(dataBuf)
if readErr != nil && readErr != io.EOF {
fmt.Printf("storage %v read data %v failed reqID %v: readErr %v\n", storage.Name(), fh.FileName, reqId, readErr)
err = readErr
break
}
copy(tmpBuf[reads:], dataBuf[:readBytes])
reads += readBytes
if readErr == io.EOF {
break
}
}
if err == nil && reads != int(len1) {
err = fmt.Errorf("wrong len %v:expected[%v]", reads, len1)
} else if err == nil {
latency = time.Since(startTime)
if t.verify {
readCrc := crc32.ChecksumIEEE(tmpBuf[:reads])
actualCrc := crc32.ChecksumIEEE(fh.bytes[from:to])
if actualCrc != readCrc {
err = fmt.Errorf("wrong crc %v:expected[%v]", actualCrc, readCrc)
}
}
}
bytespool.Free(dataBuf)
bytespool.Free(tmpBuf)
}
if err != nil && err != io.EOF {
atomic.AddInt64(&errorCount, 1)
fmt.Printf("storage %v get %v failed reqID %v: %v\n", storage.Name(), fh.FileName, reqId, err)
} else {
atomic.AddInt64(&successCount, 1)
atomic.AddInt64(&totalBytes, len1)
result.Latencies = append(result.Latencies, latency)
}
if r != nil {
r.Close()
}
bytespool.Free(dataBuf)
bytespool.Free(tmpBuf)
}
if err != nil && err != io.EOF {
atomic.AddInt64(&errorCount, 1)
fmt.Printf("storage %v get %v failed reqID %v: %v\n", storage.Name(), fh.FileName, reqId, err)
} else {
atomic.AddInt64(&successCount, 1)
atomic.AddInt64(&totalBytes, len1)
result.Latencies = append(result.Latencies, latency)
}
fh.Close()
if r != nil {
r.Close()
}
}
}(w)
}