Complete.

.github/workflows/autotests.yaml

@@ -0,0 +1,21 @@
+name: autotests
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  test:
+    name: Running test.
+    runs-on: ubuntu-latest
+    container: golang:1.22.3
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Run Unit Tests
+        run: GOOS=linux GOARCH=amd64 go test -v ./... -count=1
+
+      - name: Vet
+        run: go vet ./...

README.md

@@ -1 +1,22 @@
-# counting_concurrency
+<div align="center"> <h1 align="center"> Счётчик "многопоточности". </h1> </div>
+
+__Горутина, генерирует числа и отправляет их в канал. Далее несколько горутин читают и распределяют их по каналам. Под конец производится обратное действие, из каналов пишутся все числа в один результирующий.__
+
+- __При правильном выполнении кода:__
+    - [x] Количество и сумма входящих чисел совпадает с количеством и суммой чисел, которые получены из канала вывода.
+    - [x] Количество проходящих чисел по каналам не должно сильно отличаться.
+
+Числа генерируются с помощью ```context.WithTimeout``` в течение одной секунды.
+Количество обрабатывающих горутин зависит от числа ядер ```runtime.NumCPU()```.
+Вышеуказанные параметры можно изменить в ```cmd/api/app.go```
+
+***
+#### Инструкция по локальному запуску:
+
+Запуск производится по-умолчанию: ```go run ./...```\
+Тесты выполняются по-умолчанию: ```go test -v ./... -count=1```
+
+***
+#### Пример:
+
+![logo](/web/example.jpg)

cmd/api/app.go

@@ -0,0 +1,28 @@
+package main
+
+import (
+	"context"
+	"runtime"
+	"time"
+
+	"github.com/ra1nz0r/counting_concurrency/internal/service"
+)
+
+func main() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	var inSum int64 // сумма сгенерированных чисел
+	var inCnt int64 // количество сгенерированных чисел
+
+	chIn := service.GenNum(ctx, &inSum, &inCnt)
+
+	NumOut := runtime.NumCPU() // количество обрабатывающих горутин и каналов
+	outs := service.GenChanSliceWithNum(NumOut, chIn)
+
+	chOut, amounts := service.SendNumInResChan(NumOut, outs)
+
+	sumTOT, cntTOT := service.ReadFromResChan(chOut)
+
+	service.CheckResult(amounts, inSum, inCnt, sumTOT, cntTOT)
+}

go.mod

@@ -0,0 +1,11 @@
+module github.com/ra1nz0r/counting_concurrency
+
+go 1.22.3
+
+require github.com/stretchr/testify v1.9.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.1 // indirect
+)

go.sum

@@ -0,0 +1,10 @@
+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/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+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.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/dye/color_for_err.go

@@ -0,0 +1,4 @@
+package dye
+
+var Reset = "\033[0m"
+var Red = "\033[31m"

internal/service/all_service_test.go

@@ -0,0 +1,41 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/ra1nz0r/counting_concurrency/internal/dye"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAllService(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	var inSumT int64
+	var inCntT int64
+
+	chIn := GenNum(ctx, &inSumT, &inCntT)
+
+	NumOut := runtime.NumCPU()
+	outs := GenChanSliceWithNum(NumOut, chIn)
+
+	chOut, amounts := SendNumInResChan(NumOut, outs)
+
+	sumTestTOT, cntTestTOT := ReadFromResChan(chOut)
+
+	assert.Equal(t, inCntT, cntTestTOT,
+		fmt.Sprintf("%sCount numbers is not equal.%s", dye.Red, dye.Reset))
+	assert.Equal(t, inSumT, sumTestTOT,
+		fmt.Sprintf("%sSum numbers is not equal.%s", dye.Red, dye.Reset))
+
+	for _, v := range amounts {
+		inCntT -= v
+	}
+	assert.Zero(t, inCntT,
+		fmt.Sprintf("%sThe division of numbers by channel is incorrect.%s", dye.Red, dye.Reset))
+}

internal/service/generator_test.go

@@ -0,0 +1,47 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ra1nz0r/counting_concurrency/internal/dye"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGenerator(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	var testCntTOT int64
+	var testSumTOT int64
+
+	chInT := GenNum(ctx, &testSumTOT, &testCntTOT)
+
+	require.NotEmpty(t, <-chInT,
+		fmt.Sprintf("%sChannel empty.%s", dye.Red, dye.Reset))
+
+	var expCount int64 = 1
+	var expSum int64 = 1
+
+	for {
+		value, ok := <-chInT
+		if !ok {
+			break
+		}
+		atomic.AddInt64(&expSum, value)
+		atomic.AddInt64(&expCount, 1)
+		time.Sleep(25 * time.Microsecond)
+	}
+
+	require.NotEqual(t, expCount, expSum,
+		fmt.Sprintf("%s`Generator` does not increment the value.%s", dye.Red, dye.Reset))
+	assert.Equal(t, expCount, testCntTOT,
+		fmt.Sprintf("%sWrong count. Expect: %d, got: %d.%s", dye.Red, expCount, testCntTOT, dye.Reset))
+	assert.Equal(t, expSum, testSumTOT,
+		fmt.Sprintf("%sWrong sum. Expect: %d, got: %d.%s", dye.Red, expSum, testSumTOT, dye.Reset))
+}

internal/service/service_func.go

@@ -0,0 +1,121 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Generator генерирует последовательность чисел 1,2,3 и т.д. и
+// отправляет их в канал ch. При этом после записи в канал для каждого числа
+// вызывается функция fn. Она служит для подсчёта количества и суммы
+// сгенерированных чисел.
+func Generator(ctx context.Context, ch chan<- int64, fn func(int64)) {
+	var num int64 = 1
+	defer close(ch)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			ch <- num
+			fn(num)
+			num++
+		}
+	}
+}
+
+// Worker читает число из канала in и пишет его в канал out.
+func Worker(in <-chan int64, out chan<- int64) {
+	defer close(out)
+	for {
+		value, ok := <-in
+		if !ok {
+			break
+		}
+		out <- value
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// Генерирует числа и считает их количество и сумму.
+func GenNum(ctx context.Context, numSum, numCnt *int64) chan int64 {
+	chInput := make(chan int64)
+	go Generator(ctx, chInput, func(i int64) {
+		atomic.AddInt64((*int64)(numSum), i)
+		atomic.AddInt64((*int64)(numCnt), 1)
+	})
+	return chInput
+}
+
+// Возвращает слайс каналов, с количеством записанных чисел в каждый канал из входного канала.
+func GenChanSliceWithNum(numOutput int, chInput chan int64) []chan int64 {
+	// outsChanSlice — слайс каналов, куда будут записываться числа из chIn
+	outsChanSlice := make([]chan int64, numOutput)
+	for i := 0; i < numOutput; i++ {
+		// создаём каналы и для каждого из них вызываем горутину Worker
+		outsChanSlice[i] = make(chan int64)
+		go Worker(chInput, outsChanSlice[i])
+	}
+	return outsChanSlice
+}
+
+// Отправляет числа из слайса каналов в результирующий канал, со статистикой по отработанным горутинам.
+func SendNumInResChan(numOutput int, outsChanSlice []chan int64) (chan int64, []int64) {
+	// gorutStat — слайс, в который собирается статистика по горутинам
+	gorutStat := make([]int64, numOutput)
+	// chanRes — канал, в который будут отправляться числа из горутин `gorutStat[i]`
+	chanRes := make(chan int64, numOutput)
+
+	var wg sync.WaitGroup
+
+	// Передаем числа из каналов gorutStat в результирующий канал
+	for i := 0; i < len(outsChanSlice); i++ {
+		wg.Add(1)
+		go func(nextChan <-chan int64, numChan int64) {
+			defer wg.Done()
+			for v := range nextChan {
+				chanRes <- v
+				gorutStat[numChan]++
+			}
+		}(outsChanSlice[i], int64(i))
+	}
+
+	go func() {
+		wg.Wait()
+		close(chanRes)
+	}()
+	return chanRes, gorutStat
+}
+
+// Читаем числа из результирующего канала
+func ReadFromResChan(chanRes chan int64) (checkSum, checkCnt int64) {
+	for v := range chanRes {
+		checkCnt++
+		checkSum += v
+	}
+	return
+}
+
+// Проверка результата и вывод данных по числам, каналам.
+func CheckResult(gorutStat []int64, numSum, numCnt, checkSum, checkCnt int64) {
+	fmt.Println("Количество чисел", numCnt, checkCnt)
+	fmt.Println("Сумма чисел", numSum, checkSum)
+	fmt.Println("Разбивка по каналам", gorutStat)
+
+	if numSum != checkSum {
+		log.Fatalf("Ошибка: суммы чисел не равны: %d != %d\n", numSum, checkSum)
+	}
+	if numCnt != checkCnt {
+		log.Fatalf("Ошибка: количество чисел не равно: %d != %d\n", numCnt, checkCnt)
+	}
+	for _, v := range gorutStat {
+		numCnt -= v
+	}
+	if numCnt != 0 {
+		log.Fatalf("Ошибка: разделение чисел по каналам неверное\n")
+	}
+}

internal/service/worker_test.go

@@ -0,0 +1,51 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/ra1nz0r/counting_concurrency/internal/dye"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorker(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	chanInT := make(chan int64)
+	var num int64 = 1
+	var compareNum int64
+
+	go func() {
+		defer close(chanInT)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				chanInT <- num
+				compareNum = num
+				num++
+			}
+		}
+	}()
+
+	chanOutT := make(chan int64)
+
+	go Worker(chanInT, chanOutT)
+
+	require.NotEmpty(t, <-chanOutT,
+		fmt.Sprintf("%sChannel empty.%s", dye.Red, dye.Reset))
+
+	var count int64
+	for v := range chanOutT {
+		count = v
+	}
+
+	assert.Equal(t, compareNum, count,
+		fmt.Sprintf("%sWrong count. Expect: %d, got: %d.%s", dye.Red, compareNum, count, dye.Reset))
+}