Merge pull request #640 from metrico/fix/remove-overall-bandwith-limitation

remove bandwith -> RAM usage limitation

Author: akvlad

go.mod

@@ -24,7 +24,7 @@ replace (
 )
 
 require (
-	github.com/ClickHouse/ch-go v0.65.1
+	github.com/ClickHouse/ch-go v0.64.1
 	github.com/ClickHouse/clickhouse-go/v2 v2.30.3
 	github.com/Masterminds/sprig v2.22.0+incompatible
 	github.com/VictoriaMetrics/fastcache v1.12.2

go.sum

@@ -71,8 +71,8 @@ github.com/Azure/go-autorest/tracing v0.6.0 h1:TYi4+3m5t6K48TGI9AUdb+IzbnSxvnvUM
 github.com/Azure/go-autorest/tracing v0.6.0/go.mod h1:+vhtPC754Xsa23ID7GlGsrdKBpUA79WCAKPPZVC2DeU=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
-github.com/ClickHouse/ch-go v0.65.1 h1:SLuxmLl5Mjj44/XbINsK2HFvzqup0s6rwKLFH347ZhU=
-github.com/ClickHouse/ch-go v0.65.1/go.mod h1:bsodgURwmrkvkBe5jw1qnGDgyITsYErfONKAHn05nv4=
+github.com/ClickHouse/ch-go v0.64.1 h1:FWpP+QU4KchgzpEekuv8YoI/fUc4H2r6Bwc5WwrzvcI=
+github.com/ClickHouse/ch-go v0.64.1/go.mod h1:RBUynvczWwVzhS6Up9lPKlH1mrk4UAmle6uzCiW4Pkc=
 github.com/ClickHouse/clickhouse-go/v2 v2.30.3 h1:m0VZqUNCJ7lOmZfmOE3HZUMixZHftKmZLqcrz2+UVHk=
 github.com/ClickHouse/clickhouse-go/v2 v2.30.3/go.mod h1:V1aZaG0ctMbd8KVi+D4loXi97duWYtHiQHMCgipKJcI=
 github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=

writer/service/genericInsertService.go

@@ -11,7 +11,6 @@ import (
 	"github.com/metrico/qryn/writer/utils/logger"
 	"github.com/metrico/qryn/writer/utils/promise"
 	"github.com/metrico/qryn/writer/utils/stat"
-	"golang.org/x/sync/semaphore"
 	"math/rand"
 	"sync"
 	"sync/atomic"
@@ -106,8 +105,6 @@ type InsertServiceV2 struct {
 	client ch_wrapper.IChClient
 
 	state int32
-
-	bandwithLimiter *semaphore.Weighted
 }
 
 func (svc *InsertServiceV2) PlanFlush() {
@@ -206,14 +203,6 @@ func (svc *InsertServiceV2) Request(req helpers.SizeGetter, insertMode int) *pro
 	}
 	var size int64
 	size = req.GetSize()
-	to, cancel := context.WithTimeout(context.Background(), time.Second)
-	defer cancel()
-	err := svc.bandwithLimiter.Acquire(to, size)
-	if err != nil {
-		logger.Info("service overflow")
-		p.Done(0, fmt.Errorf("service overflow"))
-		return p
-	}
 	func() {
 		var (
 			inserted int
@@ -226,7 +215,6 @@ func (svc *InsertServiceV2) Request(req helpers.SizeGetter, insertMode int) *pro
 
 		if err != nil || inserted == 0 {
 			p.Done(0, err)
-			svc.bandwithLimiter.Release(size)
 			return
 		}
 		svc.size += size
@@ -292,7 +280,6 @@ func (svc *InsertServiceV2) fetchLoopIteration() {
 		input[i] = c.Input()
 		size += int64(svc.IngestSize(&input[i]))
 	}
-	svc.bandwithLimiter.Release(portion.size)
 
 	svc.setState(INSERT_STATE_INSERTING)
 	defer svc.setState(INSERT_STATE_IDLE)
@@ -420,25 +407,20 @@ func (svc *InsertServiceV2RoundRobin) init() {
 	logger.Info(fmt.Sprintf("creating %d services", svc.svcNum))
 	svc.services = make([]*InsertServiceV2, svc.svcNum)
 	svc.rand = rand.New(rand.NewSource(time.Now().UnixNano()))
-	var bandwidthLimit int64 = BANDWITH_LIMIT
-	if svc.maxQueueSize*2 > BANDWITH_LIMIT {
-		bandwidthLimit = svc.maxQueueSize * 2
-	}
 	for i := range svc.services {
 		svc.services[i] = &InsertServiceV2{
-			ID:              fmt.Sprintf("%s-%s-%v", svc.DatabaseNode.Node, svc.insertRequest, svc.AsyncInsert),
-			ServiceData:     ServiceData{},
-			V3Session:       svc.V3Session,
-			DatabaseNode:    svc.DatabaseNode,
-			AsyncInsert:     svc.AsyncInsert,
-			OnBeforeInsert:  svc.OnBeforeInsert,
-			pushInterval:    svc.pushInterval,
-			maxQueueSize:    svc.maxQueueSize,
-			insertRequest:   svc.insertRequest,
-			acquireColumns:  svc.acquireColumns,
-			processRequest:  svc.processRequest,
-			bandwithLimiter: semaphore.NewWeighted(bandwidthLimit),
-			serviceType:     svc.serviceType,
+			ID:             fmt.Sprintf("%s-%s-%v", svc.DatabaseNode.Node, svc.insertRequest, svc.AsyncInsert),
+			ServiceData:    ServiceData{},
+			V3Session:      svc.V3Session,
+			DatabaseNode:   svc.DatabaseNode,
+			AsyncInsert:    svc.AsyncInsert,
+			OnBeforeInsert: svc.OnBeforeInsert,
+			pushInterval:   svc.pushInterval,
+			maxQueueSize:   svc.maxQueueSize,
+			insertRequest:  svc.insertRequest,
+			acquireColumns: svc.acquireColumns,
+			processRequest: svc.processRequest,
+			serviceType:    svc.serviceType,
 		}
 		svc.services[i].Init()
 	}