[QUESTION] i think set stop in run function is not a good idea

[hibiken]

[QUESTION] i think set stop in run function is not a good idea

[yxlimo]

1. the idea from net/http seems good, but the functions name NewServer is confused.If i did not read the doc, i would think it's a server to receive request and produce a job. other names like Serve, NewServeMux also confused

2. Handler is part of Config: i prefer the api like gin

bg := asynq.NewBackgroud(...)
bg.Use(middleware)
bg.Register(handler1)
bg.Register(handler2)

[hibiken]

I see. I can see why this could be confusing. Let me try to explain.

First Question

since we have Stop method, what does Shutdown do ? this method make me think the process will exist.

Asynq took a lot of ideas from Sidekiq and they use OS signals to handle graceful shutdown (video).

Currently, when you start background workers with bg.Run it waits for signals.
Asynq handles TERM, INT, and TSTP signals.
If you send TSTP signal to the process, asynq will stop processing new tasks but doesn't exit. This is documented in Sidekiq here.

If you send TERM or INT signal to the process, asynq will wait for few seconds to let active workers finish its tasks but if they don't finish within that timeout, asynq will push the tasks back to Redis and exit the process. This is documented in Sidekiq here.

My intention was to name the first operation Stop and the second one Shutdown. But I can see why Stop is a bit confusing. Looking at sidekiq doc, maybe we can name it bg.Quiet()?

Does that makes sense?

Second Question

bg.Handler can receive options to custom each task handler, i think its important. i can't find this feature in current design, the option like Queues and Concurrency seems should be task level instead of global level

I agree with the Concurrency per Handler (i.e. Concurrency per task type). For example, if you have a task to use third-party API and if they have a limit on how many concurrent API calls they can handle, this is definitely useful. Am I understanding you correctly here?

I'm not sure if we can add an option to specify Queues per Handler though. gocraft/work has a different design in that they have one-to-one mapping between job-type and queue. Asynq's design is similar to sidekiq. It allows for multipe queues with different priority but any task can be put on any queue. So I'm not sure if we can create Queue option for a Handler. (please let me know if I'm not understanding your point correctly).

[hibiken]

Yep, just exporting just Start and Stop makes sense to me (chose Stop because of the symmetry with Start).

So effectively, if user uses Start and Stop instead of Run they don't get some nice signal handling but I guess that's okay. If we get feature request to export Quiet(i.e. stop processing new tasks), maybe we'll add that later.

I'll get started on making these changes and ping you in the PR 👍

[yxlimo]

It's fine. breaking change needs to thinking more and more.

Looking forward to your final design.

[zsiec]

+1 to exporting these. Some weeks ago I forked and exposed start/stop myself but have now stumbled on this thread and would be happy to remove my fork and use this once it’s ready!

My current implementation (importing my fork) handles the syscall termination signals for the workers as I was already handling them for other things before using asynq and it’s helpful to me to have my graceful shutdown logic collocated. It was also trivial to implement a shutdown timeout external to this library.

Thanks for your work on this very usable project, looking forward to seeing these updates!

[hibiken]

@zsiec Awesome to hear your feedback! Thank you.

I'm currently drafting a PR for this change, which includes a few breaking changes (I'll make sure to document the upgrade paths in CHANGELOG for those who are on the current version).

[hibiken]

Closed via #135

[hibiken]

@yxlimo Thank you for opening this issue!
Would exporting Background.Start and Background.Stop would resolve this?

Proposal

1. Keep Background.Run as is, since it also does signal handling.

2. Expose Background.Start and Background.Stop, if a user of the package wants to have more control over when to start and stop the background workers .

Please let me know your thoughts and feedback!

[yxlimo]

Looks good.
also export Background.WaitSignal ? more control and more convenient

[hibiken]

Yep, I agree.

So if you opt out of using Run then code would look something like this to achieve the same effect:

bg := asynq.NewBackground(redis, asynq.Config{
    Concurrency: 20,
})

bg.Start(handler)
bg.WaitForSignals()
// Do any cleanup here if needed.
bg.Stop()

would that work for your use case?

[yxlimo]

its work, thanks :)

[hibiken]

Actually I've been thinking about doing some major API changes and I think this is a great opportunity to clean up some things.
Let me run this by you and hopefully get your feedback.

func main() {
        r := asynq.RedisClientOpt{
                Addr: ":6379",
        }

        mux := asynq.NewServeMux()
        mux.HandleFunc(tasks.EmailTask, tasks.HandleEmailTask)

        // Renamed the type to more common name "Server"
        srv, err := asynq.NewServer(asynq.Config{
                RedisConnOpt: r,
                Handler:      mux, // <- Handler is part of Config
                Concurrency:  20,
                Queues: map[string]int{
                        "critical": 6,
                        "default":  3,
                        "low":      1,
                },
        })
        if err != nil {
                log.Fatal(err)
        }
      
        srv.Run()
}

Or optionally you can do Serve and Shutdown to start/stop background workers manually.
I've take an API ideas from net/http's Server type.

func main() {
        r := asynq.RedisClientOpt{
                Addr: ":6379",
        }

        mux := asynq.NewServeMux()
        mux.HandleFunc(tasks.EmailTask, tasks.HandleEmailTask)

        srv, err := asynq.NewServer(asynq.Config{
                RedisConnOpt: r,
                Handler:      mux, // <- Handler is part of Config
                Concurrency:  20,
                Queues: map[string]int{
                        "critical": 6,
                        "default":  3,
                        "low":      1,
                },
        })
        if err != nil {
                log.Fatal(err)
        }

        done := make(chan struct{})
        go func() {
                sigint := make(chan os.Signal, 1)
                signal.Notify(sigint, os.Interrupt)
                <-sigint

                // We received an interrupt signal, shut down.
                if err := srv.Shutdown(context.Background()); err != nil {
                        // Error from workers shutdown, or context timeout:
                        log.Printf("background workers Shutdown error: %v", err)

                }
                close(done)
      }()

        if err := srv.Serve(); err != asynq.ErrServerClosed {
                log.Fatal("background workers Run error: %v", err)
        }

        <-done
}

Since this is a disruptive but important change, any feedback is welcome 👍

[hibiken]

Great feedback, thank you! Let me think about this a bit more.

[hibiken]

Ok here's my next proposal, feedback appreciated!

Add

• func (bg *Background) Use( ...Middleware)
• func (bg *Background) Handle(string, Handler)
• func (bg *Background) HandleFunc(string, HandlerFunc)
• func (bg *Background) Start() error
• func (bg *Background) Stop()
• func (bg *Background) Shutdown()
• ErrBackgroundClosed: once Stop or Shutdown is called on a Background, this error will be returned from Run or Start to prevent reuse.

Remove

• ServeMux type (it's internalized in Background)

Change

• func (bg *Background) Run(Handler) to func (bg *Background) Run() error

Example

After the change, main may look something like this.

func main() {
        bg := asynq.NewBackground(redis, asynq.Config{ ... })

        bg.Use(loggingMiddleware)
        bg.Handle(task.EmailTask, tasks.EmailTaskHandler)
        bg.Handle(task.ImageProcessing, task.ImageProcessingHandler)
        // bg.HandleFunc(type, fn) for Handler functions

        bg.Run()
}

Or bg.Run() could be replaced with

        done := make(chan struct{})
        go func() {
                sigs := make(chan os.Signal, 1)
                signal.Notify(sigs, unix.SIGTERM, unix.SIGINT, unix.SIGTSTP)
                for {
                        s := <-sigs
                        if s == unix.SIGTSTP {
                                bg.Stop() // Stop processing new tasks
                                continue
                        }
                        break
                }

                // We received an interrupt or term signal, shut down.
                bg.Shutdown()
                close(done)
        }()

        if err := bg.Start(); err != asynq.ErrBackgroundClosed {
                log.Fatal("Background Start error: %v", err)
        }

        <-done

[yxlimo]

awesome! just a little confused.

• since we have Stop method, what does Shutdown do ? this method make me think the process will exist.

and here is more thinking(based on the experience of using gocraft/work and sidekiq):

• bg.Handler can receive options to custom each task handler, i think its important. i can't find this feature in current design, the option like Queues and Concurrency seems should be task level instead of global level

[yxlimo]

totally right. thanks for your explanation.

now i understand the design for Queues

For the first question

after reading the wiki, i found that TSTP just only stop processing new task, and TERM is more useful.

in go language, i found most packages have only one Stop method(or Close).i think it's enough. so stop method do not need to be exported.

func (bg *Background) Close() {
      bg.stopProcessing()
      // handle timeout
}

leave two methods maybe will let user like me confused too, we can't assume that everyone will read the doc. What do you think?

[hibiken]

@yxlimo Now that I've started working on these API changes. I have a few thoughts and want to think about them more (sorry, for going back and forth so many times 🙏).

If this change is blocking you, please feel free to fork and use that in the meantime!