nxshock/gwp
1
0
Fork 0
mirror of https://github.com/nxshock/gwp.git synced 2024-11-27 03:31:02 +05:00
Code Issues Projects Releases Wiki Activity

Some chages

Browse Source 
* Use go-eta library from calculating ETA
* Fix linter issues
* Add testify library for tests
This commit is contained in:
nxshock 2022-08-06 22:13:22 +05:00
parent 0d6ffec530
commit aba812a470
3 changed files with 53 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
go.mod
View File

@ -1,3 +1,8 @@
module github.com/nxshock/gwp module github.com/nxshock/gwp
go 1.16 go 1.16
require (
github.com/nxshock/go-eta v0.1.0
github.com/stretchr/testify v1.8.0
)

16
go.sum Normal file
View File

@ -0,0 +1,16 @@
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=
github.com/nxshock/go-eta v0.1.0 h1:rcy8rNPaVeL3f3c5rJRLvho0FErUJThD0+vihxA7t9c=
github.com/nxshock/go-eta v0.1.0/go.mod h1:hZ59FHIJSpPeZk38pd/3zUkI4H31hAQ2oFw+NgdU+SA=
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/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

48
gwp.go
View File

@ -7,6 +7,8 @@ import (
"runtime" "runtime"
"sync" "sync"
"time" "time"
"github.com/nxshock/go-eta"
) )
// WorkerPool represents pool of workers. // WorkerPool represents pool of workers.
@ -16,7 +18,7 @@ type WorkerPool struct {
stopChan chan struct{} stopChan chan struct{}
wg sync.WaitGroup wg sync.WaitGroup
EstimateCount int estimateCount int
ShowProgress bool ShowProgress bool
ShowSpeed bool ShowSpeed bool
@ -24,6 +26,8 @@ type WorkerPool struct {
errorCount int // processed jobs count that returned error errorCount int // processed jobs count that returned error
currentSpeed float64 // speed calculated for last minute currentSpeed float64 // speed calculated for last minute
eta *eta.Calculator
lastProgressMessage string lastProgressMessage string
} }
@ -37,7 +41,8 @@ func New(threadCount int) *WorkerPool {
workerPool := &WorkerPool{ workerPool := &WorkerPool{
jobChan: make(chan func() error), jobChan: make(chan func() error),
resultChan: make(chan error), resultChan: make(chan error),
stopChan: make(chan struct{})} stopChan: make(chan struct{}),
eta: eta.New(time.Minute, 0)}
workerPool.wg.Add(threadCount) workerPool.wg.Add(threadCount)
@ -58,13 +63,13 @@ func New(threadCount int) *WorkerPool {
select { select {
case <-tickerUpdateText.C: case <-tickerUpdateText.C:
if !newLined { if !newLined {
fmt.Fprintf(os.Stderr, endLine) fmt.Fprint(os.Stderr, endLine)
newLined = true newLined = true
} }
workerPool.printProgress() _ = workerPool.printProgress()
case <-tickerCalculateEta.C: case <-tickerCalculateEta.C:
workerPool.currentSpeed = float64(workerPool.processedCount-prevPos) * float64(time.Second) / float64(time.Now().Sub(prevTime)) workerPool.currentSpeed = float64(workerPool.processedCount-prevPos) * float64(time.Second) / float64(time.Since(prevTime))
prevPos = workerPool.processedCount prevPos = workerPool.processedCount
prevTime = time.Now() prevTime = time.Now()
case err := <-workerPool.resultChan: case err := <-workerPool.resultChan:
@ -72,6 +77,7 @@ func New(threadCount int) *WorkerPool {
workerPool.errorCount++ workerPool.errorCount++
} }
workerPool.processedCount++ workerPool.processedCount++
workerPool.eta.Increment(1)
case <-workerPool.stopChan: case <-workerPool.stopChan:
return return
} }
@ -91,30 +97,35 @@ func New(threadCount int) *WorkerPool {
return workerPool return workerPool
} }
func (workerPool *WorkerPool) printProgress() { func (workerPool *WorkerPool) SetEstimateCount(n int) {
workerPool.estimateCount = n
workerPool.eta.TotalCount = n
}
func (workerPool *WorkerPool) printProgress() error {
if !workerPool.ShowProgress { if !workerPool.ShowProgress {
return return nil
} }
buf := new(bytes.Buffer) buf := new(bytes.Buffer)
fmt.Fprintf(buf, newLine) fmt.Fprint(buf, newLine)
if workerPool.EstimateCount == 0 { if workerPool.estimateCount == 0 {
fmt.Fprintf(buf, "Progress: %d", workerPool.processedCount) fmt.Fprintf(buf, "Progress: %d", workerPool.processedCount)
} else { } else {
fmt.Fprintf(buf, "Progress: %.1f%% (%d / %d)", fmt.Fprintf(buf, "Progress: %.1f%% (%d / %d)",
float64(workerPool.processedCount*100)/float64(workerPool.EstimateCount), workerPool.processedCount, workerPool.EstimateCount) float64(workerPool.processedCount*100)/float64(workerPool.estimateCount), workerPool.processedCount, workerPool.estimateCount)
} }
if workerPool.errorCount > 0 { if workerPool.errorCount > 0 {
fmt.Fprintf(buf, " Errors: %d (%.1f%%)", fmt.Fprintf(buf, " Errors: %d (%.1f%%)",
workerPool.errorCount, float64(workerPool.errorCount*100)/float64(workerPool.EstimateCount)) workerPool.errorCount, float64(workerPool.errorCount*100)/float64(workerPool.estimateCount))
} }
if workerPool.currentSpeed > 0 { if workerPool.currentSpeed > 0 {
if workerPool.EstimateCount > 0 { if workerPool.estimateCount > 0 {
fmt.Fprintf(buf, " ETA: %s", fmtDuration(time.Second*time.Duration(float64(workerPool.EstimateCount-workerPool.processedCount)/workerPool.currentSpeed))) fmt.Fprintf(buf, " ETA: %s", fmtDuration(time.Until(workerPool.eta.Average())))
} }
if workerPool.ShowSpeed { if workerPool.ShowSpeed {
@ -125,12 +136,17 @@ func (workerPool *WorkerPool) printProgress() {
fmt.Fprint(buf, endLine) fmt.Fprint(buf, endLine)
if buf.String() == workerPool.lastProgressMessage { if buf.String() == workerPool.lastProgressMessage {
return return nil
} }
workerPool.lastProgressMessage = buf.String() workerPool.lastProgressMessage = buf.String()
buf.WriteTo(os.Stderr) _, err := buf.WriteTo(os.Stderr)
if err != nil {
return err
}
return nil
} }
// Add sends specified task for execution. // Add sends specified task for execution.
@ -145,7 +161,7 @@ func (workerPool *WorkerPool) CloseAndWait() {
workerPool.stopChan <- struct{}{} workerPool.stopChan <- struct{}{}
close(workerPool.resultChan) close(workerPool.resultChan)
workerPool.printProgress() _ = workerPool.printProgress()
} }
// ErrorCount returns total error count. // ErrorCount returns total error count.