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(sdk): hybrid cloud add sdk debug log

Browse Source 
Signed-off-by: baihailong <baihailong@oppo.com>
This commit is contained in:
baihailong 2024-04-17 00:13:02 -07:00 committed by AmazingChi
parent b701abdf6a
commit d33fc14119
1 changed files with 11 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
sdk/data/stream/extent_handler.go
View File

@ -152,8 +152,8 @@ func NewExtentHandler(stream *Streamer, offset int, storeMode int, size int,
// String returns the string format of the extent handler.
func (eh *ExtentHandler) String() string {
return fmt.Sprintf("ExtentHandler{ID(%v)Inode(%v)FileOffset(%v)Size(%v)StoreMode(%v)Status(%v)Dp(%v)Ver(%v)key(%v)lastKey(%v)flight(%d)dirty(%v)storageClass(%v)}",
eh.id, eh.inode, eh.fileOffset, eh.size, eh.storeMode, eh.status, eh.dp, eh.stream.verSeq, eh.key, eh.lastKey, eh.inflight, eh.dirty, eh.storageClass)
return fmt.Sprintf("ExtentHandler{ID(%v)Inode(%v)FileOffset(%v)Size(%v)StoreMode(%v)Status(%v)Dp(%v)Ver(%v)key(%v)lastKey(%v)flight(%d)dirty(%v)storageClass(%v)inflight(%v)}",
eh.id, eh.inode, eh.fileOffset, eh.size, eh.storeMode, eh.status, eh.dp, eh.stream.verSeq, eh.key, eh.lastKey, eh.inflight, eh.dirty, eh.storageClass, eh.inflight)
}
func (eh *ExtentHandler) write(data []byte, offset, size int, direct bool) (ek *proto.ExtentKey, err error) {
@ -187,6 +187,7 @@ func (eh *ExtentHandler) write(data []byte, offset, size int, direct bool) (ek *
for total < size {
if eh.packet == nil {
eh.packet = NewWritePacket(eh.inode, offset+total, eh.storeMode)
log.LogDebugf("ExtentHandler write packet nil and new packet: eh(%v)", eh)
if direct {
eh.packet.Opcode = proto.OpSyncWrite
}
@ -297,7 +298,9 @@ func (eh *ExtentHandler) receiver() {
}
func (eh *ExtentHandler) processReply(packet *Packet) {
log.LogDebugf("processReply begin: packet(%v), eh(%v)", packet, eh)
defer func() {
log.LogDebugf("processReply end: packet(%v), eh(%v)", packet, eh)
if atomic.AddInt32(&eh.inflight, -1) <= 0 {
eh.empty <- struct{}{}
}
@ -405,11 +408,14 @@ func (eh *ExtentHandler) processReply(packet *Packet) {
}
func (eh *ExtentHandler) processReplyError(packet *Packet, errmsg string) {
log.LogDebugf("processReplyError begin: eh(%v) packet(%v) errmsg(%v)", eh, packet, errmsg)
eh.setClosed()
eh.setRecovery()
if err := eh.recoverPacket(packet); err != nil {
eh.discardPacket(packet)
log.LogErrorf("processReplyError discard packet: eh(%v) packet(%v) err(%v) errmsg(%v)", eh, packet, err, errmsg)
} else {
log.LogDebugf("processReplyError end: eh(%v) packet(%v) errmsg(%v)", eh, packet, errmsg)
}
}
@ -567,6 +573,7 @@ func (eh *ExtentHandler) waitForFlush() (err error) {
}
func (eh *ExtentHandler) recoverPacket(packet *Packet) error {
log.LogDebugf("ExtentHandler recoverPacket: eh(%v), packet(%v)", eh, packet)
packet.errCount++
if packet.errCount >= MaxPacketErrorCount || proto.IsStorageClassBlobStore(eh.storageClass) {
return errors.New(fmt.Sprintf("recoverPacket failed: reach max error limit, eh(%v) packet(%v)", eh, packet))
@ -717,11 +724,13 @@ func (eh *ExtentHandler) createExtent(dp *wrapper.DataPartition) (extID int, err
// Handler lock is held by the caller.
func (eh *ExtentHandler) flushPacket() {
if eh.packet == nil {
log.LogDebugf("ExtentHandler flushPacket nil, return: eh(%v)", eh)
return
}
eh.pushToRequest(eh.packet)
eh.packet = nil
log.LogDebugf("ExtentHandler flushPacket end: eh(%v)", eh)
}
func (eh *ExtentHandler) pushToRequest(packet *Packet) {