raft: add Id to Entry

This commit is contained in:
Blake Mizerany 2014-08-26 16:19:49 -07:00 committed by Yicheng Qin
parent 3dd6280df5
commit 225e618b8f
5 changed files with 41 additions and 28 deletions

View File

@ -7,6 +7,7 @@
It is generated from these files: It is generated from these files:
entry.proto entry.proto
state.proto
It has these top-level messages: It has these top-level messages:
Entry Entry
@ -32,6 +33,7 @@ type Entry struct {
Term int64 `protobuf:"varint,2,req,name=term" json:"term"` Term int64 `protobuf:"varint,2,req,name=term" json:"term"`
Index int64 `protobuf:"varint,3,req,name=index" json:"index"` Index int64 `protobuf:"varint,3,req,name=index" json:"index"`
Data []byte `protobuf:"bytes,4,opt,name=data" json:"data,omitempty"` Data []byte `protobuf:"bytes,4,opt,name=data" json:"data,omitempty"`
Id int64 `protobuf:"varint,5,req,name=id" json:"id"`
XXX_unrecognized []byte `json:"-"` XXX_unrecognized []byte `json:"-"`
} }
@ -127,6 +129,21 @@ func (m *Entry) Unmarshal(data []byte) error {
} }
m.Data = append(m.Data, data[index:postIndex]...) m.Data = append(m.Data, data[index:postIndex]...)
index = postIndex index = postIndex
case 5:
if wireType != 0 {
return code_google_com_p_gogoprotobuf_proto.ErrWrongType
}
for shift := uint(0); ; shift += 7 {
if index >= l {
return io.ErrUnexpectedEOF
}
b := data[index]
index++
m.Id |= (int64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
default: default:
var sizeOfWire int var sizeOfWire int
for { for {
@ -160,6 +177,7 @@ func (m *Entry) Size() (n int) {
l = len(m.Data) l = len(m.Data)
n += 1 + l + sovEntry(uint64(l)) n += 1 + l + sovEntry(uint64(l))
} }
n += 1 + sovEntry(uint64(m.Id))
if m.XXX_unrecognized != nil { if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized) n += len(m.XXX_unrecognized)
} }
@ -209,6 +227,9 @@ func (m *Entry) MarshalTo(data []byte) (n int, err error) {
i = encodeVarintEntry(data, i, uint64(len(m.Data))) i = encodeVarintEntry(data, i, uint64(len(m.Data)))
i += copy(data[i:], m.Data) i += copy(data[i:], m.Data)
} }
data[i] = 0x28
i++
i = encodeVarintEntry(data, i, uint64(m.Id))
if m.XXX_unrecognized != nil { if m.XXX_unrecognized != nil {
i += copy(data[i:], m.XXX_unrecognized) i += copy(data[i:], m.XXX_unrecognized)
} }

View File

@ -12,4 +12,5 @@ message Entry {
required int64 term = 2 [(gogoproto.nullable) = false]; required int64 term = 2 [(gogoproto.nullable) = false];
required int64 index = 3 [(gogoproto.nullable) = false]; required int64 index = 3 [(gogoproto.nullable) = false];
optional bytes data = 4; optional bytes data = 4;
required int64 id = 5 [(gogoproto.nullable) = false];
} }

View File

@ -18,12 +18,12 @@ func Example_Node() {
// ReadState blocks until there is new state ready. // ReadState blocks until there is new state ready.
rd := <-n.Ready() rd := <-n.Ready()
if !prev.Equal(rd.State) { if !prev.Equal(rd.State) {
saveStateToDisk(st) saveStateToDisk(rd.State)
prev = rd.State prev = rd.State
} }
saveToDisk(ents) saveToDisk(rd.Entries)
go applyToStore(cents) go applyToStore(rd.CommittedEntries)
sendMessages(msgs) sendMessages(rd.Messages)
} }
} }

View File

@ -110,8 +110,8 @@ func (n *Node) Tick() error {
} }
// Propose proposes data be appended to the log. // Propose proposes data be appended to the log.
func (n *Node) Propose(ctx context.Context, data []byte) error { func (n *Node) Propose(ctx context.Context, id int64, data []byte) error {
return n.Step(ctx, []Message{{Type: msgProp, Entries: []Entry{{Data: data}}}}) return n.Step(ctx, []Message{{Type: msgProp, Entries: []Entry{{Id: id, Data: data}}}})
} }
// Step advances the state machine using msgs. Proposals are priotized last so // Step advances the state machine using msgs. Proposals are priotized last so

View File

@ -2,15 +2,6 @@
// source: state.proto // source: state.proto
// DO NOT EDIT! // DO NOT EDIT!
/*
Package raft is a generated protocol buffer package.
It is generated from these files:
state.proto
It has these top-level messages:
State
*/
package raft package raft
import proto "code.google.com/p/gogoprotobuf/proto" import proto "code.google.com/p/gogoprotobuf/proto"
@ -19,8 +10,8 @@ import math "math"
// discarding unused import gogoproto "code.google.com/p/gogoprotobuf/gogoproto/gogo.pb" // discarding unused import gogoproto "code.google.com/p/gogoprotobuf/gogoproto/gogo.pb"
import io "io" import io1 "io"
import code_google_com_p_gogoprotobuf_proto "code.google.com/p/gogoprotobuf/proto" import code_google_com_p_gogoprotobuf_proto1 "code.google.com/p/gogoprotobuf/proto"
// Reference proto, json, and math imports to suppress error if they are not otherwise used. // Reference proto, json, and math imports to suppress error if they are not otherwise used.
var _ = proto.Marshal var _ = proto.Marshal
@ -48,7 +39,7 @@ func (m *State) Unmarshal(data []byte) error {
var wire uint64 var wire uint64
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if index >= l { if index >= l {
return io.ErrUnexpectedEOF return io1.ErrUnexpectedEOF
} }
b := data[index] b := data[index]
index++ index++
@ -62,11 +53,11 @@ func (m *State) Unmarshal(data []byte) error {
switch fieldNum { switch fieldNum {
case 1: case 1:
if wireType != 0 { if wireType != 0 {
return code_google_com_p_gogoprotobuf_proto.ErrWrongType return code_google_com_p_gogoprotobuf_proto1.ErrWrongType
} }
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if index >= l { if index >= l {
return io.ErrUnexpectedEOF return io1.ErrUnexpectedEOF
} }
b := data[index] b := data[index]
index++ index++
@ -77,11 +68,11 @@ func (m *State) Unmarshal(data []byte) error {
} }
case 2: case 2:
if wireType != 0 { if wireType != 0 {
return code_google_com_p_gogoprotobuf_proto.ErrWrongType return code_google_com_p_gogoprotobuf_proto1.ErrWrongType
} }
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if index >= l { if index >= l {
return io.ErrUnexpectedEOF return io1.ErrUnexpectedEOF
} }
b := data[index] b := data[index]
index++ index++
@ -92,11 +83,11 @@ func (m *State) Unmarshal(data []byte) error {
} }
case 3: case 3:
if wireType != 0 { if wireType != 0 {
return code_google_com_p_gogoprotobuf_proto.ErrWrongType return code_google_com_p_gogoprotobuf_proto1.ErrWrongType
} }
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if index >= l { if index >= l {
return io.ErrUnexpectedEOF return io1.ErrUnexpectedEOF
} }
b := data[index] b := data[index]
index++ index++
@ -107,11 +98,11 @@ func (m *State) Unmarshal(data []byte) error {
} }
case 4: case 4:
if wireType != 0 { if wireType != 0 {
return code_google_com_p_gogoprotobuf_proto.ErrWrongType return code_google_com_p_gogoprotobuf_proto1.ErrWrongType
} }
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if index >= l { if index >= l {
return io.ErrUnexpectedEOF return io1.ErrUnexpectedEOF
} }
b := data[index] b := data[index]
index++ index++
@ -130,12 +121,12 @@ func (m *State) Unmarshal(data []byte) error {
} }
} }
index -= sizeOfWire index -= sizeOfWire
skippy, err := code_google_com_p_gogoprotobuf_proto.Skip(data[index:]) skippy, err := code_google_com_p_gogoprotobuf_proto1.Skip(data[index:])
if err != nil { if err != nil {
return err return err
} }
if (index + skippy) > l { if (index + skippy) > l {
return io.ErrUnexpectedEOF return io1.ErrUnexpectedEOF
} }
m.XXX_unrecognized = append(m.XXX_unrecognized, data[index:index+skippy]...) m.XXX_unrecognized = append(m.XXX_unrecognized, data[index:index+skippy]...)
index += skippy index += skippy