extracted socketbuffer to external repo (#7)

* extracted socketbuffer to external repo * multiple bw limiter goroutines
danlapid · Sep 3, 2022 · 5bd435c · 5bd435c
1 parent 98a1b2b
commit 5bd435c
Show file tree

Hide file tree

Showing 12 changed files with 15 additions and 513 deletions.
diff --git a/go.mod b/go.mod
@@ -4,12 +4,12 @@ go 1.19
 
 require (
 	github.com/BurntSushi/toml v1.2.0
+	github.com/danlapid/socketbuffer v1.0.0
 	github.com/klauspost/reedsolomon v1.10.0
 	github.com/rjeczalik/notify v0.9.3-0.20210809113154-3472d85e95cd
 	github.com/sirupsen/logrus v1.9.0
 	github.com/yeka/zip v0.0.0-20180914125537-d046722c6feb
 	github.com/zhuangsirui/binpacker v2.0.0+incompatible
-	golang.org/x/sys v0.0.0-20220829200755-d48e67d00261
 	golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9
 	gorm.io/driver/sqlite v1.3.6
 	gorm.io/gorm v1.23.8
@@ -21,4 +21,5 @@ require (
 	github.com/klauspost/cpuid/v2 v2.0.14 // indirect
 	github.com/mattn/go-sqlite3 v1.14.12 // indirect
 	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
+	golang.org/x/sys v0.0.0-20220829200755-d48e67d00261 // indirect
 )
diff --git a/go.sum b/go.sum
@@ -1,5 +1,7 @@
 github.com/BurntSushi/toml v1.2.0 h1:Rt8g24XnyGTyglgET/PRUNlrUeu9F5L+7FilkXfZgs0=
 github.com/BurntSushi/toml v1.2.0/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
+github.com/danlapid/socketbuffer v1.0.0 h1:jBfA6Sj14bWR809AeTX+Wy5ADoMO3qP7n8bM0HgxjdQ=
+github.com/danlapid/socketbuffer v1.0.0/go.mod h1:iCIDUm7ZqwK7XTCkJuybROMERiL/XmuSs9KtaR9IjfY=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -16,8 +18,6 @@ github.com/mattn/go-sqlite3 v1.14.12 h1:TJ1bhYJPV44phC+IMu1u2K/i5RriLTPe+yc68XDJ
 github.com/mattn/go-sqlite3 v1.14.12/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rjeczalik/notify v0.9.2 h1:MiTWrPj55mNDHEiIX5YUSKefw/+lCQVoAFmD6oQm5w8=
-github.com/rjeczalik/notify v0.9.2/go.mod h1:aErll2f0sUX9PXZnVNyeiObbmTlk5jnMoCa4QEjJeqM=
 github.com/rjeczalik/notify v0.9.3-0.20210809113154-3472d85e95cd h1:LHLg0gdpRUCvujg2Zol6e2Uknq5vHycLxqEzYwxt1vY=
 github.com/rjeczalik/notify v0.9.3-0.20210809113154-3472d85e95cd/go.mod h1:gF3zSOrafR9DQEWSE8TjfI9NkooDxbyT4UgRGKZA0lc=
 github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=

diff --git a/pkg/bandwidthlimiter/bandwidthlimiter.go b/pkg/bandwidthlimiter/bandwidthlimiter.go
@@ -29,11 +29,13 @@ func worker(ctx context.Context, conf *bandwidthLimiterConfig) {
 	}
 }
 
-func CreateBandwidthLimiter(ctx context.Context, bandwidth int, chunksize int, input chan *structs.Chunk, output chan *structs.Chunk) {
+func CreateBandwidthLimiter(ctx context.Context, bandwidth int, chunksize int, input chan *structs.Chunk, output chan *structs.Chunk, workercount int) {
 	conf := bandwidthLimiterConfig{
 		rl:     rate.NewLimiter(rate.Limit(bandwidth), chunksize),
 		input:  input,
 		output: output,
 	}
-	go worker(ctx, &conf)
+	for i := 0; i < workercount; i++ {
+		go worker(ctx, &conf)
+	}
 }
diff --git a/pkg/bandwidthlimiter/bandwidthlimiter_test.go b/pkg/bandwidthlimiter/bandwidthlimiter_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"oneway-filesync/pkg/bandwidthlimiter"
 	"oneway-filesync/pkg/structs"
+	"runtime"
 	"testing"
 	"time"
 )
@@ -34,7 +35,8 @@ func TestCreateBandwidthLimiter(t *testing.T) {
 
 			ctx, cancel := context.WithCancel(context.Background())
 			start := time.Now()
-			bandwidthlimiter.CreateBandwidthLimiter(ctx, tt.args.bytes_per_sec, tt.args.chunk_size, ch_in, ch_out)
+			bandwidthlimiter.CreateBandwidthLimiter(ctx, tt.args.bytes_per_sec, tt.args.chunk_size, ch_in, ch_out, runtime.GOMAXPROCS(0)*2)
+
 			for i := 0; i < tt.args.chunk_count; i++ {
 				<-ch_out
 			}

diff --git a/pkg/sender/sender.go b/pkg/sender/sender.go
@@ -25,6 +25,6 @@ func Sender(ctx context.Context, db *gorm.DB, conf config.Config) {
 	queuereader.CreateQueueReader(ctx, db, queue_chan)
 	filereader.CreateFileReader(ctx, db, conf.ChunkSize, conf.ChunkFecRequired, queue_chan, chunks_chan, maxprocs)
 	fecencoder.CreateFecEncoder(ctx, conf.ChunkSize, conf.ChunkFecRequired, conf.ChunkFecTotal, chunks_chan, shares_chan, maxprocs)
-	bandwidthlimiter.CreateBandwidthLimiter(ctx, conf.BandwidthLimit, conf.ChunkSize, shares_chan, bw_limited_chunks)
+	bandwidthlimiter.CreateBandwidthLimiter(ctx, conf.BandwidthLimit, conf.ChunkSize, shares_chan, bw_limited_chunks, maxprocs)
 	udpsender.CreateUdpSender(ctx, conf.ReceiverIP, conf.ReceiverPort, bw_limited_chunks, maxprocs)
 }
diff --git a/pkg/udpreceiver/udpreceiver.go b/pkg/udpreceiver/udpreceiver.go
@@ -5,9 +5,9 @@ import (
 	"errors"
 	"net"
 	"oneway-filesync/pkg/structs"
-	"oneway-filesync/pkg/utils"
 	"time"
 
+	"github.com/danlapid/socketbuffer"
 	"github.com/sirupsen/logrus"
 )
 
@@ -24,7 +24,7 @@ func manager(ctx context.Context, conf *udpReceiverConfig) {
 		logrus.Errorf("Error getting raw socket: %v", err)
 		return
 	}
-	bufsize, err := utils.GetReadBuffer(rawconn)
+	bufsize, err := socketbuffer.GetReadBuffer(rawconn)
 	if err != nil {
 		logrus.Errorf("Error getting read buffer size: %v", err)
 	}
@@ -34,7 +34,7 @@ func manager(ctx context.Context, conf *udpReceiverConfig) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			toread, err := utils.GetAvailableBytes(rawconn)
+			toread, err := socketbuffer.GetAvailableBytes(rawconn)
 			if err != nil {
 				logrus.Errorf("Error getting available bytes on socket: %v", err)
 			}

diff --git a/pkg/utils/darwin_ioctl.go b/pkg/utils/darwin_ioctl.go
diff --git a/pkg/utils/linux_ioctl.go b/pkg/utils/linux_ioctl.go
diff --git a/pkg/utils/linux_procudp.go b/pkg/utils/linux_procudp.go