singleflightx

README

go-singleflightx

x/sync/singleflight but better

Features

This library is inspired by x/sync/singleflight but adds many features:

• 🧬 generics
• 🍱 batching: fetch multiple keys in a single callback, with in-flight deduplication
• 📭 nullable result
• 🍕 sharded groups

🚀 Install

go get github.com/samber/go-singleflightx

This library is v0 and follows SemVer strictly. No breaking changes will be made to exported APIs before v1.0.0.

💡 Doc

GoDoc: https://pkg.go.dev/github.com/samber/go-singleflightx

Examples

Here is an example of a user retrieval in a caching layer:

import "github.com/samber/go-singleflightx"

func getUsersByID(userIDs []string) (map[string]User, error) {
    users := []User{}

    // 📍 SQL query here...
    err := sqlx.Select(&users, "SELECT * FROM users WHERE id IN (?);", userIDs...)
    if err != nil {
        return nil, err
    }

    var results = map[string]User{}
    for _, u := range users {
        results[u.ID] = u
    }

    return results, nil
}

func main() {
    var g singleflightx.Group[string, User]

    // 👇 concurrent queries will be dedup
    output := g.DoX([]string{"user-1", "user-2"}, getUsersByID)
}

output is of type map[K]singleflightx.Result[V], and will always have as many entries as requested, whatever the callback result.

type Result[V any] struct {
  	 Value  singleflightx.NullValue[V]  // 💡 A result is considered "null" if the callback did not return it.
  	 Err    error
  	 Shared bool
}

type NullValue[V any] struct {
	Value V
	Valid bool
}

Sharded groups, for high contention/concurrency environments

g := singleflightx.NewShardedGroup[K string, User](10, func (key string) uint {
    h := fnv.New64a()
    h.Write([]byte(key))
    return uint(h.Sum64())
})

// as usual, but if the keys match different shards, getUsersByID will be called twice
output := g.DoX([]string{"user-1", "user-2"}, getUsersByID) 

go-singleflightx + go-batchify

go-batchify groups concurrent tasks into a single batch. By adding go-singleflightx, you will be able to dedupe

import (
    "golang.org/x/sync/singleflight"
    "github.com/samber/go-batchify"
)

var group singleflight.Group

batch := batchify.NewBatchWithTimer(
    10,
    func (ids []int) (map[int]string, error) {
        return ..., nil
    },
    5*time.Millisecond,
)

http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
    idStr := r.URL.Query().Get("id")
    id, _ := strconv.Atoi(idStr)

    value, err, _ = group.Do(idStr, func() (interface{}, error) {
        return batch.Do(id)
    })

    // ...
})

🤝 Contributing

• Ping me on Twitter @samuelberthe (DMs, mentions, whatever :))
• Fork the project
• Fix open issues or request new features

Don't hesitate ;)

# Install some dev dependencies
make tools

# Run tests
make test
# or
make watch-test

👤 Contributors

💫 Show your support

Give a ⭐️ if this project helped you!

📝 License

Copyright © 2023 Samuel Berthe.

This project is MIT licensed.

Documentation

Index

• type Group
  • func (g *Group[K, V]) Do(key K, fn func() (V, error)) (v V, err error, shared bool)
  • func (g *Group[K, V]) DoChan(key K, fn func() (V, error)) <-chan Result[V]
  • func (g *Group[K, V]) DoChanX(keys []K, fn func([]K) (map[K]V, error)) map[K]chan Result[V]
  • func (g *Group[K, V]) DoX(keys []K, fn func([]K) (map[K]V, error)) (results map[K]Result[V])
  • func (g *Group[K, V]) Forget(key K)
  • func (g *Group[K, V]) ForgetX(keys []K)
• type Hasher
• type NullValue
• type Result
• type ShardedGroup
  • func NewShardedGroup[K comparable, V any](count uint, hasher Hasher[K]) *ShardedGroup[K, V]
  • func (sg *ShardedGroup[K, V]) Do(key K, fn func() (V, error)) (v V, err error, shared bool)
  • func (sg *ShardedGroup[K, V]) DoChan(key K, fn func() (V, error)) <-chan Result[V]
  • func (sg *ShardedGroup[K, V]) DoChanX(keys []K, fn func([]K) (map[K]V, error)) map[K]chan Result[V]
  • func (sg *ShardedGroup[K, V]) DoX(keys []K, fn func([]K) (map[K]V, error)) (results map[K]Result[V])
  • func (sg *ShardedGroup[K, V]) Forget(key K)
  • func (sg *ShardedGroup[K, V]) ForgetX(keys []K)

Types

type Group

type Group[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Group represents a class of work and forms a namespace in which units of work can be executed with duplicate suppression.

func (*Group[K, V]) Do

func (g *Group[K, V]) Do(key K, fn func() (V, error)) (v V, err error, shared bool)

Do executes and returns the results of the given function, making sure that only one execution is in-flight for a given key at a time. If a duplicate comes in, the duplicate caller waits for the original to complete and receives the same results. The return value shared indicates whether v was given to multiple callers.

func (*Group[K, V]) DoChan

func (g *Group[K, V]) DoChan(key K, fn func() (V, error)) <-chan Result[V]

DoChan is like Do but returns a channel that will receive the results when they are ready.

The returned channel will not be closed.

func (*Group[K, V]) DoChanX

func (g *Group[K, V]) DoChanX(keys []K, fn func([]K) (map[K]V, error)) map[K]chan Result[V]

DoChanX is like Do but returns a channel that will receive the results when they are ready.

The returned channel will not be closed.

func (*Group[K, V]) DoX

func (g *Group[K, V]) DoX(keys []K, fn func([]K) (map[K]V, error)) (results map[K]Result[V])

DoX executes and returns the results of the given function, making sure that only one execution is in-flight for a given key at a time. If a duplicate comes in, the duplicate caller waits for the original to complete and receives the same results. The return value shared indicates whether v was given to multiple callers. Even if fn does not return V on some keys, the results map will contain those keys with a `Valid` field set to false.

func (*Group[K, V]) Forget

func (g *Group[K, V]) Forget(key K)

Forget tells the singleflight to forget about a key. Future calls to Do for this key will call the function rather than waiting for an earlier call to complete.

func (*Group[K, V]) ForgetX

func (g *Group[K, V]) ForgetX(keys []K)

ForgetX tells the singleflight to forget about many keys. Future calls to Do for this key will call the function rather than waiting for an earlier call to complete.

type Hasher

type Hasher[K any] func(key K) uint64

Hasher is responsible for generating unsigned, 16 bit hash of provided key. Hasher should minimize collisions. For great performance, a fast function is preferable.

type NullValue

type NullValue[V any] struct {
	Value V
	Valid bool // Valid is true if Value is not NULL
}

NullValue represents a V that may be null.

type Result

type Result[V any] struct {
	Value  NullValue[V]
	Err    error
	Shared bool
}

Result holds the results of Do, so they can be passed on a channel.

type ShardedGroup

type ShardedGroup[K comparable, V any] struct {
	// contains filtered or unexported fields
}

ShardedGroup is a duplicate of singleflight.Group, but with the ability to shard the map of calls.

func NewShardedGroup

func NewShardedGroup[K comparable, V any](count uint, hasher Hasher[K]) *ShardedGroup[K, V]

func (*ShardedGroup[K, V]) Do

func (sg *ShardedGroup[K, V]) Do(key K, fn func() (V, error)) (v V, err error, shared bool)

Do executes and returns the results of the given function, making sure that only one execution is in-flight for a given key at a time. If a duplicate comes in, the duplicate caller waits for the original to complete and receives the same results. The return value shared indicates whether v was given to multiple callers. Even if fn does not return V on some keys, the results map will contain those keys with a `Valid` field set to false.

func (*ShardedGroup[K, V]) DoChan

func (sg *ShardedGroup[K, V]) DoChan(key K, fn func() (V, error)) <-chan Result[V]

DoChan is like Do but returns a channel that will receive the results when they are ready.

The returned channel will not be closed.

func (*ShardedGroup[K, V]) DoChanX

func (sg *ShardedGroup[K, V]) DoChanX(keys []K, fn func([]K) (map[K]V, error)) map[K]chan Result[V]

DoChanX is like Do but returns a channel that will receive the results when they are ready.

The returned channel will not be closed.

func (*ShardedGroup[K, V]) DoX

func (sg *ShardedGroup[K, V]) DoX(keys []K, fn func([]K) (map[K]V, error)) (results map[K]Result[V])

DoX executes and returns the results of the given function, making sure that only one execution is in-flight for a given key at a time. If a duplicate comes in, the duplicate caller waits for the original to complete and receives the same results. The return value shared indicates whether v was given to multiple callers.

func (*ShardedGroup[K, V]) Forget

func (sg *ShardedGroup[K, V]) Forget(key K)

Forget tells the singleflight to forget about a key. Future calls to Do for this key will call the function rather than waiting for an earlier call to complete.

func (*ShardedGroup[K, V]) ForgetX

func (sg *ShardedGroup[K, V]) ForgetX(keys []K)

ForgetX tells the singleflight to forget about many keys. Future calls to Do for this key will call the function rather than waiting for an earlier call to complete.