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

client(sdk): a single put operation that does not load enough data will not trigger processing. #1000151053

Browse Source 
Signed-off-by: clinx <chenlin1@oppo.com>
This commit is contained in:
clinx 2025-07-29 16:01:57 +08:00 committed by chihe
parent 3e175326ac
commit 0f3b1f8c1b
5 changed files with 134 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
proto/distributed_cache.go
View File

@ -96,6 +96,13 @@ type FlashGroupView struct {
FlashGroups []*FlashGroupInfo
}
type CoonHandler struct {
RemoteError error
Completed chan struct{}
WaitAckChanClosed bool
WaitAckChan chan struct{}
}
func (f *FlashGroupInfo) String() string {
return fmt.Sprintf("{ID: %d, Hosts: %v}", f.ID, f.Hosts)
}

17
remotecache/flashnode/cachengine/block.go
View File

@ -887,11 +887,14 @@ func (cb *CacheBlock) MaybeWriteCompleted(reqLen int64) (err error) {
return nil
}
func CalcAllocSizeV2(reqLen int) int {
func CalcAllocSizeV2(reqLen int) (int, error) {
if reqLen > proto.CACHE_OBJECT_BLOCK_SIZE {
return 0, fmt.Errorf("invalid block size: %d", reqLen)
}
if reqLen%proto.CACHE_BLOCK_PACKET_SIZE != 0 {
reqLen = (reqLen/proto.CACHE_BLOCK_PACKET_SIZE + 1) * proto.CACHE_BLOCK_PACKET_SIZE
}
return reqLen
return reqLen, nil
}
func (cb *CacheBlock) VerifyObjectReq(offset, size uint64) error {
@ -908,3 +911,13 @@ func (cb *CacheBlock) VerifyObjectReq(offset, size uint64) error {
return nil
}
// CheckSizeBoundary checks if the given size exceeds the usedSize boundary
func (cb *CacheBlock) CheckSizeBoundary(size int64) error {
usedSize := cb.getUsedSize()
if size > usedSize {
log.LogWarnf("size(%d) exceeds usedSize(%d) for block(%s)", size, usedSize, cb.blockKey)
return fmt.Errorf("size(%d) exceeds usedSize(%d)", size, usedSize)
}
return nil
}

68
remotecache/flashnode/flashnode_op.go
View File

@ -328,12 +328,15 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error)
err = fmt.Errorf("block %v already exsit", uniKey)
return err
}
var allocSize int
if allocSize, err = cachengine.CalcAllocSizeV2(int(req.BlockLen)); err != nil {
return err
}
err1 = nil
if log.EnableDebug() {
log.LogDebug(logPrefix+" create block key:"+uniKey+" logMsg:%s",
p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err))
}
allocSize := cachengine.CalcAllocSizeV2(int(req.BlockLen))
if cb, err2, created := f.cacheEngine.CreateBlockV2(pDir, uniKey, req.TTL, uint32(allocSize), conn.RemoteAddr().String()); err2 != nil || created {
if err2 != nil {
err = fmt.Errorf("create block(%v) error %v", cachengine.GenCacheBlockKeyV2(pDir, uniKey), err2.Error())
@ -358,8 +361,22 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error)
)
writeLen := proto.CACHE_BLOCK_PACKET_SIZE + 4
buf := bytespool.Alloc(writeLen)
defer bytespool.Free(buf)
ch := &proto.CoonHandler{
Completed: make(chan struct{}),
WaitAckChan: make(chan struct{}, 5),
}
defer func() {
bytespool.Free(buf)
if !ch.WaitAckChanClosed {
ch.WaitAckChanClosed = true
close(ch.WaitAckChan)
}
}()
go f.replyPutDataOk(ch, conn, p, logPrefix)
for {
if ch.RemoteError != nil {
break
}
readSize = proto.CACHE_BLOCK_PACKET_SIZE
missTaskDone := make(chan struct{})
err = f.limitWrite.TryRunAsync(context.Background(), writeLen, false, func() {
@ -400,10 +417,7 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error)
if err1 != nil || err != nil {
break
} else {
if err1 = p.WriteToConn(conn); err1 != nil {
log.LogErrorf(logPrefix+" reply to conn %v for write data", err1)
break
}
ch.WaitAckChan <- struct{}{}
}
totalWritten += int64(readSize)
if totalWritten == req.BlockLen {
@ -413,6 +427,14 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error)
break
}
}
if !ch.WaitAckChanClosed {
ch.WaitAckChanClosed = true
close(ch.WaitAckChan)
}
<-ch.Completed
if ch.RemoteError != nil {
err1 = ch.RemoteError
}
}
if err1 != nil {
f.cacheEngine.DeleteCacheBlock(blockKey)
@ -422,6 +444,16 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error)
return err
}
func (f *FlashNode) replyPutDataOk(ch *proto.CoonHandler, conn net.Conn, p *proto.Packet, logPrefix string) {
close(ch.Completed)
for range ch.WaitAckChan {
if ch.RemoteError = p.WriteToConn(conn); ch.RemoteError != nil {
log.LogErrorf(logPrefix+" reply to conn %v for write data", ch.RemoteError)
return
}
}
}
func (f *FlashNode) opCacheObjectGet(conn net.Conn, p *proto.Packet) (err error) {
var uniKey string
bgTime := stat.BeginStat()
@ -740,6 +772,10 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req
firstReply.ResultCode = proto.OpOk
firstReply.Opcode = p.Opcode
firstReply.StartT = p.StartT
end := offset + int64(req.Size_)
if err = block.CheckSizeBoundary(end); err != nil {
return
}
if err = firstReply.WriteToConn(conn); err != nil {
log.LogErrorf("testRead key:[%s] %s", req.Key,
firstReply.LogMessage(firstReply.GetOpMsg(), conn.RemoteAddr().String(), firstReply.StartT, err))
@ -747,26 +783,15 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req
}
// reply data to client
end := offset + int64(req.Size_)
var errInner error
buf := bytespool.Alloc(proto.CACHE_BLOCK_PACKET_SIZE)
defer bytespool.Free(buf)
for {
err = f.limitRead.RunNoWait(proto.CACHE_BLOCK_PACKET_SIZE, false, func() {
reply := proto.NewPacket()
reply.ReqID = p.ReqID
reply.StartT = p.StartT
var bufOnce sync.Once
buf, bufErr := proto.Buffers.Get(proto.CACHE_BLOCK_PACKET_SIZE)
bufRelease := func() {
bufOnce.Do(func() {
if bufErr == nil {
proto.Buffers.Put(reply.Data[:proto.CACHE_BLOCK_PACKET_SIZE])
}
})
}
if bufErr != nil {
buf = make([]byte, proto.CACHE_BLOCK_PACKET_SIZE)
}
reply.Data = buf
alignedOffset := offset / proto.CACHE_BLOCK_PACKET_SIZE * proto.CACHE_BLOCK_PACKET_SIZE
@ -777,7 +802,6 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req
reply.CRC, errInner = block.Read(ctx, reply.Data[:], alignedOffset, proto.CACHE_BLOCK_PACKET_SIZE, f.waitForCacheBlock, true)
if errInner != nil {
bufRelease()
return
}
p.CRC = reply.CRC
@ -790,14 +814,12 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req
bgTime := stat.BeginStat()
if errInner = reply.WriteToConnForOCS(conn); errInner != nil {
bufRelease()
log.LogErrorf("%s key:[%s] %s", action, req.Key,
reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, errInner))
return
}
stat.EndStat("HitCacheRead:ReplyToClient", errInner, bgTime, 1)
offset = alignedOffset + proto.CACHE_BLOCK_PACKET_SIZE
bufRelease()
if log.EnableInfo() {
log.LogInfof("%s ReqID[%d] key:[%s] reply[%s] block[%s]", action, p.ReqID, req.Key,
reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, errInner), block.String())

81
sdk/remotecache/client.go
View File

@ -38,7 +38,6 @@ const (
const (
ConnIdelTimeout = 30 //
PingCount = 3
SameZoneWeight = 70
RefreshFlashNodesInterval = time.Minute
RefreshHostLatencyInterval = 20 * time.Second
@ -553,17 +552,50 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re
)
writeLen := proto.CACHE_BLOCK_PACKET_SIZE + proto.CACHE_BLOCK_CRC_SIZE
buf := bytespool.Alloc(writeLen)
defer bytespool.Free(buf)
ch := &proto.CoonHandler{
Completed: make(chan struct{}),
WaitAckChan: make(chan struct{}, 5),
}
defer func() {
bytespool.Free(buf)
if !ch.WaitAckChanClosed {
ch.WaitAckChanClosed = true
close(ch.WaitAckChan)
}
}()
go rc.processPutReply(conn, ch, reqId)
for {
if ctx.Err() != nil {
return ctx.Err()
}
if ch.RemoteError != nil {
return ch.RemoteError
}
readSize := proto.CACHE_BLOCK_PACKET_SIZE
if totalWritten+int64(readSize) > length {
readSize = int(length - totalWritten)
}
n, err = r.Read(buf[:readSize])
if n != readSize {
bufferOffset := 0
for {
if ctx.Err() != nil {
return ctx.Err()
}
if ch.RemoteError != nil {
return ch.RemoteError
}
currentReadSize := readSize - bufferOffset
n, err = r.Read(buf[bufferOffset : bufferOffset+currentReadSize])
bufferOffset += n
if err == io.EOF {
break
} else if err != nil {
return err
} else if bufferOffset == readSize {
log.LogDebugf(" total written %v write size %v to fg", totalWritten, readSize)
break
}
}
if bufferOffset != readSize {
log.LogWarnf("FlashGroup put: expected to read %d bytes, but only read %d", readSize, n)
return fmt.Errorf("expected to read %d bytes, but only read %d", readSize, n)
}
@ -577,17 +609,7 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re
log.LogErrorf("wirte data and crc to flashnode get err %v", err)
return err
}
if err = replyPacket.ReadFromConnExt(conn, proto.NoReadDeadlineTime); err != nil {
log.LogWarnf("FlashGroup put data: reqId(%v) failed to ReadFromConn, replyPacket(%v), fg host(%v) , err(%v)", reqId, replyPacket, addr, err)
return
}
if replyPacket.ResultCode != proto.OpOk {
err = fmt.Errorf("%v", string(replyPacket.Data))
if !proto.IsFlashNodeLimitError(err) {
log.LogWarnf("getPutBlockReply: put data ResultCode NOK, req(%v) reply(%v) ResultCode(%v)", reqPacket, replyPacket, replyPacket.ResultCode)
}
return
}
ch.WaitAckChan <- struct{}{}
totalWritten += int64(n)
}
if err == io.EOF {
@ -600,12 +622,41 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re
break
}
}
if !ch.WaitAckChanClosed {
ch.WaitAckChanClosed = true
close(ch.WaitAckChan)
}
<-ch.Completed
if ch.RemoteError != nil {
return ch.RemoteError
}
if totalWritten != length {
return fmt.Errorf("unexpected data length: expected %d bytes, got %d", length, totalWritten)
}
return nil
}
func (rc *RemoteCacheClient) processPutReply(conn *net.TCPConn, ch *proto.CoonHandler, reqId string) {
defer close(ch.Completed)
var err error
replyPacket := NewFlashCacheReply()
for range ch.WaitAckChan {
if err = replyPacket.ReadFromConnExt(conn, int(rc.WriteTimeout)); err != nil {
log.LogWarnf("FlashGroup put data: reqId(%v) failed to ReadFromConn, replyPacket(%v) , err(%v)", reqId, replyPacket, err)
ch.RemoteError = err
return
}
if replyPacket.ResultCode != proto.OpOk {
err = fmt.Errorf("%v", string(replyPacket.Data))
if !proto.IsFlashNodeLimitError(err) {
log.LogWarnf("getPutBlockReply: put data ResultCode NOK, reqId(%v) reply(%v) ResultCode(%v)", reqId, replyPacket, replyPacket.ResultCode)
}
ch.RemoteError = err
return
}
}
}
func (rc *RemoteCacheClient) deleteRemoteBlock(key string, addr string) {
var (
err error

2
tool/remotecache-benchmark/storage/localfs.go
View File

@ -36,7 +36,7 @@ func (fs *LocalFS) Name() string { return fs.fsType }
func (fs *LocalFS) Put(ctx context.Context, reqId, key string, r io.Reader, length int64) (err error) {
pDir := cachengine.MapKeyToDirectory(key)
filePath := filepath.Join(fs.cachePath, path.Join(pDir, key))
allocSize := cachengine.CalcAllocSizeV2(int(length))
allocSize, _ := cachengine.CalcAllocSizeV2(int(length))
if _, err = os.Stat(path.Dir(filePath)); err != nil {
if !os.IsNotExist(err.(*os.PathError)) {