partition

Overview ¶

Package partition provides utilities to partition requests to a cluster.

Every server in the cluster receives requests meant for any partition but these requests are then routed to the right partition where it is handled.

Each server in the cluster also serves traffic from other servers destined for its own partition.

Usage;

Every server in the cluster creates a router at initialization time:

endpointRegistry = partition.WithEndpointRegistry(...)
router, err := partition.New(ctx, "ownIP:2222", handler, endpointRegistry)

The endpoint registry keeps track of all the servers in the cluster and their local addresses for inter-server communication

The handler is the servers own implementation of requests routed to it by other servers.

When an external request comes in, the server would hash the request and then use the router to execute it on the right server in its cluster:

resp, err := router.Run(ctx, hash, request)

Note that this call would end up being executed on another server in the cluster (or on the local server itself if this hash is mapped to the local server).

The effectiveness of this strategy depends on how uniform the hash is and the mapping strategy. The default mechanism to map hashes to specific servers in the cluster is to use a highest-random-weight algorithm which can be overridden using the WithPicker option.

The inter-server communication is via RPC and this also can be configured to alternate mechanisms using the WithNetwork option.

The requests and responses are expected to be byte slices. For stronger types, protobufs can be used to serialize structures or the runner package (see https://godoc.org/github.com/tvastar/cluster/pkg/partition/runner) for solution using gob-encoding and reflection.

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/alicebob/miniredis"
	"github.com/tvastar/cluster/pkg/partition"
)

func main() {
	minir, err := miniredis.Run()
	if err != nil {
		panic(err)
	}
	defer minir.Close()

	opt1 := partition.WithEndpointRegistry(partition.NewRedisRegistry(minir.Addr(), "prefix_"))
	opt2 := partition.WithEndpointRegistry(partition.NewRedisRegistry(minir.Addr(), "prefix_"))

	ctx := context.Background()
	h1 := handler(1)
	h2 := handler(2)

	part1, err := partition.New(ctx, ":2222", h1, opt1)
	if err != nil {
		panic(err)
	}
	defer part1.Close()

	part2, err := partition.New(ctx, ":2223", h2, opt2)
	if err != nil {
		panic(err)
	}
	defer part2.Close()

	// wait for both endpoints to be registered
	time.Sleep(100 * time.Millisecond)

	if _, err := part1.Run(ctx, 555, []byte("hello")); err != nil {
		panic(err)
	}

	if _, err := part2.Run(ctx, 555, []byte("hello")); err != nil {
		panic(err)
	}

	if _, err := part1.Run(ctx, 22222, []byte("hello")); err != nil {
		panic(err)
	}

	if _, err := part2.Run(ctx, 22222, []byte("hello")); err != nil {
		panic(err)
	}

}

type handler int

func (h handler) Run(ctx context.Context, hash uint64, input []byte) ([]byte, error) {
	fmt.Printf("[%d] Run(%d, %s)\n", int(h), hash, string(input))
	return input, nil
}

func (h handler) Close() error {
	return nil
}

Output:

[1] Run(555, hello)
[1] Run(555, hello)
[2] Run(22222, hello)
[2] Run(22222, hello)