From 9ad11066470799791bfa401cd0766a786a92e310 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 16 Apr 2020 13:18:40 -0500 Subject: [PATCH] implement transaction API layer and intra-cluster messaging also adds a "noSleep" option to the server command to avoid the 5 second sleep we introduced on startup for non-coordinator cluster nodes. The sleep doesn't seem to be needed in the tests and makes them much slower. --- api.go | 16 + broadcast.go | 5 + cluster.go | 12 + encoding/proto/proto.go | 54 +++ holder.go | 31 ++ internal/private.pb.go | 945 ++++++++++++++++++++++++++++++++++++---- internal/private.proto | 16 + server.go | 152 +++++++ server/server.go | 14 +- server/server_test.go | 109 +++++ test/pilosa.go | 2 + test/transaction.go | 36 ++ transaction.go | 25 +- transaction.md | 63 ++- transaction_test.go | 66 +-- 15 files changed, 1410 insertions(+), 136 deletions(-) create mode 100644 test/transaction.go diff --git a/api.go b/api.go index 7ced231ae..6ca4e739a 100644 --- a/api.go +++ b/api.go @@ -1585,6 +1585,22 @@ func (api *API) PrimaryReplicaNodeURL() url.URL { return node.URI.URL() } +func (api *API) StartTransaction(id string, timeout time.Duration, exclusive bool, remote bool) (Transaction, error) { + return api.server.StartTransaction(id, timeout, exclusive, remote) +} + +func (api *API) FinishTransaction(id string, remote bool) (Transaction, error) { + return api.server.FinishTransaction(id, remote) +} + +func (api *API) Transactions() (map[string]Transaction, error) { + return api.server.Transactions() +} + +func (api *API) GetTransaction(id string, remote bool) (Transaction, error) { + return api.server.GetTransaction(id, remote) +} + type serverInfo struct { ShardWidth uint64 `json:"shardWidth"` Memory uint64 `json:"memory"` diff --git a/broadcast.go b/broadcast.go index 6f2245992..37d2bb39d 100644 --- a/broadcast.go +++ b/broadcast.go @@ -69,6 +69,7 @@ const ( messageTypeRecalculateCaches messageTypeNodeEvent messageTypeNodeStatus + messageTypeTransaction ) // MarshalInternalMessage serializes the pilosa message and adds pilosa internal @@ -116,6 +117,8 @@ func getMessage(typ byte) Message { return &NodeEvent{} case messageTypeNodeStatus: return &NodeStatus{} + case messageTypeTransaction: + return &TransactionMessage{} default: panic(fmt.Sprintf("unknown message type %d", typ)) } @@ -155,6 +158,8 @@ func getMessageType(m Message) byte { return messageTypeNodeEvent case *NodeStatus: return messageTypeNodeStatus + case *TransactionMessage: + return messageTypeTransaction default: panic(fmt.Sprintf("don't have type for message %#v", m)) } diff --git a/cluster.go b/cluster.go index af0b43e77..d76365943 100644 --- a/cluster.go +++ b/cluster.go @@ -2589,3 +2589,15 @@ type FieldStatus struct { // RecalculateCaches is an internal message for recalculating all caches // within a holder. type RecalculateCaches struct{} + +// Transaction Actions +const ( + TRANSACTION_START = "start" + TRANSACTION_FINISH = "finish" + TRANSACTION_VALIDATE = "validate" +) + +type TransactionMessage struct { + Transaction Transaction + Action string +} diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 3c72e1d16..a08e3de28 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -17,6 +17,7 @@ package proto import ( "fmt" "sort" + "time" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/v2" @@ -290,6 +291,14 @@ func (Serializer) Unmarshal(buf []byte, m pilosa.Message) error { } decodeTranslateIDsResponse(msg, mt) return nil + case *pilosa.TransactionMessage: + msg := &internal.TransactionMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling TransactionMessage") + } + decodeTransactionMessage(msg, mt) + return nil default: panic(fmt.Sprintf("unhandled pilosa.Message of type %T: %#v", mt, m)) } @@ -359,6 +368,8 @@ func encodeToProto(m pilosa.Message) proto.Message { return encodeTranslateIDsRequest(mt) case *pilosa.TranslateIDsResponse: return encodeTranslateIDsResponse(mt) + case *pilosa.TransactionMessage: + return encodeTransactionMessage(mt) } return nil } @@ -829,6 +840,35 @@ func encodeTranslateIDsResponse(response *pilosa.TranslateIDsResponse) *internal } } +func encodeTransactionMessage(msg *pilosa.TransactionMessage) *internal.TransactionMessage { + return &internal.TransactionMessage{ + Action: msg.Action, + Transaction: encodeTransaction(msg.Transaction), + } +} + +func encodeTransaction(trns pilosa.Transaction) *internal.Transaction { + return &internal.Transaction{ + ID: trns.ID, + Active: trns.Active, + Exclusive: trns.Exclusive, + Timeout: int64(trns.Timeout), + Deadline: encodeTransactionDeadline(trns.Deadline), + Stats: encodeTransactionStats(trns.Stats), + } +} + +func encodeTransactionDeadline(deadline time.Time) int64 { + if deadline.Year() > 2262 || deadline.Year() < 1678 { + return 0 + } + return deadline.UnixNano() +} + +func encodeTransactionStats(stats pilosa.TransactionStats) *internal.TransactionStats { + return &internal.TransactionStats{} +} + func decodeResizeInstruction(ri *internal.ResizeInstruction, m *pilosa.ResizeInstruction) { m.JobID = ri.JobID m.Node = &pilosa.Node{} @@ -1198,6 +1238,20 @@ func decodeTranslateIDsResponse(pb *internal.TranslateIDsResponse, m *pilosa.Tra m.Keys = pb.Keys } +func decodeTransactionMessage(pb *internal.TransactionMessage, m *pilosa.TransactionMessage) { + m.Action = pb.Action + decodeTransaction(pb.Transaction, &m.Transaction) +} + +func decodeTransaction(pb *internal.Transaction, trns *pilosa.Transaction) { + trns.ID = pb.ID + trns.Active = pb.Active + trns.Exclusive = pb.Exclusive + trns.Timeout = time.Duration(pb.Timeout) + trns.Deadline = time.Unix(0, pb.Deadline) + // TODO m.Stats... once it has anything +} + // QueryResult types. const ( queryResultTypeNil uint32 = iota diff --git a/holder.go b/holder.go index 51a89826b..b345c2a96 100644 --- a/holder.go +++ b/holder.go @@ -85,6 +85,12 @@ type Holder struct { OpenTranslateStore OpenTranslateStoreFunc OpenTranslateReader OpenTranslateReaderFunc + // Func to open whatever implementation of transaction store we're using. + OpenTransactionStore OpenTransactionStoreFunc + + // transactionManager + transactionManager *TransactionManager + translationSyncer translationSyncer // Queue of fields (having a foreign index) which have @@ -98,6 +104,22 @@ type Holder struct { opening bool } +func (h *Holder) StartTransaction(id string, timeout time.Duration, exclusive bool) (Transaction, error) { + return h.transactionManager.Start(id, timeout, exclusive) +} + +func (h *Holder) FinishTransaction(id string) (Transaction, error) { + return h.transactionManager.Finish(id) +} + +func (h *Holder) Transactions() (map[string]Transaction, error) { + return h.transactionManager.List() +} + +func (h *Holder) GetTransaction(id string) (Transaction, error) { + return h.transactionManager.Get(id) +} + // lockedChan looks a little ridiculous admittedly, but exists for good reason. // The channel within is used (for example) to signal to other goroutines when // the Holder has finished opening (via closing the channel). However, it is @@ -142,6 +164,8 @@ func NewHolder(partitionN int) *Holder { OpenTranslateStore: OpenInMemTranslateStore, + OpenTransactionStore: OpenInMemTransactionStore, + translationSyncer: NopTranslationSyncer, Logger: logger.NopLogger, @@ -170,6 +194,13 @@ func (h *Holder) Open() error { return ErrCannotOpenV1TranslateFile } + if tstore, err := h.OpenTransactionStore(h.Path); err != nil { + return errors.Wrap(err, "opening transaction store") + } else { + h.transactionManager = NewTransactionManager(tstore) + h.transactionManager.Log = h.Logger + } + // Open path to read all index directories. f, err := os.Open(h.Path) if err != nil { diff --git a/internal/private.pb.go b/internal/private.pb.go index c78b91ff5..a1d0870bd 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -2187,6 +2187,187 @@ func (m *RecalculateCaches) XXX_DiscardUnknown() { var xxx_messageInfo_RecalculateCaches proto.InternalMessageInfo +type TransactionMessage struct { + Action string `protobuf:"bytes,1,opt,name=Action,proto3" json:"Action,omitempty"` + Transaction *Transaction `protobuf:"bytes,2,opt,name=Transaction,proto3" json:"Transaction,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *TransactionMessage) Reset() { *m = TransactionMessage{} } +func (m *TransactionMessage) String() string { return proto.CompactTextString(m) } +func (*TransactionMessage) ProtoMessage() {} +func (*TransactionMessage) Descriptor() ([]byte, []int) { + return fileDescriptor_d2a91b51c7bdc125, []int{35} +} +func (m *TransactionMessage) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *TransactionMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_TransactionMessage.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *TransactionMessage) XXX_Merge(src proto.Message) { + xxx_messageInfo_TransactionMessage.Merge(m, src) +} +func (m *TransactionMessage) XXX_Size() int { + return m.Size() +} +func (m *TransactionMessage) XXX_DiscardUnknown() { + xxx_messageInfo_TransactionMessage.DiscardUnknown(m) +} + +var xxx_messageInfo_TransactionMessage proto.InternalMessageInfo + +func (m *TransactionMessage) GetAction() string { + if m != nil { + return m.Action + } + return "" +} + +func (m *TransactionMessage) GetTransaction() *Transaction { + if m != nil { + return m.Transaction + } + return nil +} + +type Transaction struct { + ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"` + Active bool `protobuf:"varint,2,opt,name=Active,proto3" json:"Active,omitempty"` + Exclusive bool `protobuf:"varint,3,opt,name=Exclusive,proto3" json:"Exclusive,omitempty"` + Timeout int64 `protobuf:"varint,4,opt,name=Timeout,proto3" json:"Timeout,omitempty"` + Deadline int64 `protobuf:"varint,5,opt,name=Deadline,proto3" json:"Deadline,omitempty"` + Stats *TransactionStats `protobuf:"bytes,6,opt,name=Stats,proto3" json:"Stats,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Transaction) Reset() { *m = Transaction{} } +func (m *Transaction) String() string { return proto.CompactTextString(m) } +func (*Transaction) ProtoMessage() {} +func (*Transaction) Descriptor() ([]byte, []int) { + return fileDescriptor_d2a91b51c7bdc125, []int{36} +} +func (m *Transaction) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *Transaction) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_Transaction.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *Transaction) XXX_Merge(src proto.Message) { + xxx_messageInfo_Transaction.Merge(m, src) +} +func (m *Transaction) XXX_Size() int { + return m.Size() +} +func (m *Transaction) XXX_DiscardUnknown() { + xxx_messageInfo_Transaction.DiscardUnknown(m) +} + +var xxx_messageInfo_Transaction proto.InternalMessageInfo + +func (m *Transaction) GetID() string { + if m != nil { + return m.ID + } + return "" +} + +func (m *Transaction) GetActive() bool { + if m != nil { + return m.Active + } + return false +} + +func (m *Transaction) GetExclusive() bool { + if m != nil { + return m.Exclusive + } + return false +} + +func (m *Transaction) GetTimeout() int64 { + if m != nil { + return m.Timeout + } + return 0 +} + +func (m *Transaction) GetDeadline() int64 { + if m != nil { + return m.Deadline + } + return 0 +} + +func (m *Transaction) GetStats() *TransactionStats { + if m != nil { + return m.Stats + } + return nil +} + +type TransactionStats struct { + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *TransactionStats) Reset() { *m = TransactionStats{} } +func (m *TransactionStats) String() string { return proto.CompactTextString(m) } +func (*TransactionStats) ProtoMessage() {} +func (*TransactionStats) Descriptor() ([]byte, []int) { + return fileDescriptor_d2a91b51c7bdc125, []int{37} +} +func (m *TransactionStats) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *TransactionStats) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_TransactionStats.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *TransactionStats) XXX_Merge(src proto.Message) { + xxx_messageInfo_TransactionStats.Merge(m, src) +} +func (m *TransactionStats) XXX_Size() int { + return m.Size() +} +func (m *TransactionStats) XXX_DiscardUnknown() { + xxx_messageInfo_TransactionStats.DiscardUnknown(m) +} + +var xxx_messageInfo_TransactionStats proto.InternalMessageInfo + func init() { proto.RegisterType((*IndexMeta)(nil), "internal.IndexMeta") proto.RegisterType((*FieldOptions)(nil), "internal.FieldOptions") @@ -2224,94 +2405,103 @@ func init() { proto.RegisterType((*UpdateCoordinatorMessage)(nil), "internal.UpdateCoordinatorMessage") proto.RegisterType((*Topology)(nil), "internal.Topology") proto.RegisterType((*RecalculateCaches)(nil), "internal.RecalculateCaches") + proto.RegisterType((*TransactionMessage)(nil), "internal.TransactionMessage") + proto.RegisterType((*Transaction)(nil), "internal.Transaction") + proto.RegisterType((*TransactionStats)(nil), "internal.TransactionStats") } func init() { proto.RegisterFile("private.proto", fileDescriptor_d2a91b51c7bdc125) } var fileDescriptor_d2a91b51c7bdc125 = []byte{ - // 1298 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xcd, 0x72, 0x1b, 0xc5, - 0x13, 0xff, 0xaf, 0x56, 0x96, 0xa5, 0x96, 0xe5, 0xc8, 0x93, 0xc4, 0xd9, 0xe4, 0x4f, 0x19, 0x31, - 0xa4, 0x88, 0x48, 0x15, 0x26, 0x95, 0x70, 0xe0, 0x2b, 0x55, 0xc1, 0x96, 0x13, 0x44, 0xb0, 0x93, - 0x8c, 0x9c, 0xdc, 0x38, 0x8c, 0x57, 0x53, 0xf1, 0x96, 0x57, 0xbb, 0x62, 0x77, 0xd6, 0x91, 0x73, - 0xe0, 0x0a, 0x55, 0xbc, 0x00, 0x47, 0x1e, 0x87, 0x23, 0x8f, 0x40, 0x85, 0xa7, 0xe0, 0x46, 0x4d, - 0xcf, 0xcc, 0x7e, 0x28, 0x16, 0x0e, 0x0e, 0xb7, 0xe9, 0x5f, 0xf7, 0xf4, 0x77, 0xf7, 0xec, 0x42, - 0x67, 0x9a, 0x04, 0xc7, 0x5c, 0x8a, 0xcd, 0x69, 0x12, 0xcb, 0x98, 0x34, 0x83, 0x48, 0x8a, 0x24, - 0xe2, 0xe1, 0xb5, 0x95, 0x69, 0x76, 0x10, 0x06, 0xbe, 0xc6, 0xe9, 0x03, 0x68, 0x0d, 0xa3, 0xb1, - 0x98, 0xed, 0x0a, 0xc9, 0x09, 0x81, 0xfa, 0x43, 0x71, 0x92, 0x7a, 0x6e, 0xcf, 0xe9, 0x37, 0x19, - 0x9e, 0xc9, 0x07, 0xb0, 0xba, 0x9f, 0x70, 0xff, 0x68, 0x67, 0x16, 0xa4, 0x52, 0x44, 0xbe, 0xf0, - 0xea, 0xc8, 0x9d, 0x43, 0xe9, 0xaf, 0x2e, 0xac, 0xdc, 0x0f, 0x44, 0x38, 0x7e, 0x34, 0x95, 0x41, - 0x1c, 0xa5, 0x4a, 0xd9, 0xfe, 0xc9, 0x54, 0x78, 0xcd, 0x9e, 0xd3, 0x6f, 0x31, 0x3c, 0x93, 0x77, - 0xa0, 0xb5, 0xcd, 0xfd, 0x43, 0x81, 0x0c, 0x17, 0x19, 0x05, 0x90, 0x73, 0x47, 0xc1, 0x4b, 0x6d, - 0xa5, 0xc3, 0x0a, 0x80, 0xf4, 0xa0, 0xbd, 0x1f, 0x4c, 0xc4, 0x93, 0x8c, 0x47, 0x32, 0x9b, 0x78, - 0x4b, 0x78, 0xbb, 0x0c, 0x91, 0x75, 0x68, 0x3c, 0x0a, 0xc7, 0xbb, 0x41, 0xe4, 0xb5, 0x7a, 0x4e, - 0xdf, 0x65, 0x86, 0xb2, 0x38, 0x9f, 0x79, 0x50, 0xe0, 0x7c, 0x96, 0x87, 0xdb, 0xae, 0x86, 0xbb, - 0x17, 0x8f, 0x24, 0x8f, 0xc6, 0x3c, 0x19, 0x3f, 0x0b, 0xc4, 0x0b, 0x6f, 0x45, 0x87, 0x5b, 0x45, - 0xd5, 0xdd, 0x2d, 0x9e, 0x0a, 0xaf, 0x83, 0x1a, 0xf1, 0x4c, 0xae, 0x41, 0x73, 0x2b, 0x90, 0x03, - 0x31, 0x95, 0x87, 0xde, 0x6a, 0xcf, 0xe9, 0xd7, 0x59, 0x4e, 0x93, 0x4b, 0xb0, 0x34, 0xf2, 0x79, - 0x28, 0xbc, 0x0b, 0x78, 0x41, 0x13, 0x84, 0xc2, 0xca, 0xfd, 0x38, 0x11, 0xc1, 0xf3, 0x08, 0x8b, - 0xe0, 0x75, 0x31, 0xa8, 0x0a, 0x46, 0xde, 0x07, 0x57, 0x85, 0xb4, 0xd6, 0x73, 0xfa, 0xed, 0xdb, - 0x6b, 0x9b, 0xb6, 0x8e, 0x9b, 0x03, 0xe1, 0x07, 0x13, 0x1e, 0x32, 0xc5, 0x45, 0x21, 0x3e, 0xf3, - 0xc8, 0x62, 0x21, 0x3e, 0xa3, 0x14, 0x56, 0x87, 0x93, 0x69, 0x9c, 0x48, 0x26, 0xd2, 0x69, 0x1c, - 0xa5, 0x82, 0x74, 0xc1, 0xdd, 0x49, 0x12, 0xcf, 0x41, 0xb3, 0xea, 0x48, 0x7f, 0x80, 0xee, 0x56, - 0x18, 0xfb, 0x47, 0x03, 0x2e, 0x39, 0x13, 0xdf, 0x67, 0x22, 0x95, 0xca, 0x77, 0xed, 0x9e, 0x96, - 0xd3, 0x84, 0x42, 0xb1, 0xde, 0x5e, 0x4d, 0xa3, 0x48, 0xa8, 0xbc, 0x60, 0xd6, 0x74, 0x79, 0xf0, - 0x8c, 0xb1, 0x1f, 0xf2, 0x64, 0x8c, 0x35, 0xad, 0x33, 0x4d, 0x28, 0x14, 0x2d, 0x61, 0x1f, 0xd4, - 0x99, 0x26, 0xe8, 0x10, 0xd6, 0x4a, 0xf6, 0x8d, 0x9b, 0xeb, 0xd0, 0x60, 0xf1, 0x8b, 0xe1, 0x20, - 0xf5, 0x9c, 0x9e, 0xdb, 0xaf, 0x33, 0x43, 0x61, 0xc3, 0xc4, 0x61, 0x36, 0x89, 0x14, 0xab, 0x86, - 0xac, 0x02, 0xa0, 0x57, 0x61, 0x09, 0xbb, 0x47, 0x45, 0x59, 0xdc, 0x55, 0x47, 0xfa, 0xa3, 0x03, - 0xad, 0x5d, 0x3e, 0x43, 0x47, 0x52, 0x72, 0x17, 0x9a, 0xb6, 0xb6, 0x28, 0xd4, 0xbe, 0xfd, 0x5e, - 0x91, 0xc1, 0x5c, 0x6c, 0xd3, 0xca, 0xec, 0x44, 0x32, 0x39, 0x61, 0xf9, 0x95, 0x6b, 0x5f, 0x40, - 0xa7, 0xc2, 0x52, 0xf6, 0x8e, 0xc4, 0x89, 0xcd, 0xea, 0x91, 0x38, 0x51, 0xb1, 0x1e, 0xf3, 0x30, - 0x13, 0x98, 0xab, 0x3a, 0xd3, 0xc4, 0xe7, 0xb5, 0x4f, 0x1d, 0xfa, 0x0c, 0xc8, 0x76, 0x22, 0xb8, - 0x14, 0x68, 0x64, 0x57, 0xa4, 0x29, 0x7f, 0x2e, 0xce, 0xca, 0xb8, 0x5b, 0xce, 0x78, 0x9e, 0xdd, - 0x5a, 0x29, 0xbb, 0xf4, 0x26, 0x90, 0x81, 0x08, 0x85, 0x14, 0x66, 0xba, 0xff, 0x41, 0x2f, 0x1d, - 0x59, 0x1f, 0xce, 0x96, 0x25, 0x37, 0xa0, 0xae, 0x56, 0x05, 0x1a, 0x6b, 0xdf, 0xbe, 0x58, 0xe4, - 0x29, 0xdf, 0x22, 0x0c, 0x05, 0x68, 0x68, 0x95, 0xa2, 0x97, 0x6f, 0x18, 0x58, 0xa5, 0x95, 0x6e, - 0x1a, 0x53, 0x2e, 0x9a, 0x5a, 0x2f, 0x4c, 0x95, 0xd7, 0x8c, 0xb1, 0x76, 0xcf, 0x86, 0x7b, 0x5e, - 0x6b, 0xd4, 0x87, 0xff, 0x6b, 0x0d, 0x5f, 0x1d, 0xf3, 0x20, 0xe4, 0x07, 0xe1, 0xbf, 0xaa, 0x48, - 0xc5, 0x71, 0x0f, 0x96, 0xf1, 0xee, 0x70, 0x60, 0x7a, 0xdb, 0x92, 0xf4, 0x3b, 0x28, 0xc6, 0x64, - 0x8f, 0x4f, 0x84, 0xd1, 0x86, 0xe7, 0x3c, 0xde, 0xda, 0xd9, 0xf1, 0x2a, 0xc3, 0x6a, 0xb4, 0xd4, - 0xaa, 0x76, 0x95, 0x61, 0x24, 0xe8, 0x1d, 0x68, 0x8c, 0xfc, 0x43, 0x31, 0xe1, 0xe4, 0x43, 0x58, - 0x46, 0x0f, 0x45, 0x6a, 0x3a, 0xfa, 0xc2, 0x5c, 0xa5, 0x98, 0xe5, 0xd3, 0xd4, 0x44, 0x76, 0xaa, - 0x4f, 0x1f, 0xc1, 0xb2, 0x31, 0x8c, 0x13, 0xbd, 0xa0, 0xe2, 0x56, 0x86, 0xdc, 0x80, 0x06, 0x3a, - 0x9b, 0x7a, 0xf5, 0x79, 0xab, 0x88, 0x33, 0xc3, 0xa6, 0x3b, 0xe0, 0x3e, 0x65, 0x43, 0x35, 0xd8, - 0xe8, 0xb0, 0x35, 0x6a, 0x28, 0xe5, 0xca, 0xd7, 0x71, 0x2a, 0x4d, 0x5a, 0xf1, 0xac, 0xb0, 0xc7, - 0x71, 0x22, 0x31, 0xa5, 0x1d, 0x86, 0x67, 0x9a, 0x42, 0x7d, 0x2f, 0x1e, 0x0b, 0xb2, 0x0a, 0xb5, - 0xe1, 0xc0, 0xe8, 0xa8, 0x0d, 0x07, 0xe4, 0x5d, 0x54, 0x6f, 0x32, 0xd9, 0x29, 0x9c, 0x78, 0xca, - 0x86, 0x0c, 0x0d, 0x5f, 0x87, 0xce, 0x30, 0xdd, 0x8e, 0xe3, 0x64, 0x1c, 0x44, 0x5c, 0xc6, 0x89, - 0x79, 0xf2, 0xaa, 0x20, 0x8e, 0x96, 0xe4, 0x52, 0x3f, 0x46, 0x2d, 0xa6, 0x09, 0x7a, 0x0f, 0xba, - 0xca, 0x28, 0x12, 0xb6, 0x3d, 0xd6, 0xa1, 0xa1, 0xb0, 0xdc, 0x09, 0x43, 0x15, 0x1a, 0x6a, 0x65, - 0x0d, 0xdf, 0x6a, 0x0d, 0x3b, 0xc7, 0x22, 0x92, 0xa5, 0x06, 0x43, 0x1a, 0x15, 0x74, 0x98, 0x26, - 0x08, 0xd5, 0x01, 0x9a, 0x48, 0x56, 0x8b, 0x48, 0x14, 0xca, 0x90, 0x47, 0x7f, 0x76, 0x00, 0xac, - 0x43, 0x59, 0x9a, 0x5f, 0x71, 0x16, 0x5f, 0x21, 0x7d, 0xdb, 0x28, 0x66, 0xb8, 0xba, 0x85, 0x94, - 0xc6, 0x99, 0x6d, 0xa4, 0x8f, 0x8b, 0x46, 0xd2, 0x25, 0xbd, 0x3c, 0xd7, 0x00, 0xda, 0x6a, 0xd1, - 0x4e, 0x8f, 0xa1, 0x5d, 0xc2, 0x17, 0x34, 0x95, 0xed, 0x92, 0xda, 0xbc, 0x4a, 0xc4, 0x8d, 0x4a, - 0xdb, 0x2b, 0x0f, 0xa1, 0x5d, 0x82, 0x4f, 0xd5, 0xd8, 0x87, 0x0b, 0xd5, 0xb1, 0xb5, 0xcf, 0xc1, - 0x3c, 0x4c, 0x03, 0xe8, 0x6c, 0x87, 0x59, 0x2a, 0x45, 0x62, 0xd4, 0xa9, 0x37, 0x44, 0x03, 0x79, - 0xf1, 0x0a, 0xe0, 0xf4, 0xfa, 0x91, 0xeb, 0xb0, 0xa4, 0xd2, 0xa8, 0xa7, 0xef, 0xf5, 0x1c, 0x6b, - 0x26, 0x7d, 0x06, 0xcd, 0xad, 0xd1, 0xf0, 0x41, 0x12, 0x67, 0xd3, 0x53, 0x9d, 0xb6, 0x1f, 0x48, - 0xb5, 0xd2, 0x07, 0x52, 0x57, 0x3f, 0xf6, 0x2e, 0x7e, 0x24, 0xe0, 0xcb, 0xde, 0xd5, 0x2f, 0x7b, - 0xdd, 0x20, 0x5c, 0xad, 0xeb, 0x35, 0xbd, 0x59, 0xd5, 0xd0, 0x9f, 0x67, 0x3f, 0xd9, 0x37, 0xda, - 0x2d, 0xde, 0x68, 0xa5, 0x54, 0xaf, 0xbf, 0xff, 0x52, 0xe9, 0x5f, 0x35, 0x58, 0x63, 0x22, 0x0d, - 0x5e, 0x8a, 0x61, 0x94, 0xca, 0x24, 0xf3, 0xd5, 0x96, 0x50, 0xf7, 0xbf, 0x89, 0x0f, 0x4c, 0xb6, - 0x5d, 0xa6, 0x89, 0x37, 0xe9, 0x74, 0x72, 0x0b, 0xda, 0xf3, 0x33, 0xfb, 0xba, 0x68, 0x59, 0x84, - 0xdc, 0x82, 0xe5, 0x51, 0x9c, 0x25, 0x7e, 0xde, 0xbe, 0xa5, 0xb5, 0xaa, 0x3d, 0xd3, 0x6c, 0x66, - 0xc5, 0xc8, 0x13, 0x20, 0xfb, 0x09, 0x8f, 0xd2, 0x90, 0x2b, 0x67, 0xed, 0xe5, 0xe6, 0xfc, 0x67, - 0x41, 0x49, 0xa6, 0xa2, 0xe7, 0x94, 0xcb, 0xe4, 0x93, 0xf2, 0x7c, 0x7a, 0xcb, 0xe8, 0xf5, 0xa5, - 0xaa, 0xd7, 0xa6, 0xe5, 0xcb, 0x73, 0x7c, 0x77, 0xae, 0x53, 0xbd, 0x06, 0x5e, 0xbc, 0x52, 0x5c, - 0xac, 0xb0, 0x59, 0x55, 0x9a, 0xfe, 0xe4, 0xc0, 0x4a, 0xd9, 0xb3, 0x37, 0xda, 0x0b, 0x79, 0xc1, - 0x6b, 0x67, 0x7f, 0x77, 0xd8, 0x82, 0xd7, 0x4f, 0xfb, 0xd2, 0x5b, 0x2a, 0x7f, 0x8b, 0x64, 0x70, - 0x65, 0x41, 0xba, 0xde, 0xc2, 0xa9, 0x1e, 0xb4, 0x1f, 0xf3, 0x44, 0x06, 0x4a, 0xa5, 0x79, 0x68, - 0x97, 0x58, 0x19, 0xa2, 0x47, 0x70, 0xf5, 0xb5, 0xe6, 0xdb, 0x8e, 0x27, 0x53, 0xd5, 0xe5, 0x6f, - 0xd1, 0x84, 0x6a, 0x51, 0x27, 0x89, 0x69, 0xbf, 0x16, 0xd3, 0x04, 0xfd, 0x0c, 0x2e, 0x8f, 0x84, - 0x2c, 0xb5, 0x9e, 0x9d, 0xa1, 0x1e, 0xb8, 0x7b, 0xe2, 0xc5, 0x82, 0x00, 0x15, 0x8b, 0x7e, 0x09, - 0xde, 0xd3, 0xe9, 0x98, 0x4b, 0x71, 0xae, 0xdb, 0x5b, 0xd0, 0xdc, 0x8f, 0xa7, 0x71, 0x18, 0x3f, - 0x3f, 0x39, 0x63, 0x97, 0x79, 0xb0, 0xac, 0x5f, 0x25, 0xbd, 0x1c, 0x5b, 0xcc, 0x92, 0xf4, 0xa2, - 0x1a, 0x53, 0x9f, 0x87, 0x7e, 0x16, 0x2a, 0x37, 0xd4, 0x47, 0x73, 0xba, 0xd5, 0xfd, 0xed, 0xd5, - 0x86, 0xf3, 0xfb, 0xab, 0x0d, 0xe7, 0x8f, 0x57, 0x1b, 0xce, 0x2f, 0x7f, 0x6e, 0xfc, 0xef, 0xa0, - 0x81, 0xbf, 0x8c, 0x77, 0xfe, 0x0e, 0x00, 0x00, 0xff, 0xff, 0xa0, 0xbb, 0xce, 0xd0, 0x5b, 0x0e, - 0x00, 0x00, + // 1395 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x17, 0xcd, 0x72, 0xdb, 0x44, + 0x18, 0x59, 0x8e, 0x63, 0x7f, 0x8e, 0x53, 0x67, 0xdb, 0xa6, 0x6a, 0x60, 0x82, 0x59, 0x3a, 0xd4, + 0x74, 0x86, 0xd0, 0x69, 0x99, 0xe1, 0xb7, 0x33, 0x6d, 0xe2, 0xb4, 0x98, 0x92, 0xb4, 0x5d, 0xa7, + 0xbd, 0x71, 0xd8, 0xc8, 0x3b, 0x8d, 0x26, 0xb2, 0x64, 0xa4, 0x55, 0xea, 0xf4, 0xc0, 0x15, 0x66, + 0x78, 0x01, 0x8e, 0xbc, 0x07, 0x2f, 0xc0, 0x91, 0x47, 0x60, 0xca, 0x53, 0x70, 0x63, 0xf6, 0xdb, + 0x5d, 0x49, 0x76, 0x1c, 0x52, 0x52, 0x6e, 0xfb, 0xfd, 0xff, 0x7f, 0x9f, 0x04, 0xad, 0x71, 0x12, + 0x1c, 0x71, 0x29, 0x36, 0xc6, 0x49, 0x2c, 0x63, 0x52, 0x0f, 0x22, 0x29, 0x92, 0x88, 0x87, 0x6b, + 0x4b, 0xe3, 0x6c, 0x3f, 0x0c, 0x7c, 0x8d, 0xa7, 0x0f, 0xa0, 0xd1, 0x8f, 0x86, 0x62, 0xb2, 0x23, + 0x24, 0x27, 0x04, 0xaa, 0x0f, 0xc5, 0x71, 0xea, 0xb9, 0x1d, 0xa7, 0x5b, 0x67, 0xf8, 0x26, 0x1f, + 0xc0, 0xf2, 0x5e, 0xc2, 0xfd, 0xc3, 0xed, 0x49, 0x90, 0x4a, 0x11, 0xf9, 0xc2, 0xab, 0x22, 0x75, + 0x06, 0x4b, 0x7f, 0x75, 0x61, 0xe9, 0x7e, 0x20, 0xc2, 0xe1, 0xa3, 0xb1, 0x0c, 0xe2, 0x28, 0x55, + 0xca, 0xf6, 0x8e, 0xc7, 0xc2, 0xab, 0x77, 0x9c, 0x6e, 0x83, 0xe1, 0x9b, 0xbc, 0x03, 0x8d, 0x2d, + 0xee, 0x1f, 0x08, 0x24, 0xb8, 0x48, 0x28, 0x10, 0x39, 0x75, 0x10, 0xbc, 0xd4, 0x56, 0x5a, 0xac, + 0x40, 0x90, 0x0e, 0x34, 0xf7, 0x82, 0x91, 0x78, 0x92, 0xf1, 0x48, 0x66, 0x23, 0x6f, 0x01, 0xa5, + 0xcb, 0x28, 0xb2, 0x0a, 0xb5, 0x47, 0xe1, 0x70, 0x27, 0x88, 0xbc, 0x46, 0xc7, 0xe9, 0xba, 0xcc, + 0x40, 0x16, 0xcf, 0x27, 0x1e, 0x14, 0x78, 0x3e, 0xc9, 0xc3, 0x6d, 0x4e, 0x87, 0xbb, 0x1b, 0x0f, + 0x24, 0x8f, 0x86, 0x3c, 0x19, 0x3e, 0x0b, 0xc4, 0x0b, 0x6f, 0x49, 0x87, 0x3b, 0x8d, 0x55, 0xb2, + 0x9b, 0x3c, 0x15, 0x5e, 0x0b, 0x35, 0xe2, 0x9b, 0xac, 0x41, 0x7d, 0x33, 0x90, 0x3d, 0x31, 0x96, + 0x07, 0xde, 0x72, 0xc7, 0xe9, 0x56, 0x59, 0x0e, 0x93, 0x4b, 0xb0, 0x30, 0xf0, 0x79, 0x28, 0xbc, + 0x0b, 0x28, 0xa0, 0x01, 0x42, 0x61, 0xe9, 0x7e, 0x9c, 0x88, 0xe0, 0x79, 0x84, 0x45, 0xf0, 0xda, + 0x18, 0xd4, 0x14, 0x8e, 0xbc, 0x0f, 0xae, 0x0a, 0x69, 0xa5, 0xe3, 0x74, 0x9b, 0xb7, 0x56, 0x36, + 0x6c, 0x1d, 0x37, 0x7a, 0xc2, 0x0f, 0x46, 0x3c, 0x64, 0x8a, 0x8a, 0x4c, 0x7c, 0xe2, 0x91, 0xd3, + 0x99, 0xf8, 0x84, 0x52, 0x58, 0xee, 0x8f, 0xc6, 0x71, 0x22, 0x99, 0x48, 0xc7, 0x71, 0x94, 0x0a, + 0xd2, 0x06, 0x77, 0x3b, 0x49, 0x3c, 0x07, 0xcd, 0xaa, 0x27, 0xfd, 0x01, 0xda, 0x9b, 0x61, 0xec, + 0x1f, 0xf6, 0xb8, 0xe4, 0x4c, 0x7c, 0x9f, 0x89, 0x54, 0x2a, 0xdf, 0xb5, 0x7b, 0x9a, 0x4f, 0x03, + 0x0a, 0x8b, 0xf5, 0xf6, 0x2a, 0x1a, 0x8b, 0x80, 0xca, 0x0b, 0x66, 0x4d, 0x97, 0x07, 0xdf, 0x18, + 0xfb, 0x01, 0x4f, 0x86, 0x58, 0xd3, 0x2a, 0xd3, 0x80, 0xc2, 0xa2, 0x25, 0xec, 0x83, 0x2a, 0xd3, + 0x00, 0xed, 0xc3, 0x4a, 0xc9, 0xbe, 0x71, 0x73, 0x15, 0x6a, 0x2c, 0x7e, 0xd1, 0xef, 0xa5, 0x9e, + 0xd3, 0x71, 0xbb, 0x55, 0x66, 0x20, 0x6c, 0x98, 0x38, 0xcc, 0x46, 0x91, 0x22, 0x55, 0x90, 0x54, + 0x20, 0xe8, 0x55, 0x58, 0xc0, 0xee, 0x51, 0x51, 0x16, 0xb2, 0xea, 0x49, 0x7f, 0x74, 0xa0, 0xb1, + 0xc3, 0x27, 0xe8, 0x48, 0x4a, 0xee, 0x40, 0xdd, 0xd6, 0x16, 0x99, 0x9a, 0xb7, 0xde, 0x2b, 0x32, + 0x98, 0xb3, 0x6d, 0x58, 0x9e, 0xed, 0x48, 0x26, 0xc7, 0x2c, 0x17, 0x59, 0xfb, 0x12, 0x5a, 0x53, + 0x24, 0x65, 0xef, 0x50, 0x1c, 0xdb, 0xac, 0x1e, 0x8a, 0x63, 0x15, 0xeb, 0x11, 0x0f, 0x33, 0x81, + 0xb9, 0xaa, 0x32, 0x0d, 0x7c, 0x51, 0xf9, 0xcc, 0xa1, 0xcf, 0x80, 0x6c, 0x25, 0x82, 0x4b, 0x81, + 0x46, 0x76, 0x44, 0x9a, 0xf2, 0xe7, 0xe2, 0xac, 0x8c, 0xbb, 0xe5, 0x8c, 0xe7, 0xd9, 0xad, 0x94, + 0xb2, 0x4b, 0x6f, 0x00, 0xe9, 0x89, 0x50, 0x48, 0x61, 0xa6, 0xfb, 0x5f, 0xf4, 0xd2, 0x81, 0xf5, + 0xe1, 0x6c, 0x5e, 0x72, 0x1d, 0xaa, 0x6a, 0x55, 0xa0, 0xb1, 0xe6, 0xad, 0x8b, 0x45, 0x9e, 0xf2, + 0x2d, 0xc2, 0x90, 0x81, 0x86, 0x56, 0x29, 0x7a, 0xf9, 0x9a, 0x81, 0x4d, 0xb5, 0xd2, 0x0d, 0x63, + 0xca, 0x45, 0x53, 0xab, 0x85, 0xa9, 0xf2, 0x9a, 0x31, 0xd6, 0xee, 0xda, 0x70, 0xcf, 0x6b, 0x8d, + 0xfa, 0xf0, 0xb6, 0xd6, 0x70, 0xef, 0x88, 0x07, 0x21, 0xdf, 0x0f, 0xff, 0x53, 0x45, 0xa6, 0x1c, + 0xf7, 0x60, 0x11, 0x65, 0xfb, 0x3d, 0xd3, 0xdb, 0x16, 0xa4, 0xdf, 0x41, 0x31, 0x26, 0xbb, 0x7c, + 0x24, 0x8c, 0x36, 0x7c, 0xe7, 0xf1, 0x56, 0xce, 0x8e, 0x57, 0x19, 0x56, 0xa3, 0xa5, 0x56, 0xb5, + 0xab, 0x0c, 0x23, 0x40, 0x6f, 0x43, 0x6d, 0xe0, 0x1f, 0x88, 0x11, 0x27, 0x1f, 0xc2, 0x22, 0x7a, + 0x28, 0x52, 0xd3, 0xd1, 0x17, 0x66, 0x2a, 0xc5, 0x2c, 0x9d, 0xa6, 0x26, 0xb2, 0xb9, 0x3e, 0x7d, + 0x04, 0x8b, 0xc6, 0x30, 0x4e, 0xf4, 0x29, 0x15, 0xb7, 0x3c, 0xe4, 0x3a, 0xd4, 0xd0, 0xd9, 0xd4, + 0xab, 0xce, 0x5a, 0x45, 0x3c, 0x33, 0x64, 0xba, 0x0d, 0xee, 0x53, 0xd6, 0x57, 0x83, 0x8d, 0x0e, + 0x5b, 0xa3, 0x06, 0x52, 0xae, 0x7c, 0x1d, 0xa7, 0xd2, 0xa4, 0x15, 0xdf, 0x0a, 0xf7, 0x38, 0x4e, + 0x24, 0xa6, 0xb4, 0xc5, 0xf0, 0x4d, 0x53, 0xa8, 0xee, 0xc6, 0x43, 0x41, 0x96, 0xa1, 0xd2, 0xef, + 0x19, 0x1d, 0x95, 0x7e, 0x8f, 0xbc, 0x8b, 0xea, 0x4d, 0x26, 0x5b, 0x85, 0x13, 0x4f, 0x59, 0x9f, + 0xa1, 0xe1, 0x6b, 0xd0, 0xea, 0xa7, 0x5b, 0x71, 0x9c, 0x0c, 0x83, 0x88, 0xcb, 0x38, 0x31, 0x27, + 0x6f, 0x1a, 0x89, 0xa3, 0x25, 0xb9, 0xd4, 0xc7, 0xa8, 0xc1, 0x34, 0x40, 0xef, 0x42, 0x5b, 0x19, + 0x45, 0xc0, 0xb6, 0xc7, 0x2a, 0xd4, 0x14, 0x2e, 0x77, 0xc2, 0x40, 0x85, 0x86, 0x4a, 0x59, 0xc3, + 0xb7, 0x5a, 0xc3, 0xf6, 0x91, 0x88, 0x64, 0xa9, 0xc1, 0x10, 0x46, 0x05, 0x2d, 0xa6, 0x01, 0x42, + 0x75, 0x80, 0x26, 0x92, 0xe5, 0x22, 0x12, 0x85, 0x65, 0x48, 0xa3, 0x3f, 0x3b, 0x00, 0xd6, 0xa1, + 0x2c, 0xcd, 0x45, 0x9c, 0xd3, 0x45, 0x48, 0xd7, 0x36, 0x8a, 0x19, 0xae, 0x76, 0xc1, 0xa5, 0xf1, + 0xcc, 0x36, 0xd2, 0xc7, 0x45, 0x23, 0xe9, 0x92, 0x5e, 0x9e, 0x69, 0x00, 0x6d, 0xb5, 0x68, 0xa7, + 0xc7, 0xd0, 0x2c, 0xe1, 0x4f, 0x69, 0x2a, 0xdb, 0x25, 0x95, 0x59, 0x95, 0x88, 0x37, 0x2a, 0x6d, + 0xaf, 0x3c, 0x84, 0x66, 0x09, 0x3d, 0x57, 0x63, 0x17, 0x2e, 0x4c, 0x8f, 0xad, 0x3d, 0x07, 0xb3, + 0x68, 0x1a, 0x40, 0x6b, 0x2b, 0xcc, 0x52, 0x29, 0x12, 0xa3, 0x4e, 0xdd, 0x10, 0x8d, 0xc8, 0x8b, + 0x57, 0x20, 0xe6, 0xd7, 0x8f, 0x5c, 0x83, 0x05, 0x95, 0x46, 0x3d, 0x7d, 0x27, 0x73, 0xac, 0x89, + 0xf4, 0x19, 0xd4, 0x37, 0x07, 0xfd, 0x07, 0x49, 0x9c, 0x8d, 0xe7, 0x3a, 0x6d, 0x3f, 0x90, 0x2a, + 0xa5, 0x0f, 0xa4, 0xb6, 0x3e, 0xf6, 0x2e, 0x7e, 0x24, 0xe0, 0x65, 0x6f, 0xeb, 0xcb, 0x5e, 0x35, + 0x18, 0xae, 0xd6, 0xf5, 0x8a, 0xde, 0xac, 0x6a, 0xe8, 0xcf, 0xb3, 0x9f, 0xec, 0x8d, 0x76, 0x8b, + 0x1b, 0xad, 0x94, 0xea, 0xf5, 0xf7, 0x7f, 0x2a, 0xfd, 0xbb, 0x02, 0x2b, 0x4c, 0xa4, 0xc1, 0x4b, + 0xd1, 0x8f, 0x52, 0x99, 0x64, 0xbe, 0xda, 0x12, 0x4a, 0xfe, 0x9b, 0x78, 0xdf, 0x64, 0xdb, 0x65, + 0x1a, 0x78, 0x9d, 0x4e, 0x27, 0x37, 0xa1, 0x39, 0x3b, 0xb3, 0x27, 0x59, 0xcb, 0x2c, 0xe4, 0x26, + 0x2c, 0x0e, 0xe2, 0x2c, 0xf1, 0xf3, 0xf6, 0x2d, 0xad, 0x55, 0xed, 0x99, 0x26, 0x33, 0xcb, 0x46, + 0x9e, 0x00, 0xd9, 0x4b, 0x78, 0x94, 0x86, 0x5c, 0x39, 0x6b, 0x85, 0xeb, 0xb3, 0x9f, 0x05, 0x25, + 0x9e, 0x29, 0x3d, 0x73, 0x84, 0xc9, 0x27, 0xe5, 0xf9, 0xf4, 0x16, 0xd1, 0xeb, 0x4b, 0xd3, 0x5e, + 0x9b, 0x96, 0x2f, 0xcf, 0xf1, 0x9d, 0x99, 0x4e, 0xf5, 0x6a, 0x28, 0x78, 0xa5, 0x10, 0x9c, 0x22, + 0xb3, 0x69, 0x6e, 0xfa, 0x93, 0x03, 0x4b, 0x65, 0xcf, 0x5e, 0x6b, 0x2f, 0xe4, 0x05, 0xaf, 0x9c, + 0xfd, 0xdd, 0x61, 0x0b, 0x5e, 0x9d, 0xf7, 0xa5, 0xb7, 0x50, 0xfe, 0x16, 0xc9, 0xe0, 0xca, 0x29, + 0xe9, 0x7a, 0x03, 0xa7, 0x3a, 0xd0, 0x7c, 0xcc, 0x13, 0x19, 0x28, 0x95, 0xe6, 0xd0, 0x2e, 0xb0, + 0x32, 0x8a, 0x1e, 0xc2, 0xd5, 0x13, 0xcd, 0xb7, 0x15, 0x8f, 0xc6, 0xaa, 0xcb, 0xdf, 0xa0, 0x09, + 0xd5, 0xa2, 0x4e, 0x12, 0xd3, 0x7e, 0x0d, 0xa6, 0x01, 0xfa, 0x39, 0x5c, 0x1e, 0x08, 0x59, 0x6a, + 0x3d, 0x3b, 0x43, 0x1d, 0x70, 0x77, 0xc5, 0x8b, 0x53, 0x02, 0x54, 0x24, 0xfa, 0x15, 0x78, 0x4f, + 0xc7, 0x43, 0x2e, 0xc5, 0xb9, 0xa4, 0x37, 0xa1, 0xbe, 0x17, 0x8f, 0xe3, 0x30, 0x7e, 0x7e, 0x7c, + 0xc6, 0x2e, 0xf3, 0x60, 0x51, 0x5f, 0x25, 0xbd, 0x1c, 0x1b, 0xcc, 0x82, 0xf4, 0xa2, 0x1a, 0x53, + 0x9f, 0x87, 0x7e, 0x16, 0x2a, 0x37, 0xd4, 0x47, 0x73, 0x4a, 0x85, 0x19, 0x04, 0x8e, 0x89, 0x2b, + 0x1d, 0xba, 0x7b, 0x88, 0xb0, 0x87, 0x4e, 0x43, 0xe4, 0x53, 0x68, 0x96, 0xb8, 0x4d, 0x02, 0x2f, + 0xcf, 0xcc, 0x8b, 0x26, 0xb2, 0x32, 0x27, 0xfd, 0xcd, 0x99, 0x92, 0x3c, 0x71, 0xca, 0x8d, 0xc1, + 0x23, 0x5d, 0x94, 0x3a, 0x33, 0x90, 0x8a, 0x75, 0x7b, 0xe2, 0x87, 0x59, 0xaa, 0x48, 0xfa, 0x7a, + 0x17, 0x08, 0x15, 0xab, 0xfa, 0x33, 0x8c, 0x33, 0x69, 0x36, 0xa7, 0x05, 0xd5, 0x4f, 0x5a, 0x4f, + 0xf0, 0x61, 0x18, 0x44, 0x02, 0xbb, 0xd4, 0x65, 0x39, 0x4c, 0x6e, 0xea, 0x6d, 0x6f, 0x47, 0x6d, + 0x6d, 0xae, 0xfb, 0xc8, 0xa1, 0x2f, 0x41, 0x4a, 0x09, 0xb4, 0x67, 0x49, 0x9b, 0xed, 0xdf, 0x5f, + 0xad, 0x3b, 0x7f, 0xbc, 0x5a, 0x77, 0xfe, 0x7c, 0xb5, 0xee, 0xfc, 0xf2, 0xd7, 0xfa, 0x5b, 0xfb, + 0x35, 0xfc, 0xd7, 0xbe, 0xfd, 0x4f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x2f, 0x76, 0xf5, 0x59, 0x94, + 0x0f, 0x00, 0x00, } func (m *IndexMeta) Marshal() (dAtA []byte, err error) { @@ -4146,6 +4336,155 @@ func (m *RecalculateCaches) MarshalToSizedBuffer(dAtA []byte) (int, error) { return len(dAtA) - i, nil } +func (m *TransactionMessage) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *TransactionMessage) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *TransactionMessage) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.XXX_unrecognized != nil { + i -= len(m.XXX_unrecognized) + copy(dAtA[i:], m.XXX_unrecognized) + } + if m.Transaction != nil { + { + size, err := m.Transaction.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintPrivate(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x12 + } + if len(m.Action) > 0 { + i -= len(m.Action) + copy(dAtA[i:], m.Action) + i = encodeVarintPrivate(dAtA, i, uint64(len(m.Action))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *Transaction) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *Transaction) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *Transaction) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.XXX_unrecognized != nil { + i -= len(m.XXX_unrecognized) + copy(dAtA[i:], m.XXX_unrecognized) + } + if m.Stats != nil { + { + size, err := m.Stats.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintPrivate(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x32 + } + if m.Deadline != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.Deadline)) + i-- + dAtA[i] = 0x28 + } + if m.Timeout != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.Timeout)) + i-- + dAtA[i] = 0x20 + } + if m.Exclusive { + i-- + if m.Exclusive { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x18 + } + if m.Active { + i-- + if m.Active { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x10 + } + if len(m.ID) > 0 { + i -= len(m.ID) + copy(dAtA[i:], m.ID) + i = encodeVarintPrivate(dAtA, i, uint64(len(m.ID))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *TransactionStats) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *TransactionStats) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *TransactionStats) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.XXX_unrecognized != nil { + i -= len(m.XXX_unrecognized) + copy(dAtA[i:], m.XXX_unrecognized) + } + return len(dAtA) - i, nil +} + func encodeVarintPrivate(dAtA []byte, offset int, v uint64) int { offset -= sovPrivate(v) base := offset @@ -4988,6 +5327,70 @@ func (m *RecalculateCaches) Size() (n int) { return n } +func (m *TransactionMessage) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.Action) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Transaction != nil { + l = m.Transaction.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + if m.XXX_unrecognized != nil { + n += len(m.XXX_unrecognized) + } + return n +} + +func (m *Transaction) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.ID) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Active { + n += 2 + } + if m.Exclusive { + n += 2 + } + if m.Timeout != 0 { + n += 1 + sovPrivate(uint64(m.Timeout)) + } + if m.Deadline != 0 { + n += 1 + sovPrivate(uint64(m.Deadline)) + } + if m.Stats != nil { + l = m.Stats.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + if m.XXX_unrecognized != nil { + n += len(m.XXX_unrecognized) + } + return n +} + +func (m *TransactionStats) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if m.XXX_unrecognized != nil { + n += len(m.XXX_unrecognized) + } + return n +} + func sovPrivate(x uint64) (n int) { return (math_bits.Len64(x|1) + 6) / 7 } @@ -10137,6 +10540,382 @@ func (m *RecalculateCaches) Unmarshal(dAtA []byte) error { } return nil } +func (m *TransactionMessage) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: TransactionMessage: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: TransactionMessage: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Action", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Action = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Transaction", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Transaction == nil { + m.Transaction = &Transaction{} + } + if err := m.Transaction.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipPrivate(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *Transaction) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: Transaction: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Transaction: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ID", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.ID = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Active", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.Active = bool(v != 0) + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Exclusive", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.Exclusive = bool(v != 0) + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Timeout", wireType) + } + m.Timeout = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Timeout |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 5: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Deadline", wireType) + } + m.Deadline = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Deadline |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 6: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Stats", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Stats == nil { + m.Stats = &TransactionStats{} + } + if err := m.Stats.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipPrivate(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *TransactionStats) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: TransactionStats: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: TransactionStats: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + default: + iNdEx = preIndex + skippy, err := skipPrivate(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} func skipPrivate(dAtA []byte) (n int, err error) { l := len(dAtA) iNdEx := 0 diff --git a/internal/private.proto b/internal/private.proto index 989428756..804c3f2cd 100644 --- a/internal/private.proto +++ b/internal/private.proto @@ -207,3 +207,19 @@ message Topology { } message RecalculateCaches {} + +message TransactionMessage { + string Action = 1; + Transaction Transaction = 2; +} + +message Transaction { + string ID = 1; + bool Active = 2; + bool Exclusive = 3; + int64 Timeout = 4; + int64 Deadline = 5; + TransactionStats Stats = 6; +} + +message TransactionStats {} \ No newline at end of file diff --git a/server.go b/server.go index 936e08756..cbf580c14 100644 --- a/server.go +++ b/server.go @@ -28,6 +28,8 @@ import ( "time" "github.com/molecula/ext" + uuid "github.com/satori/go.uuid" + // extensions pulls in some extensions depending on build tags _ "github.com/pilosa/pilosa/v2/extensions" "github.com/pilosa/pilosa/v2/logger" @@ -794,11 +796,42 @@ func (s *Server) receiveMessage(m Message) error { } case *NodeStatus: s.handleRemoteStatus(obj) + case *TransactionMessage: + err := s.handleTransactionMessage(obj) + if err != nil { + return errors.Wrapf(err, "handling transaction message: %v", obj) + } } return nil } +func (s *Server) handleTransactionMessage(tm *TransactionMessage) error { + mtrns := tm.Transaction // message transaction + switch tm.Action { + case TRANSACTION_START: + _, err := s.StartTransaction(mtrns.ID, mtrns.Timeout, mtrns.Exclusive, true) + if err != nil { + return errors.Wrap(err, "starting transaction locally") + } + case TRANSACTION_FINISH: + _, err := s.FinishTransaction(mtrns.ID, true) + if err != nil { + return errors.Wrap(err, "finishing transaction locally") + } + case TRANSACTION_VALIDATE: + trns, err := s.GetTransaction(mtrns.ID, true) + if err != nil { + return errors.Wrap(err, "getting local transaction to validate") + } + err = CompareTransactions(mtrns, trns) + return errors.Wrap(err, "comparing transactions") + default: + return errors.Errorf("unknown transaction action: '%s'", tm.Action) + } + return nil +} + // SendSync represents an implementation of Broadcaster. func (s *Server) SendSync(m Message) error { var eg errgroup.Group @@ -991,6 +1024,125 @@ func (s *Server) monitorRuntime() { } } +func (srv *Server) StartTransaction(id string, timeout time.Duration, exclusive bool, remote bool) (Transaction, error) { + node := srv.node() + if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 { + return Transaction{}, ErrNodeNotCoordinator + } + if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) { + return Transaction{}, errors.New("got a remote start call to coordinator or single node cluster... shouldn't ever happen") + } + // empty string id should generate an id + + if !remote { // we are the coordinator, + if id == "" { + id = uuid.NewV4().String() + } + trns, err := srv.holder.StartTransaction(id, timeout, exclusive) + if err != nil { + return trns, errors.Wrap(err, "starting transaction") + } + err = srv.SendSync( + &TransactionMessage{ + Action: TRANSACTION_START, + Transaction: trns, + }) + if err != nil { + // try to clean up, but ignore errors + srv.holder.FinishTransaction(id) + srv.SendSync( + &TransactionMessage{ + Action: TRANSACTION_FINISH, + Transaction: trns, + }, + ) + return trns, errors.Wrap(err, "broadcasting transaction start") + } + return trns, nil + } else { // remote + return srv.holder.StartTransaction(id, timeout, exclusive) + } + +} + +func (srv *Server) FinishTransaction(id string, remote bool) (Transaction, error) { + node := srv.node() + if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 { + return Transaction{}, ErrNodeNotCoordinator + } + if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) { + return Transaction{}, errors.New("got a remote finish call to coordinator or single node cluster... shouldn't ever happen") + } + + if !remote { + trns, err := srv.holder.FinishTransaction(id) + if err != nil { + return trns, errors.Wrap(err, "finishing transaction") + } + err = srv.SendSync( + &TransactionMessage{ + Action: TRANSACTION_FINISH, + Transaction: trns, + }, + ) + if err != nil { + srv.logger.Printf("error broadcasting transaction finish: %v", err) + // TODO retry? + } + return trns, nil + } else { // remote + return srv.holder.FinishTransaction(id) + } + +} + +func (srv *Server) Transactions() (map[string]Transaction, error) { + node := srv.node() + if !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 { + return nil, ErrNodeNotCoordinator + } + + return srv.holder.Transactions() +} + +func (srv *Server) GetTransaction(id string, remote bool) (Transaction, error) { + node := srv.node() + if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 { + return Transaction{}, ErrNodeNotCoordinator + } + + if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) { + return Transaction{}, errors.New("got a remote finish call to coordinator or single node cluster... shouldn't ever happen") + } + + trns, err := srv.holder.GetTransaction(id) + if err != nil { + return Transaction{}, errors.Wrap(err, "getting transaction") + } + + // The way a client would find out that the exclusive transaction + // it requested is active is by polling the GetTransaction + // endpoint. Therefore, returning an active, exclusive + // transaction, from here is what truly makes the transaction + // "live". Before doing so, we want to make sure all nodes + // agree. (in case other nodes have activity on this transaction + // we're not aware of) + if !remote && trns.Exclusive && trns.Active { + err := srv.SendSync( + &TransactionMessage{ + Action: TRANSACTION_VALIDATE, + Transaction: trns, + }, + ) + if err != nil { + return Transaction{}, errors.Wrap(err, "contacting remote hosts") + } + return trns, nil + } else { // remote + return trns, nil + } +} + // countOpenFiles on operating systems that support lsof. func countOpenFiles() (int, error) { switch runtime.GOOS { diff --git a/server/server.go b/server/server.go index 87fc1c361..478e5cfe9 100644 --- a/server/server.go +++ b/server/server.go @@ -87,6 +87,7 @@ type Command struct { listenURI *pilosa.URI tlsConfig *tls.Config closeTimeout time.Duration + noSleep bool serverOptions []pilosa.ServerOption } @@ -114,6 +115,17 @@ func OptCommandConfig(config *Config) CommandOption { } } +// OptCommandNoSleep disables the 5 second sleep for non-coordinator +// nodes on startup. See https://github.com/molecula/pilosa/issues/266 +// This option should only be used by tests, and expect it to be +// deprecated. +func OptCommandNoSleep() CommandOption { + return func(c *Command) error { + c.noSleep = true + return nil + } +} + // NewCommand returns a new instance of Main. func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption) *Command { c := &Command{ @@ -149,7 +161,7 @@ func (m *Command) Start() (err error) { if !m.API.Node().IsCoordinator { // hack to give coordinator a head start // TODO https://github.com/molecula/pilosa/issues/266 - if len(m.Config.Gossip.Seeds) > 0 { + if len(m.Config.Gossip.Seeds) > 0 && !m.noSleep { time.Sleep(5 * time.Second) } } diff --git a/server/server_test.go b/server/server_test.go index 7809ee466..a5b3837f4 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -374,6 +374,115 @@ func TestConcurrentFieldCreation(t *testing.T) { } } +func TestTransactionsAPI(t *testing.T) { + cluster := test.MustRunCluster(t, 3) + defer cluster.Close() + + api0 := cluster[0].API + api1 := cluster[1].API + //api2 := cluster[2].API + + // can fetch empty transactions + if trnsMap, err := api0.Transactions(); err != nil { + t.Fatalf("getting transactions: %v", err) + } else if len(trnsMap) != 0 { + t.Fatalf("unexpectedly has transactions: %v", trnsMap) + } + + // can't fetch transactions from non-coordinator + if _, err := api1.Transactions(); err != pilosa.ErrNodeNotCoordinator { + t.Errorf("api1 should return ErrNodeNotCoordinator when asked for transactions but got: %v", err) + } + + // can start transaction + if trns, err := api0.StartTransaction("a", time.Minute, false, false); err != nil { + t.Errorf("couldn't start transaction: %v", err) + } else { + test.CompareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // can retrieve transaction from other nodes with remote=true + if trns, err := api1.GetTransaction("a", true); err != nil { + t.Errorf("couldn't fetch transaction from other node with remote=true: %v", err) + } else { + test.CompareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // can start transaction with blank id and get uuid back + id := "" + if trns, err := api0.StartTransaction(id, time.Minute, false, false); err != nil { + t.Errorf("couldn't start transaction: %v", err) + } else { + id = trns.ID + if len(id) != 36 { // UUID + t.Errorf("unexpected generated ID: %s", id) + } + test.CompareTransactions(t, pilosa.Transaction{ID: id, Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // can't finish transaction on non-coordinator + if _, err := api1.FinishTransaction(id, false); err != pilosa.ErrNodeNotCoordinator { + t.Errorf("unexpected error is not ErrNodeNotCoordinator: %v", err) + } + + // can finish transaction + if _, err := api0.FinishTransaction(id, false); err != nil { + t.Errorf("couldn't finish transaction: %v", err) + } + + // can finish previous transaction + if _, err := api0.FinishTransaction("a", false); err != nil { + t.Errorf("couldn't finish transaction a: %v", err) + } + + // can start exclusive transaction + if te, err := api0.StartTransaction("exc", time.Minute, true, false); err != nil { + t.Errorf("couldn't start exclusive transaction: %v", err) + } else if !te.Active { + t.Errorf("expected exclusive transaction to be active: %+v", te) + } + + // can finish exclusive transaction + if _, err := api0.FinishTransaction("exc", false); err != nil { + t.Errorf("couldn't finish exclusive transaction: %v", err) + } + + // can start transaction (with same name as previous finished transaction) + if trns, err := api0.StartTransaction("a", time.Minute, false, false); err != nil { + t.Errorf("couldn't start transaction: %v", err) + } else { + test.CompareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // can start exclusive transaction and is not immediately active + if te, err := api0.StartTransaction("exc", time.Minute, true, false); err != nil { + t.Errorf("couldn't start exclusive transaction: %v", err) + } else if te.Active { + t.Errorf("expected exclusive transaction to be inactive: %+v", te) + } + + // can finish non-exclusive transaction + if _, err := api0.FinishTransaction("a", false); err != nil { + t.Errorf("couldn't finish transaction a: %v", err) + } + + // can poll exclusive transaction and is active + if trns, err := api0.GetTransaction("exc", false); err != nil { + t.Errorf("couldn't poll exclusive transaction: %v", err) + } else { + test.CompareTransactions(t, pilosa.Transaction{ID: "exc", Active: true, Exclusive: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // transaction is active on other nodes with remote=true + if trns, err := api1.GetTransaction("exc", true); err != nil { + t.Errorf("couldn't poll exclusive transaction: %v", err) + } else { + test.CompareTransactions(t, pilosa.Transaction{ID: "exc", Active: true, Exclusive: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // LATER, test deadline extension on non-coordinator blocks active, exclusive transaction being returned +} + func TestMain_RecalculateHashes(t *testing.T) { const clusterSize = 5 cluster := test.MustRunCluster(t, clusterSize) diff --git a/test/pilosa.go b/test/pilosa.go index 8179c04e0..6903a2010 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -65,6 +65,7 @@ func newCommand(opts ...server.CommandOption) *Command { // does not fail on 32-bit systems. opts = append([]server.CommandOption{ server.OptCommandCloseTimeout(time.Millisecond * 2), + server.OptCommandNoSleep(), }, opts...) m := &Command{commandOptions: opts} m.Command = server.NewCommand(bytes.NewReader(nil), ioutil.Discard, ioutil.Discard, opts...) @@ -409,6 +410,7 @@ func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) { if err != nil { return nil, errors.Wrap(err, "new cluster") } + if err = cluster.Start(); err != nil { return nil, errors.Wrap(err, "starting cluster") } diff --git a/test/transaction.go b/test/transaction.go new file mode 100644 index 000000000..3547ad856 --- /dev/null +++ b/test/transaction.go @@ -0,0 +1,36 @@ +package test + +import ( + "testing" + "time" + + "github.com/pilosa/pilosa/v2" +) + +// CompareTransactions errors describing how the +// transactions differ (if at all). The deadlines need only be close +// (within 3ms). +func CompareTransactions(t *testing.T, trns1, trns2 pilosa.Transaction) { + t.Helper() + if trns1.ID != trns2.ID { + t.Errorf("IDs differ:\n%+v\n%+v", trns1, trns2) + } + if trns1.Active != trns2.Active { + t.Errorf("Actives differ:\n%+v\n%+v", trns1, trns2) + } + if trns1.Exclusive != trns2.Exclusive { + t.Errorf("Exclusives differ:\n%+v\n%+v", trns1, trns2) + } + if trns1.Timeout != trns2.Timeout { + t.Errorf("Timeouts differ:\n%+v\n%+v", trns1, trns2) + } + + diff := trns1.Deadline.Sub(trns2.Deadline) + + if diff > time.Millisecond*3 || diff < time.Millisecond*-3 { + t.Errorf("Deadlines differ by %v:\n%+v\n%+v", diff, trns1, trns2) + } + if trns1.Stats != trns2.Stats { + t.Errorf("Stats differ:\n%+v\n%+v", trns1, trns2) + } +} diff --git a/transaction.go b/transaction.go index 1cd6ccea9..ccfad9ed9 100644 --- a/transaction.go +++ b/transaction.go @@ -140,9 +140,8 @@ func (tm *TransactionManager) finish(id string) (Transaction, error) { // After removing, check to see if we need to activate an exclusive transaction trnsMap, err := tm.store.List() if err != nil { - // returning an error here is weird because we've already - // removed the transaction - return trns, errors.Wrap(err, "listing transactions in Finish") + tm.log().Printf("error listing transactions in Finish: %v", err) + return trns, nil } if len(trnsMap) == 1 { @@ -154,7 +153,8 @@ func (tm *TransactionManager) finish(id string) (Transaction, error) { etrans.Active = true etrans.Deadline = time.Now().Add(etrans.Timeout) if err := tm.store.Put(etrans); err != nil { - return trns, errors.Wrap(err, "activating exclusive transaction after finishing last transaction") + tm.log().Printf("activating exclusive transaction after finishing last transaction: %v", err) + return trns, nil } } } @@ -357,3 +357,20 @@ const ErrTransactionNotFound = Error("transaction not found") const ErrTransactionExclusive = Error("there is already an exclusive transaction") const ErrTransactionExists = Error("transaction with the given id already exists") const ErrTransactionInactive = Error("cannot finish an inactive transaction") + +func CompareTransactions(t1, t2 Transaction) error { + if t1.ID != t2.ID { + return errors.Errorf("transaction IDs not equal: %+v %+v", t1, t2) + } + if t1.Active != t2.Active { + return errors.Errorf("transaction Actives not equal: %+v %+v", t1, t2) + } + if t1.Exclusive != t2.Exclusive { + return errors.Errorf("transaction Exclusives not equal: %+v %+v", t1, t2) + } + if t1.Timeout != t2.Timeout { + return errors.Errorf("transaction Timeouts not equal: %+v %+v", t1, t2) + } + // don't care about Deadline or Stats + return nil +} diff --git a/transaction.md b/transaction.md index 8190a0761..d938b4a00 100644 --- a/transaction.md +++ b/transaction.md @@ -4,7 +4,7 @@ This is not full-featured transaction support with commit and rollback for now; this is a placeholder intended to allow us to solve shorter-term problems. -The primaryw purpose of this is to allow an exclusive transaction to +The primary purpose of this is to allow an exclusive transaction to block new ingest activity from starting, while permitting existing ingest operations to complete, even if a single ingest requires multiple operations. This allows users with cooperating ingest operations to ensure a stable state @@ -109,3 +109,64 @@ If multiple exclusive transactions are requested, they become active sequentially in the order the requests came in, and the snapshot queue and other transactions are not permitted to resume until the exclusive transactions all complete. + + + +### Implementation Notes + +All requests go through coordinator. + +When creating a new transaction, we'll create it on every node in the +cluster and persist it to disk. + +Only the coordinator will accept requests to start a transaction. + +Timeouts only expire when there has been *no activity* on a transaction for the timeout duration. +Any activity on the transaction may extend the deadline (unimplemented). + +When finishing a transaction, we'll finish it on the coordinator and +then broadcast the finish to the cluster before returning to the +client. + +When getting an exclusive transaction, if the transaction is active, +we'll make sure that all nodes agree before returning it. + + +Coordinator forwards all requests to every other node so they can stay +in sync. If the coordinator doesn't hear back from a node, the request +fails. The coordinator only reaches out to active nodes, so if the +cluster is in DEGRADED, things can still continue. + +If an node is down and comes back up it needs to synchronize its state +with the coordinator (unimplemented). + +There is a separate TransactionManager and TransactionStore + +The store is just responsible for persisting info about +transactions. The manager handles all the logic (at the node level). +Logic related to cluster and remote vs local node is handled by the +Server. The Holder contains the TransactionManager, and the Server +contains the logic for how to handle external vs intra cluster +requests (remote==true). + +There is intra-cluster messaging for transactions which is handled +with the new TransactionMessage and goes through the usual +SendMessage/Broadcaster stuff. + +There is also external API which is handled by the HTTP handler and +goes through API (and is passed directly to Server). (unimplemented) + + +#### TODO + +- [x] implement api layer and cluster logic, startup, etc. +- [ ] implement HTTP layer including header/transaction ID +- [ ] implement and use persistent transaction store rather than inmem. +- [ ] update go-pilosa/gpexp to actually USE transactions + - [ ] update IDK to use updated go-pilosa + +#### Testing TransactionManager +- there should never be more than one Exclusive transaction +- if the Exclusive transaction is active, there should be no other transactions + + diff --git a/transaction_test.go b/transaction_test.go index aa64ba446..6621dcda3 100644 --- a/transaction_test.go +++ b/transaction_test.go @@ -20,34 +20,34 @@ func TestTransactionManager(t *testing.T) { // can add a non-exclusive transaction trns1 := mustStart(t, tm, "a", time.Microsecond, false) - compareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns1) + test.CompareTransactions(t, pilosa.Transaction{ID: "a", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns1) // can have two non exclusive transactions trns2 := mustStart(t, tm, "b", time.Microsecond, false) - compareTransactions(t, pilosa.Transaction{ID: "b", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns2) + test.CompareTransactions(t, pilosa.Transaction{ID: "b", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns2) // trying to start a transaction with same name errors and returns previous transaction t3, err := tm.Start("a", time.Second, true) if err != pilosa.ErrTransactionExists { t.Errorf("expected transaction exists, but got: '%v'", err) } - compareTransactions(t, trns1, t3) + test.CompareTransactions(t, trns1, t3) // can get an existing transaction trns2_2 := mustGet(t, tm, "b") - compareTransactions(t, trns2, trns2_2) + test.CompareTransactions(t, trns2, trns2_2) // can list all transactions trnsMap := mustList(t, tm) if len(trnsMap) != 2 { t.Errorf("unexpected number of transactions in map: %d", len(trnsMap)) } - compareTransactions(t, trnsMap["a"], trns1) - compareTransactions(t, trnsMap["b"], trns2) + test.CompareTransactions(t, trnsMap["a"], trns1) + test.CompareTransactions(t, trnsMap["b"], trns2) // can submit an exclusive transaction trnsE := mustStart(t, tm, "ce", time.Millisecond*5, true) - compareTransactions(t, pilosa.Transaction{ID: "ce", Active: false, Exclusive: true, Timeout: time.Millisecond * 5, Deadline: time.Now().Add(time.Millisecond * 5)}, trnsE) + test.CompareTransactions(t, pilosa.Transaction{ID: "ce", Active: false, Exclusive: true, Timeout: time.Millisecond * 5, Deadline: time.Now().Add(time.Millisecond * 5)}, trnsE) // can't start new transactions while an exclusive transaction is pending if _, err := tm.Start("d", time.Millisecond, false); err != pilosa.ErrTransactionExclusive { @@ -101,7 +101,7 @@ func TestTransactionManager(t *testing.T) { // can start a new exclusive transaction and it's immediately active trnsHE := mustStart(t, tm, "he", time.Hour, true) - compareTransactions(t, pilosa.Transaction{ID: "he", Active: true, Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsHE) + test.CompareTransactions(t, pilosa.Transaction{ID: "he", Active: true, Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsHE) // can't start new transactions while an exclusive transaction is active if _, err := tm.Start("i", time.Millisecond, false); err != pilosa.ErrTransactionExclusive { @@ -110,40 +110,40 @@ func TestTransactionManager(t *testing.T) { // can finish an active exclusive transaction trnsHE_finish := mustFinish(t, tm, "he") - compareTransactions(t, trnsHE, trnsHE_finish) + test.CompareTransactions(t, trnsHE, trnsHE_finish) // can start normal transaction after finishing exclusive transaction trnsJ := mustStart(t, tm, "j", time.Hour, false) - compareTransactions(t, pilosa.Transaction{ID: "j", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsJ) + test.CompareTransactions(t, pilosa.Transaction{ID: "j", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsJ) // can finish normal transaction trnsJ_finish := mustFinish(t, tm, "j") - compareTransactions(t, trnsJ, trnsJ_finish) + test.CompareTransactions(t, trnsJ, trnsJ_finish) // can start normal transaction after finishing normal transaction trnsK := mustStart(t, tm, "k", time.Hour, false) - compareTransactions(t, pilosa.Transaction{ID: "k", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsK) + test.CompareTransactions(t, pilosa.Transaction{ID: "k", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsK) // can start new exclusive transaction, but not immediately active trnsLE := mustStart(t, tm, "le", time.Hour, true) - compareTransactions(t, pilosa.Transaction{ID: "le", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsLE) + test.CompareTransactions(t, pilosa.Transaction{ID: "le", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsLE) // finishing k should activate le trnsK_finish := mustFinish(t, tm, "k") - compareTransactions(t, trnsK, trnsK_finish) + test.CompareTransactions(t, trnsK, trnsK_finish) trnsLE_active := mustGet(t, tm, "le") trnsLE.Active = true - compareTransactions(t, trnsLE, trnsLE_active) + test.CompareTransactions(t, trnsLE, trnsLE_active) mustFinish(t, tm, "le") // can start normal transaction to test deadline reset trnsM := mustStart(t, tm, "m", time.Millisecond*4, false) - compareTransactions(t, pilosa.Transaction{ID: "m", Active: true, Timeout: time.Millisecond * 4, Deadline: time.Now().Add(time.Millisecond * 4)}, trnsM) + test.CompareTransactions(t, pilosa.Transaction{ID: "m", Active: true, Timeout: time.Millisecond * 4, Deadline: time.Now().Add(time.Millisecond * 4)}, trnsM) // start new exclusive transaction to trigger deadline check trnsNE := mustStart(t, tm, "ne", time.Hour, true) - compareTransactions(t, pilosa.Transaction{ID: "ne", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsNE) + test.CompareTransactions(t, pilosa.Transaction{ID: "ne", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsNE) // sleep for most of the deadline time.Sleep(time.Millisecond * 3) @@ -154,14 +154,14 @@ func TestTransactionManager(t *testing.T) { t.Errorf("resetting deadline: %v", err) } trnsM.Deadline = time.Now().Add(time.Millisecond * 4) - compareTransactions(t, trnsM, trnsM_reset) + test.CompareTransactions(t, trnsM, trnsM_reset) // sleep until past the original deadline time.Sleep(time.Millisecond * 2) // verify that trnsM still exists trnsM_again := mustGet(t, tm, "m") - compareTransactions(t, trnsM, trnsM_again) + test.CompareTransactions(t, trnsM, trnsM_again) } @@ -201,34 +201,6 @@ func mustList(t *testing.T, tm *pilosa.TransactionManager) map[string]pilosa.Tra return trnsMap } -// compareTransactions errors describing how the -// transactions differ (if at all). The deadlines need only be close -// (within 3ms). -func compareTransactions(t *testing.T, trns1, trns2 pilosa.Transaction) { - t.Helper() - if trns1.ID != trns2.ID { - t.Errorf("IDs differ:\n%+v\n%+v", trns1, trns2) - } - if trns1.Active != trns2.Active { - t.Errorf("Actives differ:\n%+v\n%+v", trns1, trns2) - } - if trns1.Exclusive != trns2.Exclusive { - t.Errorf("Exclusives differ:\n%+v\n%+v", trns1, trns2) - } - if trns1.Timeout != trns2.Timeout { - t.Errorf("Timeouts differ:\n%+v\n%+v", trns1, trns2) - } - - diff := trns1.Deadline.Sub(trns2.Deadline) - - if diff > time.Millisecond*3 || diff < time.Millisecond*-3 { - t.Errorf("Deadlines differ by %v:\n%+v\n%+v", diff, trns1, trns2) - } - if trns1.Stats != trns2.Stats { - t.Errorf("Stats differ:\n%+v\n%+v", trns1, trns2) - } -} - func TestInMemTransactionStore(t *testing.T) { ims := pilosa.NewInMemTransactionStore()