fix(client): fix putAheadReadBlock Blocking issue

close:#1000412030

Signed-off-by: chihe <chihe@oppo.com>
(cherry picked from commit ccb7de44bb)

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) {

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 == "" {

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

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)