nsqd: add flag --min-output-buffer-timeout with default 25ms

apps/nsqd/nsqd.go

@@ -117,6 +117,7 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet {
 	flagSet.Int64("max-rdy-count", opts.MaxRdyCount, "maximum RDY count for a client")
 	flagSet.Int64("max-output-buffer-size", opts.MaxOutputBufferSize, "maximum client configurable size (in bytes) for a client output buffer")
 	flagSet.Duration("max-output-buffer-timeout", opts.MaxOutputBufferTimeout, "maximum client configurable duration of time between flushing to a client")
+	flagSet.Duration("min-output-buffer-timeout", opts.MinOutputBufferTimeout, "minimum client configurable duration of time between flushing to a client")
 	flagSet.Duration("output-buffer-timeout", opts.OutputBufferTimeout, "default duration of time between flushing data to clients")
 
 	// statsd integration options

nsqd/client_v2.go

@@ -168,12 +168,7 @@ func (c *clientV2) Identify(data identifyDataV2) error {
 		return err
 	}
 
-	err = c.SetOutputBufferSize(data.OutputBufferSize)
-	if err != nil {
-		return err
-	}
-
-	err = c.SetOutputBufferTimeout(data.OutputBufferTimeout)
+	err = c.SetOutputBuffer(data.OutputBufferSize, data.OutputBufferTimeout)
 	if err != nil {
 		return err
 	}
@@ -413,49 +408,43 @@ func (c *clientV2) SetHeartbeatInterval(desiredInterval int) error {
 	return nil
 }
 
-func (c *clientV2) SetOutputBufferSize(desiredSize int) error {
-	var size int
+func (c *clientV2) SetOutputBuffer(desiredSize int, desiredTimeout int) error {
+	c.writeLock.Lock()
+	defer c.writeLock.Unlock()
+
+	switch {
+	case desiredTimeout == -1:
+		c.OutputBufferTimeout = 0
+	case desiredTimeout == 0:
+		// do nothing (use default)
+	case true &&
+		desiredTimeout >= int(c.ctx.nsqd.getOpts().MinOutputBufferTimeout/time.Millisecond) &&
+		desiredTimeout <= int(c.ctx.nsqd.getOpts().MaxOutputBufferTimeout/time.Millisecond):
+
+		c.OutputBufferTimeout = time.Duration(desiredTimeout) * time.Millisecond
+	default:
+		return fmt.Errorf("output buffer timeout (%d) is invalid", desiredTimeout)
+	}
 
 	switch {
 	case desiredSize == -1:
 		// effectively no buffer (every write will go directly to the wrapped net.Conn)
-		size = 1
+		c.OutputBufferSize = 1
+		c.OutputBufferTimeout = 0
 	case desiredSize == 0:
 		// do nothing (use default)
 	case desiredSize >= 64 && desiredSize <= int(c.ctx.nsqd.getOpts().MaxOutputBufferSize):
-		size = desiredSize
+		c.OutputBufferSize = desiredSize
 	default:
 		return fmt.Errorf("output buffer size (%d) is invalid", desiredSize)
 	}
 
-	if size > 0 {
-		c.writeLock.Lock()
-		defer c.writeLock.Unlock()
-		c.OutputBufferSize = size
+	if desiredSize != 0 {
 		err := c.Writer.Flush()
 		if err != nil {
 			return err
 		}
-		c.Writer = bufio.NewWriterSize(c.Conn, size)
-	}
-
-	return nil
-}
-
-func (c *clientV2) SetOutputBufferTimeout(desiredTimeout int) error {
-	c.writeLock.Lock()
-	defer c.writeLock.Unlock()
-
-	switch {
-	case desiredTimeout == -1:
-		c.OutputBufferTimeout = 0
-	case desiredTimeout == 0:
-		// do nothing (use default)
-	case desiredTimeout >= 1 &&
-		desiredTimeout <= int(c.ctx.nsqd.getOpts().MaxOutputBufferTimeout/time.Millisecond):
-		c.OutputBufferTimeout = time.Duration(desiredTimeout) * time.Millisecond
-	default:
-		return fmt.Errorf("output buffer timeout (%d) is invalid", desiredTimeout)
+		c.Writer = bufio.NewWriterSize(c.Conn, c.OutputBufferSize)
 	}
 
 	return nil

nsqd/options.go

@@ -56,6 +56,7 @@ type Options struct {
 	MaxRdyCount            int64         `flag:"max-rdy-count"`
 	MaxOutputBufferSize    int64         `flag:"max-output-buffer-size"`
 	MaxOutputBufferTimeout time.Duration `flag:"max-output-buffer-timeout"`
+	MinOutputBufferTimeout time.Duration `flag:"min-output-buffer-timeout"`
 	OutputBufferTimeout    time.Duration `flag:"output-buffer-timeout"`
 
 	// statsd integration
@@ -130,7 +131,8 @@ func NewOptions() *Options {
 		MaxHeartbeatInterval:   60 * time.Second,
 		MaxRdyCount:            2500,
 		MaxOutputBufferSize:    64 * 1024,
-		MaxOutputBufferTimeout: 1 * time.Second,
+		MaxOutputBufferTimeout: 30 * time.Second,
+		MinOutputBufferTimeout: 25 * time.Millisecond,
 		OutputBufferTimeout:    250 * time.Millisecond,
 
 		StatsdPrefix:        "nsq.%s",