InfiniSQL  v0.1.2-alpha
Massive Scale Transaction Processing
 All Classes Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
Message.h
Go to the documentation of this file.
1 /*
2  * Copyright (c) 2013 Mark Travis <mtravis15432+src@gmail.com>
3  * All rights reserved. No warranty, explicit or implicit, provided.
4  *
5  * This file is part of InfiniSQL(tm).
6 
7  * InfiniSQL is free software: you can redistribute it and/or modify
8  * it under the terms of the GNU General Public License version 3
9  * as published by the Free Software Foundation.
10  *
11  * InfiniSQL is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14  * GNU General Public License for more details.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with InfiniSQL. If not, see <http://www.gnu.org/licenses/>.
18  */
19 
30 #ifndef INFINISQLMESSAGE_H
31 #define INFINISQLMESSAGE_H
32 
33 #include "gch.h"
34 #include "defs.h"
35 #include "Topology.h"
36 
37 /* THIS STUFF PROBABLY NEEDS TO BE CONSIDERED THROUGHOUT THE CODE BASE
38  * BECAUSE SIZES CHANGED FROM int64_t TO OTHER THINGS, SUCH AS int16_t
39  Message:
40  start 34
41  Topology::addressStruct
42  nodeid -> int16_t
43  actorid -> int16_t
44  finish 10
45 
46  MessageTransaction:
47  start 65
48  tainstance -> int16_t
49  domainid -> int16_t
50  transaction_pendingcmdid -> int32_t
51  transaction_tacmdentrypoint -> int8_t
52  engineinstance -> int16_t
53  finish 36
54 
55  MessageSubtransactionCmd:
56  subtransaction_s
57  start 58
58  status -> int8_t
59  tableid -> int16_t
60  fieldid -> int16_t
61  engineid -> int16_t
62  finish 33
63 
64  start 8+16 per entry
65  indexHits (nonLockingIndexEntry_s):
66  engineid -> int16_t
67  finish 8 + 10 per entry
68 
69  MessageCommitRollback:
70  rofs, rowOrField_s
71  start 52 + fieldVal(25)
72  tableid -> int16_t
73  fieldid -> int16_t
74  engineid -> int16_t
75  newengineid -> int16_t
76  finish 28 + fieldVal(25)
77 */
78 
85 class Message
86 {
87 public:
92  struct __attribute__ ((__packed__)) message_s
93  {
94  payloadtype_e payloadtype;
95  topic_e topic;
96  Topology::addressStruct sourceAddr;
97  Topology::addressStruct destAddr;
98  };
99 
100  Message();
101  virtual ~Message();
107  size_t size();
114  std::string *ser();
120  void package(class SerializedMessage &serobj);
126  void unpack(SerializedMessage &serobj);
131  void clear();
139  static class Message *des(string *serstr);
146  string *sermsg();
154  void setEnvelope(const Topology::addressStruct &source,
155  const Topology::addressStruct &dest, class Message &msg);
156 
157  __int128 nextmsg;
158  message_s messageStruct;
159 };
160 
166 class MessageSocket : public Message
167 {
168 public:
173  struct __attribute__ ((__packed__)) socket_s
174  {
175  int socket;
176  uint32_t events;
177  listenertype_e listenertype;
178  };
179  MessageSocket();
189  MessageSocket(int socketarg, uint32_t eventsarg,
190  listenertype_e listenertype, int64_t nodeidarg,
191  topic_e topicarg);
192  virtual ~MessageSocket();
193 
199  size_t size();
206  string *ser();
212  void package(class SerializedMessage &serobj);
218  void unpack(SerializedMessage &serobj);
223  void clear();
224 
225  socket_s socketStruct;
226 };
227 
233 class MessageUserSchema : public Message
234 {
235 public:
240  struct __attribute__ ((__packed__)) userschema_s
241  {
242  operationtype_e operationtype;
243  int8_t caller;
244  int8_t callerstate;
245  int64_t argsize;
246  int64_t instance;
247  int64_t operationid;
248  int64_t domainid;
249  int64_t userid;
250  int64_t tableid;
251  int64_t fieldlen;
252  int64_t builtincmd;
253  int64_t indexid;
254  int64_t tableindexid;
255  int64_t simple;
256  int64_t fieldid;
257  int64_t numfields;
258  int64_t intdata;
259  int64_t status;
260  indextype_e indextype;
261  fieldtype_e fieldtype;
262  };
263  MessageUserSchema();
271  MessageUserSchema(topic_e topicarg);
272  virtual ~MessageUserSchema();
278  size_t size();
285  string *ser();
291  void package(class SerializedMessage &serobj);
297  void unpack(SerializedMessage &serobj);
302  void clear();
303 
304  userschema_s userschemaStruct;
305  procedures_s procs;
306  std::string argstring;
307  std::string pathname;
308  std::string procname;
309  std::string username;
310  std::string domainname;
311  std::string password;
312 };
313 
320 class MessageDeadlock : public Message
321 {
322 public:
327  struct __attribute__ ((__packed__)) deadlock_s
328  {
329  int64_t transactionid;
330  int64_t tainstance;
331  int64_t transaction_pendingcmdid;
332  int64_t deadlockchange;
333  };
334  MessageDeadlock();
335  virtual ~MessageDeadlock();
341  size_t size();
348  string *ser();
354  void package(class SerializedMessage &serobj);
360  void unpack(SerializedMessage &serobj);
365  void clear();
366 
367  deadlock_s deadlockStruct;
368  std::string deadlockNode;
369  newDeadLockLists_s nodes;
370 };
371 
378 class MessageTransaction : public Message
379 {
380 public:
385  struct __attribute__ ((__packed__)) transaction_s
386  {
387  int64_t transactionid;
388  int64_t subtransactionid;
389  int64_t previoussubtransactionid;
390  int16_t tainstance;
391  int16_t domainid;
392  enginecmd_e transaction_enginecmd;
393  int32_t transaction_pendingcmdid;
394  int8_t transaction_tacmdentrypoint;
395  int16_t engineinstance;
396  };
397  MessageTransaction();
398  virtual ~MessageTransaction();
404  size_t size();
411  string *ser();
417  void package(class SerializedMessage &serobj);
423  void unpack(SerializedMessage &serobj);
428  void clear();
429 
430  transaction_s transactionStruct;
431 };
432 
439 class MessageSubtransactionCmd : public MessageTransaction
440 {
441 public:
446  struct __attribute__ ((__packed__)) subtransaction_s
447  {
448  int8_t status;
449  bool isrow;
450  int64_t rowid;
451  int16_t tableid;
452  locktype_e locktype;
453  int64_t forward_rowid;
454  int16_t forward_engineid;
455  int16_t fieldid;
456  int16_t engineid; // index also uses rowid
457  };
458  MessageSubtransactionCmd();
459  virtual ~MessageSubtransactionCmd();
465  size_t size();
472  string *ser();
478  void package(class SerializedMessage &serobj);
484  void unpack(SerializedMessage &serobj);
489  void clear();
490 
491  subtransaction_s subtransactionStruct;
492 
493  std::string row;
494  fieldValue_s fieldVal;
495  std::vector<nonLockingIndexEntry_s> indexHits;
496  searchParams_s searchParameters;
497  std::vector<int64_t> rowids;
498  std::vector<returnRow_s> returnRows;
499 };
500 
507 class MessageCommitRollback : public MessageTransaction
508 {
509 public:
510  MessageCommitRollback();
511  virtual ~MessageCommitRollback();
517  size_t size();
524  string *ser();
530  void package(class SerializedMessage &serobj);
536  void unpack(SerializedMessage &serobj);
541  void clear();
542 
543  std::vector<rowOrField_s> rofs;
544 };
545 
552 class MessageDispatch : public Message
553 {
554 public:
559  struct __attribute__ ((__packed__)) dispatch_s
560  {
561  int64_t transactionid;
562  int64_t domainid;
563  };
564 
565  struct record_s
566  {
567  int64_t rowid;
568  pendingprimitive_e primitive;
569  int64_t tableid;
570  int64_t previoussubtransactionid;
571  std::string row;
572  std::string oldrow;
573  };
574 
575  MessageDispatch();
576  virtual ~MessageDispatch();
582  size_t size();
589  string *ser();
595  void package(class SerializedMessage &serobj);
601  void unpack(SerializedMessage &serobj);
606  void clear();
607 
608  dispatch_s dispatchStruct;
609 
610  // pidsids[partitionid] = subtransactionid
611  boost::unordered_map<int64_t, int64_t> pidsids;
612  // records[partitionid][#] = {recordinfo}
613  boost::unordered_map< int64_t, std::vector<record_s> > records;
614 };
615 
622 class MessageAckDispatch : public Message
623 {
624 public:
629  struct __attribute__ ((__packed__)) ackdispatch_s
630  {
631  int64_t transactionid;
632  int status;
633  };
634 
635  MessageAckDispatch();
644  MessageAckDispatch(int64_t transactionidarg, int statusarg);
645  virtual ~MessageAckDispatch();
651  size_t size();
658  string *ser();
664  void package(class SerializedMessage &serobj);
670  void unpack(SerializedMessage &serobj);
675  void clear();
676 
677  ackdispatch_s ackdispatchStruct;
678 };
679 
680 #define ADDFLAG 0
681 
688 class MessageApply : public Message
689 {
690 public:
695  struct __attribute__ ((__packed__)) apply_s
696  {
697  int64_t subtransactionid;
698  int64_t applierid;
699  int64_t domainid;
700  };
701  struct applyindex_s
702  {
703  fieldValue_s fieldVal;
704  nonLockingIndexEntry_s entry;
705  char flags;
706  int16_t tableid;
707  int16_t fieldid;
708  };
709 
710  MessageApply();
720  MessageApply(int64_t subtransactionidarg, int64_t applieridarg,
721  int64_t domainidarg);
722  virtual ~MessageApply();
728  size_t size();
735  string *ser();
741  void package(class SerializedMessage &serobj);
747  void unpack(SerializedMessage &serobj);
752  void clear();
753 
759  static void setisaddflag(char *c);
765  static char getisaddflag(char c);
771  static void cleariaddflag(char *c);
772 
773  apply_s applyStruct;
774 
775  std::vector<MessageDispatch::record_s> rows;
776  std::vector<applyindex_s> indices;
777 };
778 
785 class MessageAckApply : public Message
786 {
787 public:
792  struct __attribute__ ((__packed__)) ackapply_s
793  {
794  int64_t subtransactionid;
795  int64_t applierid;
796  int64_t partitionid;
797  int status;
798  };
799  MessageAckApply();
810  MessageAckApply(int64_t subtransactionidarg, int64_t applieridarg,
811  int64_t partitionidarg, int statusarg);
812  virtual ~MessageAckApply();
818  size_t size();
825  string *ser();
831  void package(class SerializedMessage &serobj);
837  void unpack(SerializedMessage &serobj);
842  void clear();
843 
844  ackapply_s ackapplyStruct;
845 };
846 
869 class SerializedMessage
870 {
871 public:
872  // source sender
873  SerializedMessage(size_t sizearg);
881  SerializedMessage(string *dataarg);
882  virtual ~SerializedMessage();
883 
884  size_t size;
885  size_t pos;
886  std::string *data;
887 
888  payloadtype_e getpayloadtype();
889  // raw
890  void ser(size_t s, void *dataptr);
891  void des(size_t s, void *dataptr);
892  // pods
893  void ser(int64_t d);
894  static size_t sersize(int64_t d);
895  void des(int64_t *d);
896  void ser(int32_t d);
897  static size_t sersize(int32_t d);
898  void des(int32_t *d);
899  void ser(int16_t d);
900  static size_t sersize(int16_t d);
901  void des(int16_t *d);
902  void ser(int8_t d);
903  static size_t sersize(int8_t d);
904  void des(int8_t *d);
905  // containers
906  void ser(const string &d);
907  static size_t sersize(const string &d);
908  void des(string &d);
909  void ser(vector<int64_t> &d);
910  static size_t sersize(vector<int64_t> &d);
911  void des(vector<int64_t> &d);
912  void ser(boost::unordered_map<int64_t, int64_t> &d);
913  static size_t sersize(boost::unordered_map<int64_t, int64_t> &d);
914  void des(boost::unordered_map<int64_t, int64_t> &d);
915  // pod structs
916  void ser(Message::message_s &d);
917  static size_t sersize(Message::message_s &d);
918  void des(Message::message_s &d);
919  void ser(MessageSocket::socket_s &d);
920  static size_t sersize(MessageSocket::socket_s &d);
921  void des(MessageSocket::socket_s &d);
922  void ser(MessageUserSchema::userschema_s &d);
923  static size_t sersize(MessageUserSchema::userschema_s &d);
924  void des(MessageUserSchema::userschema_s &d);
925  void ser(procedures_s &d);
926  static size_t sersize(procedures_s &d);
927  void des(procedures_s &d);
928  void ser(MessageDeadlock::deadlock_s &d);
929  static size_t sersize(MessageDeadlock::deadlock_s &d);
930  void des(MessageDeadlock::deadlock_s &d);
931  void ser(MessageTransaction::transaction_s &d);
932  static size_t sersize(MessageTransaction::transaction_s &d);
933  void des(MessageTransaction::transaction_s &d);
934  void ser(MessageSubtransactionCmd::subtransaction_s &d);
935  static size_t sersize(MessageSubtransactionCmd::subtransaction_s &d);
936  void des(MessageSubtransactionCmd::subtransaction_s &d);
937  void ser(nonLockingIndexEntry_s &d);
938  static size_t sersize(nonLockingIndexEntry_s &d);
939  void des(nonLockingIndexEntry_s &d);
940  void ser(MessageDispatch::dispatch_s &d);
941  static size_t sersize(MessageDispatch::dispatch_s &d);
942  void des(MessageDispatch::dispatch_s &d);
943  void ser(MessageAckDispatch::ackdispatch_s &d);
944  static size_t sersize(MessageAckDispatch::ackdispatch_s &d);
945  void des(MessageAckDispatch::ackdispatch_s &d);
946  void ser(MessageApply::apply_s &d);
947  static size_t sersize(MessageApply::apply_s &d);
948  void des(MessageApply::apply_s &d);
949  void ser(MessageAckApply::ackapply_s &d);
950  static size_t sersize(MessageAckApply::ackapply_s &d);
951  void des(MessageAckApply::ackapply_s &d);
952  // level 1
953  void ser(boost::unordered_set<string> &d);
954  static size_t sersize(boost::unordered_set<string> &d);
955  void des(boost::unordered_set<string> &d);
956  void ser(fieldValue_s &d);
957  static size_t sersize(fieldValue_s &d);
958  void des(fieldValue_s &d);
959  void ser(returnRow_s &d);
960  static size_t sersize(returnRow_s &d);
961  void des(returnRow_s &d);
962  void ser(MessageDispatch::record_s &d);
963  static size_t sersize(MessageDispatch::record_s &d);
964  void des(MessageDispatch::record_s &d);
965  void ser(vector<nonLockingIndexEntry_s> &d);
966  static size_t sersize(vector<nonLockingIndexEntry_s> &d);
967  void des(vector<nonLockingIndexEntry_s> &d);
968  // level 2
969  void ser(newDeadLockLists_s &d);
970  static size_t sersize(newDeadLockLists_s &d);
971  void des(newDeadLockLists_s &d);
972  void ser(vector<fieldValue_s> &d);
973  static size_t sersize(vector<fieldValue_s> &d);
974  void des(vector<fieldValue_s> &d);
975  void ser(vector<returnRow_s> &d);
976  static size_t sersize(vector<returnRow_s> &d);
977  void des(vector<returnRow_s> &d);
978  void ser(vector<MessageDispatch::record_s> &d);
979  static size_t sersize(vector<MessageDispatch::record_s> &d);
980  void des(vector<MessageDispatch::record_s> &d);
981  void ser(rowOrField_s &d);
982  static size_t sersize(rowOrField_s &d);
983  void des(rowOrField_s &d);
984  void ser(MessageApply::applyindex_s &d);
985  static size_t sersize(MessageApply::applyindex_s &d);
986  void des(MessageApply::applyindex_s &d);
987  // level 3
988  void ser(searchParams_s &d);
989  static size_t sersize(searchParams_s &d);
990  void des(searchParams_s &d);
991  void ser(vector<rowOrField_s> &d);
992  static size_t sersize(vector<rowOrField_s> &d);
993  void des(vector<rowOrField_s> &d);
994  void ser(vector<MessageApply::applyindex_s> &d);
995  static size_t sersize(vector<MessageApply::applyindex_s> &d);
996  void des(vector<MessageApply::applyindex_s> &d);
997  void ser(boost::unordered_map< int64_t,
998  vector<MessageDispatch::record_s> > &d);
999  static size_t sersize(boost::unordered_map< int64_t,
1000  vector<MessageDispatch::record_s> > &d);
1001  void des(boost::unordered_map< int64_t,
1002  vector<MessageDispatch::record_s> > &d);
1003 };
1004 
1012 class MessageSerialized : public Message
1013 {
1014 public:
1015  MessageSerialized(string *dataarg);
1016  virtual ~MessageSerialized();
1017 
1018  std::string *data;
1019 };
1020 
1028 class MessageBatchSerialized : public Message
1029 {
1030 public:
1035  struct msgbatch_s
1036  {
1037  int16_t nodeid;
1038  std::string *serializedmsg;
1039  };
1040 
1041  MessageBatchSerialized(int16_t nodeidarg);
1042  virtual ~MessageBatchSerialized();
1043 
1044  short nmsgs;
1045  msgbatch_s msgbatch[OBGWMSGBATCHSIZE];
1046 };
1047 
1048 #endif /* INFINISQLMESSAGE_H */