DEV Community πŸ‘©β€πŸ’»πŸ‘¨β€πŸ’»

DEV Community πŸ‘©β€πŸ’»πŸ‘¨β€πŸ’» is a community of 964,423 amazing developers

We're a place where coders share, stay up-to-date and grow their careers.

Create account Log in
Stephen Afam-Osemene
Stephen Afam-Osemene

Posted on • Originally published at stephenafamo.com on

Gracefully shutting down multiple workers in Go

#go

Managing multiple workers in Go

In this post, we’ll see how to write workers in Go, how to gracefully shut them down. We’ll also look at how to coordinate multiple workers.

Finally, I’ll introduce Orchestra, a package I built which makes it a lot easier to manage and gracefully shutdown workers.

What are workers?

When we write programs, we are often in situations when we have function that runs until either it faces an error, or we explicitly order it to stop. A good example of this is the http.Server. When we start a web server in Go, the program will block until we ask it to stop or it encounters.

Here’s a simple webserver:

package main

import (
    "io"
    "net/http"
)

func main() {
    // Hello world, the web server

    helloHandler := func(w http.ResponseWriter, req *http.Request) {
        io.WriteString(w, "Hello, world!\n")
    }

    http.HandleFunc("/hello", helloHandler)
    err := http.ListenAndServe(":8080", nil)
    if err != nil {
        panic(err)
    }
}

Enter fullscreen mode Exit fullscreen mode

We can write a function that behaves the same way:

package main

func main()
    err := myWorker()
    if err != nil {
        panic(err)
    }
}

func myWorker() error {
    for {
        err := doSomethingRepeatedly()
        if err != nil {
            return err
        }
    }

    return nil
}

Enter fullscreen mode Exit fullscreen mode

In both these programs, we can ask them to stop by killing the main process (ctrl + C) and that will forcefully close the program.

Gracefully Shutting Down Workers

To shutdown a worker grace fully we should listen for signals passed to the program. The main ones to look out for are:

  • SIGINT: The interrupt signal. The terminal sends it to the foreground process when the user presses ctrl-c. The default behavior is to terminate the process, but it can be caught or ignored. The intention is to provide a mechanism for an orderly, graceful shutdown.
  • SIGTERM: The termination signal. The default behavior is to terminate the process, but it also can be caught or ignored. The intention is to kill the process, gracefully or not, but to first allow it a chance to cleanup.

Credit for the explanation of what the signals mean

For the webserver, since Go 1.8 (which was a long time ago), the http.Server type has a Shutdown method which gracefully closes the connections.

Here’s an example of how to use it:

package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
)

func main() {
    var srv http.Server

    idleConnsClosed := make(chan struct{})
    go func() {
        signals := make(chan os.Signal, 1)
        signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
        <-signals

        // We received an interrupt signal, shut down.
        if err := srv.Shutdown(context.Background()); err != nil {
            // Error from closing listeners, or context timeout:
            log.Printf("HTTP server Shutdown: %v", err)
        }
        close(idleConnsClosed)
    }()

    if err := srv.ListenAndServe(); err != http.ErrServerClosed {
        // Error starting or closing listener:
        log.Fatalf("HTTP server ListenAndServe: %v", err)
    }

    <-idleConnsClosed
}

Enter fullscreen mode Exit fullscreen mode

Let’s create something similar for our worker:

package main

import (
    "context"
    "log"
    "os"
    "os/signal"
    "syscall"
)

var shouldStop = false

func main() {
    complete := make(chan struct{})

    go func() {
        signals := make(chan os.Signal, 1)
        signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
        <-signals

        shouldStop = true

        close(complete)
    }()

    if err := myWorker(); err != nil {
        // Error in our worker:
        log.Fatalf("Worker error: %v", err)
    }

    <-complete
}

func myWorker() error {
    for !shouldStop {
        err := doSomethingRepeatedly()
        if err != nil {
            return err
        }
    }

    return nil
}

Enter fullscreen mode Exit fullscreen mode

Multiple Workers

This seems somewhat straightforward with just a single long running process. However, it becomes trickier.

At first glance, it seems we can just do something like this:

package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
)

var shouldStop = false

func main() {
    var srv http.Server

    complete := make(chan struct{})

    go func() {
        signals := make(chan os.Signal, 1)
        signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
        <-signals

        // Ask our worker to stop
        shouldStop = true

        // Shutdown the server
        if err := srv.Shutdown(context.Background()); err != nil {
            // Error from closing listeners, or context timeout:
            log.Printf("HTTP server Shutdown: %v", err)
        }

        close(complete)
    }()

    go func() {
        if err := myWorker(); err != nil {
            // Error in our worker:
            log.Fatalf("Worker error: %v", err)
        }
    }()

    go func() {
        if err := srv.ListenAndServe(); err != http.ErrServerClosed {
            // Error starting or closing listener:
            log.Fatalf("HTTP server ListenAndServe: %v", err)
        }
    }()

    <-complete
}

func myWorker() error {
    for !shouldStop {
        err := doSomethingRepeatedly()
        if err != nil {
            return err
        }
    }

    return nil
}

Enter fullscreen mode Exit fullscreen mode

But that does not work as expected. Here’s why:

First, we did not stop the workers asynchronously. It does not seem really needed in the particular example, so let’s leave that problem for now.

More importantly, in the example above, if any of the workers encounters an error, the program rudely exits! Our entire shutdown sequence is ignored.

Coordinating with Context

In idiomatic Go, the ideal way would be to pass a context.Context to our worker, and ask it to shut itself down once the context is done.

Using this method, instead of calling log.Fatalf() when a worker encounters an error, we can simply cancel the context, and it will then initiate the shutdown sequence.

Since we are not calling the shutdown functions directly, let’s use sync.WaitGroup to know when all our workers have exited.

This means we’d have to rework our code to look like this:

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)

    var wg sync.WaitGroup

    go func() {
        signals := make(chan os.Signal, 1)
        signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
        <-signals

        cancel()
    }()

    wg.Add(1)
    go func() {
        if err := myWorker(ctx); err != nil {
            cancel()
        }
        wg.Done()
    }()

    wg.Add(1)
    go func() {
        if err := startServer(ctx); err != nil {
            cancel()
        }
        wg.Done()
    }()

    wg.Wait()
}

func myWorker(ctx context.Context) error {

    var shouldStop = false

    go func() {
        <-ctx.Done()
        shouldStop = true
    }()

    for !shouldStop {
        err := doSomethingRepeatedly()
        if err != nil {
            return err
        }
    }

    return nil
}

func startServer(ctx context.Context) error {

    var srv http.Server

    go func() {
        <-ctx.Done() // Wait for the context to be done

        // Shutdown the server
        if err := srv.Shutdown(context.Background()); err != nil {
            // Error from closing listeners, or context timeout:
            log.Printf("HTTP server Shutdown: %v", err)
        }
    }()

    if err := srv.ListenAndServe(); err != http.ErrServerClosed {
        // Error starting or closing listener:
        return fmt.Errorf("HTTP server ListenAndServe: %w", err)
    }

    return nil
}

Enter fullscreen mode Exit fullscreen mode

This neatly satisfies all our conditions. Yay!

Using Orchestra

To reduce all this boilerplate, I created a package to do most of this.

Orchestra is a library to manage long running go processes.

At the heart of the library is an interface called Player

// Player is a long running background worker
type Player interface {
    Play(context.Context) error
}

Enter fullscreen mode Exit fullscreen mode

All a type needs to do to satisfy the interface is to have a Play method that will gracefully shutdown when the context is done.

It can also return an error if it encounters a problem when playing.

Next, there’s the Conductor type (which itself is a Player)

// Conductor is a group of workers. It is also a Player itself **evil laugh**
type Conductor struct {
    Timeout time.Duration
    Players map[string]Player
}

Enter fullscreen mode Exit fullscreen mode

With the conductor, you add Players to it, and when you call the Play method on the conductor, it will start the Players under it and gracefully shut them all down when the main context is done.

The timeout is there incase there is a Player that refused to stop.

It also contains a few helper functions:

  • PlayUntilSignal(p Player, sig ...os.Signal): This will start a player with a context, and close the context once it receives any of the signals provided.
  • PlayerFunc(func(context.Context) error): This is a quick way to convert a standalone function into a type that satisfies the Player interface.
  • ServerPlayer{*http.Server}: This is a type that embeds the *http.Server and extends it to satisfy the Player interface.

More details can be found in the README.

With these, we can simplify our example code into this:

package main

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

    "github.com/stephenafamo/orchestra"
)

func main() {
    var srv = &http.Server{}

    // creates a player from a myWorker function
    workerPlayer := orchestra.PlayerFunc(myWorker)
    // A player from a server
    serverPlayer := orchestra.ServerPlayer{srv}

    // A conductor to control them all
    conductor := &orchestra.Conductor{
        Timeout: 5 * time.Second,
        Players: map[string]orchestra.Player{
            // the names are used to identify the players
            // both in logs and the returned errors
            "worker": workerPlayer,
            "server": serverPlayer,
        },
    }

    // Use the conductor as a Player
    err := orchestra.PlayUntilSignal(conductor, os.Interrupt, syscall.SIGTERM)
    if err != nil {
        panic(err)
    }
}

func myWorker(ctx context.Context) error {

    var shouldStop = false

    go func() {
        <-ctx.Done()
        shouldStop = true
    }()

    for !shouldStop {
        err := doSomethingRepeatedly()
        if err != nil {
            return err
        }
    }

    return nil
}
Enter fullscreen mode Exit fullscreen mode

Conclusion

As stated in the first paragraph, we’ve seen how to write workers in Go, how to gracefully shut them down and I’ve introduced Orchestra, a package I built which makes it a lot easier to manage and gracefully shutdown workers.

The post Gracefully shutting down multiple workers in Go appeared first on Stephen AfamO's Blog.

Oldest comments (0)

🌚 Life is too short to browse without dark mode