Function TestDMap_Fragment_Concurrent_Access

internal/dmap/fragment_test.go:72–123

(t *testing.T)

}

func TestDMap_Fragment_Concurrent_Access(t *testing.T) {
	cluster := testcluster.New(NewService)
	s := cluster.AddMember(nil).(*Service)
	dm, err := s.NewDMap("mydmap")
	if err != nil {
		t.Fatalf("Expected nil. Got: %v", err)
	}

	part := dm.getPartitionByHKey(123, partitions.PRIMARY)

	var mtx sync.RWMutex
	var wg sync.WaitGroup
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()

			f, err := dm.loadOrCreateFragment(part)
			if err != nil {
				t.Errorf("Expected nil. Got: %v", err)
			}

			e := f.storage.NewEntry()
			e.SetKey(testutil.ToKey(idx))

			mtx.Lock()
			// storage engine is not thread-safe
			err = f.storage.Put(uint64(idx), e)
			mtx.Unlock()

			if err != nil {
				t.Errorf("Expected nil. Got: %v", err)
			}
		}(i)
	}

	wg.Wait()

	f, err := dm.loadFragment(part)
	if err != nil {
		t.Errorf("Expected nil. Got: %v", err)
	}
	for i := 0; i < 1000; i++ {
		entry, err := f.storage.Get(uint64(i))
		if err != nil {
			t.Fatalf("Expected nil. Got: %v", err)
		}
		if entry.Key() != testutil.ToKey(i) {
			t.Fatalf("Expected key: %s. Got: %s", testutil.ToKey(i), entry.Key())
		}
	}
}