Function TestConcurrency

modules/backendscheduler/work/work_sharded_test.go:439–552

(t *testing.T)

}

func TestConcurrency(t *testing.T) {
	work := New(Config{})
	ctx := context.Background()

	// Test concurrent operations
	t.Run("concurrent adds", func(t *testing.T) {
		const numGoroutines = 10
		const jobsPerGoroutine = 10

		// Add jobs concurrently
		done := make(chan struct{})
		for i := range numGoroutines {
			go func(goroutineID int) {
				defer func() { done <- struct{}{} }()

				for j := range jobsPerGoroutine {
					job := createTestJob(fmt.Sprintf("concurrent-%d-%d", goroutineID, j), tempopb.JobType_JOB_TYPE_COMPACTION)
					err := work.AddJob(job)
					require.NoError(t, err)
				}
			}(i)
		}

		// Wait for all goroutines
		for range numGoroutines {
			<-done
		}

		// Verify all jobs were added (should be numGoroutines * jobsPerGoroutine)
		require.Equal(t, numGoroutines*jobsPerGoroutine, len(work.ListJobs()))
	})

	t.Run("concurrent operations", func(t *testing.T) {
		var err error

		// Add initial jobs
		for i := range 50 {
			job := createTestJob(fmt.Sprintf("ops-job-%d", i), tempopb.JobType_JOB_TYPE_COMPACTION)
			err = work.AddJob(job)
			require.NoError(t, err)
		}

		// Perform concurrent operations
		const numWorkers = 5
		done := make(chan struct{})

		for i := range numWorkers {
			go func(workerID int) {
				defer func() { done <- struct{}{} }()

				// Perform various operations
				jobs := work.ListJobs()
				if len(jobs) > 0 {
					job := jobs[workerID%len(jobs)]
					work.StartJob(job.ID)
					work.GetJob(job.ID)
					if workerID%2 == 0 {
						work.CompleteJob(job.ID)