diff --git a/batcher/README.md b/batcher/README.md new file mode 100644 index 0000000..ed83999 --- /dev/null +++ b/batcher/README.md @@ -0,0 +1,14 @@ +# batcher + +`*slow.Value` - это аналог `atomic.Value`, с двумя ограничениями: + - Вызов `Load()` всегда занимает не меньше 1ms + - Нельзя делать больше одного вызова `Load()` в один момент времени. + +Реализуйте `*Batcher`, который оборачивает `*slow.Value` и ускоряет чтения за счёт батчинга. + +Например, если 1000 горутин одновременно сделают вызов `(*Batcher).Load()`, то можно прочитать +значение один раз и раздать всем ожидающим горутинам. Это будет в 1000 раз быстрее, +чем вызывать `(*slow.Value).Load()` напрямую в каждой горутине. + +Ваша реализация не должна создавать _stale read_. Тоесть, `Load()` всегда должен возвращать значение +последнего `Store()` на момент начала вызова `Load()`. diff --git a/batcher/batcher.go b/batcher/batcher.go new file mode 100644 index 0000000..b82cfa4 --- /dev/null +++ b/batcher/batcher.go @@ -0,0 +1,12 @@ +// +build !solution + +package batcher + +import "gitlab.com/slon/shad-go/batcher/slow" + +type Batcher struct { +} + +func NewBatcher(v *slow.Value) *Batcher { + return nil +} diff --git a/batcher/batcher_test.go b/batcher/batcher_test.go new file mode 100644 index 0000000..80a355d --- /dev/null +++ b/batcher/batcher_test.go @@ -0,0 +1,97 @@ +package batcher + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "gitlab.com/slon/shad-go/batcher/slow" +) + +func TestSimple(t *testing.T) { + var value slow.Value + b := NewBatcher(&value) + + value.Store(1) + require.Equal(t, 1, b.Load()) + require.Equal(t, 1, value.Load()) + + value.Store(2) + require.Equal(t, 2, b.Load()) + require.Equal(t, 2, value.Load()) +} + +func TestStaleRead(t *testing.T) { + const ( + N = 100 + K = 100 + M = 10 + ) + + var value slow.Value + b := NewBatcher(&value) + + var counter int32 + value.Store(counter) + + var wg sync.WaitGroup + for i := 0; i < N; i++ { + wg.Add(1) + + go func(i int) { + defer wg.Done() + + time.Sleep(time.Millisecond * time.Duration(i/N)) + for j := 0; j < K; j++ { + counterValue := atomic.LoadInt32(&counter) + batcherValue := b.Load().(int32) + + if batcherValue < counterValue { + t.Errorf("load returned old value: counter=%d, batcher=%d", counterValue, batcherValue) + return + } + } + }(i) + } + + for i := 0; i < M*K; i++ { + // value is always greater than counter + value.Store(int32(i)) + atomic.StoreInt32(&counter, int32(i)) + + time.Sleep(time.Millisecond / M) + } + + wg.Wait() +} + +func TestSpeed(t *testing.T) { + const ( + N = 100 + K = 200 + ) + + var value slow.Value + b := NewBatcher(&value) + + start := time.Now() + + var wg sync.WaitGroup + for i := 0; i < N; i++ { + wg.Add(1) + + go func() { + defer wg.Done() + + for i := 0; i < K; i++ { + b.Load() + } + }() + } + wg.Wait() + + require.Truef(t, time.Since(start) < time.Second, "batching it too slow") +} diff --git a/batcher/slow/value.go b/batcher/slow/value.go new file mode 100644 index 0000000..5c0e585 --- /dev/null +++ b/batcher/slow/value.go @@ -0,0 +1,36 @@ +// +build !change + +package slow + +import ( + "sync" + "sync/atomic" + "time" +) + +type Value struct { + mu sync.Mutex + value interface{} + readRunning int32 +} + +func (s *Value) Load() interface{} { + if atomic.SwapInt32(&s.readRunning, 1) == 1 { + panic("another load is running") + } + defer atomic.StoreInt32(&s.readRunning, 0) + + s.mu.Lock() + value := s.value + s.mu.Unlock() + + time.Sleep(time.Millisecond) + return value +} + +func (s *Value) Store(v interface{}) { + s.mu.Lock() + defer s.mu.Unlock() + + s.value = v +}