Initial commit

hw06_pipeline_execution/README.md

@@ -0,0 +1,37 @@
+## Домашнее задание №6 «Пайплайн»
+Необходимо реализовать функцию для запуска конкуррентного пайплайна, состоящего из стейджей.
+
+Стейдж - функция, принимающая канал на чтение и отдающая канал на чтение, внутри в горутине берущая данные из входного канала, выполняющая полезную работу и отдающая результат в выходной канал:
+```golang
+func Stage(in <-chan interface{}) (out <-chan interface{}) {
+    out = make(chan interface{})
+    go func() { /* Some work */ }()
+    return out
+}
+```
+
+Особенность пайплайна в том, что обработка последующего элемента входных данных должна
+происходить **без ожидания завершения всего пайплайна** для текущего элемента.
+
+Т.е. пайплан из 4 функций по 100 мс каждая для 5 входных элементов **должен выполняться
+гораздо быстрее**, чем за 2 секунды (4 * 100 мс * 5).
+
+Также **должна быть реализована возможность остановить пайплайн** через
+дополнительный сигнальный канал (`done`/`terminate`/etc.).
+
+При необходимости можно выделять дополнительные функции.
+
+**Нельзя менять сигнатуры исходных функций.**
+
+Для большего понимания см. тесты.
+
+### Критерии оценки
+- CI-пайплайн зелёный - 5 баллов
+- Добавлены новые юнит-тесты - до 2 баллов
+- Понятность и чистота кода - до 3 баллов
+
+#### Зачёт от 7 баллов
+
+### Подсказки
+- https://github.com/golang/go/wiki/CommonMistakes#using-goroutines-on-loop-iterator-variables
+- `go test -v -race -count=100 .`

hw06_pipeline_execution/go.mod

@@ -0,0 +1,11 @@
+module github.com/fixme_my_friend/hw06_pipeline_execution
+
+go 1.19
+
+require github.com/stretchr/testify v1.7.0
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+)

hw06_pipeline_execution/go.sum

@@ -0,0 +1,13 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

hw06_pipeline_execution/pipeline.go

@@ -0,0 +1,14 @@
+package hw06pipelineexecution
+
+type (
+	In  = <-chan interface{}
+	Out = In
+	Bi  = chan interface{}
+)
+
+type Stage func(in In) (out Out)
+
+func ExecutePipeline(in In, done In, stages ...Stage) Out {
+	// Place your code here.
+	return nil
+}

hw06_pipeline_execution/pipeline_test.go

@@ -0,0 +1,93 @@
+package hw06pipelineexecution
+
+import (
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	sleepPerStage = time.Millisecond * 100
+	fault         = sleepPerStage / 2
+)
+
+func TestPipeline(t *testing.T) {
+	// Stage generator
+	g := func(_ string, f func(v interface{}) interface{}) Stage {
+		return func(in In) Out {
+			out := make(Bi)
+			go func() {
+				defer close(out)
+				for v := range in {
+					time.Sleep(sleepPerStage)
+					out <- f(v)
+				}
+			}()
+			return out
+		}
+	}
+
+	stages := []Stage{
+		g("Dummy", func(v interface{}) interface{} { return v }),
+		g("Multiplier (* 2)", func(v interface{}) interface{} { return v.(int) * 2 }),
+		g("Adder (+ 100)", func(v interface{}) interface{} { return v.(int) + 100 }),
+		g("Stringifier", func(v interface{}) interface{} { return strconv.Itoa(v.(int)) }),
+	}
+
+	t.Run("simple case", func(t *testing.T) {
+		in := make(Bi)
+		data := []int{1, 2, 3, 4, 5}
+
+		go func() {
+			for _, v := range data {
+				in <- v
+			}
+			close(in)
+		}()
+
+		result := make([]string, 0, 10)
+		start := time.Now()
+		for s := range ExecutePipeline(in, nil, stages...) {
+			result = append(result, s.(string))
+		}
+		elapsed := time.Since(start)
+
+		require.Equal(t, []string{"102", "104", "106", "108", "110"}, result)
+		require.Less(t,
+			int64(elapsed),
+			// ~0.8s for processing 5 values in 4 stages (100ms every) concurrently
+			int64(sleepPerStage)*int64(len(stages)+len(data)-1)+int64(fault))
+	})
+
+	t.Run("done case", func(t *testing.T) {
+		in := make(Bi)
+		done := make(Bi)
+		data := []int{1, 2, 3, 4, 5}
+
+		// Abort after 200ms
+		abortDur := sleepPerStage * 2
+		go func() {
+			<-time.After(abortDur)
+			close(done)
+		}()
+
+		go func() {
+			for _, v := range data {
+				in <- v
+			}
+			close(in)
+		}()
+
+		result := make([]string, 0, 10)
+		start := time.Now()
+		for s := range ExecutePipeline(in, done, stages...) {
+			result = append(result, s.(string))
+		}
+		elapsed := time.Since(start)
+
+		require.Len(t, result, 0)
+		require.Less(t, int64(elapsed), int64(abortDur)+int64(fault))
+	})
+}