add sync code

source/libs/sync/inc/syncIO.h

@@ -20,11 +20,38 @@
 extern "C" {
 #endif
 
+#include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
-#include <stdint.h>
+#include "os.h"
+#include "syncInt.h"
 #include "taosdef.h"
+#include "tqueue.h"
+#include "trpc.h"
 
+typedef struct SSyncIO {
+  void *      serverRpc;
+  void *      clientRpc;
+  STaosQueue *pMsgQ;
+  STaosQset * pQset;
+  pthread_t   tid;
+  int8_t      isStart;
+
+  int32_t (*start)(struct SSyncIO *ths);
+  int32_t (*stop)(struct SSyncIO *ths);
+  int32_t (*ping)(struct SSyncIO *ths);
+  int32_t (*onMessage)(struct SSyncIO *ths, void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
+  int32_t (*destroy)(struct SSyncIO *ths);
+
+} SSyncIO;
+
+SSyncIO * syncIOCreate();
+
+static int32_t syncIOStart(SSyncIO *io);
+static int32_t syncIOStop(SSyncIO *io);
+static int32_t syncIOPing(SSyncIO *io);
+static int32_t syncIOOnMessage(struct SSyncIO *io, void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
+static int32_t syncIODestroy(SSyncIO *io);
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncInt.h

@@ -23,16 +23,14 @@ extern "C" {
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
-#include "taosdef.h"
 #include "sync.h"
+#include "taosdef.h"
 #include "tlog.h"
 
 extern int32_t sDebugFlag;
 
 #define sLog(...) \
-  {                                                        \
+  { taosPrintLog("SYN FATAL ", sDebugFlag, __VA_ARGS__); }
-      taosPrintLog("SYN FATAL ", sDebugFlag, __VA_ARGS__); \
-  }
 
 #define sFatal(...)                                        \
   {                                                        \

source/libs/sync/src/syncAppendEntries.c

@@ -17,7 +17,6 @@
 #include "sync.h"
 
 void appendEntries(SRaft *pRaft, const SyncAppendEntries *pMsg) {
-
   // TLA+ Spec
   // AppendEntries(i, j) ==
   //    /\ i /= j
@@ -42,11 +41,9 @@ void appendEntries(SRaft *pRaft, const SyncAppendEntries *pMsg) {
   //                msource        |-> i,
   //                mdest          |-> j])
   //    /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
-
 }
 
 void onAppendEntries(SRaft *pRaft, const SyncAppendEntries *pMsg) {
-
   // TLA+ Spec
   // HandleAppendEntriesRequest(i, j, m) ==
   //    LET logOk == \/ m.mprevLogIndex = 0
@@ -112,6 +109,4 @@ void onAppendEntries(SRaft *pRaft, const SyncAppendEntries *pMsg) {
   //                       /\ UNCHANGED <<serverVars, commitIndex, messages>>
   //       /\ UNCHANGED <<candidateVars, leaderVars>>
   //
-
-
 }

source/libs/sync/src/syncAppendEntriesReply.c

@@ -17,7 +17,6 @@
 #include "sync.h"
 
 void onAppendEntriesReply(SRaft *pRaft, const SyncAppendEntriesReply *pMsg) {
-
   // TLA+ Spec
   // HandleAppendEntriesResponse(i, j, m) ==
   //    /\ m.mterm = currentTerm[i]
@@ -30,5 +29,4 @@ void onAppendEntriesReply(SRaft *pRaft, const SyncAppendEntriesReply *pMsg) {
   //          /\ UNCHANGED <<matchIndex>>
   //    /\ Discard(m)
   //    /\ UNCHANGED <<serverVars, candidateVars, logVars, elections>>
-
 }

source/libs/sync/src/syncIO.c

@@ -13,4 +13,191 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "sync.h"
+#include "syncIO.h"
+#include "syncOnMessage.h"
+
+void *syncConsumer(void *param) {
+  SSyncIO *io = param;
+
+  STaosQall *qall;
+  SRpcMsg   *pRpcMsg, rpcMsg;
+  int        type;
+
+  qall = taosAllocateQall();
+
+  while (1) {
+    int numOfMsgs = taosReadAllQitemsFromQset(io->pQset, qall, NULL, NULL);
+    sDebug("%d sync-io msgs are received", numOfMsgs);
+    if (numOfMsgs <= 0) break;
+
+    for (int i = 0; i < numOfMsgs; ++i) {
+      taosGetQitem(qall, (void **)&pRpcMsg);
+      sDebug("sync-io recv msg: %s", (char *)(pRpcMsg->pCont));
+    }
+
+    taosResetQitems(qall);
+    for (int i = 0; i < numOfMsgs; ++i) {
+      taosGetQitem(qall, (void **)&pRpcMsg);
+      rpcFreeCont(pRpcMsg->pCont);
+
+      /*
+          int msgSize = 128;
+          memset(&rpcMsg, 0, sizeof(rpcMsg));
+          rpcMsg.pCont = rpcMallocCont(msgSize);
+          rpcMsg.contLen = msgSize;
+          rpcMsg.handle = pRpcMsg->handle;
+          rpcMsg.code = 0;
+          rpcSendResponse(&rpcMsg);
+      */
+
+      taosFreeQitem(pRpcMsg);
+    }
+  }
+
+  taosFreeQall(qall);
+}
+
+static int retrieveAuthInfo(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey) {
+  // app shall retrieve the auth info based on meterID from DB or a data file
+  // demo code here only for simple demo
+  int ret = 0;
+  return ret;
+}
+
+static void processResponse(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
+  /*
+// SInfo *pInfo = (SInfo *)pMsg->ahandle;
+sDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen,
+       pMsg->code);
+
+if (pEpSet) pInfo->epSet = *pEpSet;
+
+rpcFreeCont(pMsg->pCont);
+// tsem_post(&pInfo->rspSem);
+tsem_post(&pInfo->rspSem);
+*/
+}
+
+static void processRequestMsg(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
+  SSyncIO *io = pParent;
+  SRpcMsg *pTemp;
+
+  pTemp = taosAllocateQitem(sizeof(SRpcMsg));
+  memcpy(pTemp, pMsg, sizeof(SRpcMsg));
+
+  sDebug("request is received, type:%d, contLen:%d, item:%p", pMsg->msgType, pMsg->contLen, pTemp);
+  taosWriteQitem(io->pMsgQ, pTemp);
+}
+
+SSyncIO *syncIOCreate() {
+  SSyncIO *io = (SSyncIO *)malloc(sizeof(SSyncIO));
+  memset(io, 0, sizeof(*io));
+
+  io->pMsgQ = taosOpenQueue();
+  io->pQset = taosOpenQset();
+  taosAddIntoQset(io->pQset, io->pMsgQ, NULL);
+
+  io->start = syncIOStart;
+  io->stop = syncIOStop;
+  io->ping = syncIOPing;
+  io->onMessage = syncIOOnMessage;
+  io->destroy = syncIODestroy;
+
+  return io;
+}
+
+static int32_t syncIOStart(SSyncIO *io) {
+  taosBlockSIGPIPE();
+
+  // cient rpc init
+  {
+    SRpcInit rpcInit;
+    memset(&rpcInit, 0, sizeof(rpcInit));
+    rpcInit.localPort = 0;
+    rpcInit.label = "SYNC-IO-CLIENT";
+    rpcInit.numOfThreads = 1;
+    rpcInit.cfp = processResponse;
+    rpcInit.sessions = 100;
+    rpcInit.idleTime = 100;
+    rpcInit.user = "sync-io";
+    rpcInit.secret = "sync-io";
+    rpcInit.ckey = "key";
+    rpcInit.spi = 1;
+    rpcInit.connType = TAOS_CONN_CLIENT;
+
+    io->clientRpc = rpcOpen(&rpcInit);
+    if (io->clientRpc == NULL) {
+      sError("failed to initialize RPC");
+      return -1;
+    }
+  }
+
+  // server rpc init
+  {
+    SRpcInit rpcInit;
+    memset(&rpcInit, 0, sizeof(rpcInit));
+    rpcInit.localPort = 38000;
+    rpcInit.label = "SYNC-IO-SERVER";
+    rpcInit.numOfThreads = 1;
+    rpcInit.cfp = processRequestMsg;
+    rpcInit.sessions = 1000;
+    rpcInit.idleTime = 2 * 1500;
+    rpcInit.afp = retrieveAuthInfo;
+    rpcInit.parent = io;
+    rpcInit.connType = TAOS_CONN_SERVER;
+
+    void *pRpc = rpcOpen(&rpcInit);
+    if (pRpc == NULL) {
+      sError("failed to start RPC server");
+      return -1;
+    }
+  }
+
+  // start consumer thread
+  {
+    if (pthread_create(&io->tid, NULL, syncConsumer, NULL) != 0) {
+      sError("failed to create sync consumer thread since %s", strerror(errno));
+      terrno = TAOS_SYSTEM_ERROR(errno);
+      return -1;
+    }
+  }
+
+  return 0;
+}
+
+static int32_t syncIOStop(SSyncIO *io) {
+  atomic_store_8(&io->isStart, 0);
+  pthread_join(io->tid, NULL);
+  return 0;
+}
+
+static int32_t syncIOPing(SSyncIO *io) { return 0; }
+
+static int32_t syncIOOnMessage(struct SSyncIO *io, void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) { return 0; }
+
+static int32_t syncIODestroy(SSyncIO *io) {
+  int8_t start = atomic_load_8(&io->isStart);
+  assert(start == 0);
+
+  if (io->serverRpc != NULL) {
+    free(io->serverRpc);
+    io->serverRpc = NULL;
+  }
+
+  if (io->clientRpc != NULL) {
+    free(io->clientRpc);
+    io->clientRpc = NULL;
+  }
+
+  if (io->pMsgQ != NULL) {
+    free(io->pMsgQ);
+    io->pMsgQ = NULL;
+  }
+
+  if (io->pQset != NULL) {
+    free(io->pQset);
+    io->pQset = NULL;
+  }
+
+  return 0;
+}

source/libs/sync/src/syncRequestVote.c

@@ -17,7 +17,6 @@
 #include "sync.h"
 
 void requestVote(SRaft *pRaft, const SyncRequestVote *pMsg) {
-
   // TLA+ Spec
   // RequestVote(i, j) ==
   //    /\ state[i] = Candidate
@@ -29,11 +28,9 @@ void requestVote(SRaft *pRaft, const SyncRequestVote *pMsg) {
   //             msource       |-> i,
   //             mdest         |-> j])
   //    /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
-
 }
 
 void onRequestVote(SRaft *pRaft, const SyncRequestVote *pMsg) {
-
   // TLA+ Spec
   // HandleRequestVoteRequest(i, j, m) ==
   //    LET logOk == \/ m.mlastLogTerm > LastTerm(log[i])
@@ -55,5 +52,4 @@ void onRequestVote(SRaft *pRaft, const SyncRequestVote *pMsg) {
   //                 mdest        |-> j],
   //                 m)
   //       /\ UNCHANGED <<state, currentTerm, candidateVars, leaderVars, logVars>>
-
 }

source/libs/sync/src/syncRequestVoteReply.c

@@ -17,7 +17,6 @@
 #include "sync.h"
 
 void onRequestVoteReply(SRaft *pRaft, const SyncRequestVoteReply *pMsg) {
-
   // TLA+ Spec
   // HandleRequestVoteResponse(i, j, m) ==
   //    \* This tallies votes even when the current state is not Candidate, but
@@ -34,5 +33,4 @@ void onRequestVoteReply(SRaft *pRaft, const SyncRequestVoteReply *pMsg) {
   //          /\ UNCHANGED <<votesGranted, voterLog>>
   //    /\ Discard(m)
   //    /\ UNCHANGED <<serverVars, votedFor, leaderVars, logVars>>
-
 }

source/libs/sync/test/syncTest.cpp

@@ -1,26 +1,19 @@
 #include <stdio.h>
+#include "syncIO.h"
 #include "syncInt.h"
 
 int main() {
-    printf("test \n");
+  tsAsyncLog = 0;
+  taosInitLog((char*)"syncTest.log", 100000, 10);
 
+  sDebug("sync test");
   syncStartEnv();
 
-    char temp[100];
+  SSyncIO *syncIO = syncIOCreate();
-    snprintf(temp, 100, "./debug.log");
+  assert(syncIO != NULL);
-    taosInitLog(temp, 10000, 1);
-    tsAsyncLog = 0;
-
-    for (int i = 0; i < 100; i++) {
-        sDebug("log:%d -------- \n", i);
-    }
-
-    fflush(NULL);
-    //taosCloseLog();
 
   while (1) {
     sleep(3);
   }
   return 0;
 }
-