Go routine- stops running process

Question

I use the following code which works ok in most cases, in case we are using some long running process which is not stops inside the program doesn't end (here I limit to 60 seconds for the example)

I want that each job will be terminated(kill the process even if it's not done the work) after 5 seconds, How can I do it without changing the function myLongRunningFunc.

I know that this is not straight-forward to solve it in go, is there any trick that I can use?

This is some minimal reproducible example

https://play.golang.org/p/a0RWY4bYWMt

package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    "github.com/gammazero/workerpool"
)

func main() {
    // here define a timeout for 5 sec,
    // the task should be terminate after 5 sec
    ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
    defer cancel()

    runner := newRunner(ctx, 10)

    runner.do(job{
        Name: "a",
        Task: func() jobResult {
            select {
            case <-ctx.Done():
                return jobResult{Error: errors.New("Timedout, exiting")}
            default:
                myLongRunningFunc("A job")
            }
            return jobResult{Data: "from a"}
        },
    })

    runner.do(job{
        Name: "b",
        Task: func() jobResult {
            select {
            case <-ctx.Done():
                return jobResult{Error: errors.New("Timeouts, exiting")}
            default:
                myLongRunningFunc("B job")
            }

            return jobResult{Data: "from b"}
        },
    })

    results := runner.getjobResults()
    fmt.Println(results)
    time.Sleep(time.Second * 60)
}

func myLongRunningFunc(name string) {
    for i := 0; i < 100000; i++ {
        time.Sleep(time.Second * 1)
        msg := "job" + name + " running..\n"
        fmt.Println(msg)
    }
}

type runner struct {
    *workerpool.WorkerPool
    ctx     context.Context
    kill    chan struct{}
    result  chan jobResult
    results []jobResult
}

func (r *runner) processResults() {
    for {
        select {
        case res, ok := <-r.result:
            if !ok {
                goto Done
            }
            r.results = append(r.results, res)
        }
    }
Done:
    <-r.kill
}

func newRunner(ctx context.Context, numRunners int) *runner {
    r := &runner{
        WorkerPool: workerpool.New(numRunners),
        ctx:        ctx,
        kill:       make(chan struct{}),
        result:     make(chan jobResult),
    }
    go r.processResults()
    return r
}

func (r *runner) do(j job) {
    r.Submit(r.wrap(&j))
}

func (r *runner) getjobResults() []jobResult {
    r.StopWait()
    close(r.result)
    r.kill <- struct{}{}
    return r.results
}

func (r *runner) wrap(job *job) func() {
    return func() {
        job.result = make(chan jobResult)
        go job.Run()
        select {
        case res := <-job.result:
            r.result <- res
        case <-r.ctx.Done():
            fmt.Printf("Job '%s' should stop here\n", job.Name)
            r.result <- jobResult{name: job.Name, Error: r.ctx.Err()}
        }
    }
}

type job struct {
    Name    string
    Task    func() jobResult
    Context context.Context
    result  chan jobResult
    stopped chan struct{}
    done    context.CancelFunc
}

func (j *job) Run() {
    result := j.Task()
    result.name = j.Name
    j.result <- result
}

type jobResult struct {
    name  string
    Error error
    Data  interface{}
}

As I use chancel channel the edit is not related

@DavidMaze - yes but not sure how to adopt my code to use it in a clean way, could you please provide example? — JME, Nov 03 '20 at 13:57

score 2 · Accepted Answer · answered Nov 03 '20 at 12:29

I want that each job will be terminated(kill the process even if it's not done the work) after 5 seconds, How can I do it without changing the function myLongRunningFunc.

Then you just add a waiter of 5 seconds then exit.

package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    "github.com/gammazero/workerpool"
)

func main() {
    go func() {
        // here define a timeout for 5 sec,
        // the task should be terminate after 5 sec
        ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
        defer cancel()

        runner := newRunner(ctx, 10)

        runner.do(job{
            Name: "a",
            Task: func() jobResult {
                select {
                case <-ctx.Done():
                    return jobResult{Error: errors.New("Timedout, exiting")}
                default:
                    myLongRunningFunc("A job")
                }
                return jobResult{Data: "from a"}
            },
        })

        runner.do(job{
            Name: "b",
            Task: func() jobResult {
                select {
                case <-ctx.Done():
                    return jobResult{Error: errors.New("Timeouts, exiting")}
                default:
                    myLongRunningFunc("B job")
                }

                return jobResult{Data: "from b"}
            },
        })

        results := runner.getjobResults()
        fmt.Println(results)
        time.Sleep(time.Second * 60)
    }()
    <-time.After(time.Second * 5)
}

func myLongRunningFunc(name string) {
    for i := 0; i < 100000; i++ {
        time.Sleep(time.Second * 1)
        msg := "job" + name + " running..\n"
        fmt.Println(msg)
    }
}

type runner struct {
    *workerpool.WorkerPool
    ctx     context.Context
    kill    chan struct{}
    result  chan jobResult
    results []jobResult
}

func (r *runner) processResults() {
    for {
        select {
        case res, ok := <-r.result:
            if !ok {
                goto Done
            }
            r.results = append(r.results, res)
        }
    }
Done:
    <-r.kill
}

func newRunner(ctx context.Context, numRunners int) *runner {
    r := &runner{
        WorkerPool: workerpool.New(numRunners),
        ctx:        ctx,
        kill:       make(chan struct{}),
        result:     make(chan jobResult),
    }
    go r.processResults()
    return r
}

func (r *runner) do(j job) {
    r.Submit(r.wrap(&j))
}

func (r *runner) getjobResults() []jobResult {
    r.StopWait()
    close(r.result)
    r.kill <- struct{}{}
    return r.results
}

func (r *runner) wrap(job *job) func() {
    return func() {
        job.result = make(chan jobResult)
        go job.Run()
        select {
        case res := <-job.result:
            r.result <- res
        case <-r.ctx.Done():
            fmt.Printf("Job '%s' should stop here\n", job.Name)
            r.result <- jobResult{name: job.Name, Error: r.ctx.Err()}
        }
    }
}

type job struct {
    Name    string
    Task    func() jobResult
    Context context.Context
    result  chan jobResult
    stopped chan struct{}
    done    context.CancelFunc
}

func (j *job) Run() {
    result := j.Task()
    result.name = j.Name
    j.result <- result
}

type jobResult struct {
    name  string
    Error error
    Data  interface{}
}

The problem is that for each task I want to configure his timeout, moreover, in my case we have a web application server that e.g. each call to route `http://localhost:3000/run` will call to this logic (inside the main and should return response), how would you do it? I dont want to kill the program, I want that each request will finish max of `5 sec` , how would you do it? — JME, Nov 03 '20 at 14:03
HamoriZ did give you an answer. I have only highlighted a soluton in regard to your description of your issue. — , Nov 03 '20 at 14:14

score -1 · Answer 2 · answered Nov 03 '20 at 09:25

-1

I do not think it is possible to stop a goroutine from outer goroutine. You can check that it timed out, however, you can not stop it.

What you could do is to send a message via a channel to the goroutine which can be monitored and stop in that case.

You can find examples here

answered Nov 03 '20 at 09:25

HamoriZ

2,370
18
38

Thanks, I saw it already but not sure that I understand how to do it in my contexts, cloud you please provide an example in my contexts ? https://play.golang.org/p/a0RWY4bYWMt – JME Nov 03 '20 at 09:30
Please check this - https://play.golang.org/p/Oa-5vlfYRMs – HamoriZ Nov 03 '20 at 10:12
Thanks a lot, but I need someting `without changing` the `longRunnigFunc` :) , if it's not possible,maybe an option is: lets assume that user providing two diff functions `longRunnigFunc1` `longRunnigFunc2` which doing totally different thing, can I `abstract` the the `select` and tell to users that implements those different function add this `abstractSelectfunc` to your `longRunnigFuncN` and you are safe , is there a way? – JME Nov 03 '20 at 10:41
It might be feasible, but the select part must be checked within the long running operation ie. you need to pass a function with select to the long running function, so might be very similar to the solution with passing a channel. – HamoriZ Nov 03 '20 at 10:54
it would be great if you can provide an example how would you do it – JME Nov 03 '20 at 10:57
After checking the details I am not sure it would be a nice solution. If you add the select logic to a function and that function is called from the long running loop, then there will be an f condition anyway whether to return from the func or not. You would just make the code more complex. For me the channel version I provided is much simpler – HamoriZ Nov 03 '20 at 14:01

Go routine- stops running process

2 Answers2