kelvin/warewulf
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #6 from ctrliq/batch-update

Browse Source 
Rework batch package
This commit is contained in:
Gregory M. Kurtzer
2020-12-18 21:34:15 -08:00
committed by GitHub
parent 6f8ec8797b c3281ec333
commit 6e4469ab17
7 changed files with 47 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
internal/app/wwctl/node/powercycle/power.go
View File

@@ -1,13 +1,14 @@
package powercycle
import (
"fmt"
"os"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/hpcng/warewulf/internal/pkg/node"
"github.com/hpcng/warewulf/internal/pkg/power"
"github.com/hpcng/warewulf/internal/pkg/wwlog"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/spf13/cobra"
"os"
"fmt"
)
func CobraRunE(cmd *cobra.Command, args []string) error {
@@ -32,7 +33,7 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
os.Exit(255)
}
batchpool := batch.New(50, 0)
batchpool := batch.New(50)
jobcount := len(nodeList)
results := make(chan power.IPMI, jobcount)
@@ -78,4 +79,3 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
return returnErr
}

10
internal/app/wwctl/node/poweroff/power.go
View File

@@ -1,13 +1,14 @@
package poweroff
import (
"fmt"
"os"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/hpcng/warewulf/internal/pkg/node"
"github.com/hpcng/warewulf/internal/pkg/power"
"github.com/hpcng/warewulf/internal/pkg/wwlog"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/spf13/cobra"
"os"
"fmt"
)
func CobraRunE(cmd *cobra.Command, args []string) error {
@@ -32,7 +33,7 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
os.Exit(255)
}
batchpool := batch.New(50, 0)
batchpool := batch.New(50)
jobcount := len(nodeList)
results := make(chan power.IPMI, jobcount)
@@ -78,4 +79,3 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
return returnErr
}

10
internal/app/wwctl/node/poweron/power.go
View File

@@ -1,13 +1,14 @@
package poweron
import (
"fmt"
"os"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/hpcng/warewulf/internal/pkg/node"
"github.com/hpcng/warewulf/internal/pkg/power"
"github.com/hpcng/warewulf/internal/pkg/wwlog"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/spf13/cobra"
"os"
"fmt"
)
func CobraRunE(cmd *cobra.Command, args []string) error {
@@ -34,7 +35,7 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
wwlog.Printf(wwlog.VERBOSE, "Found %d matching nodes for power command\n", len(nodeList))
}
batchpool := batch.New(50, 0)
batchpool := batch.New(50)
jobcount := len(nodeList)
results := make(chan power.IPMI, jobcount)
@@ -80,4 +81,3 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
return returnErr
}

10
internal/app/wwctl/node/powerstatus/power.go
View File

@@ -1,13 +1,14 @@
package powerstatus
import (
"fmt"
"os"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/hpcng/warewulf/internal/pkg/node"
"github.com/hpcng/warewulf/internal/pkg/power"
"github.com/hpcng/warewulf/internal/pkg/wwlog"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/spf13/cobra"
"os"
"fmt"
)
func CobraRunE(cmd *cobra.Command, args []string) error {
@@ -34,7 +35,7 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
wwlog.Printf(wwlog.VERBOSE, "Found %d matching nodes for power command\n", len(nodeList))
}
batchpool := batch.New(50, 0)
batchpool := batch.New(50)
jobcount := len(nodeList)
results := make(chan power.IPMI, jobcount)
@@ -80,4 +81,3 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
return returnErr
}

10
internal/app/wwctl/node/sensors/power.go
View File

@@ -1,13 +1,14 @@
package sensors
import (
"fmt"
"os"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/hpcng/warewulf/internal/pkg/node"
"github.com/hpcng/warewulf/internal/pkg/power"
"github.com/hpcng/warewulf/internal/pkg/wwlog"
"github.com/hpcng/warewulf/internal/pkg/batch"
"github.com/spf13/cobra"
"os"
"fmt"
)
func CobraRunE(cmd *cobra.Command, args []string) error {
@@ -32,7 +33,7 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
os.Exit(255)
}
batchpool := batch.New(50, 0)
batchpool := batch.New(50)
jobcount := len(nodeList)
results := make(chan power.IPMI, jobcount)
@@ -84,4 +85,3 @@ func CobraRunE(cmd *cobra.Command, args []string) error {
return returnErr
}

35
internal/pkg/batch/batch.go
View File

@@ -1,22 +1,17 @@
package batch
import (
"time"
"sync"
)
type BatchPool struct {
active int
mintime int
jobcount int
jobs []func()
jobs []func()
}
func New(active int, mintime int) *BatchPool {
func New(active int) *BatchPool {
pool := &BatchPool{
active: active,
mintime: mintime,
}
pool.jobs = []func(){}
return pool
@@ -30,28 +25,24 @@ func Min(x, y int) int {
}
func (pool *BatchPool) Submit(f func()) {
pool.jobcount++
pool.jobs = append(pool.jobs, f)
}
func wrapper(wg *sync.WaitGroup, f func()) {
defer wg.Done()
func wrapper(wg *sync.WaitGroup, limiter chan struct{}, f func()) {
f()
<-limiter
wg.Done()
}
func (pool *BatchPool) Run() {
var wg sync.WaitGroup
count := pool.jobcount
for count > 0 {
for i:=0; i<Min(count, pool.active); i++ {
wg.Add(1)
go wrapper(&wg, pool.jobs[pool.jobcount-count])
count--
}
if (pool.mintime > 0) {
time.Sleep(time.Second * time.Duration(pool.mintime))
}
wg.Wait()
limiter := make(chan struct{}, pool.active)
for i := range pool.jobs {
limiter <- struct{}{}
wg.Add(1)
go wrapper(&wg, limiter, pool.jobs[i])
}
}
wg.Wait()
close(limiter)
}

21
test/batch-test.go
View File

@@ -1,37 +1,35 @@
package main
import (
"github.com/hpcng/warewulf/internal/pkg/batch"
"fmt"
"time"
"runtime"
"time"
"github.com/hpcng/warewulf/internal/pkg/batch"
)
func exampleJob(name string, id int) string {
fmt.Printf("[%d] %s started...\n", id, name)
time.Sleep(time.Second * 5)
time.Sleep(time.Second * time.Duration(id))
fmt.Printf("[%d] %s ending...\n", id, name)
return name
}
func main() {
fmt.Printf("GOMAXPROCS=%d\n", runtime.GOMAXPROCS(0))
jobstorun := 1000
jobstorun := 100
// create batch pool to run 100 jobs at a time every 10 seconds
batchpool := batch.New(100, 10)
batchpool := batch.New(10)
retvalues := make(chan string, jobstorun)
// submit a bunch of jobs
for i:=0; i<jobstorun; i++ {
for i := 0; i < jobstorun; i++ {
name := fmt.Sprintf("job-%04d", i)
id := i;
id := i
batchpool.Submit(func() {
retvalues <- exampleJob(name, id)
})
@@ -43,9 +41,8 @@ func main() {
close(retvalues)
for s:= range retvalues {
for s := range retvalues {
fmt.Printf("Return value is %s\n", s)
}
}