move buf to log struct

http_transporter.go

@@ -7,6 +7,8 @@ import (
 	"net/http"
 )
 
+import _ "net/http/pprof"
+
 // Parts from this transporter were heavily influenced by Peter Bougon's
 // raft implementation: https://github.com/peterbourgon/raft
 
@@ -38,6 +40,7 @@ type HTTPMuxer interface {
 // Creates a new HTTP transporter with the given path prefix.
 func NewHTTPTransporter(prefix string) *HTTPTransporter {
 	return &HTTPTransporter{
+		DisableKeepAlives: false,
 		prefix:            prefix,
 		appendEntriesPath: fmt.Sprintf("%s%s", prefix, "/appendEntries"),
 		requestVotePath:   fmt.Sprintf("%s%s", prefix, "/requestVote"),

http_transporter_test.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"os"
+	"runtime/pprof"
 	"sync"
 	"testing"
 	"time"
@@ -76,6 +78,27 @@ func runTestHttpServers(t *testing.T, servers *[]*Server, transporter *HTTPTrans
 	// Wait for configuration to propagate.
 	time.Sleep(testHeartbeatTimeout * 2)
 
+	f, _ := os.Create("raftprof")
+
+	pprof.StartCPUProfile(f)
+
+	c := make(chan bool)
+	start := time.Now()
+
+	for i := 0; i < 1000; i++ {
+		go send(c, (*servers)[0])
+	}
+
+	for i := 0; i < 1000; i++ {
+		<-c
+	}
+	end := time.Now()
+	fmt.Println(end.Sub(start), "commands ", 1000*20)
+	pprof.StopCPUProfile()
+
+	// Wait for configuration to propagate.
+	time.Sleep(testHeartbeatTimeout * 2)
+
 	// Execute all the callbacks at the same time.
 	for _i, _f := range callbacks {
 		i, f := _i, _f
@@ -88,3 +111,10 @@ func runTestHttpServers(t *testing.T, servers *[]*Server, transporter *HTTPTrans
 	// Wait until everything is done.
 	wg.Wait()
 }
+
+func send(c chan bool, s *Server) {
+	for i := 0; i < 20; i++ {
+		s.Do(&NOPCommand{})
+	}
+	c <- true
+}

log.go

@@ -2,8 +2,10 @@ package raft
 
 import (
 	"bufio"
+	"code.google.com/p/goprotobuf/proto"
 	"errors"
 	"fmt"
+	"github.com/benbjohnson/go-raft/protobuf"
 	"io"
 	"os"
 	"sync"
@@ -26,6 +28,8 @@ type Log struct {
 	mutex       sync.RWMutex
 	startIndex  uint64 // the index before the first entry in the Log entries
 	startTerm   uint64
+	pBuffer     *proto.Buffer
+	pLogEntry   *protobuf.ProtoLogEntry
 }
 
 // The results of the applying a log entry.
@@ -44,6 +48,8 @@ type logResult struct {
 func newLog() *Log {
 	return &Log{
 		entries:   make([]*LogEntry, 0),
+		pBuffer:   proto.NewBuffer(nil),
+		pLogEntry: &protobuf.ProtoLogEntry{},
 	}
 }
 
@@ -470,6 +476,7 @@ func (l *Log) appendEntries(entries []*LogEntry) error {
 	var err error
 	// Append each entry but exit if we hit an error.
 	for _, entry := range entries {
+		entry.log = l
 		if size, err = l.writeEntry(entry, w); err != nil {
 			return err
 		}

log_entry.go

@@ -9,9 +9,6 @@ import (
 	"io"
 )
 
-var p = proto.NewBuffer(nil)
-var pb = &protobuf.ProtoLogEntry{}
-
 // A log entry stores a single item in the log.
 type LogEntry struct {
 	log         *Log
@@ -53,26 +50,26 @@ func newLogEntry(log *Log, index uint64, term uint64, command Command) (*LogEntr
 // Encodes the log entry to a buffer. Returns the number of bytes
 // written and any error that may have occurred.
 func (e *LogEntry) encode(w io.Writer) (int, error) {
-	defer p.Reset() 
+	defer e.log.pBuffer.Reset()
 
-	pb.Index = proto.Uint64(e.Index)
+	e.log.pLogEntry.Index = proto.Uint64(e.Index)
-	pb.Term = proto.Uint64(e.Term)
+	e.log.pLogEntry.Term = proto.Uint64(e.Term)
-	pb.CommandName = proto.String(e.CommandName)
+	e.log.pLogEntry.CommandName = proto.String(e.CommandName)
-	pb.Command = e.Command
+	e.log.pLogEntry.Command = e.Command
 
-	err := p.Marshal(pb)
+	err := e.log.pBuffer.Marshal(e.log.pLogEntry)
 
 	if err != nil {
 		return -1, err
 	}
 
-	_, err = fmt.Fprintf(w, "%8x\n", len(p.Bytes()))
+	_, err = fmt.Fprintf(w, "%8x\n", len(e.log.pBuffer.Bytes()))
 
 	if err != nil {
 		return -1, err
 	}
 
-	return w.Write(p.Bytes())
+	return w.Write(e.log.pBuffer.Bytes())
 }
 
 // Decodes the log entry from a buffer. Returns the number of bytes read and

log_test.go

@@ -62,9 +62,10 @@ func TestLogNewLog(t *testing.T) {
 
 // Ensure that we can decode and encode to an existing log.
 func TestLogExistingLog(t *testing.T) {
-	e0, _ := newLogEntry(nil, 1, 1, &testCommand1{Val: "foo", I: 20})
+	tmpLog := newLog()
-	e1, _ := newLogEntry(nil, 2, 1, &testCommand2{X: 100})
+	e0, _ := newLogEntry(tmpLog, 1, 1, &testCommand1{Val: "foo", I: 20})
-	e2, _ := newLogEntry(nil, 3, 2, &testCommand1{Val: "bar", I: 0})
+	e1, _ := newLogEntry(tmpLog, 2, 1, &testCommand2{X: 100})
+	e2, _ := newLogEntry(tmpLog, 3, 2, &testCommand1{Val: "bar", I: 0})
 	log, path := setupLog([]*LogEntry{e0, e1, e2})
 	defer log.close()
 	defer os.Remove(path)
@@ -86,9 +87,10 @@ func TestLogExistingLog(t *testing.T) {
 
 // Ensure that we can check the contents of the log by index/term.
 func TestLogContainsEntries(t *testing.T) {
-	e0, _ := newLogEntry(nil, 1, 1, &testCommand1{Val: "foo", I: 20})
+	tmpLog := newLog()
-	e1, _ := newLogEntry(nil, 2, 1, &testCommand2{X: 100})
+	e0, _ := newLogEntry(tmpLog, 1, 1, &testCommand1{Val: "foo", I: 20})
-	e2, _ := newLogEntry(nil, 3, 2, &testCommand1{Val: "bar", I: 0})
+	e1, _ := newLogEntry(tmpLog, 2, 1, &testCommand2{X: 100})
+	e2, _ := newLogEntry(tmpLog, 3, 2, &testCommand1{Val: "bar", I: 0})
 	log, path := setupLog([]*LogEntry{e0, e1, e2})
 	defer log.close()
 	defer os.Remove(path)
@@ -112,8 +114,9 @@ func TestLogContainsEntries(t *testing.T) {
 
 // Ensure that we can recover from an incomplete/corrupt log and continue logging.
 func TestLogRecovery(t *testing.T) {
-	e0, _ := newLogEntry(nil, 1, 1, &testCommand1{Val: "foo", I: 20})
+	tmpLog := newLog()
-	e1, _ := newLogEntry(nil, 2, 1, &testCommand2{X: 100})
+	e0, _ := newLogEntry(tmpLog, 1, 1, &testCommand1{Val: "foo", I: 20})
+	e1, _ := newLogEntry(tmpLog, 2, 1, &testCommand2{X: 100})
 	f, _ := ioutil.TempFile("", "raft-log-")
 
 	e0.encode(f)

server.go

@@ -29,7 +29,7 @@ const (
 )
 
 const (
-	MaxLogEntriesPerRequest         = 200
+	MaxLogEntriesPerRequest         = 2000
 	NumberOfLogEntriesAfterSnapshot = 200
 )
 

server_test.go

@@ -102,9 +102,10 @@ func TestServerRequestVoteApprovedIfAlreadyVotedInOlderTerm(t *testing.T) {
 
 // Ensure that a vote request is denied if the log is out of date.
 func TestServerRequestVoteDenyIfCandidateLogIsBehind(t *testing.T) {
-	e0, _ := newLogEntry(nil, 1, 1, &testCommand1{Val: "foo", I: 20})
+	tmpLog := newLog()
-	e1, _ := newLogEntry(nil, 2, 1, &testCommand2{X: 100})
+	e0, _ := newLogEntry(tmpLog, 1, 1, &testCommand1{Val: "foo", I: 20})
-	e2, _ := newLogEntry(nil, 3, 2, &testCommand1{Val: "bar", I: 0})
+	e1, _ := newLogEntry(tmpLog, 2, 1, &testCommand2{X: 100})
+	e2, _ := newLogEntry(tmpLog, 3, 2, &testCommand1{Val: "bar", I: 0})
 	server := newTestServerWithLog("1", &testTransporter{}, []*LogEntry{e0, e1, e2})
 
 	// start as a follower with term 2 and index 3
@@ -143,7 +144,7 @@ func TestServerRequestVoteDenyIfCandidateLogIsBehind(t *testing.T) {
 
 // // Ensure that we can self-promote a server to candidate, obtain votes and become a fearless leader.
 func TestServerPromoteSelf(t *testing.T) {
-	e0, _ := newLogEntry(nil, 1, 1, &testCommand1{Val: "foo", I: 20})
+	e0, _ := newLogEntry(newLog(), 1, 1, &testCommand1{Val: "foo", I: 20})
 	server := newTestServerWithLog("1", &testTransporter{}, []*LogEntry{e0})
 
 	// start as a follower

test.go

@@ -8,8 +8,8 @@ import (
 )
 
 const (
-	testHeartbeatTimeout = 5 * time.Millisecond
+	testHeartbeatTimeout = 50 * time.Millisecond
-	testElectionTimeout  = 20 * time.Millisecond
+	testElectionTimeout  = 200 * time.Millisecond
 )
 
 func init() {
@@ -85,7 +85,7 @@ func newTestServerWithLog(name string, transporter Transporter, entries []*LogEn
 
 func newTestCluster(names []string, transporter Transporter, lookup map[string]*Server) []*Server {
 	servers := []*Server{}
-	e0, _ := newLogEntry(nil, 1, 1, &testCommand1{Val: "foo", I: 20})
+	e0, _ := newLogEntry(newLog(), 1, 1, &testCommand1{Val: "foo", I: 20})
 
 	for _, name := range names {
 		if lookup[name] != nil {