Merge pull request #5028 from freemine/mac

bugfix on mac and pressure-test

src/mnode/src/mnodeCluster.c

@@ -138,14 +138,6 @@ void mnodeDecClusterRef(SClusterObj *pCluster) {
   sdbDecRef(tsClusterSdb, pCluster);
 }
 
-#ifdef __APPLE__
-bool taosGetSystemUid(char *uid) {
-  fprintf(stderr, "%s[%d]%s(): not implemented yet!\n", basename(__FILE__), __LINE__, __func__);
-  abort();
-  return false;
-}
-#endif // __APPLE__
-
 static int32_t mnodeCreateCluster() {
   int64_t numOfClusters = sdbGetNumOfRows(tsClusterSdb);
   if (numOfClusters != 0) return TSDB_CODE_SUCCESS;

src/os/src/darwin/darwinSemphone.c

@@ -134,7 +134,7 @@ int tsem_init(tsem_t *sem, int pshared, unsigned int value) {
     errno = ENOMEM;
     return -1;
   }
-  kern_return_t ret = semaphore_create(sem_port, &p->sem, SYNC_POLICY_FIFO, 0);
+  kern_return_t ret = semaphore_create(sem_port, &p->sem, SYNC_POLICY_FIFO, value);
   if (ret != KERN_SUCCESS) {
     fprintf(stderr, "==%s[%d]%s():[%p]==semophore_create failed\n", basename(__FILE__), __LINE__, __func__, sem);
     // we fail-fast here, because we have less-doc about semaphore_create for the moment

src/os/src/darwin/darwinSysInfo.c

@@ -103,4 +103,15 @@ int taosSystem(const char *cmd) {
 
 void taosSetCoreDump() {}
 
-char *taosGetCmdlineByPID(int pid) { return ""; }
+char *taosGetCmdlineByPID(int pid) {
+  return "[not supported yet]";
+}
+
+bool taosGetSystemUid(char *uid) {
+  uuid_t uuid = {0};
+  uuid_generate(uuid);
+  // it's caller's responsibility to make enough space for `uid`, that's 36-char + 1-null
+  uuid_unparse(uuid, uid);
+  return true;
+}
+

src/plugins/http/src/httpUtil.c

@@ -37,7 +37,7 @@ void httpTimeToString(time_t t, char *buf, int32_t buflen) {
   time_t     tt = t / 1000;
   ptm = localtime(&tt);
   strftime(ts, 31, "%Y-%m-%d %H:%M:%S", ptm);
-  sprintf(buf, "%s.%03" PRId64, ts, t % 1000);
+  sprintf(buf, "%s.%03" PRId64, ts, (int64_t)(t % 1000));
 }
 
 int32_t httpAddToSqlCmdBuffer(HttpContext *pContext, const char *const format, ...) {

src/tsdb/inc/tsdbMain.h

@@ -233,11 +233,7 @@ typedef struct {
   SMemTable*      mem;
   SMemTable*      imem;
   STsdbFileH*     tsdbFileH;
-#ifdef __APPLE__
+  tsem_t          readyToCommit;
-  sem_t          *readyToCommit;
-#else // __APPLE__
-  sem_t           readyToCommit;
-#endif // __APPLE__
   pthread_mutex_t mutex;
   bool            repoLocked;
   int32_t         code; // Commit code

src/tsdb/src/tsdbCommit.c

@@ -166,11 +166,7 @@ static void tsdbEndCommit(STsdbRepo *pRepo, int eno) {
   pRepo->imem = NULL;
   tsdbUnlockRepo(pRepo);
   tsdbUnRefMemTable(pRepo, pIMem);
-#ifdef __APPLE__
+  tsem_post(&(pRepo->readyToCommit));
-  sem_post(pRepo->readyToCommit);
-#else // __APPLE__
-  sem_post(&(pRepo->readyToCommit));
-#endif // __APPLE__
 }
 
 static int tsdbHasDataToCommit(SCommitIter *iters, int nIters, TSKEY minKey, TSKEY maxKey) {

src/tsdb/src/tsdbMain.c

@@ -146,11 +146,7 @@ int tsdbCloseRepo(TSDB_REPO_T *repo, int toCommit) {
 
   if (toCommit) {
     tsdbAsyncCommit(pRepo);
-#ifdef __APPLE__
+    tsem_wait(&(pRepo->readyToCommit));
-    sem_wait(pRepo->readyToCommit);
-#else // __APPLE__
-    sem_wait(&(pRepo->readyToCommit));
-#endif // __APPLE__
     terrno = pRepo->code;
   }
   tsdbUnRefMemTable(pRepo, pRepo->mem);
@@ -647,21 +643,12 @@ static STsdbRepo *tsdbNewRepo(char *rootDir, STsdbAppH *pAppH, STsdbCfg *pCfg) {
     goto _err;
   }
 
-#ifdef __APPLE__
+  code = tsem_init(&(pRepo->readyToCommit), 0, 1);
-  pRepo->readyToCommit = sem_open(NULL, O_CREAT, 0644, 1);
-  if (pRepo->readyToCommit==SEM_FAILED) {
-    code = errno;
-    terrno = TAOS_SYSTEM_ERROR(code);
-    goto _err;
-  }
-#else // __APPLE__
-  code = sem_init(&(pRepo->readyToCommit), 0, 1);
   if (code != 0) {
     code = errno;
     terrno = TAOS_SYSTEM_ERROR(code);
     goto _err;
   }
-#endif // __APPLE__
 
   pRepo->repoLocked = false;
 
@@ -707,11 +694,7 @@ static void tsdbFreeRepo(STsdbRepo *pRepo) {
     // tsdbFreeMemTable(pRepo->mem);
     // tsdbFreeMemTable(pRepo->imem);
     tfree(pRepo->rootDir);
-#ifdef __APPLE__
+    tsem_destroy(&(pRepo->readyToCommit));
-    sem_close(pRepo->readyToCommit);
-#else // __APPLE__
-    sem_destroy(&(pRepo->readyToCommit));
-#endif // __APPLE__
     pthread_mutex_destroy(&pRepo->mutex);
     free(pRepo);
   }

src/tsdb/src/tsdbMemTable.c

@@ -207,11 +207,7 @@ void *tsdbAllocBytes(STsdbRepo *pRepo, int bytes) {
 int tsdbAsyncCommit(STsdbRepo *pRepo) {
   if (pRepo->mem == NULL) return 0;
 
-#ifdef __APPLE__
+  tsem_wait(&(pRepo->readyToCommit));
-  sem_wait(pRepo->readyToCommit);
-#else // __APPLE__
-  sem_wait(&(pRepo->readyToCommit));
-#endif // __APPLE__
 
   ASSERT(pRepo->imem == NULL);
 
@@ -233,13 +229,8 @@ int tsdbSyncCommit(TSDB_REPO_T *repo) {
   STsdbRepo *pRepo = (STsdbRepo *)repo;
 
   tsdbAsyncCommit(pRepo);
-#ifdef __APPLE__
+  tsem_wait(&(pRepo->readyToCommit));
-  sem_wait(pRepo->readyToCommit);
+  tsem_post(&(pRepo->readyToCommit));
-  sem_post(pRepo->readyToCommit);
-#else // __APPLE__
-  sem_wait(&(pRepo->readyToCommit));
-  sem_post(&(pRepo->readyToCommit));
-#endif // __APPLE__
 
   if (pRepo->code != TSDB_CODE_SUCCESS) {
     terrno = pRepo->code;

tests/examples/c/epoll.c

@@ -13,6 +13,12 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+// how to use to do a pressure-test upon eok
+// tester:               cat /dev/urandom | nc -c <ip> <port>
+// testee:               ./debug/build/bin/epoll -l <port> > /dev/null
+// compare against:      nc -l <port> > /dev/null
+// monitor and compare : glances
+
 #ifdef __APPLE__
 #include "eok.h"
 #else // __APPLE__
@@ -68,55 +74,62 @@ static int ep_dummy = 0;
 static ep_t* ep_create(void);
 static void  ep_destroy(ep_t *ep);
 static void* routine(void* arg);
-static int open_connect(unsigned short port);
 static int open_listen(unsigned short port);
 
-typedef struct client_s          client_t;
+typedef struct fde_s          fde_t;
-struct client_s {
+struct fde_s {
   int                          skt;
-  void (*on_event)(ep_t *ep, struct epoll_event *events, client_t *client);
+  void (*on_event)(ep_t *ep, struct epoll_event *events, fde_t *client);
-  volatile unsigned int        state; // 1: listenning; 2: connected
 };
 
-static void echo_event(ep_t *ep, struct epoll_event *ev, client_t *client);
+static void listen_event(ep_t *ep, struct epoll_event *ev, fde_t *client);
+static void null_event(ep_t *ep, struct epoll_event *ev, fde_t *client);
+
+#define usage(arg0, fmt, ...)       do {                                               \
+  if (fmt[0]) {                                                                        \
+    fprintf(stderr, "" fmt "\n", ##__VA_ARGS__);                                       \
+  }                                                                                    \
+  fprintf(stderr, "usage:\n");                                                         \
+  fprintf(stderr, "  %s -l <port>             : specify listenning port\n", arg0);     \
+} while (0)
 
 int main(int argc, char *argv[]) {
+  char *prg = basename(argv[0]);
+  if (argc==1) {
+    usage(prg, "");
+    return 0;
+  }
   ep_t* ep = ep_create();
   A(ep, "failed");
-  int skt = open_connect(6789);
+  for (int i=1; i<argc; ++i) {
-  if (skt!=-1) {
+    const char *arg = argv[i];
-    client_t *client = (client_t*)calloc(1, sizeof(*client));
+    if (0==strcmp(arg, "-l")) {
-    if (client) {
+      ++i;
+      if (i>=argc) {
+        usage(prg, "expecting <port> after -l, but got nothing");
+        return 1; // confirmed potential leakage
+      }
+      arg = argv[i];
+      int port = atoi(arg);
+      int skt = open_listen(port);
+      if (skt==-1) continue;
+      fde_t *client = (fde_t*)calloc(1, sizeof(*client));
+      if (!client) {
+        E("out of memory");
+        close(skt);
+        continue;
+      }
       client->skt = skt;
-      client->on_event = echo_event;
+      client->on_event = listen_event;
-      client->state = 2;
       struct epoll_event ev = {0};
       ev.events   = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP;
       ev.data.ptr = client;
       A(0==epoll_ctl(ep->ep, EPOLL_CTL_ADD, skt, &ev), "");
+      continue;
     }
+    usage(prg, "unknown argument: [%s]", arg);
+    return 1;
   }
-  skt = open_listen(0);
-  if (skt!=-1) {
-    client_t *client = (client_t*)calloc(1, sizeof(*client));
-    if (client) {
-      client->skt = skt;
-      client->on_event = echo_event;
-      client->state = 1;
-      struct epoll_event ev = {0};
-      ev.events   = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP;
-      ev.data.ptr = client;
-      A(0==epoll_ctl(ep->ep, EPOLL_CTL_ADD, skt, &ev), "");
-    }
-  }
-  // char c = '\0';
-  // while ((c=getchar())!=EOF) {
-  //   switch (c) {
-  //     case 'q': break;
-  //     default: continue;
-  //   }
-  // }
-  // getchar();
   char *line = NULL;
   size_t linecap = 0;
   ssize_t linelen;
@@ -205,7 +218,7 @@ static void* routine(void* arg) {
         continue;
       }
       A(ev->data.ptr, "internal logic error");
-      client_t *client = (client_t*)ev->data.ptr;
+      fde_t *client = (fde_t*)ev->data.ptr;
       client->on_event(ep, ev, client);
       continue;
     }
@@ -223,7 +236,7 @@ static int open_listen(unsigned short port) {
   do {
     struct sockaddr_in si = {0};
     si.sin_family = AF_INET;
-    si.sin_addr.s_addr = inet_addr("127.0.0.1");
+    si.sin_addr.s_addr = inet_addr("0.0.0.0");
     si.sin_port = htons(port);
     r = bind(skt, (struct sockaddr*)&si, sizeof(si));
     if (r) {
@@ -249,63 +262,31 @@ static int open_listen(unsigned short port) {
   return -1;
 }
 
-static int open_connect(unsigned short port) {
+static void listen_event(ep_t *ep, struct epoll_event *ev, fde_t *client) {
-  int r = 0;
+  A(ev->events & EPOLLIN, "internal logic error");
-  int skt = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
-  if (skt==-1) {
-    E("socket() failed");
-    return -1;
-  }
-  do {
-    struct sockaddr_in si = {0};
-    si.sin_family = AF_INET;
-    si.sin_addr.s_addr = inet_addr("127.0.0.1");
-    si.sin_port = htons(port);
-    r = connect(skt, (struct sockaddr*)&si, sizeof(si));
-    if (r) {
-      E("connect(%u) failed", port);
-      break;
-    }
-    memset(&si, 0, sizeof(si));
-    socklen_t len = sizeof(si);
-    r = getsockname(skt, (struct sockaddr *)&si, &len);
-    if (r) {
-      E("getsockname() failed");
-    }
-    A(len==sizeof(si), "internal logic error");
-    D("connected: %d", ntohs(si.sin_port));
-    return skt;
-  } while (0);
-  close(skt);
-  return -1;
-}
-
-static void echo_event(ep_t *ep, struct epoll_event *ev, client_t *client) {
-  if (ev->events & EPOLLIN) {
-    if (client->state==1) {
   struct sockaddr_in si = {0};
   socklen_t silen = sizeof(si);
   int skt = accept(client->skt, (struct sockaddr*)&si, &silen);
-      if (skt!=-1) {
+  A(skt!=-1, "internal logic error");
-        client_t *server = (client_t*)calloc(1, sizeof(*server));
+  fde_t *server = (fde_t*)calloc(1, sizeof(*server));
-        if (server) {
+  if (!server) {
+    close(skt);
+    return;
+  }
   server->skt = skt;
-          server->on_event = echo_event;
+  server->on_event = null_event;
-          server->state = 2;
+  struct epoll_event ee = {0};
-          struct epoll_event ev = {0};
+  ee.events   = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP;
-          ev.events   = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP;
+  ee.data.ptr = server;
-          ev.data.ptr = server;
+  A(0==epoll_ctl(ep->ep, EPOLL_CTL_ADD, skt, &ee), "");
-          A(0==epoll_ctl(ep->ep, EPOLL_CTL_ADD, skt, &ev), "");
+}
-        }
+
-      }
+static void null_event(ep_t *ep, struct epoll_event *ev, fde_t *client) {
-    }
+  if (ev->events & EPOLLIN) {
-    if (client->state==2) {
+    char buf[8192];
-      char buf[4];
+    int n = recv(client->skt, buf, sizeof(buf), 0);
-      int n = recv(client->skt, buf, sizeof(buf)-1, 0);
+    A(n>=0 && n<=sizeof(buf), "internal logic error:[%d]", n);
-      A(n>=0 && n<sizeof(buf), "internal logic error:[%d]", n);
+    A(n==fwrite(buf, 1, n, stdout), "internal logic error");
-      buf[n] = '\0';
-      fprintf(stderr, "events[%x]:%s\n", ev->events, buf);
-    }
   }
   if (ev->events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) {
     A(0==pthread_mutex_lock(&ep->lock), "");