Function TestConcurrentOperations

pkg/task/queue/queue_test.go:13–141

(t *testing.T)

)

func TestConcurrentOperations(t *testing.T) {
	queue := NewTasksQueue(
		"test",
		metricsstorage.NewMetricStorage(metricsstorage.WithNewRegistry()),
		WithContext(context.Background()),
	)

	// Use a smaller number of goroutines and operations to avoid overwhelming the system
	numGoroutines := 10
	opsPerGoroutine := 10

	var wg sync.WaitGroup

	// 10 goroutines adding tasks
	for i := 0; i < numGoroutines; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < opsPerGoroutine; j++ {
				newTask := task.NewTask("test")
				queue.AddLast(newTask)
			}
		}()
	}

	// 10 goroutines adding tasks before a random existing task
	for i := 0; i < numGoroutines; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < opsPerGoroutine; j++ {
				newTask := task.NewTask("test")
				// Try to add before a task that might exist
				existingTask := queue.GetFirst()
				if existingTask != nil {
					queue.AddBefore(existingTask.GetId(), newTask)
				} else {
					queue.AddLast(newTask)
				}
			}
		}()
	}

	// 10 goroutines adding tasks after a random existing task
	for i := 0; i < numGoroutines; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < opsPerGoroutine; j++ {
				newTask := task.NewTask("test")
				// Try to add after a task that might exist
				existingTask := queue.GetFirst()
				if existingTask != nil {
					queue.AddAfter(existingTask.GetId(), newTask)
				} else {
					queue.AddLast(newTask)
				}
			}