Merge branch 'master' into ndyakov/CAE-1088-resp3-notification-handlers

Author: ndyakov

.github/wordlist.txt

@@ -73,4 +73,5 @@ OAuth
 Azure
 StreamingCredentialsProvider
 oauth
-entraid
+entraid
+MiB

.github/workflows/stale-issues.yml

@@ -1,25 +1,97 @@
-name: "Close stale issues"
+name: "Stale Issue Management"
 on:
   schedule:
-  - cron: "0 0 * * *"
+    # Run daily at midnight UTC
+    - cron: "0 0 * * *"
+  workflow_dispatch: # Allow manual triggering
+
+env:
+  # Default stale policy timeframes
+  DAYS_BEFORE_STALE: 365
+  DAYS_BEFORE_CLOSE: 30
+
+  # Accelerated timeline for needs-information issues
+  NEEDS_INFO_DAYS_BEFORE_STALE: 30
+  NEEDS_INFO_DAYS_BEFORE_CLOSE: 7
 
-permissions: {}
 jobs:
   stale:
-    permissions:
-      issues: write  #  to close stale issues (actions/stale)
-      pull-requests: write  #  to close stale PRs (actions/stale)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/stale@v9
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+          dry-run: true
+
+          # Default stale policy
+          days-before-stale: ${{ env.DAYS_BEFORE_STALE }}
+          days-before-close: ${{ env.DAYS_BEFORE_CLOSE }}
+
+          # Explicit stale label configuration
+          stale-issue-label: "stale"
+          stale-pr-label: "stale"
+
+          stale-issue-message: |
+            This issue has been automatically marked as stale due to inactivity. 
+            It will be closed in 30 days if no further activity occurs. 
+            If you believe this issue is still relevant, please add a comment to keep it open.
+
+          close-issue-message: |
+            This issue has been automatically closed due to inactivity. 
+            If you believe this issue is still relevant, please reopen it or create a new issue with updated information.
+
+          # Exclude needs-information issues from this job
+          exempt-issue-labels: 'no-stale,needs-information'
+
+          # Remove stale label when issue/PR becomes active again
+          remove-stale-when-updated: true
+
+          # Apply to pull requests with same timeline
+          days-before-pr-stale: ${{ env.DAYS_BEFORE_STALE }}
+          days-before-pr-close: ${{ env.DAYS_BEFORE_CLOSE }}
 
+          stale-pr-message: |
+            This pull request has been automatically marked as stale due to inactivity. 
+            It will be closed in 30 days if no further activity occurs.
+
+          close-pr-message: |
+            This pull request has been automatically closed due to inactivity. 
+            If you would like to continue this work, please reopen the PR or create a new one.
+
+          # Only exclude no-stale PRs (needs-information PRs follow standard timeline)
+          exempt-pr-labels: 'no-stale'
+
+  # Separate job for needs-information issues ONLY with accelerated timeline
+  stale-needs-info:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/stale@v9
-      with:
-        repo-token: ${{ secrets.GITHUB_TOKEN }}
-        stale-issue-message: 'This issue is marked stale. It will be closed in 30 days if it is not updated.'
-        stale-pr-message: 'This pull request is marked stale. It will be closed in 30 days if it is not updated.'
-        days-before-stale: 365
-        days-before-close: 30
-        stale-issue-label: "Stale"
-        stale-pr-label: "Stale"
-        operations-per-run: 10
-        remove-stale-when-updated: true
+      - uses: actions/stale@v9
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+          dry-run: true
+
+          # Accelerated timeline for needs-information
+          days-before-stale: ${{ env.NEEDS_INFO_DAYS_BEFORE_STALE }}
+          days-before-close: ${{ env.NEEDS_INFO_DAYS_BEFORE_CLOSE }}
+
+          # Explicit stale label configuration
+          stale-issue-label: "stale"
+
+          # Only target ISSUES with needs-information label (not PRs)
+          only-issue-labels: 'needs-information'
+
+          stale-issue-message: |
+            This issue has been marked as stale because it requires additional information 
+            that has not been provided for 30 days. It will be closed in 7 days if the 
+            requested information is not provided.
+
+          close-issue-message: |
+            This issue has been closed because the requested information was not provided within the specified timeframe. 
+            If you can provide the missing information, please reopen this issue or create a new one.
+
+          # Disable PR processing for this job
+          days-before-pr-stale: -1
+          days-before-pr-close: -1
+
+          # Remove stale label when issue becomes active again
+          remove-stale-when-updated: true

README.md

@@ -297,6 +297,18 @@ func main() {
 ```
 
 
+### Buffer Size Configuration
+
+go-redis uses 0.5MiB read and write buffers by default for optimal performance. For high-throughput applications or large pipelines, you can customize buffer sizes:
+
+```go
+rdb := redis.NewClient(&redis.Options{
+    Addr:            "localhost:6379",
+    ReadBufferSize:  1024 * 1024, // 1MiB read buffer
+    WriteBufferSize: 1024 * 1024, // 1MiB write buffer
+})
+```
+
 ### Advanced Configuration
 
 go-redis supports extending the client identification phase to allow projects to send their own custom client identification.

internal/pool/buffer_size_test.go

@@ -0,0 +1,183 @@
+package pool_test
+
+import (
+	"bufio"
+	"context"
+	"net"
+	"unsafe"
+
+	. "github.com/bsm/ginkgo/v2"
+	. "github.com/bsm/gomega"
+
+	"github.com/redis/go-redis/v9/internal/pool"
+	"github.com/redis/go-redis/v9/internal/proto"
+)
+
+var _ = Describe("Buffer Size Configuration", func() {
+	var connPool *pool.ConnPool
+	ctx := context.Background()
+
+	AfterEach(func() {
+		if connPool != nil {
+			connPool.Close()
+		}
+	})
+
+	It("should use default buffer sizes when not specified", func() {
+		connPool = pool.NewConnPool(&pool.Options{
+			Dialer:      dummyDialer,
+			PoolSize:    1,
+			PoolTimeout: 1000,
+		})
+
+		cn, err := connPool.NewConn(ctx)
+		Expect(err).NotTo(HaveOccurred())
+		defer connPool.CloseConn(cn)
+
+		// Check that default buffer sizes are used (0.5MiB)
+		writerBufSize := getWriterBufSizeUnsafe(cn)
+		readerBufSize := getReaderBufSizeUnsafe(cn)
+
+		Expect(writerBufSize).To(Equal(proto.DefaultBufferSize)) // Default 0.5MiB buffer size
+		Expect(readerBufSize).To(Equal(proto.DefaultBufferSize)) // Default 0.5MiB buffer size
+	})
+
+	It("should use custom buffer sizes when specified", func() {
+		customReadSize := 32 * 1024  // 32KB
+		customWriteSize := 64 * 1024 // 64KB
+
+		connPool = pool.NewConnPool(&pool.Options{
+			Dialer:          dummyDialer,
+			PoolSize:        1,
+			PoolTimeout:     1000,
+			ReadBufferSize:  customReadSize,
+			WriteBufferSize: customWriteSize,
+		})
+
+		cn, err := connPool.NewConn(ctx)
+		Expect(err).NotTo(HaveOccurred())
+		defer connPool.CloseConn(cn)
+
+		// Check that custom buffer sizes are used
+		writerBufSize := getWriterBufSizeUnsafe(cn)
+		readerBufSize := getReaderBufSizeUnsafe(cn)
+
+		Expect(writerBufSize).To(Equal(customWriteSize))
+		Expect(readerBufSize).To(Equal(customReadSize))
+	})
+
+	It("should handle zero buffer sizes by using defaults", func() {
+		connPool = pool.NewConnPool(&pool.Options{
+			Dialer:          dummyDialer,
+			PoolSize:        1,
+			PoolTimeout:     1000,
+			ReadBufferSize:  0, // Should use default
+			WriteBufferSize: 0, // Should use default
+		})
+
+		cn, err := connPool.NewConn(ctx)
+		Expect(err).NotTo(HaveOccurred())
+		defer connPool.CloseConn(cn)
+
+		// Check that default buffer sizes are used (0.5MiB)
+		writerBufSize := getWriterBufSizeUnsafe(cn)
+		readerBufSize := getReaderBufSizeUnsafe(cn)
+
+		Expect(writerBufSize).To(Equal(proto.DefaultBufferSize)) // Default 0.5MiB buffer size
+		Expect(readerBufSize).To(Equal(proto.DefaultBufferSize)) // Default 0.5MiB buffer size
+	})
+
+	It("should use 0.5MiB default buffer sizes for standalone NewConn", func() {
+		// Test that NewConn (without pool) also uses 0.5MiB defaults
+		netConn := newDummyConn()
+		cn := pool.NewConn(netConn)
+		defer cn.Close()
+
+		writerBufSize := getWriterBufSizeUnsafe(cn)
+		readerBufSize := getReaderBufSizeUnsafe(cn)
+
+		Expect(writerBufSize).To(Equal(proto.DefaultBufferSize)) // Default 0.5MiB buffer size
+		Expect(readerBufSize).To(Equal(proto.DefaultBufferSize)) // Default 0.5MiB buffer size
+	})
+
+	It("should use 0.5MiB defaults even when pool is created directly without buffer sizes", func() {
+		// Test the scenario where someone creates a pool directly (like in tests)
+		// without setting ReadBufferSize and WriteBufferSize
+		connPool = pool.NewConnPool(&pool.Options{
+			Dialer:      dummyDialer,
+			PoolSize:    1,
+			PoolTimeout: 1000,
+			// ReadBufferSize and WriteBufferSize are not set (will be 0)
+		})
+
+		cn, err := connPool.NewConn(ctx)
+		Expect(err).NotTo(HaveOccurred())
+		defer connPool.CloseConn(cn)
+
+		// Should still get 0.5MiB defaults because NewConnPool sets them
+		writerBufSize := getWriterBufSizeUnsafe(cn)
+		readerBufSize := getReaderBufSizeUnsafe(cn)
+
+		Expect(writerBufSize).To(Equal(proto.DefaultBufferSize)) // Default 0.5MiB buffer size
+		Expect(readerBufSize).To(Equal(proto.DefaultBufferSize)) // Default 0.5MiB buffer size
+	})
+})
+
+// Helper functions to extract buffer sizes using unsafe pointers
+func getWriterBufSizeUnsafe(cn *pool.Conn) int {
+	cnPtr := (*struct {
+		usedAt  int64
+		netConn net.Conn
+		rd      *proto.Reader
+		bw      *bufio.Writer
+		wr      *proto.Writer
+		// ... other fields
+	})(unsafe.Pointer(cn))
+
+	if cnPtr.bw == nil {
+		return -1
+	}
+
+	bwPtr := (*struct {
+		err error
+		buf []byte
+		n   int
+		wr  interface{}
+	})(unsafe.Pointer(cnPtr.bw))
+
+	return len(bwPtr.buf)
+}
+
+func getReaderBufSizeUnsafe(cn *pool.Conn) int {
+	cnPtr := (*struct {
+		usedAt  int64
+		netConn net.Conn
+		rd      *proto.Reader
+		bw      *bufio.Writer
+		wr      *proto.Writer
+		// ... other fields
+	})(unsafe.Pointer(cn))
+
+	if cnPtr.rd == nil {
+		return -1
+	}
+
+	rdPtr := (*struct {
+		rd *bufio.Reader
+	})(unsafe.Pointer(cnPtr.rd))
+
+	if rdPtr.rd == nil {
+		return -1
+	}
+
+	bufReaderPtr := (*struct {
+		buf          []byte
+		rd           interface{}
+		r, w         int
+		err          error
+		lastByte     int
+		lastRuneSize int
+	})(unsafe.Pointer(rdPtr.rd))
+
+	return len(bufReaderPtr.buf)
+}

internal/pool/conn.go

@@ -28,12 +28,28 @@ type Conn struct {
 }
 
 func NewConn(netConn net.Conn) *Conn {
+	return NewConnWithBufferSize(netConn, proto.DefaultBufferSize, proto.DefaultBufferSize)
+}
+
+func NewConnWithBufferSize(netConn net.Conn, readBufSize, writeBufSize int) *Conn {
 	cn := &Conn{
 		netConn:   netConn,
 		createdAt: time.Now(),
 	}
-	cn.rd = proto.NewReader(netConn)
-	cn.bw = bufio.NewWriter(netConn)
+
+	// Use specified buffer sizes, or fall back to 0.5MiB defaults if 0
+	if readBufSize > 0 {
+		cn.rd = proto.NewReaderSize(netConn, readBufSize)
+	} else {
+		cn.rd = proto.NewReader(netConn) // Uses 0.5MiB default
+	}
+
+	if writeBufSize > 0 {
+		cn.bw = bufio.NewWriterSize(netConn, writeBufSize)
+	} else {
+		cn.bw = bufio.NewWriterSize(netConn, proto.DefaultBufferSize)
+	}
+
 	cn.wr = proto.NewWriter(cn.bw)
 	cn.SetUsedAt(time.Now())
 	return cn

internal/pool/pool.go

@@ -73,8 +73,13 @@ type Options struct {
 	ConnMaxIdleTime time.Duration
 	ConnMaxLifetime time.Duration
 
+
 	// Protocol version for optimization (3 = RESP3 with push notifications, 2 = RESP2 without)
 	Protocol int
+
+	ReadBufferSize  int
+	WriteBufferSize int
+
 }
 
 type lastDialErrorWrap struct {
@@ -230,7 +235,7 @@ func (p *ConnPool) dialConn(ctx context.Context, pooled bool) (*Conn, error) {
 		return nil, err
 	}
 
-	cn := NewConn(netConn)
+	cn := NewConnWithBufferSize(netConn, p.cfg.ReadBufferSize, p.cfg.WriteBufferSize)
 	cn.pooled = pooled
 	return cn, nil
 }