Skip to content

N
neoppod
Commit bc16d2ee authored by Kirill Smelkov's avatar Kirill Smelkov
Browse files
Options

.

parent 878bf8ca
Hide whitespace changes
Inline Side-by-side
Showing with 147 additions and 106 deletions
go/neo/proto.go
View file @ bc16d2ee
......@@ -29,7 +29,11 @@ package neo
// several messages and does not itself denote a separate message, its
// definition is prefixed with `//neo:proto typeonly` comment.
//
// XXX neo:proto answerto x? (btw just needs "answer" flag)
// The order of message definitions is significant - messages are assigned
// message IDs in the same order they are defined.
//
// For compatibility with neo/py is a message should have its ID assigned with
// "answer" bit set its definition is prefixed with `//neo:proto answer` comment.
// TODO regroup messages definitions to stay more close to 1 communication topic
// TODO document protocol itself better (who sends who what with which semantic)
......@@ -60,7 +64,7 @@ const (
MAX_PACKET_SIZE = 0x4000000
RESPONSE_MASK = 0x8000
answerBit = 0x8000
)
type ErrorCode uint32
......@@ -192,7 +196,7 @@ type Msg interface {
neoMsgDecode(data []byte) (nread int, err error)
}
// FIXME not pkt
//neo:proto typeonly
type Address struct {
Host string
Port uint16
......@@ -303,7 +307,7 @@ type RequestIdentification struct {
IdTimestamp float64
}
// XXX -> ReplyIdentification? RequestIdentification.Answer somehow ?
//neo:proto answer
type AcceptIdentification struct {
NodeType NodeType // XXX name
MyUUID NodeUUID
......
go/neo/protogen.go
View file @ bc16d2ee
......@@ -168,6 +168,7 @@ func loadPkg(pkgPath string, sources ...string) *types.Package {
// `//neo:proto ...` annotations
type Annotation struct {
typeonly bool
answer bool
}
// parse checks doc for specific comment annotations and, if present, loads them.
......@@ -188,16 +189,45 @@ func (a *Annotation) parse(doc *ast.CommentGroup) {
switch arg {
case "typeonly":
if a.typeonly {
log.Fatalf("%v: duplicate typeonly", cpos)
log.Fatalf("%v: duplicate `typeonly`", cpos)
}
a.typeonly = true
case "answer":
if a.answer {
log.Fatalf("%v: duplicate `answer`", cpos)
}
a.answer = true
default:
log.Fatalf("%v: unknown neo:proto directive %q", cpos, arg)
}
}
}
// MsgCode represents message code in symbolic form: `serial (| answerBit)?`
type MsgCode struct {
msgSerial int
answer bool
}
func (c MsgCode) String() string {
s := fmt.Sprintf("%d", c.msgSerial)
if c.answer {
s += " | answerBit"
}
return s
}
// sort MsgCode by serial
type BySerial []MsgCode
func (v BySerial) Less(i, j int) bool { return v[i].msgSerial < v[j].msgSerial }
func (v BySerial) Swap(i, j int) { v[i], v[j] = v[j], v[i] }
func (v BySerial) Len() int { return len(v) }
// ----------------------------------------
func main() {
var err error
......@@ -223,11 +253,11 @@ import (
"lab.nexedi.com/kirr/neo/go/zodb"
)`)
msgTypeRegistry := map[int]string{} // msgCode -> typename
msgTypeRegistry := map[MsgCode]string{} // msgCode -> typename
// go over message types declaration and generate marshal code for them
buf.emit("// messages marshalling\n")
msgCode := 0
msgSerial := 0
for _, decl := range f.Decls {
// we look for types (which can be only under GenDecl)
gendecl, ok := decl.(*ast.GenDecl)
......@@ -262,10 +292,17 @@ import (
continue
}
fmt.Fprintf(&buf, "// %d. %s\n\n", msgCode, typename)
// generate code for this type to implement neo.Msg
msgCode := MsgCode{msgSerial, specAnnotation.answer}
fmt.Fprintf(&buf, "// %d. %s", msgSerial, typename)
if specAnnotation.answer {
fmt.Fprintf(&buf, " (answer)")
}
fmt.Fprintf(&buf, "\n\n")
buf.emit("func (*%s) neoMsgCode() uint16 {", typename)
buf.emit("return %d", msgCode)
buf.emit("return %s", msgCode)
buf.emit("}\n")
buf.WriteString(generateCodecCode(typespec, &sizer{}))
......@@ -273,20 +310,20 @@ import (
buf.WriteString(generateCodecCode(typespec, &decoder{}))
msgTypeRegistry[msgCode] = typename
msgCode++
msgSerial++
}
}
// now generate message types registry
buf.emit("\n// registry of message types")
buf.emit("var msgTypeRegistry = map[uint16]reflect.Type {") // XXX key -> MsgCode ?
buf.emit("var msgTypeRegistry = map[uint16]reflect.Type {")
// ordered by msgCode
msgCodeV := []int{}
msgCodeV := []MsgCode{}
for msgCode := range msgTypeRegistry {
msgCodeV = append(msgCodeV, msgCode)
}
sort.Ints(msgCodeV)
sort.Sort(BySerial(msgCodeV))
for _, msgCode := range msgCodeV {
buf.emit("%v: reflect.TypeOf(%v{}),", msgCode, msgTypeRegistry[msgCode])
......@@ -545,8 +582,8 @@ type sizer struct {
// encode<typ2>(data[n2:], path2)
// ...
//
// TODO encode have to care in neoMsgEncode to emit preambule such that bound
// checking is performed only once (currenty compiler emits many of them)
// TODO encode have to care in neoMsgEncode to emit preamble such that bound
// checking is performed only once (currently compiler emits many of them)
type encoder struct {
commonCodeGen
n int // current write position in data
......@@ -635,7 +672,7 @@ func (d *decoder) resetPos() {
// mark current place for insertion of overflow check code
//
// The check will be acutally inserted later.
// The check will be actually inserted later.
//
// later: because first we go forward in decode path scanning ahead as far as
// we can - until first seeing variable-size encoded something, and then -
......
go/neo/zproto-marshal.go
View file @ bc16d2ee
......@@ -123,10 +123,10 @@ overflow:
return 0, ErrDecodeOverflow
}
// 2. AcceptIdentification
// 2. AcceptIdentification (answer)
func (*AcceptIdentification) neoMsgCode() uint16 {
return 2
return 2 | answerBit
}
func (p *AcceptIdentification) neoMsgEncodedLen() int {
......@@ -3415,92 +3415,92 @@ func (p *NotifyReady) neoMsgDecode(data []byte) (int, error) {
// registry of message types
var msgTypeRegistry = map[uint16]reflect.Type{
0: reflect.TypeOf(Error{}),
1: reflect.TypeOf(RequestIdentification{}),
2: reflect.TypeOf(AcceptIdentification{}),
3: reflect.TypeOf(Ping{}),
4: reflect.TypeOf(CloseClient{}),
5: reflect.TypeOf(PrimaryMaster{}),
6: reflect.TypeOf(AnswerPrimary{}),
7: reflect.TypeOf(NotPrimaryMaster{}),
8: reflect.TypeOf(Recovery{}),
9: reflect.TypeOf(AnswerRecovery{}),
10: reflect.TypeOf(LastIDs{}),
11: reflect.TypeOf(AnswerLastIDs{}),
12: reflect.TypeOf(AskPartitionTable{}),
13: reflect.TypeOf(AnswerPartitionTable{}),
14: reflect.TypeOf(NotifyPartitionTable{}),
15: reflect.TypeOf(NotifyPartitionChanges{}),
16: reflect.TypeOf(StartOperation{}),
17: reflect.TypeOf(StopOperation{}),
18: reflect.TypeOf(UnfinishedTransactions{}),
19: reflect.TypeOf(AnswerUnfinishedTransactions{}),
20: reflect.TypeOf(LockedTransactions{}),
21: reflect.TypeOf(AnswerLockedTransactions{}),
22: reflect.TypeOf(FinalTID{}),
23: reflect.TypeOf(AnswerFinalTID{}),
24: reflect.TypeOf(ValidateTransaction{}),
25: reflect.TypeOf(BeginTransaction{}),
26: reflect.TypeOf(AnswerBeginTransaction{}),
27: reflect.TypeOf(FailedVote{}),
28: reflect.TypeOf(FinishTransaction{}),
29: reflect.TypeOf(AnswerFinishTransaction{}),
30: reflect.TypeOf(NotifyTransactionFinished{}),
31: reflect.TypeOf(LockInformation{}),
32: reflect.TypeOf(AnswerLockInformation{}),
33: reflect.TypeOf(InvalidateObjects{}),
34: reflect.TypeOf(UnlockInformation{}),
35: reflect.TypeOf(GenerateOIDs{}),
36: reflect.TypeOf(AnswerGenerateOIDs{}),
37: reflect.TypeOf(Deadlock{}),
38: reflect.TypeOf(RebaseTransaction{}),
39: reflect.TypeOf(AnswerRebaseTransaction{}),
40: reflect.TypeOf(RebaseObject{}),
41: reflect.TypeOf(AnswerRebaseObject{}),
42: reflect.TypeOf(StoreObject{}),
43: reflect.TypeOf(AnswerStoreObject{}),
44: reflect.TypeOf(AbortTransaction{}),
45: reflect.TypeOf(StoreTransaction{}),
46: reflect.TypeOf(VoteTransaction{}),
47: reflect.TypeOf(GetObject{}),
48: reflect.TypeOf(AnswerGetObject{}),
49: reflect.TypeOf(TIDList{}),
50: reflect.TypeOf(AnswerTIDList{}),
51: reflect.TypeOf(TIDListFrom{}),
52: reflect.TypeOf(AnswerTIDListFrom{}),
53: reflect.TypeOf(TransactionInformation{}),
54: reflect.TypeOf(AnswerTransactionInformation{}),
55: reflect.TypeOf(ObjectHistory{}),
56: reflect.TypeOf(AnswerObjectHistory{}),
57: reflect.TypeOf(PartitionList{}),
58: reflect.TypeOf(AnswerPartitionList{}),
59: reflect.TypeOf(NodeList{}),
60: reflect.TypeOf(AnswerNodeList{}),
61: reflect.TypeOf(SetNodeState{}),
62: reflect.TypeOf(AddPendingNodes{}),
63: reflect.TypeOf(TweakPartitionTable{}),
64: reflect.TypeOf(NotifyNodeInformation{}),
65: reflect.TypeOf(NodeInformation{}),
66: reflect.TypeOf(SetClusterState{}),
67: reflect.TypeOf(repairFlags{}),
68: reflect.TypeOf(Repair{}),
69: reflect.TypeOf(RepairOne{}),
70: reflect.TypeOf(NotifyClusterState{}),
71: reflect.TypeOf(AskClusterState{}),
72: reflect.TypeOf(AnswerClusterState{}),
73: reflect.TypeOf(ObjectUndoSerial{}),
74: reflect.TypeOf(AnswerObjectUndoSerial{}),
75: reflect.TypeOf(CheckCurrentSerial{}),
76: reflect.TypeOf(Pack{}),
77: reflect.TypeOf(AnswerPack{}),
78: reflect.TypeOf(CheckReplicas{}),
79: reflect.TypeOf(CheckPartition{}),
80: reflect.TypeOf(CheckTIDRange{}),
81: reflect.TypeOf(AnswerCheckTIDRange{}),
82: reflect.TypeOf(CheckSerialRange{}),
83: reflect.TypeOf(AnswerCheckSerialRange{}),
84: reflect.TypeOf(PartitionCorrupted{}),
85: reflect.TypeOf(LastTransaction{}),
86: reflect.TypeOf(AnswerLastTransaction{}),
87: reflect.TypeOf(NotifyReady{}),
0: reflect.TypeOf(Error{}),
1: reflect.TypeOf(RequestIdentification{}),
2 | answerBit: reflect.TypeOf(AcceptIdentification{}),
3: reflect.TypeOf(Ping{}),
4: reflect.TypeOf(CloseClient{}),
5: reflect.TypeOf(PrimaryMaster{}),
6: reflect.TypeOf(AnswerPrimary{}),
7: reflect.TypeOf(NotPrimaryMaster{}),
8: reflect.TypeOf(Recovery{}),
9: reflect.TypeOf(AnswerRecovery{}),
10: reflect.TypeOf(LastIDs{}),
11: reflect.TypeOf(AnswerLastIDs{}),
12: reflect.TypeOf(AskPartitionTable{}),
13: reflect.TypeOf(AnswerPartitionTable{}),
14: reflect.TypeOf(NotifyPartitionTable{}),
15: reflect.TypeOf(NotifyPartitionChanges{}),
16: reflect.TypeOf(StartOperation{}),
17: reflect.TypeOf(StopOperation{}),
18: reflect.TypeOf(UnfinishedTransactions{}),
19: reflect.TypeOf(AnswerUnfinishedTransactions{}),
20: reflect.TypeOf(LockedTransactions{}),
21: reflect.TypeOf(AnswerLockedTransactions{}),
22: reflect.TypeOf(FinalTID{}),
23: reflect.TypeOf(AnswerFinalTID{}),
24: reflect.TypeOf(ValidateTransaction{}),
25: reflect.TypeOf(BeginTransaction{}),
26: reflect.TypeOf(AnswerBeginTransaction{}),
27: reflect.TypeOf(FailedVote{}),
28: reflect.TypeOf(FinishTransaction{}),
29: reflect.TypeOf(AnswerFinishTransaction{}),
30: reflect.TypeOf(NotifyTransactionFinished{}),
31: reflect.TypeOf(LockInformation{}),
32: reflect.TypeOf(AnswerLockInformation{}),
33: reflect.TypeOf(InvalidateObjects{}),
34: reflect.TypeOf(UnlockInformation{}),
35: reflect.TypeOf(GenerateOIDs{}),
36: reflect.TypeOf(AnswerGenerateOIDs{}),
37: reflect.TypeOf(Deadlock{}),
38: reflect.TypeOf(RebaseTransaction{}),
39: reflect.TypeOf(AnswerRebaseTransaction{}),
40: reflect.TypeOf(RebaseObject{}),
41: reflect.TypeOf(AnswerRebaseObject{}),
42: reflect.TypeOf(StoreObject{}),
43: reflect.TypeOf(AnswerStoreObject{}),
44: reflect.TypeOf(AbortTransaction{}),
45: reflect.TypeOf(StoreTransaction{}),
46: reflect.TypeOf(VoteTransaction{}),
47: reflect.TypeOf(GetObject{}),
48: reflect.TypeOf(AnswerGetObject{}),
49: reflect.TypeOf(TIDList{}),
50: reflect.TypeOf(AnswerTIDList{}),
51: reflect.TypeOf(TIDListFrom{}),
52: reflect.TypeOf(AnswerTIDListFrom{}),
53: reflect.TypeOf(TransactionInformation{}),
54: reflect.TypeOf(AnswerTransactionInformation{}),
55: reflect.TypeOf(ObjectHistory{}),
56: reflect.TypeOf(AnswerObjectHistory{}),
57: reflect.TypeOf(PartitionList{}),
58: reflect.TypeOf(AnswerPartitionList{}),
59: reflect.TypeOf(NodeList{}),
60: reflect.TypeOf(AnswerNodeList{}),
61: reflect.TypeOf(SetNodeState{}),
62: reflect.TypeOf(AddPendingNodes{}),
63: reflect.TypeOf(TweakPartitionTable{}),
64: reflect.TypeOf(NotifyNodeInformation{}),
65: reflect.TypeOf(NodeInformation{}),
66: reflect.TypeOf(SetClusterState{}),
67: reflect.TypeOf(repairFlags{}),
68: reflect.TypeOf(Repair{}),
69: reflect.TypeOf(RepairOne{}),
70: reflect.TypeOf(NotifyClusterState{}),
71: reflect.TypeOf(AskClusterState{}),
72: reflect.TypeOf(AnswerClusterState{}),
73: reflect.TypeOf(ObjectUndoSerial{}),
74: reflect.TypeOf(AnswerObjectUndoSerial{}),
75: reflect.TypeOf(CheckCurrentSerial{}),
76: reflect.TypeOf(Pack{}),
77: reflect.TypeOf(AnswerPack{}),
78: reflect.TypeOf(CheckReplicas{}),
79: reflect.TypeOf(CheckPartition{}),
80: reflect.TypeOf(CheckTIDRange{}),
81: reflect.TypeOf(AnswerCheckTIDRange{}),
82: reflect.TypeOf(CheckSerialRange{}),
83: reflect.TypeOf(AnswerCheckSerialRange{}),
84: reflect.TypeOf(PartitionCorrupted{}),
85: reflect.TypeOf(LastTransaction{}),
86: reflect.TypeOf(AnswerLastTransaction{}),
87: reflect.TypeOf(NotifyReady{}),
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7