close EC streams when idle, reopen on demand

* refactor and amend housekeeper - use UnregInterval consistently across - add `UnregIf` - reduce work chan capacity to 48 (was 512); add "channel full" check * target: implement on-EC/off-EC handler - TODO: revisit 1m delay * part three, prev. commit: 1114868 Signed-off-by: Alex Aizman <[email protected]>
NVIDIA · Sep 10, 2024 · 0642c85 · 0642c85
1 parent a91636b
commit 0642c85
Show file tree

Hide file tree

Showing 10 changed files with 126 additions and 96 deletions.
diff --git a/ais/daemon.go b/ais/daemon.go
@@ -193,7 +193,7 @@ func initDaemon(version, buildTime string) cos.Runner {
 
 	daemon.rg = &rungroup{rs: make(map[string]cos.Runner, 6)}
 	hk.Init()
-	daemon.rg.add(hk.DefaultHK)
+	daemon.rg.add(hk.HK)
 
 	// K8s
 	k8s.Init()
@@ -361,11 +361,11 @@ func (g *rungroup) runAll(mainRunner cos.Runner) error {
 	g.errCh = make(chan runRet, len(g.rs))
 
 	// run all, housekeeper first
-	go g.run(hk.DefaultHK)
+	go g.run(hk.HK)
 	runtime.Gosched()
 	hk.WaitStarted()
 	for _, r := range g.rs {
-		if r.Name() == hk.DefaultHK.Name() {
+		if r.Name() == hk.HK.Name() {
 			continue
 		}
 		go g.run(r)

diff --git a/ais/tgtec.go b/ais/tgtec.go
@@ -9,12 +9,14 @@ import (
 	"net/http"
 	"os"
 	"strconv"
+	"time"
 
 	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/ec"
+	"github.com/NVIDIA/aistore/hk"
 )
 
 func (t *target) ecHandler(w http.ResponseWriter, r *http.Request) {
@@ -67,21 +69,31 @@ func (t *target) sendECMetafile(w http.ResponseWriter, r *http.Request, bck *met
 }
 
 func (t *target) httpecpost(w http.ResponseWriter, r *http.Request) {
-	apiItems, err := t.parseURL(w, r, apc.URLPathEC.L, 1, false)
+	const (
+		hkname   = "close-ec-streams" + hk.NameSuffix
+		postpone = time.Minute
+	)
+	items, err := t.parseURL(w, r, apc.URLPathEC.L, 1, false)
 	if err != nil {
 		return
 	}
-	action := apiItems[0]
+	action := items[0]
 	switch action {
 	case apc.ActEcOpen:
+		hk.UnregIf(hkname, closeEc) // just in case
 		ec.ECM.OpenStreams(false /*with refc*/)
 	case apc.ActEcClose:
-		ec.ECM.CloseStreams(false /*with refc*/)
+		hk.Reg(hkname, closeEc, postpone)
 	default:
 		t.writeErr(w, r, errActEc(action))
 	}
 }
 
+func closeEc(int64) time.Duration {
+	ec.ECM.CloseStreams(false /*with refc*/)
+	return hk.UnregInterval
+}
+
 func errActEc(act string) error {
 	return fmt.Errorf(fmtErrInvaldAction, act, []string{apc.ActEcOpen, apc.ActEcClose})
 }
diff --git a/bench/tools/aisloader/run.go b/bench/tools/aisloader/run.go
@@ -351,7 +351,7 @@ func Start(version, buildtime string) (err error) {
 	// empty config to use memsys constants;
 	// alternatively: "memsys": { "min_free": "2gb", ... }
 	hk.Init()
-	go hk.DefaultHK.Run()
+	go hk.HK.Run()
 	hk.WaitStarted()
 
 	config := &cmn.Config{}

diff --git a/cmd/cli/go.mod b/cmd/cli/go.mod
@@ -3,7 +3,7 @@ module github.com/NVIDIA/aistore/cmd/cli
 go 1.22.3
 
 require (
-	github.com/NVIDIA/aistore v1.3.24-0.20240906170600-713920710479
+	github.com/NVIDIA/aistore v1.3.24-0.20240909232046-a91636bc334a
 	github.com/fatih/color v1.17.0
 	github.com/json-iterator/go v1.1.12
 	github.com/onsi/ginkgo/v2 v2.20.0

diff --git a/cmd/cli/go.sum b/cmd/cli/go.sum
@@ -1,7 +1,7 @@
 code.cloudfoundry.org/bytefmt v0.0.0-20190710193110-1eb035ffe2b6/go.mod h1:wN/zk7mhREp/oviagqUXY3EwuHhWyOvAdsn5Y4CzOrc=
 github.com/BurntSushi/toml v1.3.2/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
-github.com/NVIDIA/aistore v1.3.24-0.20240906170600-713920710479 h1:JUPW/hrRVa9zqxjJNBXrWdscGyVBv1dhsT9nNPi4V+8=
-github.com/NVIDIA/aistore v1.3.24-0.20240906170600-713920710479/go.mod h1:si83S9r29vwIC0f0CE2Mk+25bFiaN6mmVlmuBpP4hHM=
+github.com/NVIDIA/aistore v1.3.24-0.20240909232046-a91636bc334a h1:3pqUBkkTtvMa+6+YqA8o199Wvucqc9uJaAtndLJk2vo=
+github.com/NVIDIA/aistore v1.3.24-0.20240909232046-a91636bc334a/go.mod h1:si83S9r29vwIC0f0CE2Mk+25bFiaN6mmVlmuBpP4hHM=
 github.com/OneOfOne/xxhash v1.2.8 h1:31czK/TI9sNkxIKfaUfGlU47BAxQ0ztGgd9vPyqimf8=
 github.com/OneOfOne/xxhash v1.2.8/go.mod h1:eZbhyaAYD41SGSSsnmcpxVoRiQ/MPUTjUdIIOT9Um7Q=
 github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA=

diff --git a/ec/manager.go b/ec/manager.go
@@ -122,9 +122,10 @@ func (mgr *Manager) OpenStreams(withRefc bool) {
 	mgr.respBundle.Store(bundle.New(client, respSbArgs))
 }
 
-func (mgr *Manager) CloseStreams(withRefc bool) {
-	if withRefc {
+func (mgr *Manager) CloseStreams(justRefc bool) {
+	if justRefc {
 		mgr._refc.Dec()
+		return
 	}
 	if !mgr.bundleEnabled.CAS(true, false) {
 		return

diff --git a/hk/common_durations.go b/hk/common_durations.go
@@ -1,7 +1,7 @@
 // Package hk provides mechanism for registering cleanup
 // functions which are invoked at specified intervals.
 /*
- * Copyright (c) 2024, NVIDIA CORPORATION. All rights reserved.
+ * Copyright (c) 2023-2024, NVIDIA CORPORATION. All rights reserved.
  */
 package hk