futriix/internal/cluster/worker_pool.go

/*
 * 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()
    }()
}