From b47bc52042331118e25e3cecb3f743b4588a6dbd Mon Sep 17 00:00:00 2001 From: chihe Date: Tue, 9 Dec 2025 10:46:46 +0800 Subject: [PATCH] fix(client): fix putAheadReadBlock Blocking issue close:#1000412030 Signed-off-by: chihe (cherry picked from commit ccb7de44bbc49b5e01a3d8b8ceba85fe9866dc19) --- client/fuse.go | 20 +++++++++++++++++++- master/server.go | 9 +++++++++ sdk/data/stream/stream_aheadread.go | 25 +++++++++++++++++-------- util/stat/statistic.go | 5 +++++ 4 files changed, 50 insertions(+), 9 deletions(-) diff --git a/client/fuse.go b/client/fuse.go index 9e76f8b28..da247157b 100644 --- a/client/fuse.go +++ b/client/fuse.go @@ -22,6 +22,7 @@ package main import ( "bufio" + "encoding/json" "flag" "fmt" "io" @@ -37,6 +38,8 @@ import ( "path/filepath" "runtime" "runtime/debug" + + //"runtime/debug" runtimepprof "runtime/pprof" "strings" "syscall" @@ -92,7 +95,8 @@ const ( ControlCommandSetRate = "/rate/set" ControlCommandGetRate = "/rate/get" - ControlCommandFreeOSMemory = "/debug/freeosmemory" + ControlCommandFreeOSMemory = "/freeosmemory" + ControlCommandMemStats = "/memstats" ControlCommandSuspend = "/suspend" ControlCommandResume = "/resume" ControlCommandStopWarmMeta = "/stopWarmMeta" @@ -767,6 +771,7 @@ func mount(opt *proto.MountOptions) (fsConn *fuse.Conn, super *cfs.Super, err er http.HandleFunc(ControlCommandGetRate, super.GetRate) http.HandleFunc(log.SetLogLevelPath, log.SetLogLevel) http.HandleFunc(ControlCommandFreeOSMemory, freeOSMemory) + http.HandleFunc(ControlCommandMemStats, memStats) http.HandleFunc(log.GetLogPath, log.GetLog) http.HandleFunc(ControlCommandSuspend, super.SetSuspend) http.HandleFunc(ControlCommandResume, super.SetResume) @@ -1136,6 +1141,19 @@ func changeRlimit(val uint64) { func freeOSMemory(w http.ResponseWriter, r *http.Request) { debug.FreeOSMemory() + w.WriteHeader(http.StatusOK) + w.Write([]byte("freeOSMemory is called")) +} + +func memStats(w http.ResponseWriter, r *http.Request) { + var m runtime.MemStats + runtime.ReadMemStats(&m) + + w.Header().Set("Content-Type", "application/json; charset=utf-8") + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + enc.SetEscapeHTML(false) + _ = enc.Encode(m) } func loadConfFromMaster(opt *proto.MountOptions) (err error) { diff --git a/master/server.go b/master/server.go index 4ea8e70b1..f6e0141d2 100644 --- a/master/server.go +++ b/master/server.go @@ -20,6 +20,8 @@ import ( syslog "log" "net/http" "net/http/httputil" + "os" + "path/filepath" "regexp" "strconv" "sync" @@ -235,6 +237,13 @@ func (m *Server) checkConfig(cfg *config.Config) (err error) { m.bindIp = cfg.GetBool(proto.BindIpKey) m.port = cfg.GetString(proto.ListenPort) m.logDir = cfg.GetString(LogDir) + // make logDir absolute to avoid dependency on process working directory + if m.logDir != "" && !filepath.IsAbs(m.logDir) { + if exe, e := os.Executable(); e == nil { + base := filepath.Dir(exe) + m.logDir = filepath.Join(base, m.logDir) + } + } m.walDir = cfg.GetString(WalDir) m.bStoreAddr = cfg.GetString(BStoreAddrKey) if m.bStoreAddr == "" { diff --git a/sdk/data/stream/stream_aheadread.go b/sdk/data/stream/stream_aheadread.go index a7beb186a..470d7ed85 100644 --- a/sdk/data/stream/stream_aheadread.go +++ b/sdk/data/stream/stream_aheadread.go @@ -147,9 +147,13 @@ func (arc *AheadReadCache) getAheadReadBlock() (block *AheadReadBlock, err error } func (arc *AheadReadCache) putAheadReadBlock(key string, block *AheadReadBlock) { - arc.availableBlockC <- struct{}{} + select { + case arc.availableBlockC <- struct{}{}: + atomic.AddInt64(&arc.availableBlockCnt, 1) + default: + // channel is full: duplicate return detected, drop token to avoid blocking + } putAheadReadBlock(block) - atomic.AddInt64(&arc.availableBlockCnt, 1) } func (arc *AheadReadCache) checkBlockTimeOut() { @@ -185,11 +189,13 @@ func (arc *AheadReadCache) doCheckBlockTimeOut() { if atomic.LoadUint32(&bv.readed) == AheadReadedInit { log.LogInfof("doCheckBlockTimeOut delete unreaded block: key(%v) addr(%p)", key, bv) } - arc.blockCache.Delete(key.(string)) - // block is ready to recycle + // block is ready to recycle; ensure return-once semantics + keyStr := key.(string) bv.key = "" bv.lock.Unlock() - arc.putAheadReadBlock(key.(string), bv) + if actual, loaded := arc.blockCache.LoadAndDelete(keyStr); loaded { + arc.putAheadReadBlock(keyStr, actual.(*AheadReadBlock)) + } } else { bv.lock.Unlock() } @@ -747,12 +753,15 @@ func (arw *AheadReadWindow) evictAllBlocks() { bv.lock.Lock() bv.key = "" if atomic.LoadUint32(&bv.state) == AheadReadBlockStateInit { - arw.cache.blockCache.Delete(key) - arw.cache.putAheadReadBlock(key.(string), bv) + // ensure only one goroutine returns the token once + bv.lock.Unlock() + if actual, loaded := arw.cache.blockCache.LoadAndDelete(key.(string)); loaded { + arw.cache.putAheadReadBlock(key.(string), actual.(*AheadReadBlock)) + } } else { atomic.StoreUint32(&bv.state, AheadReadBlockStateClear) + bv.lock.Unlock() } - bv.lock.Unlock() return true }) // wait for all blockCache is cleared diff --git a/util/stat/statistic.go b/util/stat/statistic.go index a70f96fcf..399810e80 100644 --- a/util/stat/statistic.go +++ b/util/stat/statistic.go @@ -21,6 +21,7 @@ import ( "io/ioutil" "os" "path" + "path/filepath" "regexp" "runtime/debug" "sort" @@ -98,6 +99,10 @@ var gSt *Statistic = nil func NewStatistic(dir, logModule string, logMaxSize int64, timeOutUs [MaxTimeoutLevel]uint32, useMutex bool) (*Statistic, error) { dir = path.Join(dir, logModule) + // normalize to absolute path to avoid cwd-related issues + if absDir, err := filepath.Abs(dir); err == nil { + dir = absDir + } fi, err := os.Stat(dir) if err != nil { os.MkdirAll(dir, 0o755)