Golang Test: R. Kochnev

golang/main.go

@@ -1,114 +1,197 @@
 package main
 
 import (
+	"context"
 	"fmt"
+	"math/rand"
+	"runtime"
+	"sync"
 	"time"
 )
 
-// Приложение эмулирует получение и обработку неких тасков. Пытается и получать, и обрабатывать в многопоточном режиме.
-// Приложение должно генерировать таски 10 сек. Каждые 3 секунды должно выводить в консоль результат всех обработанных к этому моменту тасков (отдельно успешные и отдельно с ошибками).
+const (
+	TimeFormat      = time.RFC3339
+	ApproxBufSize   = 256
+	ResultBufSize   = ApproxBufSize / 2
+	TotalWorkTime   = 10 * time.Second
+	NewTaskInterval = 100 * time.Millisecond
+	PrintInterval   = 3 * time.Second
+)
 
-// ЗАДАНИЕ: сделать из плохого кода хороший и рабочий - as best as you can.
-// Важно сохранить логику появления ошибочных тасков.
-// Важно оставить асинхронные генерацию и обработку тасков.
-// Сделать правильную мультипоточность обработки заданий.
-// Обновленный код отправить через pull-request в github
-// Как видите, никаких привязок к внешним сервисам нет - полный карт-бланш на модификацию кода.
+// ============================================================================
 
-// Мы даем тестовое задание чтобы:
-// * уменьшить время технического собеседования - лучше вы потратите пару часов в спокойной домашней обстановке, чем будете волноваться, решая задачи под взором наших ребят;
-// * увеличить вероятность прохождения испытательного срока - видя сразу стиль и качество кода, мы можем быть больше уверены в выборе;
-// * снизить число коротких собеседований, когда мы отказываем сразу же.
+type Task struct {
+	ID      int64
+	Created time.Time
+}
 
-// Выполнение тестового задания не гарантирует приглашение на собеседование, т.к. кроме качества выполнения тестового задания, оцениваются и другие показатели вас как кандидата.
+type TaskResult struct {
+	Task
+	Duration time.Duration
+}
 
-// Мы не даем комментариев по результатам тестового задания. Если в случае отказа вам нужен наш комментарий по результатам тестового задания, то просим об этом написать вместе с откликом.
+type TaskError struct {
+	Task
+	Message string
+}
 
-// A Ttype represents a meaninglessness of our life
-type Ttype struct {
-	id         int
-	cT         string // время создания
-	fT         string // время выполнения
-	taskRESULT []byte
+func NewTask(id int64, created time.Time) Task {
+	return Task{ID: id, Created: created}
 }
 
-func main() {
-	taskCreturer := func(a chan Ttype) {
-		go func() {
-			for {
-				ft := time.Now().Format(time.RFC3339)
-				if time.Now().Nanosecond()%2 > 0 { // вот такое условие появления ошибочных тасков
-					ft = "Some error occured"
-				}
-				a <- Ttype{cT: ft, id: int(time.Now().Unix())} // передаем таск на выполнение
-			}
-		}()
+func (t Task) Exec() (TaskResult, error) {
+	if time.Now().Nanosecond()/1000%2 > 0 {
+		return TaskResult{}, TaskError{Task: Task{ID: t.ID, Created: t.Created}, Message: "Bad Nanoseconds"}
 	}
 
-	superChan := make(chan Ttype, 10)
+	sleepTime := time.Duration(rand.Intn(135)+85) * time.Millisecond
+	time.Sleep(sleepTime)
 
-	go taskCreturer(superChan)
+	duration := time.Since(t.Created)
 
-	task_worker := func(a Ttype) Ttype {
-		tt, _ := time.Parse(time.RFC3339, a.cT)
-		if tt.After(time.Now().Add(-20 * time.Second)) {
-			a.taskRESULT = []byte("task has been successed")
-		} else {
-			a.taskRESULT = []byte("something went wrong")
-		}
-		a.fT = time.Now().Format(time.RFC3339Nano)
+	res := TaskResult{
+		Task:     Task{ID: t.ID, Created: t.Created},
+		Duration: duration,
+	}
 
-		time.Sleep(time.Millisecond * 150)
+	return res, nil
+}
 
-		return a
-	}
+func (t TaskResult) String() string {
+	return fmt.Sprintf("(Ok) Task %-19d: [Created: %s, Duration: %s]", t.ID, t.Created.Format(TimeFormat), t.Duration)
+}
+
+func (e TaskError) Error() string {
+	return fmt.Sprintf("(Err) Task %-19d, [Created: %s, Message: %s]", e.ID, e.Created.Format(TimeFormat), e.Message)
+}
 
-	doneTasks := make(chan Ttype)
-	undoneTasks := make(chan error)
+// ============================================================================
 
-	tasksorter := func(t Ttype) {
-		if string(t.taskRESULT[14:]) == "successed" {
-			doneTasks <- t
-		} else {
-			undoneTasks <- fmt.Errorf("Task id %d time %s, error %s", t.id, t.cT, t.taskRESULT)
+func taskProducer(
+	ctx context.Context,
+	tasksChan chan<- Task,
+	wg *sync.WaitGroup,
+) {
+	defer wg.Done()
+
+	ticker := time.NewTicker(NewTaskInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-ticker.C:
+			newTask := NewTask(rand.Int63(), time.Now())
+
+			select {
+			case tasksChan <- newTask:
+			default:
+			}
 		}
 	}
+}
 
-	go func() {
-		// получение тасков
-		for t := range superChan {
-			t = task_worker(t)
-			go tasksorter(t)
-		}
-		close(superChan)
-	}()
-
-	result := map[int]Ttype{}
-	err := []error{}
-	go func() {
-		for r := range doneTasks {
-			go func() {
-				result[r.id] = r
-			}()
+func taskWorker(
+	ctx context.Context,
+	tasksChan <-chan Task,
+	resChan chan<- TaskResult,
+	errChan chan<- error,
+	wg *sync.WaitGroup,
+) {
+	defer wg.Done()
+
+	for task := range tasksChan {
+		res, err := task.Exec()
+
+		if ctx.Err() != nil {
+			return
 		}
-		for r := range undoneTasks {
-			go func() {
-				err = append(err, r)
-			}()
+
+		if err != nil {
+			errChan <- err
+			continue
 		}
-		close(doneTasks)
-		close(undoneTasks)
-	}()
 
-	time.Sleep(time.Second * 3)
+		resChan <- res
+	}
+
+}
 
-	println("Errors:")
-	for r := range err {
-		println(r)
+func printErr(buf []error) {
+	fmt.Println("Errors:")
+	for _, err := range buf {
+		fmt.Println(err)
 	}
+}
 
-	println("Done tasks:")
-	for r := range result {
-		println(r)
+func printRes(buf []TaskResult) {
+	fmt.Println("Done:")
+	for _, res := range buf {
+		fmt.Println(res)
 	}
 }
+
+// ============================================================================
+
+func main() {
+	ctx, cancel := context.WithTimeout(context.Background(), TotalWorkTime)
+	defer cancel()
+
+	ticker := time.NewTicker(PrintInterval)
+	defer ticker.Stop()
+
+	var wg sync.WaitGroup
+
+	tasksChan := make(chan Task, ApproxBufSize)
+	resChan := make(chan TaskResult, ResultBufSize)
+	errChan := make(chan error, ResultBufSize)
+
+	// --------------------------------------
+
+	numCPU := runtime.NumCPU()
+	numWorkers := 1
+	if numCPU > 2 {
+		numWorkers = numCPU - 2
+	}
+
+	wg.Add(1)
+	go taskProducer(ctx, tasksChan, &wg)
+
+	for i := 0; i < numWorkers; i++ {
+		wg.Add(1)
+		go taskWorker(ctx, tasksChan, resChan, errChan, &wg)
+	}
+
+	// --------------------------------------
+
+	resBuf := make([]TaskResult, 0, ResultBufSize)
+	errBuf := make([]error, 0, ResultBufSize)
+
+	for {
+		select {
+		case <-ctx.Done():
+      		close(tasksChan)
+			wg.Wait()
+			close(resChan)
+			close(errChan)
+			fmt.Println("Finished.")
+			return
+
+		case <-ticker.C:
+			printErr(errBuf)
+			printRes(resBuf)
+			errBuf = errBuf[:0]
+			resBuf = resBuf[:0]
+
+		case res := <-resChan:
+			resBuf = append(resBuf, res)
+
+		case err := <-errChan:
+			errBuf = append(errBuf, err)
+
+		}
+	}
+
+}