From 0f8b3c857cf30c98ef1b3a28347f233c53e85236 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=93=D1=80=D0=B8=D0=B3=D0=BE=D1=80=D0=B8=D0=B9=20=D0=A1?= =?UTF-8?q?=D0=B0=D1=84=D1=80=D0=BE=D0=BD=D0=BE=D0=B2?= Date: Fri, 12 Jun 2026 22:55:13 +0000 Subject: [PATCH] Delete internal/cluster/worker_pool.go --- internal/cluster/worker_pool.go | 216 -------------------------------- 1 file changed, 216 deletions(-) delete mode 100644 internal/cluster/worker_pool.go diff --git a/internal/cluster/worker_pool.go b/internal/cluster/worker_pool.go deleted file mode 100644 index 2b18e6b..0000000 --- a/internal/cluster/worker_pool.go +++ /dev/null @@ -1,216 +0,0 @@ -/* - * Copyright 2026 Safronov Grigorii - * - * Licensed under the CDDL, Version 1.0 (the "License"); - * you may not use this file except in compliance with the License. - * - * You may obtain a copy of the License at - * https://opensource.org/licenses/CDDL-1.0 - */ - -// Файл: internal/cluster/worker_pool.go -// Назначение: Пул воркеров для ограничения количества горутин - -package cluster - -import ( - "fmt" - "runtime/debug" - "sync" - "sync/atomic" - "time" -) - -// LoggerInterface определяет интерфейс для логирования -type LoggerInterface interface { - Debug(msg string) - Info(msg string) - Error(msg string) - Warn(msg string) -} - -// WorkerTask представляет задачу для выполнения в пуле -type WorkerTask struct { - ID string - Handler func() error - CreatedAt int64 -} - -// WorkerPool управляет пулом воркеров -type WorkerPool struct { - maxWorkers int - taskQueue chan *WorkerTask - wg sync.WaitGroup - stopChan chan struct{} - activeTasks atomic.Int32 - totalTasks atomic.Uint64 - failedTasks atomic.Uint64 - logger LoggerInterface - panicHandler func(interface{}) -} - -// NewWorkerPool создаёт новый пул воркеров -func NewWorkerPool(maxWorkers int, logger LoggerInterface) *WorkerPool { - if maxWorkers <= 0 { - maxWorkers = 100 // ограничение по умолчанию - } - - pool := &WorkerPool{ - maxWorkers: maxWorkers, - taskQueue: make(chan *WorkerTask, maxWorkers*10), - stopChan: make(chan struct{}), - logger: logger, - panicHandler: defaultPanicHandler, - } - - for i := 0; i < maxWorkers; i++ { - pool.wg.Add(1) - go pool.worker(i) - } - - return pool -} - -// defaultPanicHandler обрабатывает панику в горутине -func defaultPanicHandler(r interface{}) { - fmt.Printf("PANIC recovered: %v\n%s\n", r, debug.Stack()) -} - -// SetPanicHandler устанавливает обработчик паник -func (wp *WorkerPool) SetPanicHandler(handler func(interface{})) { - if handler != nil { - wp.panicHandler = handler - } -} - -// worker запускает воркер -func (wp *WorkerPool) worker(id int) { - defer func() { - if r := recover(); r != nil { - if wp.panicHandler != nil { - wp.panicHandler(r) - } - if wp.logger != nil { - wp.logger.Error(fmt.Sprintf("Worker %d panicked: %v", id, r)) - } - // Перезапускаем воркер - wp.wg.Add(1) - go wp.worker(id) - } - wp.wg.Done() - }() - - for { - select { - case <-wp.stopChan: - return - case task, ok := <-wp.taskQueue: - if !ok { - return - } - - wp.activeTasks.Add(1) - wp.totalTasks.Add(1) - - func() { - defer func() { - if r := recover(); r != nil { - wp.failedTasks.Add(1) - if wp.panicHandler != nil { - wp.panicHandler(r) - } - if wp.logger != nil { - wp.logger.Error(fmt.Sprintf("Task %s panicked: %v", task.ID, r)) - } - } - wp.activeTasks.Add(-1) - }() - - if err := task.Handler(); err != nil { - wp.failedTasks.Add(1) - if wp.logger != nil { - wp.logger.Error(fmt.Sprintf("Task %s failed: %v", task.ID, err)) - } - } - }() - } - } -} - -// Submit отправляет задачу в пул -func (wp *WorkerPool) Submit(task *WorkerTask) error { - select { - case wp.taskQueue <- task: - return nil - case <-time.After(5 * time.Second): - return fmt.Errorf("task queue full, task rejected: %s", task.ID) - } -} - -// SubmitFunc отправляет функцию как задачу -func (wp *WorkerPool) SubmitFunc(id string, handler func() error) error { - return wp.Submit(&WorkerTask{ - ID: id, - Handler: handler, - CreatedAt: time.Now().UnixMilli(), - }) -} - -// GetStats возвращает статистику пула -func (wp *WorkerPool) GetStats() map[string]interface{} { - return map[string]interface{}{ - "max_workers": wp.maxWorkers, - "active_tasks": wp.activeTasks.Load(), - "total_tasks": wp.totalTasks.Load(), - "failed_tasks": wp.failedTasks.Load(), - "queue_size": len(wp.taskQueue), - "queue_capacity": cap(wp.taskQueue), - } -} - -// GetWorkerPoolStats возвращает статистику в структурированном виде -func (wp *WorkerPool) GetWorkerPoolStats() *WorkerPoolStats { - return &WorkerPoolStats{ - MaxWorkers: wp.maxWorkers, - ActiveTasks: wp.activeTasks.Load(), - TotalTasks: wp.totalTasks.Load(), - FailedTasks: wp.failedTasks.Load(), - QueueSize: len(wp.taskQueue), - QueueCapacity: cap(wp.taskQueue), - } -} - -// Stop останавливает пул воркеров -func (wp *WorkerPool) Stop() { - close(wp.stopChan) - close(wp.taskQueue) - wp.wg.Wait() -} - -// SafeGo безопасно запускает горутину с обработкой паники -func SafeGo(fn func(), panicHandler func(interface{})) { - go func() { - defer func() { - if r := recover(); r != nil { - if panicHandler != nil { - panicHandler(r) - } - } - }() - fn() - }() -} - -// SafeGoWithLogger безопасно запускает горутину с логгером -func SafeGoWithLogger(fn func(), logger LoggerInterface, name string) { - go func() { - defer func() { - if r := recover(); r != nil { - if logger != nil { - logger.Error(fmt.Sprintf("Goroutine %s panicked: %v\n%s", name, r, debug.Stack())) - } - } - }() - fn() - }() -}