fix: channel min size is 1

lang/channel/channel.go

@@ -23,7 +23,7 @@ import (
 
 const (
 	defaultThrottleWindow = time.Millisecond * 100
-	defaultSize           = 0
+	defaultMinSize        = 1
 )
 
 type item struct {
@@ -45,12 +45,13 @@ type Option func(c *channel)
 // Throttle define channel Throttle function
 type Throttle func(c Channel) bool
 
-// WithSize define the size of channel.
+// WithSize define the size of channel. If channel is full, it will block.
+// Min channel size is 1.
 // It conflicts with WithNonBlock option.
 func WithSize(size int) Option {
 	return func(c *channel) {
 		// with non block mode, no need to change size
-		if !c.nonblock {
+		if size >= defaultMinSize && !c.nonblock {
 			c.size = size
 		}
 	}
@@ -61,7 +62,6 @@ func WithSize(size int) Option {
 func WithNonBlock() Option {
 	return func(c *channel) {
 		c.nonblock = true
-		c.size = 1024
 	}
 }
 
@@ -163,7 +163,7 @@ type Channel interface {
 
 // channel implements a safe and feature-rich channel struct for the real world.
 type channel struct {
-	size             int
+	size             int // min size is 1
 	state            int32
 	producer         chan interface{}
 	consumer         chan interface{}
@@ -186,13 +186,13 @@ type channel struct {
 // New create a new channel.
 func New(opts ...Option) Channel {
 	c := new(channel)
-	c.size = defaultSize
+	c.size = defaultMinSize
 	c.throttleWindow = defaultThrottleWindow
 	c.bufferCond = sync.NewCond(&c.bufferLock)
 	for _, opt := range opts {
 		opt(c)
 	}
-	c.producer = make(chan interface{}, c.size)
+	c.producer = make(chan interface{})
 	c.consumer = make(chan interface{})
 	c.buffer = list.New()
 	go c.produce()
@@ -242,10 +242,6 @@ func (c *channel) Stats() (uint64, uint64) {
 
 // produce used to process input channel
 func (c *channel) produce() {
-	capacity := c.size
-	if c.size == 0 {
-		capacity = 1
-	}
 	for p := range c.producer {
 		// only check throttle function in blocking mode
 		if !c.nonblock {
@@ -267,7 +263,8 @@ func (c *channel) produce() {
 		c.enqueueBuffer(it)
 		c.bufferCond.Signal()
 		if !c.nonblock {
-			for c.buffer.Len() >= capacity && !c.isClosed() {
+			for c.Len() >= c.size && !c.isClosed() {
+				// wait for consuming
 				c.bufferCond.Wait()
 			}
 		}
@@ -297,7 +294,6 @@ func (c *channel) consume() {
 		}
 		it, ok := c.dequeueBuffer()
 		c.bufferLock.Unlock()
-		c.bufferCond.Signal()
 		if !ok {
 			// in fact, this case will never happen
 			continue
@@ -309,11 +305,13 @@ func (c *channel) consume() {
 				c.timeoutCallback(it.value)
 			}
 			atomic.AddUint64(&c.consumed, 1)
+			c.bufferCond.Signal()
 			continue
 		}
 		// consuming, if block here means consumer is busy
 		c.consumer <- it.value
 		atomic.AddUint64(&c.consumed, 1)
+		c.bufferCond.Signal()
 	}
 }
 

lang/channel/channel_test.go

@@ -93,8 +93,6 @@ func TestChannelDefaultSize(t *testing.T) {
 	}
 	ch.Input() <- 0 // wait for be consumed
 	t.Logf("put 0-1")
-	ch.Input() <- 0 // wait for be buffered
-	t.Logf("put 0-2")
 	timeout := false
 	select {
 	case ch.Input() <- 0: // block
@@ -272,41 +270,82 @@ func TestChannelGoroutinesThrottle(t *testing.T) {
 }
 
 func TestChannelNoConsumer(t *testing.T) {
+	done := make(chan struct{})
+	defer close(done)
+	// zero size channel
 	ch := New()
-
 	var sum int32
 	go func() {
 		for i := 1; i <= 20; i++ {
-			ch.Input() <- i
+			select {
+			case ch.Input() <- i:
+			case <-done:
+				return
+			}
 			tlogf(t, "producer=%d finished", i)
 			atomic.AddInt32(&sum, 1)
 		}
 	}()
 	time.Sleep(time.Millisecond * 100)
-	assert.Equal(t, int32(2), atomic.LoadInt32(&sum))
+	assert.Equal(t, int32(1), atomic.LoadInt32(&sum))
+
+	// 1 size channel
+	ch = New(WithSize(1))
+	atomic.StoreInt32(&sum, 0)
+	go func() {
+		for i := 1; i <= 20; i++ {
+			select {
+			case ch.Input() <- i:
+			case <-done:
+				return
+			}
+			tlogf(t, "producer=%d finished", i)
+			atomic.AddInt32(&sum, 1)
+		}
+	}()
+	time.Sleep(time.Millisecond * 100)
+	assert.Equal(t, int32(1), atomic.LoadInt32(&sum))
+
+	// 10 size channel
+	ch = New(WithSize(10))
+	atomic.StoreInt32(&sum, 0)
+	go func() {
+		for i := 1; i <= 20; i++ {
+			select {
+			case ch.Input() <- i:
+			case <-done:
+				return
+			}
+			tlogf(t, "producer=%d finished", i)
+			atomic.AddInt32(&sum, 1)
+		}
+	}()
+	time.Sleep(time.Millisecond * 100)
+	assert.Equal(t, int32(10), atomic.LoadInt32(&sum))
 }
 
 func TestChannelOneSlowTask(t *testing.T) {
-	ch := New(WithTimeout(time.Millisecond*500), WithSize(0))
+	ch := New(WithTimeout(time.Millisecond*100), WithSize(20))
 	defer ch.Close()
 
 	var total int32
 	go func() {
 		for c := range ch.Output() {
 			id := c.(int)
 			if id == 10 {
-				time.Sleep(time.Second)
+				time.Sleep(time.Millisecond * 200)
 			}
 			atomic.AddInt32(&total, 1)
+			tlogf(t, "consumer=%d finished", id)
 		}
 	}()
 
 	for i := 1; i <= 20; i++ {
 		ch.Input() <- i
 		tlogf(t, "producer=%d finished", i)
 	}
-	time.Sleep(time.Second)
-	assert.Equal(t, int32(19), atomic.LoadInt32(&total))
+	time.Sleep(time.Millisecond * 300)
+	assert.Equal(t, int32(11), atomic.LoadInt32(&total))
 }
 
 func TestChannelProduceRateControl(t *testing.T) {