Merge branch 'develop'

README.md

@@ -183,3 +183,4 @@ Gmail: wyk4true@gmail.com
 ## Feedback
 
 Gmail: siddontang@gmail.com
+Skype: live:siddontang_1

client/go/ledis/client.go

@@ -14,6 +14,8 @@ const (
 type Config struct {
 	Addr            string
 	MaxIdleConns    int
+	ReadBufferSize  int
+	WriteBufferSize int
 }
 
 type Client struct {
@@ -36,6 +38,12 @@ func NewClient(cfg *Config) *Client {
 	c := new(Client)
 
 	c.cfg = cfg
+	if c.cfg.ReadBufferSize == 0 {
+		c.cfg.ReadBufferSize = 4096
+	}
+	if c.cfg.WriteBufferSize == 0 {
+		c.cfg.WriteBufferSize = 4096
+	}
 
 	c.conns = list.New()
 

client/go/ledis/conn.go

@@ -384,7 +384,7 @@ func (c *Conn) readReply() (interface{}, error) {
 }
 
 func (c *Client) newConn(addr string) *Conn {
-	co := NewConn(addr)
+	co := NewConnSize(addr, c.cfg.ReadBufferSize, c.cfg.WriteBufferSize)
 	co.client = c
 
 	return co

cmd/ledis-benchmark/main.go

@@ -268,6 +268,8 @@ func main() {
 	cfg := new(ledis.Config)
 	cfg.Addr = addr
 	cfg.MaxIdleConns = *clients
+	cfg.ReadBufferSize = 10240
+	cfg.WriteBufferSize = 10240
 	client = ledis.NewClient(cfg)
 
 	if *round <= 0 {

cmd/ledis-storebench/main.go

@@ -46,8 +46,8 @@ func bench(cmd string, f func()) {
 	t2 := time.Now()
 
 	d := t2.Sub(t1)
-	fmt.Printf("%s: %0.3f micros/op, %0.2fmb/s\n", cmd, float64(d.Nanoseconds()/1e3)/float64(*number),
-		float64((*valueSize+16)*(*number))/(1024.0*1024.0*(d.Seconds())))
+	fmt.Printf("%s: %0.3f micros/op, %0.2fmb/s %0.2fop/s\n", cmd, float64(d.Nanoseconds()/1e3)/float64(*number),
+		float64((*valueSize+16)*(*number))/(1024.0*1024.0*(d.Seconds())), float64(*number)/d.Seconds())
 }
 
 var kvSetBase int64 = 0

config/config.go

@@ -59,6 +59,7 @@ type RocksDBConfig struct {
 	StatsDumpPeriodSec             int  `toml:"stats_dump_period_sec"`
 	BackgroundThreads              int  `toml:"background_theads"`
 	HighPriorityBackgroundThreads  int  `toml:"high_priority_background_threads"`
+	DisableWAL                     bool `toml:"disable_wal"`
 }
 
 type LMDBConfig struct {
@@ -109,6 +110,9 @@ type Config struct {
 	Replication    ReplicationConfig `toml:"replication"`
 
 	Snapshot SnapshotConfig `toml:"snapshot"`
+
+	ConnReadBufferSize  int `toml:"conn_read_buffer_size"`
+	ConnWriteBufferSize int `toml:"conn_write_buffer_size"`
 }
 
 func NewConfigWithFile(fileName string) (*Config, error) {
@@ -169,6 +173,7 @@ func NewConfigDefault() *Config {
 	cfg.RocksDB.UseFsync = false
 	cfg.RocksDB.DisableAutoCompactions = false
 	cfg.RocksDB.AllowOsBuffer = true
+	cfg.RocksDB.DisableWAL = false
 
 	cfg.adjust()
 
@@ -189,6 +194,8 @@ func (cfg *Config) adjust() {
 	cfg.RocksDB.adjust()
 
 	cfg.Replication.ExpiredLogDays = getDefault(7, cfg.Replication.ExpiredLogDays)
+	cfg.ConnReadBufferSize = getDefault(4*KB, cfg.ConnReadBufferSize)
+	cfg.ConnWriteBufferSize = getDefault(4*KB, cfg.ConnWriteBufferSize)
 }
 
 func (cfg *LevelDBConfig) adjust() {

config/config.toml

@@ -43,6 +43,10 @@ db_sync_commit = 0
 # enable replication or not
 use_replication = true
 
+# set connection buffer, you can increase them appropriately
+conn_read_buffer_size = 10240
+conn_write_buffer_size = 10240
+
 [leveldb]
 # for leveldb and goleveldb
 compression = false
@@ -82,7 +86,10 @@ max_background_flushes = 1
 allow_os_buffer = true
 enable_statistics = false
 stats_dump_period_sec = 3600
-
+# dangerous to set true, write may got lost after a crash
+# you can set true if replication opened, we may recover from replication log,
+# but it is still not a easy work.
+disable_wal = false
 
 [lmdb]
 map_size = 524288000

etc/ledis.conf

@@ -82,7 +82,10 @@ max_background_flushes = 1
 allow_os_buffer = true
 enable_statistics = false
 stats_dump_period_sec = 3600
-
+# dangerous to set true, write may got lost after a crash
+# you can set true if replication opened, we may recover from replication log,
+# but it is still not a easy work.
+disable_wal = false
 
 [lmdb]
 map_size = 524288000

server/client_resp.go

@@ -33,9 +33,13 @@ func newClientRESP(conn net.Conn, app *App) {
 	c.client = newClient(app)
 	c.conn = conn
 
-	c.rb = bufio.NewReaderSize(conn, 4096)
+	if tcpConn, ok := conn.(*net.TCPConn); ok {
+		tcpConn.SetReadBuffer(app.cfg.ConnReadBufferSize)
+		tcpConn.SetWriteBuffer(app.cfg.ConnWriteBufferSize)
+	}
+	c.rb = bufio.NewReaderSize(conn, app.cfg.ConnReadBufferSize)
 
-	c.resp = newWriterRESP(conn)
+	c.resp = newWriterRESP(conn, app.cfg.ConnWriteBufferSize)
 	c.remoteAddr = conn.RemoteAddr().String()
 
 	go c.run()
@@ -158,9 +162,9 @@ func (c *respClient) handleRequest(reqData [][]byte) {
 
 //	response writer
 
-func newWriterRESP(conn net.Conn) *respWriter {
+func newWriterRESP(conn net.Conn, size int) *respWriter {
 	w := new(respWriter)
-	w.buff = bufio.NewWriterSize(conn, 256)
+	w.buff = bufio.NewWriterSize(conn, size)
 	return w
 }
 

store/rocksdb/db.go

@@ -150,9 +150,11 @@ func (db *DB) initOptions(cfg *config.RocksDBConfig) {
 
 	db.readOpts = NewReadOptions()
 	db.writeOpts = NewWriteOptions()
+	db.writeOpts.DisableWAL(cfg.DisableWAL)
 
 	db.syncOpts = NewWriteOptions()
 	db.syncOpts.SetSync(true)
+	db.syncOpts.DisableWAL(cfg.DisableWAL)
 
 	db.iteratorOpts = NewReadOptions()
 	db.iteratorOpts.SetFillCache(false)