Use job with time and timeout

Question

I use the following sample code which works, now I want that each job be able to print the time it took to execute (it's better as generic that not each job will need to use the code of

start := time.Now()
took := time.Since(start).Milliseconds()

And also provide a timeout for a job, for example, if it takes more then 10 seconds to kill it or stop it.

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 {
           return job1()
        },
        func() MyReturnType {
           return job2()
        },
    }

    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))
}

Here is an example of what I try to avoid and to provide some general solution for printing time for each job:

func job1() {
   start := time.Now()
   ...
   // running some code
   took := time.Since(start).Milliseconds()
}

func job2(){
   start := time.Now()
   ...
   // running some code
   took := time.Since(start).Milliseconds()
}

Answer 1

Update

Scroll to section "This is the accepted answer" to view the accepted answer

I went ahead and wrote a little library from the code in the accepted answer...

You can find it here or the code below:

// How to use the library
package main

import (
    "fmt"
    "time"

    "github.com/oze4/reactor"
)

func main() {
    timeoutForJobs := time.Duration(time.Second * 10)
    numOfWorkers := 10

    myreactor := reactor.New(numOfWorkers, timeoutForJobs)

    // You can also create a Reactor with a custom Client
    // myreactor := reactor.NewWithClient(numOfWorkers, timeoutForJobs, &reactor.Client{...})

    // Add job(s)
    myreactor.Add(reactor.Job{
        Name: "job1",
        Runner: func(c *reactor.Client) reactor.React {
            // do something with client `c`
            res, _ := c.HTTP.Get("xyz.com")
            return reactor.React{Info: res}
        },
    })

    // All results will be here
    results := myreactor.GetResults()
    
    for _, result := range results {
        fmt.Println(result)
    }
}

Code for library

// Library code
package reactor

import (
    "context"
    "net/http"
    "time"

    "github.com/gammazero/workerpool"
    "k8s.io/client-go/kubernetes"
)

// New creates a new Reactor
func New(maxWorkers int, jobTimeout time.Duration) Reactor {
    // Do whatever you need to here to create default client
    defaultClient := &Client{
        HTTP:       http.Client{},
        Kubernetes: kubernetes.Clientset{},
    }

    return &reactor{
        workerPool:  workerpool.New(maxWorkers),
        jobTimeout:  jobTimeout,
        transport:   defaultClient,
        resultsChan: make(chan React, 100),
    }
}

// NewWithClient creates a new Reactor with a custom client
func NewWithClient(client *Client, maxWorkers int, jobTimeout time.Duration) Reactor {
    return &reactor{
        workerPool:  workerpool.New(maxWorkers),
        jobTimeout:  jobTimeout,
        transport:   client,
        resultsChan: make(chan React, 100),
    }
}

// Reactor knows how to handle jobs
type Reactor interface {
    Add(j Job)                          // Add puts a job on the queue
    Client() *Client                    // I dont know if you want the consumer to have access to this or not
    GetResults() []React                // Get results
    Timeout() time.Duration             // I dont know if you want the consumer to have access to this or not
    WorkerPool() *workerpool.WorkerPool // I dont know if you want the consumer to have access to this or not
}

type reactor struct {
    jobTimeout  time.Duration
    workerPool  *workerpool.WorkerPool
    resultsChan chan React
    transport   *Client
}

// Add submits a job
func (r *reactor) Add(j Job) {
    r.workerPool.Submit(r.wrapper(j))
}

// I dont know if you want the consumer to have access to this or not
func (r *reactor) Client() *Client {
    return r.transport
}

// Get results gets results
func (r *reactor) GetResults() []React {
    return r.getResults()
}

func (r *reactor) getResults() []React {
    r.workerPool.StopWait()
    close(r.resultsChan)

    var allReacts []React
    for jobreact := range r.resultsChan {
        allReacts = append(allReacts, jobreact)
    }

    return allReacts
}

func (r *reactor) Timeout() time.Duration {
    return r.jobTimeout
}

// I dont know if you want the consumer to have access to this or not
func (r *reactor) WorkerPool() *workerpool.WorkerPool {
    return r.workerPool
}

// worker should be private
func (r *reactor) worker(ctx context.Context, done context.CancelFunc, job Job, start time.Time) {
    runner := job.Runner(r.transport)
    runner.duration = time.Since(start)
    runner.name = job.Name

    if ctx.Err() == nil {
        r.resultsChan <- runner
    }

    done()
}

// wrapper should be private
func (r *reactor) wrapper(job Job) func() {
    ctx, cancel := context.WithTimeout(context.Background(), r.jobTimeout)

    return func() {
        start := time.Now()
        go r.worker(ctx, cancel, job, start)

        select {
        case <-ctx.Done():
            switch ctx.Err() {
            case context.DeadlineExceeded:
                r.resultsChan <- React{
                    Error:    context.DeadlineExceeded,
                    name:     job.Name,
                    duration: time.Since(start),
                }
            }
        }
    }
}

// React holds response data
type React struct {
    // This should be public so the consumer can set it
    Info  interface{}
    Error error

    // These fields should be private and handled via public methods
    duration time.Duration
    name     string
}

// Duration returns duration
func (r *React) Duration() time.Duration {
    return r.duration
}

// Name returns the job name
func (r *React) Name() string {
    return r.name
}

// Client holds http and kubernetes clients
type Client struct {
    HTTP       http.Client
    Kubernetes kubernetes.Clientset
}

// Job holds job data
type Job struct {
    Name   string
    Runner func(*Client) React
}

---

This is the accepted answer

The following example shows how you can gather execution time as well as set a timeout..

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/gammazero/workerpool"
)

var (
    //
    // Set timeout for all jobs here
    //
    jobTimeout = time.Duration(time.Second * 1)
)

// MyReturnType could be anything you want it to be
type MyReturnType struct {
    name              string
    Data              interface{}
    Error             error
    ExecutionDuration time.Duration
}

// Name returns name. It is written like this so the consumer
// cannot change the name outside of supplying one via the Job
func (m *MyReturnType) Name() string {
    return m.name
}

// Job holds job data
type Job struct {
    Name string
    Task func() MyReturnType
}

func wrapJob(timeout time.Duration, resultsChan chan MyReturnType, job Job) func() {
    timeoutContext, timeoutCancel := context.WithTimeout(context.Background(), timeout)

    return func() {
        timerStart := time.Now()
        go func(ctx context.Context, done context.CancelFunc, resChan chan MyReturnType, todo Job, startTime time.Time) {
            result := todo.Task()
            result.ExecutionDuration = time.Since(startTime)
            result.name = todo.Name
            if timeoutContext.Err() == nil {
                resChan <- result
            }
            done()
        }(timeoutContext, timeoutCancel, resultsChan, job, timerStart)

        select {
        case <-timeoutContext.Done():
            switch timeoutContext.Err() {
            case context.DeadlineExceeded:
                resultsChan <- MyReturnType{
                    name:              job.Name,
                    Error:             context.DeadlineExceeded,
                    ExecutionDuration: time.Since(timerStart),
                }
            }
        }
    }
}

func main() {
    jobs := []Job{
        {
            Name: "job1",
            Task: func() MyReturnType {
                // This will surpass our timeout and should get cancelled
                time.Sleep(time.Second * 3)
                // Don't have to set the name here
                return MyReturnType{Data: map[string]string{"Whatever": "You want"}}
            },
        },
        {
            Name: "job2",
            Task: func() MyReturnType {
                // This job will succeed
                time.Sleep(time.Millisecond * 300)
                resultFromCurl := "i am a result"
                return MyReturnType{Data: resultFromCurl}
            },
        },
    }

    jobResultsChannel := make(chan MyReturnType, len(jobs))
    pool := workerpool.New(10)

    for _, job := range jobs {
        pool.Submit(wrapJob(jobTimeout, jobResultsChannel, job))
    }

    pool.StopWait()
    close(jobResultsChannel)

    // Do whatever you want with results
    for jobResult := range jobResultsChannel {
        if jobResult.Error != nil {
            fmt.Printf("[took '%d' ms] '%s' : JobError : %s\n", jobResult.ExecutionDuration, jobResult.Name(), jobResult.Error)
        } else {
            fmt.Printf("[took '%d' ms] '%s' : JobSuccess : %s\n", jobResult.ExecutionDuration, jobResult.Name(), jobResult.Data)
        }
    }
}

Which returns:

// [took '305182398' ms] 'job2' : JobSuccess : i am a result
// [took '1001045539' ms] 'job1' : JobError : context deadline exceeded

---

Original Answer

You should be able to use context for timeouts/cancellation (as Peter mentioned).

As far as recording execution time, you could do what you stated in your comment, or something like this:

package main

import (
    "fmt"
    "time"

    "github.com/gammazero/workerpool"
)

type MyReturnType struct {
    Name string
    Data interface{}
    Time time.Duration
}

func wrapJob(rc chan MyReturnType, f func() MyReturnType) func() {
    return func() {
        start := time.Now()
        result := f()
        result.Time = time.Since(start)
        rc <- result
    }
}

func main() {
    jobs := []func() MyReturnType{
        func() MyReturnType {
            time.Sleep(time.Millisecond*400)
            return MyReturnType{Name: "job1", Data: map[string]string{"Whatever": "You want"}}
        },
        func() MyReturnType {
            resultFromCurl := "i am a result"
            return MyReturnType{Name: "job2", Data: resultFromCurl}
        },
    }

    results := make(chan MyReturnType, len(jobs))
    pool := workerpool.New(10)

    for _, job := range jobs {
        j := job
        pool.Submit(wrapJob(results, j))
    }

    pool.StopWait()
    close(results)

    for res := range results {
        fmt.Printf("[took '%d' ms] ", res.Time)
        fmt.Println(res)
    }
}