Dynamic workspace threshold

pkg/vm/engine/tae/db/merge/meminfo_darwin.go → pkg/objectio/meminfo_darwin.go

@@ -4,22 +4,22 @@
 // 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
+//	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 merge
+package objectio
 
 import (
 	"syscall"
 	"unsafe"
 )
 
-func totalMem() uint64 {
+func TotalMem() uint64 {
 	s, err := syscall.Sysctl("hw.memsize")
 	if err != nil {
 		return 0

pkg/vm/engine/tae/db/merge/meminfo_linux.go → pkg/objectio/meminfo_linux.go

@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package merge
+package objectio
 
 import "syscall"
 
-func totalMem() uint64 {
+func TotalMem() uint64 {
 	in := new(syscall.Sysinfo_t)
 	err := syscall.Sysinfo(in)
 	if err != nil {

pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go

@@ -53,6 +53,7 @@ func (c *DashboardCreator) initTxnDashboard() error {
 			c.initTxnShowAccountsRow(),
 			c.initCNCommittedObjectQuantityRow(),
 			c.initTombstoneTransferRow(),
+			c.initTxnExtraWorkspaceQuota(),
 		)...)
 	if err != nil {
 		return err
@@ -570,3 +571,14 @@ func (c *DashboardCreator) initTombstoneTransferRow() dashboard.Option {
 		),
 	)
 }
+
+func (c *DashboardCreator) initTxnExtraWorkspaceQuota() dashboard.Option {
+	return dashboard.Row(
+		"Extra Workspace Quota",
+		c.withGraph(
+			"Extra Workspace Quota",
+			12,
+			`sum(`+c.getMetricWithFilter("mo_txn_extra_workspace_quota", ``)+`)`,
+			"{{ "+c.by+" }}", axis.Unit("decbytes")),
+	)
+}

pkg/util/metric/v2/metrics.go

@@ -161,6 +161,7 @@ func initTxnMetrics() {
 	registry.MustRegister(txnReaderTombstoneSelectivityHistogram)
 	registry.MustRegister(txnTransferDurationHistogram)
 	registry.MustRegister(TransferTombstonesCountHistogram)
+	registry.MustRegister(TxnExtraWorkspaceQuotaGauge)
 }
 
 func initRPCMetrics() {

pkg/util/metric/v2/txn.go

@@ -397,3 +397,13 @@ var (
 	TransferTombstonesDurationHistogram      = txnTransferDurationHistogram.WithLabelValues("tombstones")
 	BatchTransferTombstonesDurationHistogram = txnTransferDurationHistogram.WithLabelValues("batch")
 )
+
+var (
+	TxnExtraWorkspaceQuotaGauge = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: "mo",
+			Subsystem: "txn",
+			Name:      "extra_workspace_quota",
+			Help:      "Extra workspace quota for txn.",
+		})
+)

pkg/vm/engine/disttae/engine.go

@@ -39,6 +39,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/lockservice"
 	"github.com/matrixorigin/matrixone/pkg/logservice"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/statsinfo"
@@ -166,12 +167,18 @@ func (e *Engine) fillDefaults() {
 	if e.config.cnTransferTxnLifespanThreshold <= 0 {
 		e.config.cnTransferTxnLifespanThreshold = CNTransferTxnLifespanThreshold
 	}
+	if e.Limiter.quota.Load() <= 0 {
+		mem := objectio.TotalMem() / 100 * 5
+		e.Limiter.quota.Store(mem)
+		v2.TxnExtraWorkspaceQuotaGauge.Set(float64(mem))
+	}
 
 	logutil.Info(
 		"INIT-ENGINE-CONFIG",
 		zap.Int("InsertEntryMaxCount", e.config.insertEntryMaxCount),
 		zap.Uint64("CommitWorkspaceThreshold", e.config.commitWorkspaceThreshold),
 		zap.Uint64("WriteWorkspaceThreshold", e.config.writeWorkspaceThreshold),
+		zap.Uint64("ExtraWorkspaceThresholdQuota", e.Limiter.quota.Load()),
 		zap.Duration("CNTransferTxnLifespanThreshold", e.config.cnTransferTxnLifespanThreshold),
 	)
 }
@@ -191,6 +198,40 @@ func (e *Engine) SetWorkspaceThreshold(commitThreshold, writeThreshold uint64) (
 	return
 }
 
+func (e *limiter) AcquireQuota(v uint64, quota *MemoryQuota) (uint64, bool) {
+	for {
+		oldRemaining := e.quota.Load()
+		if oldRemaining < v {
+			return 0, false
+		}
+		remaining := oldRemaining - v
+		if e.quota.CompareAndSwap(oldRemaining, remaining) {
+			quota.Apply(v)
+			v2.TxnExtraWorkspaceQuotaGauge.Set(float64(remaining))
+			logutil.Info(
+				"WORKSPACE-QUOTA-ACQUIRE",
+				zap.Uint64("quota", v),
+				zap.Uint64("remaining", remaining),
+			)
+			return remaining, true
+		}
+	}
+}
+
+func (e *limiter) ReleaseQuota(quota *MemoryQuota) {
+	size := quota.Release()
+	if size == 0 {
+		return
+	}
+	e.quota.Add(size)
+	v2.TxnExtraWorkspaceQuotaGauge.Set(float64(e.quota.Load()))
+	logutil.Info(
+		"WORKSPACE-QUOTA-RELEASE",
+		zap.Uint64("quota", size),
+		zap.Uint64("remaining", e.quota.Load()),
+	)
+}
+
 func (e *Engine) GetService() string {
 	return e.service
 }

pkg/vm/engine/disttae/txn.go

@@ -482,6 +482,17 @@ func (txn *Transaction) dumpBatchLocked(ctx context.Context, offset int) error {
 		if size < txn.writeWorkspaceThreshold {
 			return nil
 		}
+
+		// try to increase the write threshold from quota
+		threshold := txn.writeWorkspaceThreshold
+		for size >= threshold {
+			threshold *= 2
+		}
+		_, acquired := txn.engine.Limiter.AcquireQuota(threshold-txn.writeWorkspaceThreshold, &txn.quota)
+		if acquired {
+			txn.writeWorkspaceThreshold = threshold
+			return nil
+		}
 		size = 0
 	}
 	txn.hasS3Op.Store(true)
@@ -1462,6 +1473,8 @@ func (txn *Transaction) delTransaction() {
 	txn.transfer.timestamps = nil
 	txn.transfer.lastTransferred = types.TS{}
 	txn.transfer.pendingTransfer = false
+
+	txn.engine.Limiter.ReleaseQuota(&txn.quota)
 }
 
 func (txn *Transaction) rollbackTableOpLocked() {

pkg/vm/engine/disttae/types.go

@@ -168,6 +168,12 @@ func WithWriteWorkspaceThreshold(th uint64) EngineOptions {
 	}
 }
 
+func WithExtraWorkspaceThresholdQuota(quota uint64) EngineOptions {
+	return func(e *Engine) {
+		e.Limiter.quota.Store(quota)
+	}
+}
+
 func WithInsertEntryMaxCount(th int) EngineOptions {
 	return func(e *Engine) {
 		e.config.insertEntryMaxCount = th
@@ -198,6 +204,33 @@ func WithSQLExecFunc(f func() ie.InternalExecutor) EngineOptions {
 	}
 }
 
+type MemoryQuota struct {
+	state uint8 // active 0, released 1
+	size  uint64
+}
+
+func (q *MemoryQuota) Apply(size uint64) bool {
+	if q.state != 0 {
+		return false
+	}
+	q.size += size
+	return true
+}
+
+func (q *MemoryQuota) Release() (size uint64) {
+	if q.state == 1 || q.size == 0 {
+		return 0
+	}
+	q.state = 1
+	size = q.size
+	q.size = 0
+	return size
+}
+
+type limiter struct {
+	quota atomic.Uint64
+}
+
 type Engine struct {
 	sync.RWMutex
 	service  string
@@ -259,6 +292,8 @@ type Engine struct {
 	dynamicCtx
 	// for test only.
 	skipConsume bool
+	// workspace extra threshold limit
+	Limiter limiter
 }
 
 func (e *Engine) SetService(svr string) {
@@ -366,6 +401,8 @@ type Transaction struct {
 
 	writeWorkspaceThreshold  uint64
 	commitWorkspaceThreshold uint64
+
+	quota MemoryQuota
 }
 
 type Pos struct {

pkg/vm/engine/tae/db/merge/utils.go

@@ -228,7 +228,7 @@ func (c *resourceController) setMemLimit(total uint64) {
 
 func (c *resourceController) refresh() {
 	if c.limit == 0 {
-		c.setMemLimit(totalMem())
+		c.setMemLimit(objectio.TotalMem())
 	}
 
 	if c.proc == nil {