HW5 is completed

hw05_parallel_execution/go.mod

@@ -1,4 +1,4 @@
-module github.com/fixme_my_friend/hw05_parallel_execution
+module github.com/DimVlas/otus_hw/hw05_parallel_execution
 
 go 1.19
 

hw05_parallel_execution/run.go

@@ -2,14 +2,83 @@ package hw05parallelexecution
 
 import (
 	"errors"
+	"sync"
 )
 
-var ErrErrorsLimitExceeded = errors.New("errors limit exceeded")
+var (
+	ErrErrorsLimitExceeded = errors.New("errors limit exceeded")
+	ErrEmptyTasks          = errors.New("empty task list")
+)
 
 type Task func() error
 
-// Run starts tasks in n goroutines and stops its work when receiving m errors from tasks.
+// Run Tasks with channels.
 func Run(tasks []Task, n, m int) error {
-	// Place your code here.
+	if len(tasks) == 0 {
+		return ErrEmptyTasks
+	}
+
+	if m <= 0 {
+		return ErrErrorsLimitExceeded
+	}
+
+	if len(tasks) < n { // если кол-во задач меньше кол-ва воркеров, ограничиваем кол-во воркеров
+		n = len(tasks)
+	}
+
+	tasksCh := make(chan Task)       // канал задач.
+	stopErrCh := make(chan struct{}) // сигнальный канал.
+	errorCh := make(chan error)      // канал ошибок.
+	wg := sync.WaitGroup{}
+	wgErr := sync.WaitGroup{}
+
+	defer func() {
+		close(tasksCh)
+		close(stopErrCh)
+		wg.Wait()
+		close(errorCh)
+		wgErr.Wait()
+	}()
+
+	// запускаем воркеры
+	for i := 0; i < n; i++ {
+		wg.Add(1)
+		go doWork(&wg, tasksCh, errorCh)
+	}
+
+	wgErr.Add(1)
+	go func(wg *sync.WaitGroup) {
+		defer wg.Done()
+
+		errCount := 0 // текущее кол-во ошибок
+		for range errorCh {
+			errCount++
+			if errCount == m {
+				stopErrCh <- struct{}{}
+			}
+		}
+	}(&wgErr)
+
+	for _, task := range tasks {
+		select {
+		case <-stopErrCh:
+			return ErrErrorsLimitExceeded
+		case tasksCh <- task:
+		}
+	}
+
 	return nil
 }
+
+func doWork(wg *sync.WaitGroup, tasks <-chan Task, errs chan<- error) {
+	defer func() {
+		wg.Done()
+	}()
+
+	for task := range tasks {
+		err := task()
+		if err != nil {
+			errs <- err
+		}
+	}
+}

hw05_parallel_execution/run_test.go

@@ -67,4 +67,91 @@ func TestRun(t *testing.T) {
 		require.Equal(t, runTasksCount, int32(tasksCount), "not all tasks were completed")
 		require.LessOrEqual(t, int64(elapsedTime), int64(sumTime/2), "tasks were run sequentially?")
 	})
+
+	t.Run("all tasks without elapsedTime", func(t *testing.T) {
+		tasksCount := 50
+		tasks := make([]Task, 0, tasksCount)
+
+		var runTasksCount int32
+		var sumTime time.Duration
+		var err error
+
+		for i := 0; i < tasksCount; i++ {
+			taskSleep := time.Millisecond * time.Duration(rand.Intn(100))
+			sumTime += taskSleep
+
+			tasks = append(tasks, func() error {
+				time.Sleep(taskSleep)
+				atomic.AddInt32(&runTasksCount, 1)
+				return nil
+			})
+		}
+
+		workersCount := 5
+		maxErrorsCount := 1
+
+		require.Eventually(t, func() bool {
+			err = Run(tasks, workersCount, maxErrorsCount)
+			return err == nil
+		}, time.Duration(int64(sumTime/2)), time.Duration(int64(sumTime/1000)), "tasks were run sequentially?")
+
+		require.NoError(t, err)
+	})
+
+	t.Run("empty tasks", func(t *testing.T) {
+		tasks := make([]Task, 0)
+
+		err := Run(tasks, 1, 1)
+
+		require.Truef(t, errors.Is(err, ErrEmptyTasks), "actual err - %v", err)
+	})
+
+	t.Run("task count less workers", func(t *testing.T) {
+		tasksCount := 5
+		tasks := make([]Task, 0, tasksCount)
+
+		var runTasksCount int32
+		var sumTime time.Duration
+
+		for i := 0; i < tasksCount; i++ {
+			taskSleep := time.Millisecond * time.Duration(rand.Intn(100))
+			sumTime += taskSleep
+
+			tasks = append(tasks, func() error {
+				time.Sleep(taskSleep)
+				atomic.AddInt32(&runTasksCount, 1)
+				return nil
+			})
+		}
+
+		workersCount := 10
+		maxErrorsCount := 10
+		err := Run(tasks, workersCount, maxErrorsCount)
+
+		require.NoError(t, err)
+		require.Equal(t, runTasksCount, int32(tasksCount), "not all tasks were completed")
+	})
+
+	t.Run("zero errors", func(t *testing.T) {
+		tasksCount := 50
+		tasks := make([]Task, 0, tasksCount)
+
+		var runTasksCount int32
+
+		for i := 0; i < tasksCount; i++ {
+			if i == 5 {
+				tasks = append(tasks, func() error {
+					time.Sleep(time.Millisecond * time.Duration(rand.Intn(100)))
+					atomic.AddInt32(&runTasksCount, 1)
+					return nil
+				})
+			}
+		}
+
+		workersCount := 5
+		maxErrorsCount := 0
+		err := Run(tasks, workersCount, maxErrorsCount)
+
+		require.Truef(t, errors.Is(err, ErrErrorsLimitExceeded), "actual err - %v", err)
+	})
 }