Usa il lavoro con tempo e timeout

Question

Uso il seguente codice di esempio che funziona, ora voglio che ogni lavoro sia in grado di stampare il tempo impiegato per l'esecuzione (è meglio come generico che non ogni lavoro avrà bisogno di usare il codice di

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

Fornisci anche un timeout per un lavoro, ad esempio, se sono necessari più di 10 secondi per ucciderlo o interromperlo.

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

Ecco un esempio di ciò che cerco di evitare e di fornire alcune soluzioni generali per il tempo di stampa per ogni lavoro:

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

Aggiornare

Scorri fino alla sezione "Questa è la risposta accettata" per visualizzare la risposta accettata

Sono andato avanti e ho scritto una piccola libreria dal codice nella risposta accettata ...

Puoi trovarlo qui o il codice qui sotto:

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

Codice per la libreria

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

---

Questa è la risposta accettata

L'esempio seguente mostra come raccogliere il tempo di esecuzione e impostare un 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)
        }
    }
}

Che ritorna:

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

---

Risposta originale

Dovresti essere in grado di utilizzare il contesto per timeout / cancellazione ( come menzionato da Peter ).

Per quanto riguarda il tempo di esecuzione della registrazione, potresti fare quello che hai dichiarato nel tuo commento , o qualcosa del genere:

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