Function TestDB_ConcurrentPut

engine/db_test.go:95–209

(t *testing.T)

}

func TestDB_ConcurrentPut(t *testing.T) {
	opts := config.DefaultOptions
	dir, _ := os.MkdirTemp("", "flydb-ConcurrentPut")
	opts.DirPath = dir
	opts.DataFileSize = 64 * 1024 * 1024
	db, err := NewDB(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	var wg sync.WaitGroup

	putTestWorker := func(id int, db *DB) {
		defer func() {
			// fmt.Printf("put Worker %d done\n", id)
			wg.Done()
		}()
		// fmt.Printf("Worker %d processing\n", id)
		err = db.Put(randkv.GetTestKey(id), randkv.GetTestKey(id))
		assert.Nil(t, err)
		// fmt.Printf("Worker %d resumed\n", id)
	}

	getTestWorker := func(id int, db *DB) {
		defer func() {
			wg.Done()
		}()
		val, err := db.Get(randkv.GetTestKey(id))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, randkv.GetTestKey(id), val)
	}

	var workerNum = 100

	// Put workerNum data in parallel
	wg.Add(workerNum)
	go func() {
		for i := 0; i < workerNum; i++ {
			go putTestWorker(i, db)
		}
	}()

	// Parallel Put workerNum to the same data as the previous parallel process
	wg.Add(workerNum)
	go func() {
		for i := 0; i < workerNum; i++ {
			go putTestWorker(i, db)
		}
	}()

	// Wait for all Put to be completed and perform subsequent tests
	wg.Wait()

	// Get all the previously inserted data in parallel
	wg.Add(workerNum)
	go func() {
		for i := 0; i < workerNum; i++ {