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): cleanup eh when flush is completed

Browse Source 
close:#1000310987

Signed-off-by: chihe <chihe@oppo.com>
(cherry picked from commit d1e7e412e0)
This commit is contained in:
chihe 2025-09-09 19:14:50 +08:00
parent 1cf8acde1c
commit 99b040a620
4 changed files with 14 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
client/fs/file.go
View File

@ -583,8 +583,8 @@ func (f *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.Wri
}
}
elapsed := time.Since(start)
log.LogDebugf("TRACE Write: ino(%v) offset(%v) len(%v) flags(%v) fileflags(%v) req(%v) (%v)ns ",
ino, req.Offset, reqlen, req.Flags, req.FileFlags, req, elapsed.Nanoseconds())
log.LogDebugf("TRACE Write: ino(%v) offset(%v) len(%v) flags(%v) fileflags(%v) req(%v) (%v) ",
ino, req.Offset, reqlen, req.Flags, req.FileFlags, req, elapsed.String())
return nil
}

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

@ -288,7 +288,8 @@ func (eh *ExtentHandler) write(data []byte, offset, size int, direct bool) (ek *
func (eh *ExtentHandler) sender() {
var err error
ticker := time.NewTicker(10 * time.Minute)
defer ticker.Stop()
for {
select {
case packet := <-eh.request:
@ -353,6 +354,8 @@ func (eh *ExtentHandler) sender() {
eh.setClosed()
log.LogDebugf("sender: done, eh(%v) size(%v) ek(%v)", eh, eh.size, eh.key)
return
case <-ticker.C:
log.LogErrorf("eh(%v) sender is still working", eh)
}
}
}

57
sdk/data/stream/stream_reader.go
View File

@ -575,43 +575,8 @@ func (s *Streamer) completeAsyncFlush(req *AsyncFlushRequest) {
handler := req.handler
log.LogDebugf("completeAsyncFlush: for streamer(%v) eh(%v)", s.inode, handler)
// Check if this request is the next one to be processed
nextReq := s.getNextPendingAsyncFlush()
if nextReq == nil {
log.LogWarnf("completeAsyncFlush: No pending async flush requests found for streamer(%v) handler(%v)",
s.inode, handler)
req.done <- errors.New("no pending async flush requests")
return
}
// If this is not the next request to be processed, wait
if nextReq.handler.id != handler.id {
log.LogDebugf("Async flush request streamer(%v) eh(%v) id(%v) is not next in sequence (next: %v), waiting...",
s.inode, handler, handler.id, nextReq.handler.id)
// Wait for the correct request to be processed
// This is a simple polling approach - in a production system, you might want to use channels or condition variables
start := time.Now()
for {
time.Sleep(time.Duration(asyncFlushCheckIntervalMs) * time.Millisecond)
nextReq = s.getNextPendingAsyncFlush()
if nextReq == nil {
log.LogErrorf("completeAsyncFlush: No pending async flush requests found while waiting for "+
"streamer(%v) handler(%v)", s.inode, handler)
req.done <- errors.New("no pending async flush requests")
return
}
if nextReq.handler.id == handler.id {
break
}
if time.Since(start) > 5*time.Minute {
start = time.Now()
log.LogErrorf("completeAsyncFlush: streamer(%v) handler(%v) id(%v) wait to long for next(%v) nextid(%v) len(%v)",
s.inode, handler, handler.id, nextReq.handler, nextReq.handler.id, len(s.asyncFlushCh))
}
}
}
// extentHandler may be sync flushed by streamer.flush when inflight is 0
// the flush order cannot be guaranteed
err := handler.flush()
if err != nil {
log.LogWarnf("completeAsyncFlush: completed successfully for handler(%v)", handler)
@ -697,24 +662,6 @@ func (s *Streamer) removePendingAsyncFlush(handlerID uint64) {
log.LogDebugf("removePendingAsyncFlush streamer(%v) handler(%v) trace(%v)", s.inode, handlerID, string(debug.Stack()))
}
// getNextPendingAsyncFlush returns the next pending request that should be processed
func (s *Streamer) getNextPendingAsyncFlush() *AsyncFlushRequest {
var oldestHandlerID uint64 = ^uint64(0) // Max uint64
var oldestReq *AsyncFlushRequest
s.pendingAsyncFlushMap.Range(func(key, value interface{}) bool {
handlerID := key.(uint64)
req := value.(*AsyncFlushRequest)
if handlerID < oldestHandlerID {
oldestHandlerID = handlerID
oldestReq = req
}
return true // continue iteration
})
return oldestReq
}
// getPendingRequestsCount returns the number of pending requests
func (s *Streamer) getPendingRequests() []uint64 {
ids := make([]uint64, 0)

30
sdk/data/stream/stream_writer.go
View File

@ -50,9 +50,8 @@ const (
streamWriterFlushPeriod = 3
streamWriterIdleTimeoutPeriod = 10
// Async flush constants
asyncFlushCheckIntervalMs = 100 // Check every 100ms
asyncFlushQueueSize = 128 // Buffer size for async flush channel
asyncFlushSemaphoreSize = 4 // Capacity of semaphore to limit concurrent processAsyncFlushRequest executions
asyncFlushQueueSize = 128 // Buffer size for async flush channel
asyncFlushSemaphoreSize = 4 // Capacity of semaphore to limit concurrent processAsyncFlushRequest executions
)
// Note: asyncFlushSequencer is now per-streamer to avoid global lock contention
@ -380,6 +379,7 @@ begin:
s.client.LimitManager.WriteAlloc(ctx, size)
requests := s.extents.PrepareWriteRequests(offset, size, data)
// requests contain offset
log.LogDebugf("Streamer write: ino(%v) prepared requests(%v)", s.inode, requests)
isChecked := false
@ -1051,7 +1051,7 @@ func (s *Streamer) flush(wait bool, id string) (err error) {
clearFunc()
}
}
log.LogDebugf("Streamer(%v) wait(%v) id(%v)", s.inode, wait, id)
log.LogDebugf("Streamer(%v) wait(%v) pending(%v) id(%v)", s.inode, wait, pending, id)
if wait {
for len(pending) > 0 {
progressed := false
@ -1137,9 +1137,8 @@ func (s *Streamer) closeOpenHandler(wait bool) (err error) {
cleanFunc := func(h *ExtentHandler) func() {
return func() {
h.setClosed()
if !s.dirty {
h.cleanup()
}
// cleanup eh when flush is completed
h.cleanup()
}
}(handler)
@ -1148,25 +1147,8 @@ func (s *Streamer) closeOpenHandler(wait bool) (err error) {
log.LogDebugf("closeOpenHandler: using async flush for handler(%v) with inflight(%v) id(%v)",
handler, atomic.LoadInt32(&handler.inflight), id)
s.requestAsyncFlush(handler, cleanFunc)
// Don't set handler to nil here - let async flush complete first
//if wait {
// err = <-ch
// s.removePendingAsyncFlush(handler.id)
// if err != nil {
// log.LogErrorf("closeOpenHandler: eh(%v) async flush failed, err %s,id(%v)", handler, err.Error(), id)
// return
// }
//} else {
// // handler may be not put into dirtyList
// s.dirtylist.Put(handler)
//}
s.dirtylist.Put(handler)
} else {
// Check if this handler is protected by a write operation
//if s.isHandlerProtected(handler) && !wait {
// log.LogDebugf("closeOpenHandler: handler(%v) is protected by write operation, skipping (%v)", handler, string(debug.Stack()))
// return nil
//}
log.LogDebugf("closeOpenHandler: try flush (%v)id(%v)", handler, id)
err = handler.flush()
if err != nil {