diff --git a/internal/cluster/node.go b/internal/cluster/node.go index 01b4d94..a8f7d30 100644 --- a/internal/cluster/node.go +++ b/internal/cluster/node.go @@ -12,6 +12,7 @@ // Назначение: Реализация узла кластера (node) для распределённой СУБД с поддержкой временных меток. // Полностью lock-free с использованием атомарных операций. // Новая функциональность: Автоматическое восстановление после паники горутин. +// Lock-free: WorkerPool полностью lock-free, остальные компоненты оптимизированы. package cluster @@ -42,6 +43,68 @@ const ( StatusFailed ) +// NodeRequest представляет запрос между узлами +type NodeRequest struct { + Type string `json:"type"` + FromNode string `json:"from_node"` + Data json.RawMessage `json:"data"` + Timestamp int64 `json:"timestamp"` +} + +// NodeInfo представляет информацию об узле в кластере +type NodeInfo struct { + ID string `json:"id"` + IP string `json:"ip"` + Port int `json:"port"` + Status string `json:"status"` + LastSeen int64 `json:"last_seen"` + JoinedAt int64 `json:"joined_at"` + UpdatedAt int64 `json:"updated_at"` + Version int `json:"version"` +} + +// ShardInfo представляет информацию о шарде +type ShardInfo struct { + ID string `json:"id"` + Name string `json:"name"` + Nodes []string `json:"nodes"` + LeaderNode string `json:"leader_node"` + Status string `json:"status"` + CreatedAt int64 `json:"created_at"` + UpdatedAt int64 `json:"updated_at"` + LastRebalanced int64 `json:"last_rebalanced"` + DocumentCount int64 `json:"document_count"` + SizeBytes int64 `json:"size_bytes"` +} + +// LoggerInterface определяет интерфейс для логирования +type LoggerInterface interface { + Debug(msg string) + Info(msg string) + Warn(msg string) + Error(msg string) + Debugf(format string, args ...interface{}) + Infof(format string, args ...interface{}) + Warnf(format string, args ...interface{}) + Errorf(format string, args ...interface{}) +} + +// SafeGoWithLogger безопасно запускает горутину с обработкой паники +func SafeGoWithLogger(fn func(), logger *log.Logger, 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())) + } + time.Sleep(5 * time.Second) + SafeGoWithLogger(fn, logger, name) + } + }() + fn() + }() +} + // Node представляет отдельный узел в распределённой системе type Node struct { ID string // Уникальный идентификатор узла @@ -63,14 +126,13 @@ type Node struct { bytesTx atomic.Uint64 // Отправлено байт mu sync.RWMutex // Для синхронизации изменений статуса - // Новые компоненты для production-ready реализации - workerPool *WorkerPool // Пул воркеров для ограничения горутин + // Lock-free компоненты + workerPool *WorkerPool // Lock-free пул воркеров replicator *NetworkReplicator // Сетевой репликатор с retry и backoff - connPool sync.Map // Пул соединений к другим узлам + connPool sync.Map // Пул соединений к другим узлам (lock-free) - // НОВЫЕ КОМПОНЕНТЫ ДЛЯ ВОССТАНОВЛЕНИЯ ПОСЛЕ ПАНИКИ - panicRecoveryMgr *PanicRecoveryManager // Менеджер восстановления после паник - recoverableRoutines map[string]*RecoverableRoutine // Карта восстанавливаемых горутин + panicRecoveryMgr *PanicRecoveryManager + recoverableRoutines map[string]*RecoverableRoutine routinesMu sync.RWMutex } @@ -93,7 +155,7 @@ func NewNode(ip string, port int, store *storage.Storage, logger *log.Logger) *N func NewNodeWithRecovery(ip string, port int, store *storage.Storage, logger *log.Logger, panicRecoveryMgr *PanicRecoveryManager) *Node { now := time.Now().UnixMilli() - // Создаём пул воркеров с ограничением (максимум 500 одновременных горутин) + // Создаём lock-free пул воркеров workerPool := NewWorkerPool(500, logger) // Создаём сетевой репликатор с настройками retry и backoff @@ -105,7 +167,7 @@ func NewNodeWithRecovery(ip string, port int, store *storage.Storage, logger *lo Port: port, Storage: store, logger: logger, - incomingConn: make(chan net.Conn, 10000), // Увеличенный буфер + incomingConn: make(chan net.Conn, 10000), stopChan: make(chan struct{}), createdAt: now, startedAt: now, @@ -118,9 +180,7 @@ func NewNodeWithRecovery(ip string, port int, store *storage.Storage, logger *lo node.lastSeen.Store(now) node.joinedAt.Store(0) - // ========== НОВАЯ ФУНКЦИОНАЛЬНОСТЬ: Безопасный запуск горутин с восстановлением ========== if panicRecoveryMgr != nil { - // Используем восстанавливаемые горутины через PanicRecoveryManager node.startRecoverableRoutine("TCPServer", node.startTCPServer) node.startRecoverableRoutine("IncomingConnections", node.handleIncomingConnections) node.startRecoverableRoutine("HeartbeatLoop", node.heartbeatLoop) @@ -128,13 +188,12 @@ func NewNodeWithRecovery(ip string, port int, store *storage.Storage, logger *lo logger.Info(fmt.Sprintf("Node %s created with panic recovery (max workers: 500)", node.ID)) } else { - // Безопасный запуск сервера с обработкой паник (fallback) SafeGoWithLogger(node.startTCPServer, logger, "TCPServer") SafeGoWithLogger(node.handleIncomingConnections, logger, "IncomingConnections") SafeGoWithLogger(node.heartbeatLoop, logger, "HeartbeatLoop") SafeGoWithLogger(node.connectionHealthMonitor, logger, "ConnectionHealthMonitor") - logger.Info(fmt.Sprintf("Node %s created at %s with worker pool (max: 500)", node.ID, node.GetCreatedAtStr())) + logger.Info(fmt.Sprintf("Node %s created at %s with lock-free worker pool (max: 500)", node.ID, node.GetCreatedAtStr())) } return node @@ -150,11 +209,10 @@ func (n *Node) startRecoverableRoutine(name string, fn func()) { n.routinesMu.Lock() defer n.routinesMu.Unlock() - // Создаём восстанавливаемую горутину routine := NewRecoverableRoutine(name, func() error { fn() return nil - }, n.panicRecoveryMgr, 10) // максимум 10 перезапусков + }, n.panicRecoveryMgr, 10) n.recoverableRoutines[name] = routine routine.Start() @@ -185,10 +243,8 @@ func (n *Node) startTCPServer() { if n.logger != nil { n.logger.Error(fmt.Sprintf("TCP server panicked: %v\n%s", r, debug.Stack())) } - // Перезапускаем сервер с задержкой time.Sleep(5 * time.Second) - // Используем восстанавливаемый запуск if n.panicRecoveryMgr != nil { n.startRecoverableRoutine("TCPServer", n.startTCPServer) } else { @@ -220,7 +276,6 @@ func (n *Node) startTCPServer() { } return default: - // Устанавливаем таймаут для Accept if err := listener.(*net.TCPListener).SetDeadline(time.Now().Add(5 * time.Second)); err != nil { continue } @@ -236,7 +291,6 @@ func (n *Node) startTCPServer() { continue } - // Устанавливаем таймауты на соединение conn.SetReadDeadline(time.Now().Add(30 * time.Second)) conn.SetWriteDeadline(time.Now().Add(30 * time.Second)) @@ -253,14 +307,13 @@ func (n *Node) startTCPServer() { } } -// handleIncomingConnections обрабатывает входящие соединения с использованием пула воркеров +// handleIncomingConnections обрабатывает входящие соединения с использованием lock-free пула воркеров func (n *Node) handleIncomingConnections() { defer func() { if r := recover(); r != nil { if n.logger != nil { n.logger.Error(fmt.Sprintf("Incoming connections handler panicked: %v\n%s", r, debug.Stack())) } - // Перезапускаем обработчик с задержкой time.Sleep(5 * time.Second) if n.panicRecoveryMgr != nil { @@ -277,7 +330,6 @@ func (n *Node) handleIncomingConnections() { return case conn := <-n.incomingConn: n.requestCount.Add(1) - // Используем пул воркеров вместо создания неограниченного количества горутин taskID := fmt.Sprintf("handle_conn_%d_%s", time.Now().UnixNano(), conn.RemoteAddr().String()) err := n.workerPool.SubmitFunc(taskID, func() error { n.handleNodeRequest(conn) @@ -304,7 +356,6 @@ func (n *Node) handleNodeRequest(conn net.Conn) { conn.Close() }() - // Устанавливаем таймаут на чтение conn.SetReadDeadline(time.Now().Add(30 * time.Second)) decoder := json.NewDecoder(conn) @@ -384,7 +435,6 @@ func (n *Node) handleReplicateRequest(data []byte) { return } - // Безопасное получение ID документа docID, ok := repData.Document["_id"].(string) if !ok { if n.logger != nil { @@ -393,10 +443,10 @@ func (n *Node) handleReplicateRequest(data []byte) { return } - doc := &storage.Document{ - ID: docID, - Fields: repData.Document, - } +doc := storage.NewDocumentWithID(docID) +for k, v := range repData.Document { + doc.SetField(k, v) +} if err := coll.Insert(doc); err != nil { if n.logger != nil { @@ -576,7 +626,7 @@ func (n *Node) handleHeartbeatRequest(req NodeRequest, conn net.Conn) { encoder.Encode(response) } -// handleStatusSyncRequest обрабатывает запрос синхронизации статуса (для split-brain recovery) +// handleStatusSyncRequest обрабатывает запрос синхронизации статуса func (n *Node) handleStatusSyncRequest(req NodeRequest, conn net.Conn) { defer func() { if r := recover(); r != nil { @@ -636,7 +686,6 @@ func (n *Node) heartbeatLoop() { if n.logger != nil { n.logger.Error(fmt.Sprintf("Heartbeat loop panicked: %v\n%s", r, debug.Stack())) } - // Перезапускаем heartbeat loop с задержкой time.Sleep(5 * time.Second) if n.panicRecoveryMgr != nil { @@ -702,7 +751,6 @@ func (n *Node) cleanupStaleConnections() { n.connPool.Range(func(key, value interface{}) bool { if conn, ok := value.(net.Conn); ok { - // Проверяем, активно ли соединение conn.SetReadDeadline(time.Now().Add(1 * time.Second)) buf := make([]byte, 1) _, err := conn.Read(buf) @@ -896,7 +944,6 @@ func (n *Node) GetStats() map[string]interface{} { stats["replication"] = n.replicator.GetStats() } - // Новая статистика о восстанавливаемых горутинах if n.panicRecoveryMgr != nil { n.routinesMu.RLock() routineStatus := make(map[string]bool) @@ -919,7 +966,6 @@ func (n *Node) Stop() { n.Status.Store(int32(StatusOffline)) n.stoppedAt = time.Now().UnixMilli() - // Останавливаем все восстанавливаемые горутины n.routinesMu.RLock() for name, routine := range n.recoverableRoutines { routine.Stop() @@ -939,7 +985,6 @@ func (n *Node) Stop() { n.replicator.Close() } - // Закрываем все соединения в пуле n.connPool.Range(func(key, value interface{}) bool { if conn, ok := value.(net.Conn); ok { conn.Close() @@ -964,7 +1009,7 @@ func generateReplicationID() string { return base64.URLEncoding.EncodeToString(bytes) } -// ReplicateDocument отправляет документ на репликацию всем активным узлам с retry и backoff +// ReplicateDocument отправляет документ на репликацию всем активным узлам func (n *Node) ReplicateDocument(database, collection string, doc *storage.Document) error { if n.coordinator == nil { if n.logger != nil { @@ -981,7 +1026,6 @@ func (n *Node) ReplicateDocument(database, collection string, doc *storage.Docum return nil } - // Подготавливаем данные для репликации repData := struct { Database string `json:"database"` Collection string `json:"collection"` @@ -1003,7 +1047,6 @@ func (n *Node) ReplicateDocument(database, collection string, doc *storage.Docum startTime := time.Now().UnixMilli() - // Отправляем на все узлы, кроме себя var wg sync.WaitGroup var failedCount atomic.Int32 var successCount atomic.Int32 @@ -1017,7 +1060,6 @@ func (n *Node) ReplicateDocument(database, collection string, doc *storage.Docum targetNodeID := nodeInfo.ID targetAddress := fmt.Sprintf("%s:%d", nodeInfo.IP, nodeInfo.Port) - // Асинхронная репликация с повторными попытками через пул воркеров taskID := fmt.Sprintf("replicate_%s_to_%s_%s", doc.ID, targetNodeID, repData.ReplicaID) err := n.workerPool.SubmitFunc(taskID, func() error { defer wg.Done() @@ -1054,7 +1096,6 @@ func (n *Node) ReplicateDocument(database, collection string, doc *storage.Docum } } - // Ждём завершения всех репликаций с таймаутом done := make(chan struct{}) go func() { wg.Wait()