Update 5 (accepted answer)
*Since you are interested in this question, you may also be interested in this one as well. See here for more on how to run each job with an auto-cancelling timeout.*
To answer your question how would you add random functions..
I have no idea what type you are wanting to return, but you can do anything you wanted.
There are about a million different ways you could do this, this is just one example:
package main
import (
"encoding/json"
"fmt"
"github.com/gammazero/workerpool"
)
var (
numWorkers = 10
)
type MyReturnType struct {
Name string
Data interface{}
}
func wrapJob(rc chan MyReturnType, f func() MyReturnType) func() {
return func() {
rc <- f()
}
}
func main() {
// create results chan and worker pool
// should prob make your results channel typed to what you need
jobs := []func() MyReturnType {
func() MyReturnType {
// whatever you want to do here
return MyReturnType{Name: "job1", Data: map[string]string{"Whatever": "You want"}}
},
func() MyReturnType {
// whatever you want to do here
// do a curl or a kubectl or whatever you want
resultFromCurl := "i am a result"
return MyReturnType{Name: "job2", Data: resultFromCurl}
},
}
results := make(chan MyReturnType, len(jobs))
pool := workerpool.New(numWorkers)
for _, job := range jobs {
j := job
pool.Submit(wrapJob(results, j))
}
// Wait for all jobs to finish
pool.StopWait()
// Close results chan
close(results)
// Iterate over results, printing to console
for res := range results {
prettyPrint(res)
}
}
func prettyPrint(i interface{}) {
prettyJSON, err := json.MarshalIndent(i, "", " ")
if err != nil {
fmt.Printf("Error : %s \n", err.Error())
}
fmt.Printf("MyReturnType %s\n", string(prettyJSON))
}
Which returns:
// MyReturnType {
// "Name": "job2",
// "Data": "i am a result"
// }
// MyReturnType {
// "Name": "job1",
// "Data": {
// "Whatever": "You want"
// }
// }
Update 4
After digging into this for a number of hours, I would recommend using something like workerpool, which you can find here. Honestly, using workerpool seems to make the most sense here.. It looks production ready and is used by a handful of rather big names (see the readme in their repo).
I wrote a little example that shows how you can use workerpool:
package main
import (
"fmt"
"net/http"
"time"
"github.com/gammazero/workerpool"
)
var (
numWorkers = 10
urls = []string{"yahoo.com", "example.com", "google.com"}
)
func main() {
// create results chan and worker pool
// should prob make your results channel typed to what you need
results := make(chan interface{}, len(urls))
pool := workerpool.New(numWorkers)
// Create jobs by iterating over urls
for i, u := range urls {
url := u
jobNum := i
// Create job
f := func() {
start := time.Now()
c := &http.Client{}
r, e := c.Get("http://" + url)
if e != nil {
fmt.Println(e.Error())
}
took := time.Since(start).Milliseconds()
o := fmt.Sprintf("completed job '%d' to '%s' in '%dms' with status code '%d'\n", jobNum, url, took, r.StatusCode)
results <- o
}
// Add job to workerpool
pool.Submit(f)
}
// Wait for all jobs to finish
pool.StopWait()
// Close results chan
close(results)
// Iterate over results, printing to console
for res := range results {
fmt.Printf(res.(string))
}
}
Which outputs:
// completed job '1' to 'example.com' in '81ms' with status code '200'
// completed job '2' to 'google.com' in '249ms' with status code '200'
// completed job '0' to 'yahoo.com' in '816ms' with status code '200'
Update 3
I went ahead and wrote a worker pool library (with the help of workerpool) as I also wanted to dive deeper into channels and concurrency design.
You can find the repo here and the code below.
How to use:
pool := New(3)
pool.Job(func() {
c := &http.Client{}
r, e := c.Get("http://google.com")
if e != nil {
panic(e.Error())
}
fmt.Printf("To google.com %d\n", r.StatusCode)
})
pool.Job(func() {
c := &http.Client{}
r, e := c.Get("http://yahoo.com")
if e != nil {
panic(e.Error())
}
fmt.Printf("To yahoo.com %d\n", r.StatusCode)
})
pool.Job(func() {
c := &http.Client{}
r, e := c.Get("http://example.com")
if e != nil {
panic(e.Error())
}
fmt.Printf("To example.com %d\n", r.StatusCode)
})
pool.Seal()
Code for worker pool (puddle)
package puddle
import (
"container/list"
"fmt"
"net/http"
"sync"
"time"
)
const (
idleTimeout = time.Second * 2
)
// New creates a new puddle (aka worker pool)
func New(maxWorkers int) Puddle {
// There must be at least one worker
if maxWorkers < 1 {
maxWorkers = 1
}
p := &puddle{
maxWorkers: maxWorkers,
jobs: make(chan func(), 1),
workers: make(chan func()),
killswitch: make(chan struct{}),
}
// Start accepting/working jobs as they come in
go p.serve()
return p
}
// Puddle knows how to interact with worker pools
type Puddle interface {
Job(f func())
Seal()
}
// puddle is a worker pool that holds workers, tasks, and misc metadata
type puddle struct {
maxWorkers int
jobs chan func()
workers chan func()
killswitch chan struct{}
queue List
once sync.Once
stopped int32
waiting int32
wait bool
}
// Job submits a new task to the worker pool
func (p *puddle) Job(f func()) {
if f != nil {
p.jobs <- f
}
}
// Seal stops worker pool and waits for queued tasks to complete
func (p *puddle) Seal() {
p.stop(true)
}
func (p *puddle) stop(wait bool) {
p.once.Do(func() {
p.wait = wait
// Close task queue and wait for currently running tasks to finish
close(p.jobs)
})
<-p.killswitch
}
func (p *puddle) killWorkerIfIdle() bool {
select {
case p.workers <- nil:
// Kill worker
return true
default:
// No ready workers
return false
}
}
// process puts new jobs onto the queue, and removes jobs from the queue as workers become available.
// Returns false if puddle is stopped.
func (p *puddle) process() bool {
select {
case task, ok := <-p.jobs:
if !ok {
return false
}
p.queue.PushBack(task)
case p.workers <- p.queue.Front().Value.(func()):
// Give task to ready worker
p.queue.PopFront()
}
return true
}
func (p *puddle) serve() {
defer close(p.killswitch)
timeout := time.NewTimer(idleTimeout)
var workerCount int
var idle bool
Serving:
for {
if p.queue.Len() != 0 {
if !p.process() {
break Serving
}
continue
}
select {
case job, ok := <-p.jobs:
if !ok {
break Serving
}
// Give a task to our workers
select {
case p.workers <- job:
default:
// If we are not maxed on workers, create a new one
if workerCount < p.maxWorkers {
go startJob(job, p.workers)
workerCount++
} else {
// Place a task on the back of the queue
p.queue.PushBack(job)
}
}
idle = false
case <-timeout.C:
// Timed out waiting for work to arrive. Kill a ready worker if
// pool has been idle for a whole timeout.
if idle && workerCount > 0 {
if p.killWorkerIfIdle() {
workerCount--
}
}
idle = true
timeout.Reset(idleTimeout)
}
}
// Allow queued jobs to complete
if p.wait {
p.work()
}
// Stop all workers before shutting down
for workerCount > 0 {
p.workers <- nil
workerCount--
}
timeout.Stop()
}
// work removes each task from the waiting queue and gives it to
// workers until queue is empty.
func (p *puddle) work() {
for p.queue.Len() != 0 {
// A worker is ready, so give task to worker.
p.workers <- p.queue.PopFront()
}
}
// startJob runs initial task, then starts a worker waiting for more.
func startJob(job func(), workerQueue chan func()) {
job()
go worker(workerQueue)
}
// worker executes tasks and stops when it receives a nil task.
func worker(queue chan func()) {
for job := range queue {
if job == nil {
return
}
job()
}
}
// List wraps `container/list`
type List struct {
list.List
}
// PopFront removes then returns first element in list as func()
func (l *List) PopFront() func() {
f := l.Front()
l.Remove(f)
return f.Value.(func())
}
Update 2
Since you were asking about how to use the code, this is how you would do so.
I turned worker into it's own package, as well as wrote another repo to show how to use that package.
worker package
package worker
import "fmt"
type JobResponse struct {
err error
name string
res int
url string
}
type Job interface {
Name() string
Callback() JobResponse
}
func Do(jobs []Job, maxWorkers int) {
jobsPool := make(chan Job, len(jobs))
resultsPool := make(chan JobResponse, len(jobs))
for i := 0; i < maxWorkers; i++ {
go worker(jobsPool, resultsPool)
}
makeJobs(jobsPool, jobs)
getResults(resultsPool, jobs)
}
func worker(jobs <-chan Job, response chan<- JobResponse) {
for n := range jobs {
response <- n.Callback()
}
}
func makeJobs(jobs chan<- Job, queue []Job) {
for _, t := range queue {
jobs <- t
}
}
func getResults(response <-chan JobResponse, queue []Job) {
for range queue {
job := <-response
status := fmt.Sprintf("[result] '%s' to '%s' was fetched with status '%d'\n", job.name, job.url, job.res)
if job.err != nil {
status = fmt.Sprintf(job.err.Error())
}
fmt.Printf(status)
}
}
How to use worker package
package main
import (
"github.com/oze4/worker"
)
func main() {
jobs := []worker.Job{
AddedByUser{name: "1"},
AddedByUser{name: "2"},
AddedByUser{name: "3"},
AddedByUser{name: "4"},
AddedByUser{name: "5"},
AddedByUser{name: "6"},
}
worker.Do(jobs, 5)
}
type AddedByUser struct {
name string
}
func (a AddedByUser) Name() string {
return a.name
}
func (a AddedByUser) Callback() worker.JobResponse {
// User added func/callback goes here
return worker.JobResponse{}
}
Update
I renamed some things to hopefully help make it a little more clear.
This is the basics of what you need:
package main
import (
"fmt"
)
func main() {
fmt.Println("Hello, playground")
}
type JobResponse struct {
err error
name string
res int
url string
}
type Job interface {
Name() string
Callback() JobResponse
}
func worker(jobs <-chan Job, response chan<- JobResponse) {
for n := range jobs {
response <- n.Callback()
}
}
func makeJobs(jobs chan<- Job, queue []Job) {
for _, t := range queue {
jobs <- t
}
}
func getResults(response <-chan JobResponse, queue []Job) {
for range queue {
j := <-response
status := fmt.Sprintf("[result] '%s' to '%s' was fetched with status '%d'\n", j.name, j.url, j.res)
if j.err != nil {
status = fmt.Sprintf(j.err.Error())
}
fmt.Printf(status)
}
}
As long as I satisfy the Job interface I can pass it into worker, makeJobs, and getResults:
type AddedByUser struct {
name string
}
func (a AddedByUser) Name() string {
return a.name
}
func (a AddedByUser) Callback() JobResponse {
// User added func/callback goes here
return JobResponse{}
}
Like so:
package main
import (
"fmt"
)
func main() {
jobsPool := make(chan Job, len(testers))
resultsPool := make(chan JobResponse, len(testers))
maxWorkers := 5
for i := 0; i < maxWorkers; i++ {
go worker(jobsPool, resultsPool)
}
makeJobs(jobsPool, testers)
getResults(resultsPool, testers)
}
var testers = []Job{
AddedByUser{name: "abu"}, // Using different types in Job
Tap{name: "tap"}, // Using different types in Job
}
type AddedByUser struct {
name string
}
func (a AddedByUser) Name() string {
return a.name
}
func (a AddedByUser) Callback() JobResponse {
// User added func/callback goes here
return JobResponse{}
}
type Tap struct {
name string
}
func (t Tap) Name() string {
return t.name
}
func (t Tap) Callback() JobResponse {
// User added func/callback goes here
return JobResponse{}
}
type JobResponse struct {
err error
name string
res int
url string
}
type Job interface {
Name() string
Callback() JobResponse
}
func worker(jobs <-chan Job, response chan<- JobResponse) {
for n := range jobs {
response <- n.Callback()
}
}
func makeJobs(jobs chan<- Job, queue []Job) {
for _, t := range queue {
jobs <- t
}
}
func getResults(response <-chan JobResponse, queue []Job) {
for range queue {
job := <-response
status := fmt.Sprintf("[result] '%s' to '%s' was fetched with status '%d'\n", job.name, job.url, job.res)
if job.err != nil {
status = fmt.Sprintf(job.err.Error())
}
fmt.Printf(status)
}
}
Original Answer
[adding this answer because OP and I have been talking outside of this thread]
You had a couple errors in your code, but at the end of the day all you had to do was take the advice people were giving you. You just had to connect the dots. I suggest troubleshooting your code and trying to fully understand what the issue is. That is honestly the only way to learn.
The biggest issues I can remember were:
- Needed to modify your
HT interface so the Check(...) signature matched each method
- Otherwise, those structs (
Tap, Tap1, Tap2) do not satisfy the HT interface and therefore do not implement HT
- The param type(s) in funcs
worker(...), makeJobs(...), and getResults(...) were changed from []Tap to []HT
- You were not aggregating all Taps into one slice
- The only reason we can use all different Taps as HT is because they all implement HT
Is something like this what you're looking for?
https://play.golang.org/p/zLmKOKAnX4C
package main
import (
"fmt"
"net/http"
// "os/exec"
"time"
)
type HT interface {
Name() string
Check() testerResponse
}
type testerResponse struct {
err error
name string
//res http.Response
res int
url string
}
type Tap struct {
url string
name string
timeout time.Duration
client *http.Client
}
func (p *Tap) Check() testerResponse {
fmt.Printf("[job][Tap1] Fetching %s %s \n", p.name, p.url)
p.client = &http.Client{Timeout: p.timeout}
res, err := p.client.Get(p.url)
if err != nil {
return testerResponse{err: err}
}
// need to close body
res.Body.Close()
return testerResponse{name: p.name, res: res.StatusCode, url: p.url}
}
func (p *Tap) Name() string {
return p.name
}
// ---- CUSTOM CHECKS-------------
// ---- 1. NEW specific function -------------
type Tap2 struct {
url string
name string
timeout time.Duration
client *http.Client
}
func (p *Tap2) Check() testerResponse {
// Do some request here.....
fmt.Printf("[job][Tap2] Fetching %s %s \n", p.name, p.url)
return testerResponse{res: 200, url: p.url, name: p.name}
}
func (p *Tap2) Name() string {
return "yahoo custom check"
}
// ---- 2. NEW specific function which is not running http
type Tap3 struct {
url string
name string
timeout time.Duration
client *http.Client
}
func (p *Tap3) Check() testerResponse {
// Do some request here....
fmt.Printf("[job][Tap3] Fetching %s %s \n", p.name, p.url)
return testerResponse{res: 200, url: p.url, name: p.name}
}
func (p *Tap3) Name() string {
return "custom check2"
}
// makeJobs fills up our jobs channel
func makeJobs(jch chan<- HT, jobs []HT) {
for _, t := range jobs {
jch <- t
}
}
// getResults takes a job from our jobs channel, gets the result, and
// places it on the results channel
func getResults(tr <-chan testerResponse, jobs []HT) []testerResponse {
var rts []testerResponse
var r testerResponse
for range jobs {
r = <-tr
status := fmt.Sprintf("[result] '%s' to '%s' was fetched with status '%d'\n", r.name, r.url, r.res)
if r.err != nil {
status = fmt.Sprintf(r.err.Error())
}
fmt.Printf(status)
rts = append(rts, r)
}
return rts
}
// worker defines our worker func. as long as there is a job in the
// "queue" we continue to pick up the "next" job
func worker(jobs <-chan HT, results chan<- testerResponse) {
for n := range jobs {
results <- n.Check()
}
}
var (
testers1 = []Tap{
{
name: "First Tap1",
url: "http://google.com",
timeout: time.Second * 20,
},
{
name: "Second Tap1",
url: "http://stackoverflow.com",
timeout: time.Second * 20,
},
}
testers2 = []Tap2{
{
name: "First Tap2",
url: "http://1.tap2.com",
},
{
name: "Second Tap2",
url: "http://2.tap2.com",
},
}
testers3 = []Tap3{
{
name: "First Tap3",
url: "http://1.tap3.com",
},
{
name: "Second Tap3",
url: "http://2.tap3.com",
},
}
)
func main() {
// Aggregate all testers into one slice
var testers []HT
for _, t1 := range testers1 {
testers = append(testers, &t1)
}
for _, t2 := range testers2 {
testers = append(testers, &t2)
}
for _, t3 := range testers3 {
testers = append(testers, &t3)
}
// Make buffered channels
buffer := len(testers)
jobsPipe := make(chan HT, buffer) // Jobs will be of type `HT`
resultsPipe := make(chan testerResponse, buffer) // Results will be of type `testerResponse`
// Create worker pool
// Max workers default is 5
maxWorkers := 5
for i := 0; i < maxWorkers; i++ {
go worker(jobsPipe, resultsPipe)
}
makeJobs(jobsPipe, testers)
getResults(resultsPipe, testers)
//fmt.Println("at the end",tr)
}
Which outputs:
// [job][Tap1] Fetching Second Tap1 http://stackoverflow.com
// [job][Tap2] Fetching Second Tap2 http://2.tap2.com
// [job][Tap3] Fetching Second Tap3 http://2.tap3.com
// [job][Tap3] Fetching Second Tap3 http://2.tap3.com
// [result] 'Second Tap2' to 'http://2.tap2.com' was fetched with status '200'
// [result] 'Second Tap3' to 'http://2.tap3.com' was fetched with status '200'
// [result] 'Second Tap3' to 'http://2.tap3.com' was fetched with status '200'
// [job][Tap2] Fetching Second Tap2 http://2.tap2.com
// [job][Tap1] Fetching Second Tap1 http://stackoverflow.com
// [result] 'Second Tap2' to 'http://2.tap2.com' was fetched with status '200'
// [result] 'Second Tap1' to 'http://stackoverflow.com' was fetched with status '200'
// [result] 'Second Tap1' to 'http://stackoverflow.com' was fetched with status '200'
