pool

Overview ¶

Package pool is a generic solution for async job dispatching from web server. While Go natively supports async jobs by using the keyword "go", but this may lead to several unwanted consequences. Suppose we have a typical http handler:

func Handle(req *http.Request, resp http.ResponseWriter) {}

If we dispatch async jobs using "go" like this:

  func Handle(req *http.Request, resp http.ResponseWriter) {
    go AsyncWork()
	   resp.Write([]byte("ok"))
  }

Let's go through all the disadvantages. First, the backpressure is lost. There is no way to limit the maximum goroutine the handler can create. clients can easily flood the server. Secondly, the graceful shutdown process is ruined. The http server can shutdown itself without losing any request, but the async jobs created with "go" are not protected by the server. You will lose all unfinished jobs once the server shuts down and program exits. lastly, the async job may want to access the original request context, maybe for tracing purposes. The request context terminates at the end of the request, so if you are not careful, the async jobs may be relying on a dead context.

Package pool creates a goroutine worker pool at beginning of the program, limits the maximum concurrency for you, shuts it down at the end of the request without losing any async jobs, and manages the context conversion for you.

Add the dependency to core:

var c *core.C = core.New()
c.Provide(pool.Providers())

Then you can inject the pool into your http handler:

type Handler struct {
    pool *pool.Pool
}

func (h *Handler) ServeHTTP(req *http.Request, resp http.ResponseWriter) {
   pool.Go(request.Context(), AsyncWork(asyncContext))
   resp.Write([]byte("ok"))
}

package main

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

	"github.com/yanghp/core"
	"github.com/yanghp/core/contract/lifecycle"
	"github.com/yanghp/core/control/pool"

	"github.com/gorilla/mux"
)

func main() {
	c := core.Default(
		core.WithInline("http.addr", ":9777"),
		core.WithInline("log.level", "none"),
	)
	c.Provide(pool.Providers())

	c.Invoke(func(m *pool.Manager, dispatcher lifecycle.HTTPServerStart) {
		p := pool.NewPool(m, 10)
		dispatcher.On(func(ctx context.Context, payload lifecycle.HTTPServerStartPayload) error {
			go func() {
				if _, err := http.Get("http://localhost:9777/"); err != nil {
					panic(err)
				}
			}()
			return nil
		})
		c.AddModule(core.HttpFunc(func(router *mux.Router) {
			router.HandleFunc("/", func(writer http.ResponseWriter, request *http.Request) {
				p.Go(request.Context(), func(asyncContext context.Context) {
					select {
					case <-asyncContext.Done():
						fmt.Println("async context cancelled")
					case <-time.After(time.Second):
						fmt.Println("async context will not be cancelled")
					}
				})
				writer.Write(nil)
			})
		}))
	})

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	c.Serve(ctx)

}

Output:
async context will not be cancelled