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: right aligned for reading block

Browse Source 
with: #1000303671

Signed-off-by: clinx <chenlin1@oppo.com>
This commit is contained in:
clinx 2025-08-29 10:55:30 +08:00 committed by Victor1319
parent 5dca2183da
commit e5b4de9fdb
3 changed files with 16 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -617,6 +617,12 @@ func (cb *CacheBlock) updateAllocSize(size int64) {
cb.allocSize = size
}
func (cb *CacheBlock) LoadDataSize() (size int64) {
cb.sizeLock.Lock()
defer cb.sizeLock.Unlock()
return cb.usedSize
}
func (cb *CacheBlock) GetRootPath() string {
return cb.rootPath
}

5
remotecache/flashnode/flashnode_op.go
View File

@ -778,9 +778,10 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req
}()
// first reply to client
dataSize := block.LoadDataSize()
firstReply := proto.NewPacket()
firstReply.ReqID = p.ReqID
firstReply.Size = uint32(req.Size_)
firstReply.Size = uint32(dataSize)
firstReply.ResultCode = proto.OpOk
firstReply.Opcode = p.Opcode
firstReply.StartT = p.StartT
@ -833,7 +834,7 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req
var keepAlive bool
for {
alignedOffset = offset / proto.CACHE_BLOCK_PACKET_SIZE * proto.CACHE_BLOCK_PACKET_SIZE
readDiskSize = uint32(util.Min(proto.CACHE_BLOCK_PACKET_SIZE, int(end-alignedOffset)))
readDiskSize = uint32(util.Min(proto.CACHE_BLOCK_PACKET_SIZE, int(dataSize-alignedOffset)))
if !keepAlive {
err = f.limitRead.RunNoWait(int(readDiskSize), false, readAndReply)
keepAlive = true

11
sdk/remotecache/client.go
View File

@ -1124,6 +1124,7 @@ type RemoteCacheReader struct {
flashIp string
currOffset uint32
endOffset uint32
blockDataSize uint32
ctx context.Context
buffer []byte
localData []byte
@ -1182,7 +1183,7 @@ func (reader *RemoteCacheReader) read(p []byte) (n int, err error) {
return
}
alignedOffset := reader.currOffset / proto.CACHE_BLOCK_PACKET_SIZE * proto.CACHE_BLOCK_PACKET_SIZE
readPackageSize := uint32(util.Min(proto.CACHE_BLOCK_PACKET_SIZE, int(reader.endOffset-alignedOffset)))
readPackageSize := uint32(util.Min(proto.CACHE_BLOCK_PACKET_SIZE, int(reader.blockDataSize-alignedOffset)))
_, err = io.ReadFull(reader.conn, p[:readPackageSize])
if err != nil {
log.LogErrorf("RemoteCacheReader:reqID(%v) Read err(%v)", reader.reqID, err)
@ -1280,7 +1281,7 @@ func (reader *RemoteCacheReader) Close() error {
return nil
}
func (rc *RemoteCacheClient) ReadObjectFirstReply(conn *net.TCPConn, p *proto.Packet, reqId string) (length int64, err error) {
func (rc *RemoteCacheClient) ReadObjectFirstReply(conn *net.TCPConn, p *proto.Packet, reqId string) (length uint32, err error) {
header, err := proto.Buffers.Get(util.PacketHeaderSize)
if err != nil {
header = make([]byte, util.PacketHeaderSize)
@ -1321,7 +1322,7 @@ func (rc *RemoteCacheClient) ReadObjectFirstReply(conn *net.TCPConn, p *proto.Pa
}
return
}
return int64(p.Size), nil
return p.Size, nil
}
func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, reqId string, req *proto.CacheReadRequestBase) (reader *RemoteCacheReader, length int64, err error) {
@ -1386,7 +1387,8 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req
reply := new(proto.Packet)
// set time out for first packet
conn.SetReadDeadline(time.Now().Add(time.Millisecond * time.Duration(rc.firstPacketTimeout)))
length, err = rc.ReadObjectFirstReply(conn, reply, reqId)
var blockDataSize uint32
blockDataSize, err = rc.ReadObjectFirstReply(conn, reply, reqId)
if err != nil {
log.LogWarnf("%v ReadObject getReadObjectReply from(%v) failed, reply(%v) error(%v)",
logPrefix, conn.RemoteAddr(), reply, err)
@ -1397,6 +1399,7 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req
rcvTime = time.Now().UnixNano()
reader = rc.NewRemoteCacheReader(ctx, conn, reqId, flashIp, uint32(req.Offset), uint32(req.Offset+req.Size_))
reader.needReadLen = int64(req.Size_)
reader.blockDataSize = blockDataSize
break
}
if log.EnableDebug() {