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

feat(data): limit memory used

Browse Source 
Signed-off-by: NaturalSelect <huangzhibin1@oppo.com>
This commit is contained in:
NaturalSelect 2024-06-17 17:12:33 +08:00 committed by chihe
parent c523bb4e0c
commit 17a523f7af
9 changed files with 473 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
cmd/cmd.go
View File

@ -25,9 +25,11 @@ import (
"path"
"path/filepath"
"runtime"
"runtime/debug"
"strconv"
"strings"
"syscall"
"time"
"github.com/cubefs/cubefs/authnode"
"github.com/cubefs/cubefs/cmd/common"
@ -126,6 +128,16 @@ func modifyOpenFiles() (err error) {
return
}
func releaseMemory(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
w.Header().Set("Content-Type", "text/plain")
begin := time.Now()
debug.FreeOSMemory()
interval := time.Since(begin)
msg := fmt.Sprintf("Success to release memory using %v", interval)
w.Write([]byte(msg))
}
func main() {
flag.Parse()
@ -320,6 +332,7 @@ func main() {
mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
mux.Handle("/debug/", http.HandlerFunc(pprof.Index))
mux.Handle("/releaseMemory", http.HandlerFunc(releaseMemory))
mainHandler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
if strings.HasPrefix(req.URL.Path, "/debug/") {
mux.ServeHTTP(w, req)

51
datanode/server.go
View File

@ -44,6 +44,7 @@ import (
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/loadutil"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/strutil"
"github.com/xtaci/smux"
)
@ -127,12 +128,19 @@ const (
ConfigServiceIDKey = "serviceIDKey"
ConfigEnableGcTimer = "enableGcTimer"
// disk status becomes unavailable if disk error partition count reaches this value
ConfigKeyDiskUnavailablePartitionErrorCount = "diskUnavailablePartitionErrorCount"
)
const cpuSampleDuration = 1 * time.Second
const (
gcTimerDuration = 10 * time.Second
gcRecyclePercent = 0.90
)
// DataNode defines the structure of a data node.
type DataNode struct {
space *SpaceManager
@ -186,6 +194,9 @@ type DataNode struct {
cpuUtil atomicutil.Float64
cpuSamplerDone chan struct{}
enableGcTimer bool
gcTimer *util.RecycleTimer
diskUnavailablePartitionErrorCount uint64 // disk status becomes unavailable when disk error partition count reaches this value
started int32
}
@ -285,6 +296,8 @@ func doStart(server common.Server, cfg *config.Config) (err error) {
// start cpu sampler
s.startCpuSample()
s.startGcTimer()
s.setStart()
return
@ -321,6 +334,9 @@ func doShutdown(server common.Server) {
MasterClient.Stop()
// stop cpu sample
close(s.cpuSamplerDone)
if s.gcTimer != nil {
s.gcTimer.Stop()
}
}
func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
@ -376,6 +392,8 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
s.serviceIDKey = cfg.GetString(ConfigServiceIDKey)
s.enableGcTimer = cfg.GetBoolWithDefault(ConfigEnableGcTimer, false)
diskUnavailablePartitionErrorCount := cfg.GetInt64(ConfigKeyDiskUnavailablePartitionErrorCount)
if diskUnavailablePartitionErrorCount <= 0 || diskUnavailablePartitionErrorCount > 100 {
diskUnavailablePartitionErrorCount = DefaultDiskUnavailablePartitionErrorCount
@ -1018,6 +1036,39 @@ func (s *DataNode) startCpuSample() {
}()
}
func (s *DataNode) startGcTimer() {
if !s.enableGcTimer {
return
}
defer func() {
if r := recover(); r != nil {
log.LogErrorf("[startGcTimer] panic(%v)", r)
}
}()
enable, err := loadutil.IsEnableSwapMemory()
if err != nil {
log.LogErrorf("[startGcTimer] failed to get swap memory info, err(%v)", err)
return
}
if enable {
log.LogWarnf("[startGcTimer] swap memory is enable")
return
}
s.gcTimer, err = util.NewRecycleTimer(gcTimerDuration, gcRecyclePercent, 1*util.GB)
if err != nil {
log.LogErrorf("[startGcTimer] failed to start gc timer, err(%v)", err)
return
}
s.gcTimer.SetPanicHook(func(r interface{}) {
log.LogErrorf("[startGcTimer] gc timer panic, err(%v)", r)
})
s.gcTimer.SetStatHook(func(totalPercent float64, currentProcess, currentGoHeap uint64) {
log.LogWarnf("[startGcTimer] host use too many memory, percent(%v), current process(%v), current process go heap(%v)", strutil.FormatPercent(totalPercent), strutil.FormatSize(currentProcess), strutil.FormatSize(currentGoHeap))
})
}
func (s *DataNode) scheduleToCheckLackPartitions() {
go func() {
for {

1
metanode/const.go
View File

@ -227,6 +227,7 @@ const (
cfgRetainLogs = "retainLogs" // string, raft RetainLogs
cfgRaftSyncSnapFormatVersion = "raftSyncSnapFormatVersion" // int, format version of snapshot that raft leader sent to follower
cfgServiceIDKey = "serviceIDKey"
cfgEnableGcTimer = "enableGcTimer" // bool
metaNodeDeleteBatchCountKey = "batchCount"
configNameResolveInterval = "nameResolveInterval" // int

56
metanode/manager.go
View File

@ -36,6 +36,7 @@ import (
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/loadutil"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/strutil"
)
const (
@ -47,6 +48,11 @@ const sampleDuration = 1 * time.Second
const UpdateVolTicket = 2 * time.Minute
const (
gcTimerDuration = 10 * time.Second
gcRecyclePercent = 0.90
)
// MetadataManager manages all the meta partitions.
type MetadataManager interface {
Start() error
@ -61,10 +67,11 @@ type MetadataManager interface {
// MetadataManagerConfig defines the configures in the metadata manager.
type MetadataManagerConfig struct {
NodeID uint64
RootDir string
ZoneName string
RaftStore raftstore.RaftStore
NodeID uint64
RootDir string
ZoneName string
EnableGcTimer bool
RaftStore raftstore.RaftStore
}
type verOp2Phase struct {
@ -94,6 +101,8 @@ type metadataManager struct {
stopC chan struct{}
volUpdating *sync.Map // map[string]*verOp2Phase
verUpdateChan chan string
enableGcTimer bool
gcTimer *util.RecycleTimer
}
func (m *metadataManager) GetAllVolumes() (volumes *util.Set) {
@ -422,6 +431,39 @@ func (m *metadataManager) startCpuSample() {
}()
}
func (m *metadataManager) startGcTimer() {
if !m.enableGcTimer {
log.LogDebugf("[startGcTimer] gc timer disable")
return
}
defer func() {
if r := recover(); r != nil {
log.LogErrorf("[startGcTimer] panic(%v)", r)
}
}()
enable, err := loadutil.IsEnableSwapMemory()
if err != nil {
log.LogErrorf("[startGcTimer] failed to get swap memory info, err(%v)", err)
return
}
if enable {
log.LogWarnf("[startGcTimer] swap memory is enable")
return
}
if m.gcTimer, err = util.NewRecycleTimer(gcTimerDuration, gcRecyclePercent, 1*util.GB); err != nil {
log.LogErrorf("[startGcTimer] failed to start gc timer, err(%v)", err)
return
}
m.gcTimer.SetPanicHook(func(r interface{}) {
log.LogErrorf("[startGcTimer] gc timer panic, err(%v)", r)
})
m.gcTimer.SetStatHook(func(totalPercent float64, currentProcess, currentGoHeap uint64) {
log.LogWarnf("[startGcTimer] host use too many memory, percent(%v), current process(%v), current process go heap(%v)", strutil.FormatPercent(totalPercent), strutil.FormatSize(currentProcess), strutil.FormatSize(currentGoHeap))
})
}
func (m *metadataManager) startSnapshotVersionPromote() {
m.verUpdateChan = make(chan string, 1000)
go func() {
@ -448,6 +490,7 @@ func (m *metadataManager) onStart() (err error) {
m.startCpuSample()
m.startSnapshotVersionPromote()
m.startUpdateVolumes()
m.startGcTimer()
return
}
@ -458,6 +501,10 @@ func (m *metadataManager) onStop() {
partition.Stop()
}
}
if m.gcTimer != nil {
m.gcTimer.Stop()
}
return
}
@ -776,6 +823,7 @@ func NewMetadataManager(conf MetadataManagerConfig, metaNode *MetaNode) Metadata
metaNode: metaNode,
maxQuotaGoroutineNum: defaultMaxQuotaGoroutine,
volUpdating: new(sync.Map),
enableGcTimer: conf.EnableGcTimer,
}
}

13
metanode/metanode.go
View File

@ -141,7 +141,7 @@ func doStart(s common.Server, cfg *config.Config) (err error) {
if err = m.startRaftServer(cfg); err != nil {
return
}
if err = m.newMetaManager(); err != nil {
if err = m.newMetaManager(cfg); err != nil {
return
}
if err = m.startServer(); err != nil {
@ -392,7 +392,7 @@ func (m *MetaNode) validConfig() (err error) {
return
}
func (m *MetaNode) newMetaManager() (err error) {
func (m *MetaNode) newMetaManager(cfg *config.Config) (err error) {
if _, err = os.Stat(m.metadataDir); err != nil {
if err = os.MkdirAll(m.metadataDir, 0o755); err != nil {
return
@ -419,10 +419,11 @@ func (m *MetaNode) newMetaManager() (err error) {
// load metadataManager
conf := MetadataManagerConfig{
NodeID: m.nodeId,
RootDir: m.metadataDir,
RaftStore: m.raftStore,
ZoneName: m.zoneName,
NodeID: m.nodeId,
RootDir: m.metadataDir,
RaftStore: m.raftStore,
ZoneName: m.zoneName,
EnableGcTimer: cfg.GetBoolWithDefault(cfgEnableGcTimer, false),
}
m.metadataManager = NewMetadataManager(conf, m)
return

96
util/loadutil/mem.go
View File

@ -12,9 +12,97 @@
package loadutil
import "github.com/shirou/gopsutil/mem"
import (
"bufio"
"fmt"
"os"
"runtime"
"strconv"
"strings"
func GetMemUsedPercent() float64 {
memInfo, _ := mem.VirtualMemory()
return memInfo.UsedPercent
"github.com/shirou/gopsutil/mem"
)
const PRO_MEM = "/proc/%d/status"
func GetMemoryUsedPercent() (used float64, err error) {
memInfo, err := mem.VirtualMemory()
if err != nil {
return
}
used = memInfo.UsedPercent
return
}
func GetTotalMemory() (total uint64, err error) {
memInfo, err := mem.VirtualMemory()
if err != nil {
return
}
total = memInfo.Total
return
}
func GetUsedMemory() (used uint64, err error) {
memInfo, err := mem.VirtualMemory()
if err != nil {
return
}
used = memInfo.Used
return
}
func GetTotalSwapMemory() (total uint64, err error) {
memInfo, err := mem.SwapMemory()
if err != nil {
return
}
total = memInfo.Total
return
}
func IsEnableSwapMemory() (enable bool, err error) {
total, err := GetTotalSwapMemory()
if err != nil {
return
}
enable = total != 0
return
}
var procMemFile = fmt.Sprintf(PRO_MEM, os.Getpid())
func GetCurrentProcessMemory() (used uint64, err error) {
fp, err := os.Open(procMemFile)
if err != nil {
return
}
defer fp.Close()
scan := bufio.NewScanner(fp)
for scan.Scan() {
line := scan.Text()
if !strings.HasPrefix(line, "VmRSS:") {
continue
}
value := strings.TrimPrefix(line, "VmRSS:")
value = strings.TrimSpace(value)
value = strings.TrimSuffix(value, " kB")
used, err = strconv.ParseUint(value, 10, 64)
if err != nil {
return
}
used = used * 1024
break
}
return
}
func GetGoMemStats() (heapInfo runtime.MemStats) {
runtime.ReadMemStats(&heapInfo)
return
}
func GetGoInUsedHeap() (size uint64) {
size = GetGoMemStats().HeapInuse
return
}

60
util/loadutil/mem_test.go
View File

@ -15,12 +15,70 @@
package loadutil_test
import (
"runtime"
"runtime/debug"
"testing"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/loadutil"
"github.com/cubefs/cubefs/util/strutil"
"github.com/stretchr/testify/require"
)
func TestMemoryUsed(t *testing.T) {
used := loadutil.GetMemUsedPercent()
used, err := loadutil.GetMemoryUsedPercent()
require.NoError(t, err)
t.Logf("Mem Used:%v\n", used)
}
func TestGetTotalMem(t *testing.T) {
total, err := loadutil.GetTotalMemory()
require.NoError(t, err)
t.Logf("Mem Total:%v\n", total)
}
func TestGetCurrMemory(t *testing.T) {
curr, err := loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
require.Greater(t, curr, uint64(0))
}
func TestAlloc(t *testing.T) {
allocSize := uint64(4 * util.GB)
total, err := loadutil.GetTotalMemory()
t.Logf("Total Memory %v", strutil.FormatSize(total))
require.NoError(t, err)
for allocSize > total {
allocSize /= 2
}
buff := make([]byte, allocSize)
for i := 0; i < int(allocSize); i++ {
buff[i] = 0
}
curr, err := loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
require.Greater(t, curr, allocSize)
t.Logf("Memory %v Inused %v", strutil.FormatSize(curr), strutil.FormatSize(loadutil.GetGoInUsedHeap()))
buff = nil
curr, err = loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
require.Greater(t, curr, allocSize)
runtime.GC()
t.Logf("Memory %v Inused %v", strutil.FormatSize(curr), strutil.FormatSize(loadutil.GetGoInUsedHeap()))
old := curr
debug.FreeOSMemory()
curr, err = loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
require.Less(t, curr, old)
t.Logf("Memory %v", strutil.FormatSize(curr))
}
func TestGetTotalSwapedMemory(t *testing.T) {
total, err := loadutil.GetTotalSwapMemory()
require.NoError(t, err)
t.Logf("Swap Memory %v", strutil.FormatSize(total))
}

111
util/recycle_timer.go Normal file
View File

@ -0,0 +1,111 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package util
import (
"runtime/debug"
"time"
"github.com/cubefs/cubefs/util/loadutil"
)
type RecycleTimer struct {
interval time.Duration
recyclePercent float64
minRecycleLimit uint64
panicHook func(r interface{})
statHook func(totalPercent float64, currentProcess uint64, currentGoHeap uint64)
stopC chan interface{}
}
func (r *RecycleTimer) emitPanicHook(p interface{}) {
if handle := r.panicHook; handle != nil {
handle(p)
}
}
func (r *RecycleTimer) emitStatHook(totalPercent float64, currentProcess uint64, currentGoHeap uint64) {
if handle := r.statHook; handle != nil {
handle(totalPercent, currentProcess, currentGoHeap)
}
}
func (r *RecycleTimer) start() (err error) {
total, err := loadutil.GetTotalMemory()
if err != nil {
return
}
go func() {
defer func() {
if p := recover(); p != nil {
r.emitPanicHook(p)
}
}()
timer := time.NewTicker(r.interval)
defer timer.Stop()
for {
select {
case <-r.stopC:
return
case <-timer.C:
used, err := loadutil.GetUsedMemory()
if err != nil {
return
}
percent := float64(used) / float64(total)
if percent >= r.recyclePercent {
used, err = loadutil.GetCurrentProcessMemory()
if err != nil {
return
}
goHeapSize := loadutil.GetGoInUsedHeap()
r.emitStatHook(percent, used, goHeapSize)
if used > goHeapSize+r.minRecycleLimit {
debug.FreeOSMemory()
}
}
}
}
}()
return
}
func (r *RecycleTimer) Stop() {
close(r.stopC)
}
func (r *RecycleTimer) SetPanicHook(handle func(r interface{})) {
r.panicHook = handle
}
func (r *RecycleTimer) SetStatHook(handle func(totalPercent float64, currentProcess uint64, currentGoHeap uint64)) {
r.statHook = handle
}
func NewRecycleTimer(interval time.Duration, recyclePercent float64, minRecycleLimit uint64) (r *RecycleTimer, err error) {
r = &RecycleTimer{
interval: interval,
recyclePercent: recyclePercent,
minRecycleLimit: minRecycleLimit,
stopC: make(chan interface{}),
}
err = r.start()
if err != nil {
r.Stop()
return
}
return
}

87
util/recycle_timer_test.go Normal file
View File

@ -0,0 +1,87 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package util_test
import (
"runtime"
"testing"
"time"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/loadutil"
"github.com/cubefs/cubefs/util/strutil"
"github.com/stretchr/testify/require"
)
func TestRecycleTimer(t *testing.T) {
const smallBufSize uint64 = 1 * util.GB
const largeBufSize uint64 = 2 * util.GB
buf := make([]byte, smallBufSize)
for i := 0; i < len(buf); i++ {
buf[i] = 0
}
actual, err := loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
inUsed := loadutil.GetGoInUsedHeap()
t.Logf("Alloc: Total used %v heap %v", strutil.FormatSize(actual), strutil.FormatSize(inUsed))
require.GreaterOrEqual(t, actual, smallBufSize)
require.GreaterOrEqual(t, inUsed, smallBufSize)
timer, err := util.NewRecycleTimer(1*time.Second, 0, largeBufSize)
require.NoError(t, err)
defer timer.Stop()
buf = nil
runtime.GC()
actual, err = loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
inUsed = loadutil.GetGoInUsedHeap()
t.Logf("After GC: Total used %v heap %v", strutil.FormatSize(actual), strutil.FormatSize(inUsed))
require.GreaterOrEqual(t, actual, smallBufSize)
require.Less(t, inUsed, smallBufSize)
buf = make([]byte, largeBufSize)
for i := 0; i < len(buf); i++ {
buf[i] = 0
}
actual, err = loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
inUsed = loadutil.GetGoInUsedHeap()
t.Logf("Alloc: Total used %v heap %v", strutil.FormatSize(actual), strutil.FormatSize(inUsed))
require.GreaterOrEqual(t, actual, largeBufSize)
require.GreaterOrEqual(t, inUsed, largeBufSize)
buf = nil
runtime.GC()
actual, err = loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
inUsed = loadutil.GetGoInUsedHeap()
t.Logf("After GC: Total used %v heap %v", strutil.FormatSize(actual), strutil.FormatSize(inUsed))
require.GreaterOrEqual(t, actual, smallBufSize)
require.Less(t, inUsed, smallBufSize)
time.Sleep(2 * time.Second)
actual, err = loadutil.GetCurrentProcessMemory()
require.NoError(t, err)
inUsed = loadutil.GetGoInUsedHeap()
t.Logf("After Recycle: Total used %v heap %v", strutil.FormatSize(actual), strutil.FormatSize(inUsed))
require.Less(t, actual, largeBufSize)
require.Less(t, inUsed, largeBufSize)
}