add bufio, bytes.Buffer, bytes.Pool improve GC performance

comet/bucket.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"github.com/Terry-Mao/goim/libs/define"
+	"github.com/Terry-Mao/goim/libs/time"
 	"sync"
 )
 
@@ -30,7 +31,7 @@ func NewBucket(boptions BucketOptions, roptions RoomOptions) (b *Bucket) {
 }
 
 // Put put a channel according with sub key.
-func (b *Bucket) Put(key string, ch *Channel, tr *Timer) {
+func (b *Bucket) Put(key string, ch *Channel, tr *time.Timer) {
 	var (
 		room *Room
 		ok   bool

comet/channel.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"github.com/Terry-Mao/goim/libs/bufio"
 	"sync"
 	"time"
 )
@@ -12,7 +13,8 @@ type Channel struct {
 	CliProto Proto
 	SvrProto Ring
 	cLock    sync.Mutex
-	Buf      [RawHeaderSize]byte
+	Writer   bufio.Writer
+	Reader   bufio.Reader
 }
 
 func NewChannel(proto int, rid int32) *Channel {
@@ -32,7 +34,9 @@ func (c *Channel) Reply(p *Proto) (err error) {
 		c.SvrProto.SetAdv()
 	}
 	c.cLock.Unlock()
-	c.Signal()
+	if err == nil {
+		c.Signal()
+	}
 	return
 }
 
@@ -47,7 +51,9 @@ func (c *Channel) Push(ver int16, operation int32, body []byte) (err error) {
 		c.SvrProto.SetAdv()
 	}
 	c.cLock.Unlock()
-	c.Signal()
+	if err == nil {
+		c.Signal()
+	}
 	return
 }
 

comet/comet-example.conf

@@ -117,25 +117,39 @@ rcvbuf 256
 
 keepalive 0
 
-# Sets the reader buffer instance.
+# Sets the reader number, used in round-robin selection.
 #
 # Examples:
 #
-# read.timeout 5s
-readbuf 2048
+# reader 1024
+reader 1024
 
-# Sets the writer buffer instance.
+# Sets the reader buffer instance.
 #
 # Examples:
 #
-# read.timeout 5s
-writebuf 2048
+# readbuf 1024
+readbuf 1024
 
 # read buffer size
-readbuf.size 1024
+readbuf.size 512
+
+# Sets the writer number, used in round-robin selection.
+#
+# Examples:
+#
+# writer 1024
+writer 1024
+
+# Sets the writer buffer instance.
+#
+# Examples:
+#
+# writebuf 1024
+writebuf 1024
 
 # write buffer size
-writebuf.size 1024
+writebuf.size 4096
 
 [websocket]
 # By default comet websocket listens for connections from all the network interfaces

comet/config.go

@@ -48,9 +48,11 @@ type Config struct {
 	TCPSndbuf       int      `goconf:"tcp:sndbuf:memory"`
 	TCPRcvbuf       int      `goconf:"tcp:rcvbuf:memory"`
 	TCPKeepalive    bool     `goconf:"tcp:keepalive"`
+	TCPReader       int      `goconf:"tcp:reader"`
 	TCPReadBuf      int      `goconf:"tcp:readbuf"`
-	TCPWriteBuf     int      `goconf:"tcp:writebuf"`
 	TCPReadBufSize  int      `goconf:"tcp:readbuf.size"`
+	TCPWriter       int      `goconf:"tcp:writer"`
+	TCPWriteBuf     int      `goconf:"tcp:writebuf"`
 	TCPWriteBufSize int      `goconf:"tcp:writebuf.size"`
 	// websocket
 	WebsocketBind        []string `goconf:"websocket:bind:,"`

comet/http.go

@@ -5,6 +5,7 @@ import (
 	log "code.google.com/p/log4go"
 	"encoding/json"
 	"github.com/Terry-Mao/goim/libs/define"
+	itime "github.com/Terry-Mao/goim/libs/time"
 	"math/rand"
 	"net"
 	"net/http"
@@ -55,15 +56,15 @@ func serveHTTP(w http.ResponseWriter, r *http.Request) {
 	DefaultServer.serveHTTP(w, r, tr)
 }
 
-func (server *Server) serveHTTP(w http.ResponseWriter, r *http.Request, tr *Timer) {
+func (server *Server) serveHTTP(w http.ResponseWriter, r *http.Request, tr *itime.Timer) {
 	var (
 		b    *Bucket
 		ok   bool
 		hb   time.Duration // heartbeat
 		key  string
 		cb   string
 		err  error
-		trd  *TimerData
+		trd  *itime.TimerData
 		conn net.Conn
 		rwr  *bufio.ReadWriter
 		hj   http.Hijacker

comet/main.go

@@ -40,16 +40,23 @@ func main() {
 			SignalTime:  Conf.RoomSignal,
 		})
 	}
-	round := NewRound(Conf.TCPReadBuf, Conf.TCPWriteBuf, Conf.Timer, Conf.TimerSize)
+	round := NewRound(RoundOptions{
+		Reader:       Conf.TCPReader,
+		ReadBuf:      Conf.TCPReadBuf,
+		ReadBufSize:  Conf.TCPReadBufSize,
+		Writer:       Conf.TCPWriter,
+		WriteBuf:     Conf.TCPWriteBuf,
+		WriteBufSize: Conf.TCPWriteBufSize,
+		Timer:        Conf.Timer,
+		TimerSize:    Conf.TimerSize,
+	})
 	operator := new(DefaultOperator)
 	DefaultServer = NewServer(buckets, round, operator, ServerOptions{
 		Proto:            Conf.Proto,
 		HandshakeTimeout: Conf.HandshakeTimeout,
 		TCPKeepalive:     Conf.TCPKeepalive,
 		TCPRcvbuf:        Conf.TCPRcvbuf,
 		TCPSndbuf:        Conf.TCPSndbuf,
-		TCPReadBufSize:   Conf.TCPReadBufSize,
-		TCPWriteBufSize:  Conf.TCPWriteBufSize,
 	})
 	// tcp comet
 	if err := InitTCP(Conf.TCPBind, Conf.MaxProc); err != nil {

comet/room.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	log "code.google.com/p/log4go"
+	itime "github.com/Terry-Mao/goim/libs/time"
 	"sync"
 	"time"
 )
@@ -19,13 +20,13 @@ type Room struct {
 	proto   Ring
 	signal  chan int
 	chs     map[*Channel]struct{} // map room id with channels
-	timer   *Timer
+	timer   *itime.Timer
 	sigTime time.Duration
 	options RoomOptions
 }
 
 // NewRoom new a room struct, store channel room info.
-func NewRoom(id int32, t *Timer, options RoomOptions) (r *Room) {
+func NewRoom(id int32, t *itime.Timer, options RoomOptions) (r *Room) {
 	r = new(Room)
 	r.id = id
 	r.options = options
@@ -61,7 +62,7 @@ func (r *Room) push() {
 		p     *Proto
 		ch    *Channel
 		last  time.Time
-		td    *TimerData
+		td    *itime.TimerData
 		least time.Duration
 		vers  = make([]int32, r.options.BatchNum)
 		ops   = make([]int32, r.options.BatchNum)

comet/round.go

@@ -1,44 +1,66 @@
 package main
 
 import (
-	log "code.google.com/p/log4go"
-	"sync"
+	"github.com/Terry-Mao/goim/libs/bytes"
+	"github.com/Terry-Mao/goim/libs/time"
 )
 
+type RoundOptions struct {
+	Timer        int
+	TimerSize    int
+	Reader       int
+	ReadBuf      int
+	ReadBufSize  int
+	Writer       int
+	WriteBuf     int
+	WriteBufSize int
+}
+
+// Ronnd userd for connection round-robin get a reader/writer/timer for split big lock.
 type Round struct {
-	readers   []sync.Pool
-	writers   []sync.Pool
-	timers    []Timer
+	readers   []bytes.Pool
+	writers   []bytes.Pool
+	timers    []time.Timer
+	options   RoundOptions
 	readerIdx int
 	writerIdx int
 	timerIdx  int
 }
 
-func NewRound(readBuf, writeBuf, timer, timerSize int) *Round {
-	r := new(Round)
-	log.Debug("create %d reader buffer pool", readBuf)
-	log.Debug("create %d writer buffer pool", writeBuf)
-	r.readerIdx = readBuf
-	r.writerIdx = writeBuf
-	r.readers = make([]sync.Pool, readBuf)
-	r.writers = make([]sync.Pool, writeBuf)
-	log.Debug("create %d timer", timer)
-	r.timerIdx = timer
-	r.timers = make([]Timer, timer)
-	for i := 0; i < timer; i++ {
-		r.timers[i].Init(timerSize)
+// NewRound new a round struct.
+func NewRound(options RoundOptions) (r *Round) {
+	var i int
+	r = new(Round)
+	r.options = options
+	// reader
+	r.readers = make([]bytes.Pool, options.Reader)
+	for i = 0; i < options.Reader; i++ {
+		r.readers[i].Init(options.ReadBuf, options.ReadBufSize)
+	}
+	// writer
+	r.writers = make([]bytes.Pool, options.Writer)
+	for i = 0; i < options.Writer; i++ {
+		r.writers[i].Init(options.WriteBuf, options.WriteBufSize)
+	}
+	// timer
+	r.timers = make([]time.Timer, options.Timer)
+	for i = 0; i < options.Timer; i++ {
+		r.timers[i].Init(options.TimerSize)
 	}
-	return r
+	return
 }
 
-func (r *Round) Timer(rn int) *Timer {
-	return &(r.timers[rn%r.timerIdx])
+// Timer get a timer.
+func (r *Round) Timer(rn int) *time.Timer {
+	return &(r.timers[rn%r.options.Timer])
 }
 
-func (r *Round) Reader(rn int) *sync.Pool {
-	return &(r.readers[rn%r.readerIdx])
+// Reader get a reader memory buffer.
+func (r *Round) Reader(rn int) *bytes.Pool {
+	return &(r.readers[rn%r.options.Reader])
 }
 
-func (r *Round) Writer(rn int) *sync.Pool {
-	return &(r.writers[rn%r.writerIdx])
+// Writer get a writer memory buffer pool.
+func (r *Round) Writer(rn int) *bytes.Pool {
+	return &(r.writers[rn%r.options.Writer])
 }

comet/server.go

@@ -17,8 +17,6 @@ type ServerOptions struct {
 	TCPKeepalive     bool
 	TCPRcvbuf        int
 	TCPSndbuf        int
-	TCPReadBufSize   int
-	TCPWriteBufSize  int
 }
 
 type Server struct {