cond_test.go

package cond

import (
	"context"
	"strconv"
	"sync"
	"sync/atomic"
	"testing"
)

// https://github.com/golang/go/issues/20491

// BenchmarkCond/Serial/Cond1-24   	 5000000	       205 ns/op	       0 B/op	       0 allocs/op
// BenchmarkCond/Serial/Map1-24    	  500000	      2033 ns/op	     424 B/op	       8 allocs/op
// BenchmarkCond/Serial/Map2-24    	 5000000	       306 ns/op	       0 B/op	       0 allocs/op
// BenchmarkCond/Parallel_3/Cond1-24         	 1000000	      1026 ns/op	       0 B/op	       0 allocs/op
// BenchmarkCond/Parallel_3/Map1-24          	 5000000	       343 ns/op	     147 B/op	       4 allocs/op
// BenchmarkCond/Parallel_3/Map2-24          	 3000000	       429 ns/op	       0 B/op	       0 allocs/op
// BenchmarkCond/Parallel_512/Cond1-24       	 1000000	      1341 ns/op	       0 B/op	       0 allocs/op
// BenchmarkCond/Parallel_512/Map1-24        	 1000000	      1767 ns/op	     198 B/op	       4 allocs/op
// BenchmarkCond/Parallel_512/Map2-24        	10000000	       147 ns/op	       0 B/op	       0 allocs/op
// BenchmarkCond/Parallel_8192/Cond1-24      	 1000000	      1289 ns/op	       0 B/op	       0 allocs/op
// BenchmarkCond/Parallel_8192/Map1-24       	 1000000	      1817 ns/op	     199 B/op	       4 allocs/op
// BenchmarkCond/Parallel_8192/Map2-24       	10000000	       106 ns/op	       0 B/op	       0 allocs/op

type Cond1 struct {
	c *sync.Cond
	s map[string]struct{}
}

func NewCond1() *Cond1 {
	return &Cond1{
		c: sync.NewCond(new(sync.Mutex)),
		s: make(map[string]struct{}),
	}
}

func (p *Cond1) Lock(id string) {
	p.c.L.Lock()
	defer p.c.L.Unlock()
	for _, ok := p.s[id]; ok; _, ok = p.s[id] {
		p.c.Wait()
	}
	p.s[id] = struct{}{}
	return
}

func (p *Cond1) Unlock(id string) {
	p.c.L.Lock()
	defer p.c.L.Unlock()
	delete(p.s, id)
	p.c.Broadcast()
}

type Map1 struct {
	s sync.Map // map[string]chan struct{}
}

func NewMap1() *Map1 { return new(Map1) }

func (p *Map1) Lock(ctx context.Context, id string) error {
	c := make(chan struct{})
	for {
		wait, loaded := p.s.LoadOrStore(id, c)
		if !loaded {
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-wait.(chan struct{}):
		}
	}
}

func (p *Map1) Unlock(id string) {
	c, _ := p.s.Load(id)
	p.s.Delete(id)
	close(c.(chan struct{}))
}

type Map2 struct {
	s sync.Map // map[string]chan struct{}
}

func NewMap2() *Map2 {
	return new(Map2)
}

func (p *Map2) Lock(ctx context.Context, id string) error {
	ci, ok := p.s.Load(id)
	if !ok {
		ci, _ = p.s.LoadOrStore(id, make(chan struct{}, 1))
	}
	c := ci.(chan struct{})
	select {
	case <-ctx.Done():
		return ctx.Err()
	case c <- struct{}{}:
	}
	return nil
}

func (p *Map2) Unlock(id string) {
	c, _ := p.s.Load(id)
	<-c.(chan struct{})
}

func BenchmarkCond(b *testing.B) {
	ctx := context.Background()

	const count = 8192
	parts := make([]string, count)
	for i := range parts {
		parts[i] = strconv.Itoa(i)
	}

	b.Run(`Serial`, func(b *testing.B) {
		b.Run(`Cond1`, func(b *testing.B) {
			pl := NewCond1()
			b.ResetTimer()
			b.ReportAllocs()
			for i := 0; i < b.N; i++ {
				part := parts[i%count]
				pl.Lock(part)
				pl.Unlock(part)
			}
		})
		b.Run(`Map1`, func(b *testing.B) {
			pl := NewMap1()
			b.ResetTimer()
			b.ReportAllocs()
			for i := 0; i < b.N; i++ {
				part := parts[i%count]
				if err := pl.Lock(context.Background(), part); err != nil {
					b.Fatal(err)
				}
				pl.Unlock(part)
			}
		})
		b.Run(`Map2`, func(b *testing.B) {
			pl := NewMap2()
			b.ResetTimer()
			b.ReportAllocs()
			for i := 0; i < b.N; i++ {
				part := parts[i%count]
				if err := pl.Lock(context.Background(), part); err != nil {
					b.Fatal(err)
				}
				pl.Unlock(part)
			}
		})
	})

	counts := []int{3, 512, count}
	for _, y := range counts {
		b.Run(`Parallel_`+strconv.Itoa(y), func(b *testing.B) {
			b.Run(`Cond1`, func(b *testing.B) {
				var i uint64
				pl := NewCond1()
				b.ResetTimer()
				b.ReportAllocs()

				b.RunParallel(func(pb *testing.PB) {
					for pb.Next() {
						part := parts[int(atomic.AddUint64(&i, 1))%y]
						pl.Lock(part)
						pl.Unlock(part)
					}
				})
			})
			b.Run(`Map1`, func(b *testing.B) {
				var i uint64
				pl := NewMap1()
				b.ResetTimer()
				b.ReportAllocs()

				b.RunParallel(func(pb *testing.PB) {
					for pb.Next() {
						part := parts[int(atomic.AddUint64(&i, 1))%y]
						if err := pl.Lock(ctx, part); err != nil {
							b.Fatal(err)
						}
						pl.Unlock(part)
					}
				})
			})
			b.Run(`Map2`, func(b *testing.B) {
				var i uint64
				pl := NewMap2()
				b.ResetTimer()
				b.ReportAllocs()

				b.RunParallel(func(pb *testing.PB) {
					for pb.Next() {
						part := parts[int(atomic.AddUint64(&i, 1))%y]
						if err := pl.Lock(ctx, part); err != nil {
							b.Fatal(err)
						}
						pl.Unlock(part)
					}
				})
			})
		})
	}
}