osbase/event/memory/memory_test.go - monogon - Gitiles

 // Copyright 2020 The Monogon Project Authors.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 package memory

 import (
 	"context"
 	"errors"
 	"fmt"
 	"testing"
 	"time"

 	"source.monogon.dev/osbase/event"
 )

 // TestAsync exercises the high-level behaviour of a Value, in which a
 // watcher is able to catch up to the newest Set value.
 func TestAsync(t *testing.T) {
 	p := Value[int]{}
 	p.Set(0)

 	ctx := context.Background()

 	// The 0 from Set() should be available via .Get().
 	watcher := p.Watch()
 	val, err := watcher.Get(ctx)
 	if err != nil {
 		t.Fatalf("Get: %v", err)
 	}
 	if want, got := 0, val; want != got {
 		t.Fatalf("Value: got %d, wanted %d", got, want)
 	}

 	// Send a large amount of updates that the watcher does not actively .Get().
 	for i := 1; i <= 100; i++ {
 		p.Set(i)
 	}

 	// The watcher should still end up with the newest .Set() value on the next
 	// .Get() call.
 	val, err = watcher.Get(ctx)
 	if err != nil {
 		t.Fatalf("Get: %v", err)
 	}
 	if want, got := 100, val; want != got {
 		t.Fatalf("Value: got %d, wanted %d", got, want)
 	}
 }

 // TestMultipleGets verifies that calling .Get() on a single watcher from two
 // goroutines is prevented by returning an error in exactly one of them.
 func TestMultipleGets(t *testing.T) {
 	p := Value[int]{}
 	ctx := context.Background()

 	w := p.Watch()

 	tryError := func(errs chan error) {
 		_, err := w.Get(ctx)
 		errs <- err
 	}
 	errs := make(chan error, 2)
 	go tryError(errs)
 	go tryError(errs)

 	for err := range errs {
 		if err == nil {
 			t.Fatalf("A Get call succeeded, while it should have blocked or returned an error")
 		} else {
 			// Found the error, test succeeded.
 			break
 		}
 	}
 }

 // TestConcurrency attempts to stress the Value/Watcher
 // implementation to design limits (a hundred simultaneous watchers), ensuring
 // that the watchers all settle to the final set value.
 func TestConcurrency(t *testing.T) {
 	ctx := context.Background()

 	p := Value[int]{}
 	p.Set(0)

 	// Number of watchers to create.
 	watcherN := 100
 	// Expected final value to be Set().
 	final := 100
 	// Result channel per watcher.
 	resC := make([]chan error, watcherN)

 	// Spawn watcherN watchers.
 	for i := 0; i < watcherN; i++ {
 		resC[i] = make(chan error, 1)
 		go func(id int) {
 			// done is a helper function that will put an error on the
 			// respective watcher's resC.
 			done := func(err error) {
 				resC[id] <- err
 				close(resC[id])
 			}

 			watcher := p.Watch()
 			// prev is used to ensure the values received are monotonic.
 			prev := -1
 			for {
 				val, err := watcher.Get(ctx)
 				if err != nil {
 					done(err)
 					return
 				}

 				// Ensure monotonicity of received data.
 				if val <= prev {
 					done(fmt.Errorf("received out of order data: %d after %d", val, prev))
 				}
 				prev = val

 				// Quit when the final value is received.
 				if val == final {
 					done(nil)
 					return
 				}

 				// Sleep a bit, depending on the watcher. This makes each
 				// watcher behave slightly differently, and attempts to
 				// exercise races dependent on sleep time between subsequent
 				// Get calls.
 				time.Sleep(time.Millisecond * time.Duration(id))
 			}
 		}(i)
 	}

 	// Set 1..final on the value.
 	for i := 1; i <= final; i++ {
 		p.Set(i)
 	}

 	// Ensure all watchers exit with no error.
 	for i, c := range resC {
 		err := <-c
 		if err != nil {
 			t.Errorf("Watcher %d returned %v", i, err)
 		}
 	}
 }

 // TestCanceling exercises whether a context canceling in a .Get() gracefully
 // aborts that particular Get call, but also allows subsequent use of the same
 // watcher.
 func TestCanceling(t *testing.T) {
 	p := Value[int]{}

 	ctx, ctxC := context.WithCancel(context.Background())

 	watcher := p.Watch()

 	// errs will contain the error returned by Get.
 	errs := make(chan error, 1)
 	go func() {
 		// This Get will block, as no initial data has been Set on the value.
 		_, err := watcher.Get(ctx)
 		errs <- err
 	}()

 	// Cancel the context, and expect that context error to propagate to the .Get().
 	ctxC()
 	if want, got := ctx.Err(), <-errs; !errors.Is(got, want) {
 		t.Fatalf("Get should've returned %v, got %v", want, got)
 	}

 	// Do another .Get() on the same watcher with a new context. Even though the
 	// call was aborted via a context cancel, the watcher should continue working.
 	ctx = context.Background()
 	go func() {
 		_, err := watcher.Get(ctx)
 		errs <- err
 	}()

 	// Unblock the .Get now.
 	p.Set(1)
 	if want, got := error(nil), <-errs; !errors.Is(got, want) {
 		t.Fatalf("Get should've returned %v, got %v", want, got)
 	}
 }

 // TestSetAfterWatch ensures that if a value is updated between a Watch and the
 // initial Get, only the newest Set value is returns.
 func TestSetAfterWatch(t *testing.T) {
 	ctx := context.Background()

 	p := Value[int]{}
 	p.Set(0)

 	watcher := p.Watch()
 	p.Set(1)

 	data, err := watcher.Get(ctx)
 	if err != nil {
 		t.Fatalf("Get: %v", err)
 	}
 	if want, got := 1, data; want != got {
 		t.Errorf("Get should've returned %v, got %v", want, got)
 	}
 }

 // TestWatchersList ensures that the list of watchers is managed correctly,
 // i.e. there is no memory leak and closed watchers are removed while
 // keeping all non-closed watchers.
 func TestWatchersList(t *testing.T) {
 	ctx := context.Background()
 	p := Value[int]{}

 	var watchers []event.Watcher[int]
 	for i := 0; i < 100; i++ {
 		watchers = append(watchers, p.Watch())
 	}
 	for i := 0; i < 10000; i++ {
 		watchers[10].Close()
 		watchers[10] = p.Watch()
 	}

 	if want, got := 1000, cap(p.watchers); want <= got {
 		t.Fatalf("Got capacity %d, wanted less than %d", got, want)
 	}

 	p.Set(1)
 	if want, got := 100, len(p.watchers); want != got {
 		t.Fatalf("Got %d watchers, wanted %d", got, want)
 	}

 	for _, watcher := range watchers {
 		data, err := watcher.Get(ctx)
 		if err != nil {
 			t.Fatalf("Get: %v", err)
 		}
 		if want, got := 1, data; want != got {
 			t.Errorf("Get should've returned %v, got %v", want, got)
 		}
 	}
 }
	// Copyright 2020 The Monogon Project Authors.
	//
	// SPDX-License-Identifier: Apache-2.0
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	package memory

	import (
	"context"
	"errors"
	"fmt"
	"testing"
	"time"

	"source.monogon.dev/osbase/event"
	)

	// TestAsync exercises the high-level behaviour of a Value, in which a
	// watcher is able to catch up to the newest Set value.
	func TestAsync(t *testing.T) {
	p := Value[int]{}
	p.Set(0)

	ctx := context.Background()

	// The 0 from Set() should be available via .Get().
	watcher := p.Watch()
	val, err := watcher.Get(ctx)
	if err != nil {
	t.Fatalf("Get: %v", err)
	}
	if want, got := 0, val; want != got {
	t.Fatalf("Value: got %d, wanted %d", got, want)
	}

	// Send a large amount of updates that the watcher does not actively .Get().
	for i := 1; i <= 100; i++ {
	p.Set(i)
	}

	// The watcher should still end up with the newest .Set() value on the next
	// .Get() call.
	val, err = watcher.Get(ctx)
	if err != nil {
	t.Fatalf("Get: %v", err)
	}
	if want, got := 100, val; want != got {
	t.Fatalf("Value: got %d, wanted %d", got, want)
	}
	}

	// TestMultipleGets verifies that calling .Get() on a single watcher from two
	// goroutines is prevented by returning an error in exactly one of them.
	func TestMultipleGets(t *testing.T) {
	p := Value[int]{}
	ctx := context.Background()

	w := p.Watch()

	tryError := func(errs chan error) {
	_, err := w.Get(ctx)
	errs <- err
	}
	errs := make(chan error, 2)
	go tryError(errs)
	go tryError(errs)

	for err := range errs {
	if err == nil {
	t.Fatalf("A Get call succeeded, while it should have blocked or returned an error")
	} else {
	// Found the error, test succeeded.
	break
	}
	}
	}

	// TestConcurrency attempts to stress the Value/Watcher
	// implementation to design limits (a hundred simultaneous watchers), ensuring
	// that the watchers all settle to the final set value.
	func TestConcurrency(t *testing.T) {
	ctx := context.Background()

	p := Value[int]{}
	p.Set(0)

	// Number of watchers to create.
	watcherN := 100
	// Expected final value to be Set().
	final := 100
	// Result channel per watcher.
	resC := make([]chan error, watcherN)

	// Spawn watcherN watchers.
	for i := 0; i < watcherN; i++ {
	resC[i] = make(chan error, 1)
	go func(id int) {
	// done is a helper function that will put an error on the
	// respective watcher's resC.
	done := func(err error) {
	resC[id] <- err
	close(resC[id])
	}

	watcher := p.Watch()
	// prev is used to ensure the values received are monotonic.
	prev := -1
	for {
	val, err := watcher.Get(ctx)
	if err != nil {
	done(err)
	return
	}

	// Ensure monotonicity of received data.
	if val <= prev {
	done(fmt.Errorf("received out of order data: %d after %d", val, prev))
	}
	prev = val

	// Quit when the final value is received.
	if val == final {
	done(nil)
	return
	}

	// Sleep a bit, depending on the watcher. This makes each
	// watcher behave slightly differently, and attempts to
	// exercise races dependent on sleep time between subsequent
	// Get calls.
	time.Sleep(time.Millisecond * time.Duration(id))
	}
	}(i)
	}

	// Set 1..final on the value.
	for i := 1; i <= final; i++ {
	p.Set(i)
	}

	// Ensure all watchers exit with no error.
	for i, c := range resC {
	err := <-c
	if err != nil {
	t.Errorf("Watcher %d returned %v", i, err)
	}
	}
	}

	// TestCanceling exercises whether a context canceling in a .Get() gracefully
	// aborts that particular Get call, but also allows subsequent use of the same
	// watcher.
	func TestCanceling(t *testing.T) {
	p := Value[int]{}

	ctx, ctxC := context.WithCancel(context.Background())

	watcher := p.Watch()

	// errs will contain the error returned by Get.
	errs := make(chan error, 1)
	go func() {
	// This Get will block, as no initial data has been Set on the value.
	_, err := watcher.Get(ctx)
	errs <- err
	}()

	// Cancel the context, and expect that context error to propagate to the .Get().
	ctxC()
	if want, got := ctx.Err(), <-errs; !errors.Is(got, want) {
	t.Fatalf("Get should've returned %v, got %v", want, got)
	}

	// Do another .Get() on the same watcher with a new context. Even though the
	// call was aborted via a context cancel, the watcher should continue working.
	ctx = context.Background()
	go func() {
	_, err := watcher.Get(ctx)
	errs <- err
	}()

	// Unblock the .Get now.
	p.Set(1)
	if want, got := error(nil), <-errs; !errors.Is(got, want) {
	t.Fatalf("Get should've returned %v, got %v", want, got)
	}
	}

	// TestSetAfterWatch ensures that if a value is updated between a Watch and the
	// initial Get, only the newest Set value is returns.
	func TestSetAfterWatch(t *testing.T) {
	ctx := context.Background()

	p := Value[int]{}
	p.Set(0)

	watcher := p.Watch()
	p.Set(1)

	data, err := watcher.Get(ctx)
	if err != nil {
	t.Fatalf("Get: %v", err)
	}
	if want, got := 1, data; want != got {
	t.Errorf("Get should've returned %v, got %v", want, got)
	}
	}

	// TestWatchersList ensures that the list of watchers is managed correctly,
	// i.e. there is no memory leak and closed watchers are removed while
	// keeping all non-closed watchers.
	func TestWatchersList(t *testing.T) {
	ctx := context.Background()
	p := Value[int]{}

	var watchers []event.Watcher[int]
	for i := 0; i < 100; i++ {
	watchers = append(watchers, p.Watch())
	}
	for i := 0; i < 10000; i++ {
	watchers[10].Close()
	watchers[10] = p.Watch()
	}

	if want, got := 1000, cap(p.watchers); want <= got {
	t.Fatalf("Got capacity %d, wanted less than %d", got, want)
	}

	p.Set(1)
	if want, got := 100, len(p.watchers); want != got {
	t.Fatalf("Got %d watchers, wanted %d", got, want)
	}

	for _, watcher := range watchers {
	data, err := watcher.Get(ctx)
	if err != nil {
	t.Fatalf("Get: %v", err)
	}
	if want, got := 1, data; want != got {
	t.Errorf("Get should've returned %v, got %v", want, got)
	}
	}
	}