add many

.gitignore

@@ -0,0 +1,30 @@
+# Compiled Object files, Static and Dynamic libs (Shared Objects)
+*.o
+*.a
+*.so
+
+# Folders
+_obj
+_test
+
+# Architecture specific extensions/prefixes
+*.[568vq]
+[568vq].out
+
+*.cgo1.go
+*.cgo2.c
+_cgo_defun.c
+_cgo_gotypes.go
+_cgo_export.*
+
+_testmain.go
+
+*.exe
+*.test
+*.prof
+
+sabo
+vendor/
+dist/
+foo/
+*~

.goreleaser.yml

@@ -0,0 +1,16 @@
+builds:
+  - binary: sabo
+    goos:
+      - darwin
+      - linux
+    goarch:
+      - amd64
+    ldflags:
+     - -X main.Version={{.Version}}
+archive:
+  format: zip
+  name_template: "{{ .ProjectName }}_{{ .Os }}_{{ .Arch }}"
+release:
+  github:
+    owner: kazeburo
+    name: sabo

Gopkg.toml

@@ -0,0 +1,42 @@
+# Gopkg.toml example
+#
+# Refer to https://golang.github.io/dep/docs/Gopkg.toml.html
+# for detailed Gopkg.toml documentation.
+#
+# required = ["github.com/user/thing/cmd/thing"]
+# ignored = ["github.com/user/project/pkgX", "bitbucket.org/user/project/pkgA/pkgY"]
+#
+# [[constraint]]
+#   name = "github.com/user/project"
+#   version = "1.0.0"
+#
+# [[constraint]]
+#   name = "github.com/user/project2"
+#   branch = "dev"
+#   source = "github.com/myfork/project2"
+#
+# [[override]]
+#   name = "github.com/x/y"
+#   version = "2.4.0"
+#
+# [prune]
+#   non-go = false
+#   go-tests = true
+#   unused-packages = true
+
+
+[[constraint]]
+  name = "github.com/dustin/go-humanize"
+  version = "1.0.0"
+
+[[constraint]]
+  name = "github.com/jessevdk/go-flags"
+  version = "1.4.0"
+
+[[constraint]]
+  branch = "master"
+  name = "golang.org/x/time"
+
+[prune]
+  go-tests = true
+  unused-packages = true

Makefile

@@ -0,0 +1,29 @@
+VERSION=0.0.1
+LDFLAGS=-ldflags "-X main.Version=${VERSION}"
+all: sabo
+
+.PHONY: sabo
+
+bundle:
+	dep ensure
+
+update:
+	dep ensure -update
+
+sabo: sabo.go
+	go build $(LDFLAGS) -o sabo
+
+linux: sabo.go
+	GOOS=linux GOARCH=amd64 go build $(LDFLAGS) -o sabo
+
+fmt:
+	go fmt ./...
+
+clean:
+	rm -rf sabo
+
+tag:
+	git tag v${VERSION}
+	git push origin v${VERSION}
+	git push origin master
+	goreleaser --rm-dist

README.md

@@ -1,5 +1,39 @@
-# sabo
+# sabo - bandwidth limiting pipe with collaborative capability
+
+sabo is bandwidth limiting pipe like throttle command. And has collaborative capability like trickle
+
+throttle - bandwidth limiting pipe https://linux.die.net/man/1/throttle
+trickle - a lightweight userspace bandwidth shaper https://www.systutorials.com/docs/linux/man/1-trickle/
+
+## Usage
 
 ```
-cat /dev/urandom|  go run sabo.go --max-bandwidth 100 --work-dir foo > /dev/null
+% sabo -h
+Usage:
+  sabo [OPTIONS]
+
+Application Options:
+      --max-bandwidth= max bandwidth (Bytes/sec)
+      --work-dir=      directory for control bandwidth
+  -v, --version        Show version
+
+Help Options:
+  -h, --help           Show this help message
 ```
+
+sabo creates a lock file in work dir. sabo checks number of files in work dirr in 1 sec.
+Each process's bandwidth limitation become `max-bandwidth / number of file`.
+
+## example
+
+```
+MAX_BW=100M
+WORK_DIR=/tmp/sabo_for_dump
+mkdir -p /tmp/sabo_for_dump
+mysqldump -h backup1 db table1  | sabo --max-bandwidth $MAX_BW --work-dir $WORK_DIR > /tmp/sabo_for_dump/table1.sql &
+mysqldump -h backup2 db table2  | sabo --max-bandwidth $MAX_BW --work-dir $WORK_DIR > /tmp/sabo_for_dump/table2.sql &
+wait
+```
+
+Each mysqldump's bandwidth limitation is 50MB/sec.
+When either of mysqldump finishes, the bandwidth of the remaining process becomes 100 MB/s.

sabo.go

@@ -6,8 +6,10 @@ import (
 	"io"
 	"log"
 	"os"
+	"os/signal"
 	"path/filepath"
 	"runtime"
+	"syscall"
 
 	humanize "github.com/dustin/go-humanize"
 	flags "github.com/jessevdk/go-flags"
@@ -17,6 +19,8 @@ import (
 // Version set in compile
 var Version string
 
+const minimumBandwidh = 32 * 1000
+
 type cmdOpts struct {
 	MaxBandWidth string `long:"max-bandwidth" description:"max bandwidth (Bytes/sec)" required:"true"`
 	WorkDir      string `long:"work-dir" description:"directory for control bandwidth" required:"true"`
@@ -32,7 +36,7 @@ func main() {
 	}
 
 	if opts.Version {
-		fmt.Printf(`motarei %s
+		fmt.Printf(`sabo %s
 Compiler: %s %s
 `,
 			Version,
@@ -44,7 +48,11 @@ Compiler: %s %s
 
 	bw, err := humanize.ParseBytes(opts.MaxBandWidth)
 	if err != nil {
-		fmt.Println("Cannot parse -max-bandwidth", err)
+		log.Printf("Cannot parse -max-bandwidth: %v", err)
+		os.Exit(1)
+	}
+	if bw < minimumBandwidh {
+		log.Printf("max-bandwidth > 32K required: %d", bw)
 		os.Exit(1)
 	}
 
@@ -58,7 +66,14 @@ Compiler: %s %s
 	if err != nil {
 		log.Fatalf("Cannot create initial bandwidth:%s", err)
 	}
-	reader.RunRefresh(ctx)
+	go reader.RunRefresh(ctx)
 
-	io.Copy(os.Stdout, reader)
+	sc := make(chan os.Signal, 1)
+	signal.Notify(sc, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		<-sc
+		os.Stdin.Close()
+	}()
+	buf := make([]byte, minimumBandwidh)
+	io.CopyBuffer(os.Stdout, reader, buf)
 }

saboreader/saboreader.go

@@ -28,7 +28,8 @@ type Reader struct {
 	limiter *rate.Limiter
 	ctx     context.Context
 	mu      *sync.RWMutex
-	lf      *os.File
+	lfh     *os.File
+	lfn     string
 	wd      string
 	bw      uint64
 }
@@ -39,8 +40,10 @@ func NewReaderWithContext(ctx context.Context, r io.Reader, workDir string, bw u
 	if err != nil {
 		return nil, fmt.Errorf("Cannot open workdir: %v", err)
 	}
-	lockfile := fmt.Sprintf("_sabo_%d_%d.lock", bw, os.Getpid())
-	file, err := os.OpenFile(filepath.Join(workDir, lockfile), syscall.O_RDWR|syscall.O_CREAT, 0600)
+	lockfileName := fmt.Sprintf("sabo_%d_%d.lock", bw, os.Getpid())
+	lockfile := filepath.Join(workDir, lockfileName)
+	tmpfile := filepath.Join(workDir, "_"+lockfileName)
+	file, err := os.OpenFile(tmpfile, syscall.O_RDWR|syscall.O_CREAT, 0600)
 	if err != nil {
 		return nil, fmt.Errorf("Cannot create lockfile in workdir: %v", err)
 	}
@@ -49,21 +52,25 @@ func NewReaderWithContext(ctx context.Context, r io.Reader, workDir string, bw u
 	if err != nil {
 		return nil, fmt.Errorf("Cannot lock lockfile in workdir: %v", err)
 	}
-	err = os.Rename(filepath.Join(workDir, lockfile), filepath.Join(workDir, fmt.Sprintf("sabo_%d_%d.lock", bw, os.Getpid())))
+	err = os.Rename(tmpfile, lockfile)
+	if err != nil {
+		return nil, fmt.Errorf("Cannot rename lockfile: %v", err)
+	}
 	return &Reader{
 		r:   r,
 		ctx: ctx,
 		mu:  new(sync.RWMutex),
-		lf:  file,
+		lfh: file,
+		lfn: lockfile,
 		wd:  workDir,
 		bw:  bw,
 	}, nil
 }
 
 // CleanUp clean up lockfile
 func (s *Reader) CleanUp() {
-	defer os.Remove(filepath.Join(s.wd, s.lf.Name()))
-	s.lf.Close()
+	defer os.Remove(s.lfn)
+	s.lfh.Close()
 }
 
 func (s *Reader) getRateLimit() *rate.Limiter {
@@ -83,7 +90,7 @@ func (s *Reader) Read(p []byte) (int, error) {
 	if err != nil {
 		return n, err
 	}
-
+	//log.Printf("read: %d", n)
 	if err := limiter.WaitN(s.ctx, n); err != nil {
 		return n, err
 	}
@@ -102,7 +109,7 @@ func (s *Reader) RefreshLimiter(ctx context.Context) error {
 			continue
 		}
 
-		if m, _ := regexp.MatchString(fmt.Sprintf("^sabo_%d", s.bw), file.Name()); !m {
+		if m, _ := regexp.MatchString(fmt.Sprintf("^sabo_%d_", s.bw), file.Name()); !m {
 			continue
 		}
 
@@ -125,7 +132,7 @@ func (s *Reader) RefreshLimiter(ctx context.Context) error {
 	if locked > 0 {
 		bytesPerSec = float64(s.bw / locked)
 	}
-	fmt.Fprintf(os.Stderr, "new limit %f\n", bytesPerSec)
+	//fmt.Fprintf(os.Stderr, "new limit %f\n", bytesPerSec)
 	limiter := rate.NewLimiter(rate.Limit(bytesPerSec), burstLimit)
 	limiter.AllowN(time.Now(), burstLimit) // spend initial burst
 	s.mu.Lock()