gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Upload files to "internal/cluster"

Browse Source
This commit is contained in:
Григорий Сафронов
2026-06-12 22:56:18 +00:00
parent 062d5b8ba4
commit e2d101b522
1 changed files with 79 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
internal/cluster/node.go
View File

@@ -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()