Initial import

Author: elgopher

README.md

@@ -0,0 +1,89 @@
+[![Build](https://github.com/elgopher/batch/actions/workflows/build.yml/badge.svg)](https://github.com/elgopher/batch/actions/workflows/build.yml)
+[![Go Reference](https://pkg.go.dev/badge/github.com/elgopher/batch.svg)](https://pkg.go.dev/github.com/elgopher/batch)
+[![Go Report Card](https://goreportcard.com/badge/github.com/elgopher/batch)](https://goreportcard.com/report/github.com/elgopher/batch)
+[![codecov](https://codecov.io/gh/elgopher/batch/branch/master/graph/badge.svg)](https://codecov.io/gh/elgopher/batch)
+[![Project Status: Active – The project has reached a stable, usable state and is being actively developed.](https://www.repostatus.org/badges/latest/active.svg)](https://www.repostatus.org/#active)
+
+## What it can be used for?
+
+To speed up application performance without sacrificing *data consistency* or *durability* and making source code or architecture complex.
+
+The **batch** package simplifies writing Go applications that process incoming requests (HTTP, GRPC etc.) in a batch manner:
+instead of processing each request separately, group incoming requests to a batch and run whole group at once.
+This method of processing can significantly speed up the application and reduce the consumption of disk, network or CPU.
+
+The **batch** package can be used to write any type of *servers* that handle thousands of requests per second. 
+Thanks to this small library, you can create relatively simple code without the need to use low-level data structures.
+
+## Why batch processing improves performance?
+
+Normally a web application is using following pattern to modify data in the database:
+
+1. **Load resource** from database. Resource is some portion of data 
+such as record, document etc. Lock the entire resource pessimistically
+or optimistically (by reading version number).
+2. **Apply change** to data
+3. **Save resource** to database. Release the pessimistic lock. Or run
+atomic update with version check (optimistic lock).
+
+But such architecture does not scale well if number of requests 
+for a single resource is very high
+(meaning hundreds or thousands of requests per second). 
+The lock contention in such case is very high and database is significantly 
+overloaded. Practically, the number of concurrent requests is limited.  
+
+One solution to this problem is to reduce the number of costly operations.
+Because a single resource is loaded and saved thousands of times per second 
+we can instead:
+
+1. Load the resource **once** (let's say once per second) 
+2. Execute all the requests from this period of time on an already loaded resource. Run them all sequentially.
+3. Save the resource and send responses to all clients if data was stored successfully.
+
+Such solution could improve the performance by a factor of 1000. And resource is still stored in a consistent state. 
+
+The **batch** package does exactly that. You configure the duration of window, provide functions
+to load and save resource and once the request comes in - you run a function:
+
+```go
+// set up the batch processor:
+processor := batch.StartProcessor(
+    batch.Options[*YourResource]{ // YourResource is your Go struct
+        MinDuration:  100 * time.Millisecond,
+        LoadResource: ...,
+        SaveResource: ...,
+    },
+)
+
+// following code is run from http/grpc handler
+// resourceKey uniquely identifies the resource
+err := s.BatchProcessor.Run(resourceKey, func(r *YourResource) {
+    // here is the code which is executed inside batch  
+})
+```
+
+For real-life example see [example web application](_example).
+
+## Installation
+
+```sh
+# Add batch to your Go module:
+go get github.com/elgopher/batch
+```
+Please note that at least **Go 1.18** is required.
+
+## Scaling out
+
+Single Go http server is able to handle up to 10-50k of requests per second on a commodity hardware. This is a lot, but very often you also need:
+
+* high availability (if one server goes down you want other to handle the traffic)
+* you want to handle hundred thousands or millions of requests per second
+
+For both cases you need to deploy **multiple servers** and put a **load balancer** in front of them. 
+Please note though, that you have to carefully configure the load balancing algorithm. 
+Round-robin is not an option here, because sooner or later you will have problems with locking 
+(multiple server instances will run batches on the same resource). 
+Ideal solution is to route requests based on parameters or URL. 
+For example some http parameter could be a resource key. You can instruct load balancer
+to calculate hash on this parameter and always route requests with this param value 
+to the same backend (of course if all backends are still available).

_example/http/http.go

@@ -0,0 +1,60 @@
+// (c) 2022 Jacek Olszak
+// This code is licensed under MIT license (see LICENSE for details)
+
+package http
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+
+	"github.com/elgopher/batch/_example/train"
+)
+
+type TrainService interface {
+	Book(train string, seatNumber int, person string) error
+}
+
+func ListenAndServe(trainService TrainService) error {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/book", bookHandler(trainService))
+
+	server := &http.Server{Addr: ":8080", Handler: mux}
+	return server.ListenAndServe()
+}
+
+// example request: /book?train=batchy&person=Jacek&seat=3
+func bookHandler(trainService TrainService) func(http.ResponseWriter, *http.Request) {
+	return func(writer http.ResponseWriter, request *http.Request) {
+		if err := request.ParseForm(); err != nil {
+			writer.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		trainKey := request.Form.Get("train")
+		person := request.Form.Get("person")
+		seat, err := strconv.Atoi(request.Form.Get("seat"))
+		if err != nil {
+			writer.WriteHeader(http.StatusBadRequest)
+			_, _ = writer.Write([]byte("invalid seat number"))
+			return
+		}
+
+		err = trainService.Book(trainKey, seat, person)
+
+		if errors.Is(err, train.ErrValidation("")) {
+			writer.WriteHeader(http.StatusBadRequest)
+			_, _ = writer.Write([]byte(err.Error()))
+			return
+		}
+
+		if err != nil {
+			writer.WriteHeader(http.StatusInternalServerError)
+			fmt.Println("internal server error", err)
+			return
+		}
+
+		writer.WriteHeader(http.StatusOK)
+	}
+}

_example/main.go

@@ -0,0 +1,34 @@
+// (c) 2022 Jacek Olszak
+// This code is licensed under MIT license (see LICENSE for details)
+
+package main
+
+import (
+	"time"
+
+	"github.com/elgopher/batch"
+	"github.com/elgopher/batch/_example/http"
+	"github.com/elgopher/batch/_example/store"
+	"github.com/elgopher/batch/_example/train"
+)
+
+func main() {
+	db := store.File{Dir: "/tmp/"}
+
+	processor := batch.StartProcessor(
+		batch.Options[*train.Train]{
+			MinDuration:  100 * time.Millisecond,
+			MaxDuration:  3 * time.Second,
+			LoadResource: db.LoadTrain,
+			SaveResource: db.SaveTrain,
+		},
+	)
+
+	trainService := train.Service{
+		BatchProcessor: processor,
+	}
+
+	if err := http.ListenAndServe(trainService); err != nil {
+		panic(err)
+	}
+}

_example/store/file.go

@@ -0,0 +1,48 @@
+// (c) 2022 Jacek Olszak
+// This code is licensed under MIT license (see LICENSE for details)
+
+package store
+
+import (
+	"context"
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"path"
+
+	"github.com/elgopher/batch/_example/train"
+)
+
+// File is database implementation which stores data to a file. Real world app would use real database instead, which supports
+// atomic write with some kind of optimistic locking (version-check, compare and swap etc.) to always have data in consistent state.
+type File struct {
+	Dir string
+}
+
+func (d File) LoadTrain(_ context.Context, key string) (*train.Train, error) {
+	jason, err := ioutil.ReadFile(d.filename(key))
+	if os.IsNotExist(err) {
+		return train.New(30), nil
+	}
+
+	t := &train.Train{}
+	err = json.Unmarshal(jason, t)
+	if err != nil {
+		return nil, err
+	}
+
+	return t, nil
+}
+
+func (d File) filename(key string) string {
+	return path.Join(d.Dir, key+".json")
+}
+
+func (d File) SaveTrain(_ context.Context, key string, t *train.Train) error {
+	jason, err := json.Marshal(t)
+	if err != nil {
+		return err
+	}
+
+	return ioutil.WriteFile(d.filename(key), jason, 0644)
+}

_example/train/error.go

@@ -0,0 +1,15 @@
+// (c) 2022 Jacek Olszak
+// This code is licensed under MIT license (see LICENSE for details)
+
+package train
+
+type ErrValidation string
+
+func (e ErrValidation) Error() string {
+	return string(e)
+}
+
+func (e ErrValidation) Is(err error) bool {
+	_, ok := err.(ErrValidation)
+	return ok
+}

_example/train/service.go

@@ -0,0 +1,27 @@
+// (c) 2022 Jacek Olszak
+// This code is licensed under MIT license (see LICENSE for details)
+
+package train
+
+// BatchProcessor is an optional interface to decouple your code from `batch` package.
+type BatchProcessor interface {
+	Run(key string, operation func(*Train)) error
+}
+
+type Service struct {
+	BatchProcessor BatchProcessor
+}
+
+func (s Service) Book(trainName string, seatNumber int, person string) error {
+	var operationError error
+
+	batchError := s.BatchProcessor.Run(trainName, func(train *Train) {
+		operationError = train.Book(seatNumber, person)
+	})
+
+	if operationError != nil {
+		return operationError
+	}
+
+	return batchError
+}

_example/train/train.go

@@ -0,0 +1,48 @@
+// (c) 2022 Jacek Olszak
+// This code is licensed under MIT license (see LICENSE for details)
+
+package train
+
+import (
+	"fmt"
+)
+
+type Train struct {
+	Seats []string
+}
+
+func New(maxSeats int) *Train {
+	return &Train{
+		Seats: make([]string, maxSeats),
+	}
+}
+
+func (t *Train) Book(seatNumber int, person string) error {
+	// first validate if action is possible (`try phase` of a method)
+	// such validation is needed to always keep Train in a consistent state
+	// (data consistency is required to properly handle next request in a batch)
+	if err := t.validateBooking(seatNumber, person); err != nil {
+		return err
+	}
+
+	// then mutate state (`do phase` of method)
+	t.Seats[seatNumber] = person
+
+	return nil
+}
+
+func (t *Train) validateBooking(seatNumber int, person string) error {
+	if seatNumber < 0 || seatNumber >= len(t.Seats) {
+		return ErrValidation(fmt.Sprintf("train does not have seat number %d", seatNumber))
+	}
+
+	if person == "" {
+		return ErrValidation("empty person name")
+	}
+
+	if t.Seats[seatNumber] != "" && t.Seats[seatNumber] != person {
+		return ErrValidation("seat number is already booked by another person")
+	}
+
+	return nil
+}

async_test.go

@@ -0,0 +1,36 @@
+// (c) 2022 Jacek Olszak
+// This code is licensed under MIT license (see LICENSE for details)
+
+package batch_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func FutureValue[V any]() Value[V] {
+	return Value[V]{
+		done: make(chan V, 1),
+	}
+}
+
+type Value[V any] struct {
+	done chan V
+}
+
+func (d Value[V]) Set(result V) {
+	d.done <- result
+}
+
+func (d Value[V]) Get(t *testing.T) V {
+	select {
+	case r, _ := <-d.done:
+		return r
+	case <-time.After(time.Second):
+		assert.FailNow(t, "timeout waiting for value")
+		var r V
+		return r
+	}
+}