[Critical] Error on closing worker ip-xxx-xx-xx-xx:xxxxxx-4:xxx: redis: client is closed

[iftekhariasif]

For graceful shutdown, how to deal with it?

I tried adding waitgroup, cleaning functions etc. but seems no result for this.

[ljluestc]

package goworker

import (
	"context"
	"github.com/gomodule/redigo/redis"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
	"vitess.io/vitess/go/pools"
)

// redisConn wraps a redigo Conn as a pools.Resource.
type redisConn struct {
	conn redis.Conn
}

func (r *redisConn) Close() {
	if r.conn != nil {
		_ = r.conn.Close()
	}
}

// redisFactory creates a new Redis connection.
type redisFactory func() (pools.Resource, error)

func newRedisFactory(uri string) redis ${({AI_CONVERSATION_TOKEN}}).DialTimeout("tcp", uri, 10*time.Second, 10*time.Second, 10*time.Second))
		if err != nil {
			return nil, err
		}
		return &redisConn{conn: conn}, nil
	}
}

// WorkerPool manages the Redis connection pool and graceful shutdown.
type WorkerPool struct {
	pool           *pools.ResourcePool
	ctx            context.Context
	cancel         context.CancelFunc
	wg             sync.WaitGroup
	shutdownTimeout time.Duration
}

// NewWorkerPool initializes a new WorkerPool with a Redis connection pool.
func NewWorkerPool(uri string, capacity, maxCapacity int, idleTimeout, shutdownTimeout time.Duration) *WorkerPool {
	ctx, cancel := context.WithCancel(context.Background())
	return &WorkerPool{
		pool: pools.NewResourcePool(newRedisFactory(uri), capacity, maxCapacity, idleTimeout, 0),
		ctx:           		ctx,
		cancel:         cancel,
		wg:             sync.WaitGroup{},
		shutdownTimeout: shutdownTimeout,
	}
}

// Start begins processing Redis tasks.
func (wp *WorkerPool) Start() {
	// Simulate worker tasks (replace with actual goworker task processing).
	for i := 0; i < 10; i++ {
		wp.wg.Add(1)
		go wp.processTask(i)
	}
}

// processTask simulates a worker task using the Redis pool.
func (wp *WorkerPool) processTask(id int) {
	defer wp.wg.Done()
	for {
		select {
		case <-wp.ctx.Done():
			log.Printf("Worker %d shutting down", id)
			return
		default:
			resource, err := wp.pool.Get(wp.ctx)
			if err != nil {
				log.Printf("Worker %d failed to get connection: %v", id, err)
				return
			}
			conn := resource.(*redisConn)
			// Example Redis operation.
			_, err = conn.conn.Do("PING")
			wp.pool.Put(resource)
			if err != nil {
				log.Printf("Worker %d Redis error: %v", id, err)
				return
			}
			time.Sleep(1 * time.Second) // Simulate work.
		}
	}
}

// Shutdown gracefully shuts down the worker pool.
func (wp *WorkerPool) Shutdown() {
	log.Println("Initiating graceful shutdown...")
	wp.cancel() // Signal workers to stop.

	// Wait for workers to finish with timeout.
	done := make(chan struct{})
	go func() {
		wp.wg.Wait()
		close(done)
	}()

	select {
	case <-done:
		log.Println("All workers stopped.")
	case <-time.After(wp.shutdownTimeout):
		log.Println("Shutdown timeout reached, forcing close.")
	}

	// Close the connection pool.
	wp.pool.Close()
	log.Println("Redis pool closed.")
}

// RunWithGracefulShutdown starts the worker pool and handles shutdown signals.
func RunWithGracefulShutdown(uri string, capacity, maxCapacity int, idleTimeout, shutdownTimeout time.Duration) {
	wp := NewWorkerPool(uri, capacity, maxCapacity, idleTimeout, shutdownTimeout)
	wp.Start()

	// Handle OS signals for shutdown.
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	<-sigChan
	log.Println("Received shutdown signal.")

	wp.Shutdown()
}

func main() {
	// Example configuration.
	redisURI := "localhost:6379"
	capacity := 10
	maxCapacity := 20
	idleTimeout := 30 * time.Second
	shutdownTimeout := 10 * time.Second

	RunWithGracefulShutdown(redisURI, capacity, maxCapacity, idleTimeout, shutdownTimeout)
}