Optimize for performance

Author: limpo1989

Diff for: bootstrap.go

@@ -30,7 +30,7 @@ type Bootstrap interface {
 	// Context return context
 	Context() context.Context
 	// Listen create a listener
-	Listen(url string, option ...transport.Option) Listener
+	Listen(url string, attachment Attachment, option ...transport.Option) Listener
 	// Connect to remote endpoint
 	Connect(url string, attachment Attachment, option ...transport.Option) (Channel, error)
 	// Shutdown boostrap
@@ -43,8 +43,9 @@ func NewBootstrap(option ...Option) Bootstrap {
 	opts := &bootstrapOptions{
 		channelIDFactory: SequenceID(),
 		pipelineFactory:  NewPipeline,
-		channelFactory:   NewChannel(128),
+		channelFactory:   NewChannel(64),
 		transportFactory: tcp.New(),
+		executor:         AsyncExecutor(),
 	}
 	opts.bootstrapCtx, opts.bootstrapCancel = context.WithCancel(context.Background())
 
@@ -67,7 +68,7 @@ func (bs *bootstrap) Context() context.Context {
 }
 
 // serveTransport to serve channel
-func (bs *bootstrap) serveTransport(transport transport.Transport, attachment Attachment, childChannel bool) Channel {
+func (bs *bootstrap) serveTransport(ctx context.Context, transport transport.Transport, attachment Attachment, childChannel bool) Channel {
 
 	// create a new pipeline
 	pl := bs.pipelineFactory()
@@ -76,7 +77,7 @@ func (bs *bootstrap) serveTransport(transport transport.Transport, attachment At
 	cid := bs.channelIDFactory()
 
 	// create a channel
-	ch := bs.channelFactory(cid, bs.bootstrapCtx, pl, transport)
+	ch := bs.channelFactory(cid, ctx, pl, transport, bs.executor)
 
 	// set the attachment if necessary
 	if nil != attachment {
@@ -110,12 +111,12 @@ func (bs *bootstrap) Connect(url string, attachment Attachment, option ...transp
 	}
 
 	// serve client transport
-	return bs.serveTransport(t, attachment, false), nil
+	return bs.serveTransport(options.Context, t, attachment, false), nil
 }
 
 // Listen to the address with options
-func (bs *bootstrap) Listen(url string, option ...transport.Option) Listener {
-	l := &listener{bs: bs, url: url, option: option}
+func (bs *bootstrap) Listen(url string, attachment Attachment, option ...transport.Option) Listener {
+	l := &listener{bs: bs, url: url, attachment: attachment, option: option}
 	bs.listeners.Store(url, l)
 	return l
 }
@@ -146,11 +147,12 @@ type Listener interface {
 
 // impl Listener
 type listener struct {
-	bs       *bootstrap
-	url      string
-	option   []transport.Option
-	options  *transport.Options
-	acceptor transport.Acceptor
+	bs         *bootstrap
+	url        string
+	attachment Attachment
+	option     []transport.Option
+	options    *transport.Options
+	acceptor   transport.Acceptor
 }
 
 // Close listener
@@ -191,14 +193,14 @@ func (l *listener) Sync() error {
 			return t.Close()
 		default:
 			// serve child transport
-			l.bs.serveTransport(t, nil, true)
+			l.bs.serveTransport(l.options.Context, t, l.attachment, true)
 		}
 	}
 }
 
 // Async accept new transport from listener
 func (l *listener) Async(fn func(err error)) {
-	go func() {
+	l.bs.executor.Exec(func() {
 		fn(l.Sync())
-	}()
+	})
 }

Diff for: bootstrap_test.go

@@ -50,7 +50,6 @@ func TestBootstrap(t *testing.T) {
 	}
 
 	bs := NewBootstrap(
-		WithChannel(NewBufferedChannel(128, 1024)),
 		WithChildInitializer(pipelineInitializer),
 		WithClientInitializer(pipelineInitializer),
 		WithTransport(tcp.New()),

Diff for: channel.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"net"
-	"runtime/debug"
 	"sync"
 	"sync/atomic"
 
@@ -34,7 +33,7 @@ type Channel interface {
 	ID() int64
 
 	// Write a message through the Pipeline
-	Write(Message) bool
+	Write(Message) error
 
 	// Trigger user event
 	Trigger(event Event)
@@ -75,43 +74,46 @@ type Channel interface {
 
 // NewChannel create a ChannelFactory
 func NewChannel(capacity int) ChannelFactory {
-	return func(id int64, ctx context.Context, pipeline Pipeline, transport transport.Transport) Channel {
-		return newChannelWith(ctx, pipeline, transport, id, capacity)
-	}
-}
-
-// NewBufferedChannel create a ChannelFactory with buffered transport
-func NewBufferedChannel(capacity int, sizeRead int) ChannelFactory {
-	return func(id int64, ctx context.Context, pipeline Pipeline, tran transport.Transport) Channel {
-		tran = transport.BufferedTransport(tran, sizeRead)
-		return newChannelWith(ctx, pipeline, tran, id, capacity)
+	return func(id int64, ctx context.Context, pipeline Pipeline, transport transport.Transport, executor Executor) Channel {
+		return newChannelWith(ctx, pipeline, transport, executor, id, capacity)
 	}
 }
 
 // newChannelWith internal method for NewChannel & NewBufferedChannel
-func newChannelWith(ctx context.Context, pipeline Pipeline, transport transport.Transport, id int64, capacity int) Channel {
+func newChannelWith(ctx context.Context, pipeline Pipeline, transport transport.Transport, executor Executor, id int64, capacity int) Channel {
 	childCtx, cancel := context.WithCancel(ctx)
 	return &channel{
-		id:        id,
-		ctx:       childCtx,
-		cancel:    cancel,
-		pipeline:  pipeline,
-		transport: transport,
-		sendQueue: make(chan [][]byte, capacity),
+		id:           id,
+		ctx:          childCtx,
+		cancel:       cancel,
+		pipeline:     pipeline,
+		transport:    transport,
+		executor:     executor,
+		sendQueue:    utils.NewRingBuffer(uint64(capacity)),
+		writeBuffers: make(net.Buffers, 0, capacity/3+8),
+		writeIndexes: make([]int, 0, capacity/3+8),
 	}
 }
 
+const idle = 0
+const running = 1
+
 // implement of Channel
 type channel struct {
-	id         int64
-	ctx        context.Context
-	cancel     context.CancelFunc
-	transport  transport.Transport
-	pipeline   Pipeline
-	attachment Attachment
-	sendQueue  chan [][]byte
-	activeWait sync.WaitGroup
-	closed     int32
+	id           int64
+	ctx          context.Context
+	cancel       context.CancelFunc
+	transport    transport.Transport
+	executor     Executor
+	pipeline     Pipeline
+	attachment   Attachment
+	sendQueue    *utils.RingBuffer
+	writeBuffers net.Buffers
+	writeIndexes []int
+	activeWait   sync.WaitGroup
+	closed       int32
+	running      int32
+	closeErr     error
 }
 
 // ID get channel id
@@ -120,22 +122,22 @@ func (c *channel) ID() int64 {
 }
 
 // Write a message through the Pipeline
-func (c *channel) Write(message Message) bool {
-
-	select {
-	case <-c.ctx.Done():
-		return false
-	default:
-		c.invokeMethod(func() {
-			c.pipeline.FireChannelWrite(message)
-		})
-		return true
+func (c *channel) Write(message Message) error {
+	if !c.IsActive() {
+		select {
+		case <-c.ctx.Done():
+			return c.closeErr
+		}
 	}
+
+	c.invokeMethod(func() {
+		c.pipeline.FireChannelWrite(message)
+	})
+	return nil
 }
 
 // Trigger trigger event
 func (c *channel) Trigger(event Event) {
-
 	c.invokeMethod(func() {
 		c.pipeline.FireChannelEvent(event)
 	})
@@ -144,27 +146,34 @@ func (c *channel) Trigger(event Event) {
 // Close through the Pipeline
 func (c *channel) Close(err error) {
 	if atomic.CompareAndSwapInt32(&c.closed, 0, 1) {
-		c.cancel()
+		c.closeErr = err
+		c.sendQueue.Dispose()
 		c.transport.Close()
+		c.cancel()
 
 		c.invokeMethod(func() {
-			c.pipeline.FireChannelInactive(AsException(err, debug.Stack()))
+			c.pipeline.FireChannelInactive(err)
 		})
 	}
 }
 
 // Writev to write [][]byte for optimize syscall
 func (c *channel) Writev(p [][]byte) (n int64, err error) {
-
-	select {
-	case <-c.ctx.Done():
-		return 0, errors.New("broken pipe")
-	case c.sendQueue <- p:
-		for _, d := range p {
-			n += int64(len(d))
+	if !c.IsActive() {
+		select {
+		case <-c.ctx.Done():
+			return 0, c.closeErr
 		}
-		return
 	}
+	// put packet to send queue
+	if err = c.sendQueue.Put(p); nil != err {
+		return 0, err
+	}
+	// try send
+	if atomic.CompareAndSwapInt32(&c.running, idle, running) {
+		c.executor.Exec(c.writeOnce)
+	}
+	return utils.CountOf(p), nil
 }
 
 // IsActive return true if the Channel is active and so connected
@@ -210,20 +219,19 @@ func (c *channel) Context() context.Context {
 // serveChannel start write & read routines
 func (c *channel) serveChannel() {
 	c.activeWait.Add(1)
-	go c.readLoop()
-	go c.writeLoop()
+	c.executor.Exec(c.readLoop)
 	c.activeWait.Wait()
 }
 
 func (c *channel) invokeMethod(fn func()) {
 
 	defer func() {
 		if err := recover(); nil != err && 0 == atomic.LoadInt32(&c.closed) {
-			c.pipeline.FireChannelException(AsException(err, debug.Stack()))
+			c.pipeline.FireChannelException(AsException(err))
 
 			if e, ok := err.(error); ok {
 				var ne net.Error
-				if errors.As(e, &ne) && !ne.Temporary() {
+				if errors.As(e, &ne) && !ne.Timeout() {
 					c.Close(e)
 				}
 			}
@@ -237,11 +245,7 @@ func (c *channel) invokeMethod(fn func()) {
 func (c *channel) readLoop() {
 
 	defer func() {
-		if err := recover(); nil != err {
-			c.Close(AsException(err, debug.Stack()))
-		} else {
-			c.Close(nil)
-		}
+		c.Close(AsException(recover()))
 	}()
 
 	func() {
@@ -261,56 +265,56 @@ func (c *channel) readLoop() {
 	}
 }
 
-// writeLoop sending message of channel
-func (c *channel) writeLoop() {
+// writeOnce sending messages of channel
+func (c *channel) writeOnce() {
 
 	defer func() {
 		if err := recover(); nil != err {
-			c.Close(AsException(err, debug.Stack()))
-		} else {
-			c.Close(nil)
+			c.Close(AsException(err))
 		}
 	}()
 
-	var bufferCap = cap(c.sendQueue)
-	var buffers = make(net.Buffers, 0, bufferCap)
-	var indexes = make([]int, 0, bufferCap)
+	for {
+		// reuse buffer.
+		sendBuffers := c.writeBuffers[:0]
+		sendIndexes := c.writeIndexes[:0]
+
+		// more packet will be merged
+		for c.sendQueue.Len() > 0 && len(sendBuffers) < cap(sendBuffers) {
+			// poll packet
+			item, err := c.sendQueue.Poll(-1)
+			if nil != err {
+				break
+			}
 
-	// Try to combine packet sending to optimize sending performance
-	sendWithWritev := func(data [][]byte, queue <-chan [][]byte) (int64, error) {
+			// combine send bytes to reduce syscall.
+			pkts := item.([][]byte)
+			sendBuffers = append(sendBuffers, pkts...)
+			sendIndexes = append(sendIndexes, len(sendBuffers))
+		}
 
-		// reuse buffer.
-		sendBuffers := buffers[:0]
-		sendIndexes := indexes[:0]
-
-		// append first packet.
-		sendBuffers = append(sendBuffers, data...)
-		sendIndexes = append(sendIndexes, len(sendBuffers))
-
-		// more packet will be merged.
-		for {
-			select {
-			case data := <-queue:
-				sendBuffers = append(sendBuffers, data...)
-				sendIndexes = append(sendIndexes, len(sendBuffers))
-				if len(sendIndexes) >= bufferCap {
-					return c.transport.Writev(transport.Buffers{Buffers: sendBuffers, Indexes: sendIndexes})
+		if len(sendBuffers) > 0 {
+			utils.AssertLong(c.transport.Writev(transport.Buffers{Buffers: sendBuffers, Indexes: sendIndexes}))
+			utils.Assert(c.transport.Flush())
+
+			// clear buffer ref
+			for index := range sendBuffers {
+				sendBuffers[index] = nil // avoid memory leak
+				if index < len(sendIndexes) {
+					sendIndexes[index] = -1 // for safety
 				}
-			default:
-				return c.transport.Writev(transport.Buffers{Buffers: sendBuffers, Indexes: sendIndexes})
 			}
 		}
-	}
 
-	for {
-		select {
-		case buf := <-c.sendQueue:
-			// combine send bytes to reduce syscall.
-			utils.AssertLong(sendWithWritev(buf, c.sendQueue))
-			// flush buffer
-			utils.Assert(c.transport.Flush())
-		case <-c.ctx.Done():
-			return
+		// double check
+		atomic.StoreInt32(&c.running, idle)
+		if size := c.sendQueue.Len(); size > 0 {
+			if atomic.CompareAndSwapInt32(&c.running, idle, running) {
+				continue
+			}
 		}
+
+		// no packets to send
+		break
 	}
 }