Delete internal/cluster/worker_pool.go

2026-06-12 22:55:13 +00:00
parent f67a810dc6
commit 0f8b3c857c
1 changed files with 0 additions and 216 deletions
--- a/internal/cluster/worker_pool.go
+++ b/internal/cluster/worker_pool.go
@@ -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()
    }()
 }