Merge remote-tracking branch 'freemine/mac' into feature/mac

cmake/define.inc

@@ -128,6 +128,8 @@ IF (TD_DARWIN_64)
   SET(COMMON_FLAGS "-std=gnu99 -Wall -Werror -Wno-missing-braces -fPIC -msse4.2 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE")
   SET(DEBUG_FLAGS "-O0 -g3 -DDEBUG")
   SET(RELEASE_FLAGS "-Og")
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/cJson/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/lz4/inc)
 ENDIF ()
 
 IF (TD_WINDOWS)

deps/CMakeLists.txt

@@ -12,4 +12,8 @@ ADD_SUBDIRECTORY(MsvcLibX)
 
 IF (TD_LINUX AND TD_MQTT)
   ADD_SUBDIRECTORY(MQTT-C)
-ENDIF ()
\ No newline at end of file
+ENDIF ()
+
+IF (TD_DARWIN AND TD_MQTT)
+  ADD_SUBDIRECTORY(MQTT-C)
+ENDIF ()

packaging/tools/install_client.sh

@@ -21,7 +21,7 @@ else
     cd ${script_dir}
     script_dir="$(pwd)"
     data_dir="/var/lib/taos"
-    log_dir="~/TDengineLog"
+    log_dir=~/TDengineLog
 fi
 
 log_link_dir="/usr/local/taos/log"

packaging/tools/make_install.sh

@@ -24,7 +24,7 @@ data_dir="/var/lib/taos"
 if [ "$osType" != "Darwin" ]; then
     log_dir="/var/log/taos"
 else
-    log_dir="~/TDengineLog"
+    log_dir=~/TDengineLog
 fi
 
 data_link_dir="/usr/local/taos/data"
@@ -178,7 +178,9 @@ function install_bin() {
 function install_lib() {
     # Remove links
     ${csudo} rm -f ${lib_link_dir}/libtaos.*     || :
-    ${csudo} rm -f ${lib64_link_dir}/libtaos.*   || :
+    if [ "$osType" != "Darwin" ]; then
+      ${csudo} rm -f ${lib64_link_dir}/libtaos.*   || :
+    fi
     
     if [ "$osType" != "Darwin" ]; then
         ${csudo} cp ${binary_dir}/build/lib/libtaos.so.${verNumber} ${install_main_dir}/driver && ${csudo} chmod 777 ${install_main_dir}/driver/*
@@ -190,12 +192,14 @@ function install_lib() {
           ${csudo} ln -sf ${lib64_link_dir}/libtaos.so.1 ${lib64_link_dir}/libtaos.so
         fi
     else
-        ${csudo} cp ${binary_dir}/build/lib/libtaos.* ${install_main_dir}/driver && ${csudo} chmod 777 ${install_main_dir}/driver/*
-        ${csudo} ln -sf ${install_main_dir}/driver/libtaos.* ${lib_link_dir}/libtaos.1.dylib
+        ${csudo} cp -Rf ${binary_dir}/build/lib/libtaos.* ${install_main_dir}/driver && ${csudo} chmod 777 ${install_main_dir}/driver/*
+        ${csudo} ln -sf ${install_main_dir}/driver/libtaos.1.dylib ${lib_link_dir}/libtaos.1.dylib
         ${csudo} ln -sf ${lib_link_dir}/libtaos.1.dylib ${lib_link_dir}/libtaos.dylib
     fi
     
-    ${csudo} ldconfig
+    if [ "$osType" != "Darwin" ]; then
+        ${csudo} ldconfig
+    fi
 }
 
 function install_header() {

src/balance/CMakeLists.txt

@@ -11,3 +11,7 @@ AUX_SOURCE_DIRECTORY(src SRC)
 IF (TD_LINUX)
   ADD_LIBRARY(balance ${SRC})
 ENDIF ()
+
+IF (TD_DARWIN)
+  ADD_LIBRARY(balance ${SRC})
+ENDIF ()

src/client/CMakeLists.txt

@@ -28,6 +28,28 @@ IF (TD_LINUX)
 
   ADD_SUBDIRECTORY(tests)
 
+ELSEIF (TD_DARWIN)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/jni/linux)
+
+  # set the static lib name
+  ADD_LIBRARY(taos_static STATIC ${SRC})
+  TARGET_LINK_LIBRARIES(taos_static common query trpc tutil pthread m)
+  SET_TARGET_PROPERTIES(taos_static PROPERTIES OUTPUT_NAME "taos_static")
+  SET_TARGET_PROPERTIES(taos_static PROPERTIES CLEAN_DIRECT_OUTPUT 1)
+
+  # generate dynamic library (*.dylib)
+  ADD_LIBRARY(taos SHARED ${SRC})
+  TARGET_LINK_LIBRARIES(taos common query trpc tutil pthread m)
+  SET_TARGET_PROPERTIES(taos PROPERTIES CLEAN_DIRECT_OUTPUT 1)
+
+  #set version of .dylib
+  #VERSION    dylib version
+  #SOVERSION  dylib version
+  #MESSAGE(STATUS "build version ${TD_VER_NUMBER}")
+  SET_TARGET_PROPERTIES(taos PROPERTIES VERSION ${TD_VER_NUMBER} SOVERSION 1)
+
+  ADD_SUBDIRECTORY(tests)
+
 ELSEIF (TD_WINDOWS)
   INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/jni/windows)
   INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/jni/windows/win32)
@@ -49,12 +71,12 @@ ELSEIF (TD_DARWIN)
   INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/jni/linux)
 
   ADD_LIBRARY(taos_static STATIC ${SRC})
-  TARGET_LINK_LIBRARIES(taos_static trpc tutil pthread m)
+  TARGET_LINK_LIBRARIES(taos_static query trpc tutil pthread m)
   SET_TARGET_PROPERTIES(taos_static PROPERTIES OUTPUT_NAME "taos_static")
 
   # generate dynamic library (*.dylib)
   ADD_LIBRARY(taos SHARED ${SRC})
-  TARGET_LINK_LIBRARIES(taos trpc tutil pthread m)
+  TARGET_LINK_LIBRARIES(taos query trpc tutil pthread m)
 
   SET_TARGET_PROPERTIES(taos PROPERTIES CLEAN_DIRECT_OUTPUT 1)
 

src/client/src/tscSQLParser.c

@@ -13,10 +13,12 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#ifndef __APPLE__
 #define _BSD_SOURCE
 #define _XOPEN_SOURCE 500
 #define _DEFAULT_SOURCE
 #define _GNU_SOURCE
+#endif // __APPLE__
 
 #include "os.h"
 #include "ttype.h"

src/client/src/tscSql.c

@@ -295,6 +295,10 @@ void taos_close(TAOS *taos) {
 
       tscDebug("%p HB is freed", pHb);
       taosReleaseRef(tscObjRef, pHb->self);
+#ifdef __APPLE__
+      // to satisfy later tsem_destroy in taos_free_result
+      tsem_init(&pHb->rspSem, 0, 0);
+#endif // __APPLE__
       taos_free_result(pHb);
     }
   }

src/client/src/tscUtil.c

@@ -1942,6 +1942,10 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, __async_cb_func_t fp, void* param, in
   }
 
   if (tscAddSubqueryInfo(pCmd) != TSDB_CODE_SUCCESS) {
+#ifdef __APPLE__
+    // to satisfy later tsem_destroy in taos_free_result
+    tsem_init(&pNew->rspSem, 0, 0);
+#endif // __APPLE__
     tscFreeSqlObj(pNew);
     return NULL;
   }
@@ -2508,7 +2512,11 @@ bool tscSetSqlOwner(SSqlObj* pSql) {
   SSqlRes* pRes = &pSql->res;
 
   // set the sql object owner
+#ifdef __APPLE__
+  pthread_t threadId = (pthread_t)taosGetSelfPthreadId();
+#else // __APPLE__
   uint64_t threadId = taosGetSelfPthreadId();
+#endif // __APPLE__
   if (atomic_val_compare_exchange_64(&pSql->owner, 0, threadId) != 0) {
     pRes->code = TSDB_CODE_QRY_IN_EXEC;
     return false;

src/cq/CMakeLists.txt

@@ -10,7 +10,17 @@ IF (TD_LINUX)
   ADD_LIBRARY(tcq ${SRC})
   IF (TD_SOMODE_STATIC)
     TARGET_LINK_LIBRARIES(tcq tutil common taos_static)
-  ELSE ()  
+  ELSE ()
+    TARGET_LINK_LIBRARIES(tcq tutil common taos)
+  ENDIF ()
+  ADD_SUBDIRECTORY(test)
+ENDIF ()
+
+IF (TD_DARWIN)
+  ADD_LIBRARY(tcq ${SRC})
+  IF (TD_SOMODE_STATIC)
+    TARGET_LINK_LIBRARIES(tcq tutil common taos_static)
+  ELSE ()
     TARGET_LINK_LIBRARIES(tcq tutil common taos)
   ENDIF ()
   ADD_SUBDIRECTORY(test)

src/dnode/CMakeLists.txt

@@ -23,7 +23,7 @@ IF (TD_LINUX)
   IF (TD_ACCOUNT)
     TARGET_LINK_LIBRARIES(taosd account)
   ENDIF ()
-  
+
   IF (TD_GRANT)
     TARGET_LINK_LIBRARIES(taosd grant)
   ENDIF ()
@@ -47,3 +47,42 @@ IF (TD_LINUX)
       COMMENT "prepare taosd environment")
   ADD_CUSTOM_TARGET(${PREPARE_ENV_TARGET} ALL WORKING_DIRECTORY ${TD_EXECUTABLE_OUTPUT_PATH} DEPENDS ${PREPARE_ENV_CMD})
 ENDIF ()
+
+IF (TD_DARWIN)
+  ADD_EXECUTABLE(taosd ${SRC})
+  TARGET_LINK_LIBRARIES(taosd mnode monitor http tsdb twal vnode cJson lz4 balance sync)
+
+  IF (TD_SOMODE_STATIC)
+    TARGET_LINK_LIBRARIES(taosd taos_static)
+  ELSE ()
+    TARGET_LINK_LIBRARIES(taosd taos)
+  ENDIF ()
+
+  IF (TD_ACCOUNT)
+    TARGET_LINK_LIBRARIES(taosd account)
+  ENDIF ()
+
+  IF (TD_GRANT)
+    TARGET_LINK_LIBRARIES(taosd grant)
+  ENDIF ()
+
+  IF (TD_MQTT)
+    TARGET_LINK_LIBRARIES(taosd mqtt)
+  ENDIF ()
+
+  # SET(PREPARE_ENV_CMD "prepare_env_cmd")
+  # SET(PREPARE_ENV_TARGET "prepare_env_target")
+  # ADD_CUSTOM_COMMAND(OUTPUT ${PREPARE_ENV_CMD}
+  #     POST_BUILD
+  #     COMMAND echo "make test directory"
+  #     DEPENDS taosd
+  #     COMMAND ${CMAKE_COMMAND} -E make_directory ${TD_TESTS_OUTPUT_DIR}/cfg/
+  #     COMMAND ${CMAKE_COMMAND} -E make_directory ${TD_TESTS_OUTPUT_DIR}/log/
+  #     COMMAND ${CMAKE_COMMAND} -E make_directory ${TD_TESTS_OUTPUT_DIR}/data/
+  #     COMMAND ${CMAKE_COMMAND} -E echo dataDir ${TD_TESTS_OUTPUT_DIR}/data > ${TD_TESTS_OUTPUT_DIR}/cfg/taos.cfg
+  #     COMMAND ${CMAKE_COMMAND} -E echo logDir ${TD_TESTS_OUTPUT_DIR}/log  >> ${TD_TESTS_OUTPUT_DIR}/cfg/taos.cfg
+  #     COMMAND ${CMAKE_COMMAND} -E echo charset UTF-8  >> ${TD_TESTS_OUTPUT_DIR}/cfg/taos.cfg
+  #     COMMENT "prepare taosd environment")
+  # ADD_CUSTOM_TARGET(${PREPARE_ENV_TARGET} ALL WORKING_DIRECTORY ${TD_EXECUTABLE_OUTPUT_PATH} DEPENDS ${PREPARE_ENV_CMD})
+ENDIF ()
+

src/dnode/src/dnodeSystem.c

@@ -160,7 +160,11 @@ static void signal_handler(int32_t signum, siginfo_t *sigInfo, void *context) {
   syslog(LOG_INFO, "Shut down signal is %d", signum);
   syslog(LOG_INFO, "Shutting down TDengine service...");
   // clean the system.
+#ifdef __APPLE__
+  dInfo("shut down signal is %d, sender PID:%d", signum, sigInfo->si_pid);
+#else // __APPLE__
   dInfo("shut down signal is %d, sender PID:%d cmdline:%s", signum, sigInfo->si_pid, taosGetCmdlineByPID(sigInfo->si_pid));
+#endif // __APPLE__
 
   // protect the application from receive another signal
   struct sigaction act = {{0}};

src/dnode/src/dnodeTelemetry.c

@@ -236,6 +236,13 @@ static void sendTelemetryReport() {
   taosCloseSocket(fd);
 }
 
+#ifdef __APPLE__
+static int sem_timedwait(tsem_t *sem, struct timespec *to) {
+  fprintf(stderr, "%s[%d]%s(): not implemented yet!\n", basename(__FILE__), __LINE__, __func__);
+  abort();
+}
+#endif // __APPLE__
+
 static void* telemetryThread(void* param) {
   struct timespec end = {0};
   clock_gettime(CLOCK_REALTIME, &end);

src/kit/shell/CMakeLists.txt

@@ -9,14 +9,14 @@ IF (TD_LINUX)
   AUX_SOURCE_DIRECTORY(./src SRC)
   LIST(REMOVE_ITEM SRC ./src/shellWindows.c)
   LIST(REMOVE_ITEM SRC ./src/shellDarwin.c)
-  ADD_EXECUTABLE(shell ${SRC})  
-  
+  ADD_EXECUTABLE(shell ${SRC})
+
   IF (TD_SOMODE_STATIC)
     TARGET_LINK_LIBRARIES(shell taos_static)
   ELSE ()
     TARGET_LINK_LIBRARIES(shell taos)
   ENDIF ()
-  
+
   SET_TARGET_PROPERTIES(shell PROPERTIES OUTPUT_NAME taos)
 ELSEIF (TD_WINDOWS)
   LIST(APPEND SRC ./src/shellEngine.c)
@@ -27,7 +27,7 @@ ELSEIF (TD_WINDOWS)
 
   IF (TD_POWER)
     SET_TARGET_PROPERTIES(shell PROPERTIES OUTPUT_NAME power)
-  ELSE ()   
+  ELSE ()
     SET_TARGET_PROPERTIES(shell PROPERTIES OUTPUT_NAME taos)
   ENDIF ()
 ELSEIF (TD_DARWIN)
@@ -37,7 +37,10 @@ ELSEIF (TD_DARWIN)
   LIST(APPEND SRC ./src/shellCommand.c)
   LIST(APPEND SRC ./src/shellImport.c)
   ADD_EXECUTABLE(shell ${SRC})
-  TARGET_LINK_LIBRARIES(shell taos_static)
+  # linking with dylib
+  TARGET_LINK_LIBRARIES(shell taos)
+  # linking taos statically
+  # TARGET_LINK_LIBRARIES(shell taos_static)
   SET_TARGET_PROPERTIES(shell PROPERTIES OUTPUT_NAME taos)
 ENDIF ()
 

src/kit/shell/src/shellDarwin.c

@@ -21,6 +21,8 @@
 #include "shellCommand.h"
 #include "tkey.h"
 
+#include "tscLog.h"
+
 #define OPT_ABORT 1 /* �Cabort */
 
 int indicator = 1;
@@ -348,6 +350,9 @@ void *shellLoopQuery(void *arg) {
       reset_terminal_mode();
     } while (shellRunCommand(con, command) == 0);
 
+  tfree(command);
+  exitShell();
+
   pthread_cleanup_pop(1);
 
   return NULL;

src/kit/taosdemo/CMakeLists.txt

@@ -7,7 +7,7 @@ INCLUDE_DIRECTORIES(inc)
 IF (TD_LINUX)
   AUX_SOURCE_DIRECTORY(. SRC)
   ADD_EXECUTABLE(taosdemo ${SRC})
-  
+
   IF (TD_SOMODE_STATIC)
     TARGET_LINK_LIBRARIES(taosdemo taos_static)
   ELSE ()
@@ -17,4 +17,13 @@ ELSEIF (TD_WINDOWS)
   AUX_SOURCE_DIRECTORY(. SRC)
   ADD_EXECUTABLE(taosdemo ${SRC})
   TARGET_LINK_LIBRARIES(taosdemo taos_static)
+ELSEIF (TD_DARWIN)
+  AUX_SOURCE_DIRECTORY(. SRC)
+  ADD_EXECUTABLE(taosdemo ${SRC})
+
+  IF (TD_SOMODE_STATIC)
+    TARGET_LINK_LIBRARIES(taosdemo taos_static)
+  ELSE ()
+    TARGET_LINK_LIBRARIES(taosdemo taos)
+  ENDIF ()
 ENDIF ()

src/kit/taosdemox/CMakeLists.txt

@@ -4,22 +4,44 @@ PROJECT(TDengine)
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/libcurl/include)
 
-IF (TD_LINUX)         
+IF (TD_LINUX)
   AUX_SOURCE_DIRECTORY(. SRC)
   ADD_EXECUTABLE(taosdemox ${SRC})
-  
-  #find_program(HAVE_CURL NAMES curl)   
-  IF ((NOT TD_ARM_64) AND (NOT TD_ARM_32))    
+
+  #find_program(HAVE_CURL NAMES curl)
+  IF ((NOT TD_ARM_64) AND (NOT TD_ARM_32))
     ADD_DEFINITIONS(-DTD_LOWA_CURL)
     LINK_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/libcurl/lib)
     ADD_LIBRARY(curl STATIC IMPORTED)
     SET_PROPERTY(TARGET curl PROPERTY IMPORTED_LOCATION ${TD_COMMUNITY_DIR}/deps/libcurl/lib/libcurl.a)
-    TARGET_LINK_LIBRARIES(taosdemox curl) 
+    TARGET_LINK_LIBRARIES(taosdemox curl)
   ENDIF ()
-  
+
   IF (TD_SOMODE_STATIC)
     TARGET_LINK_LIBRARIES(taosdemox taos_static cJson)
   ELSE ()
     TARGET_LINK_LIBRARIES(taosdemox taos cJson)
   ENDIF ()
 ENDIF ()
+
+IF (TD_DARWIN)
+  # missing a few dependencies, such as <argp.h>
+  # AUX_SOURCE_DIRECTORY(. SRC)
+  # ADD_EXECUTABLE(taosdemox ${SRC})
+  #
+  # #find_program(HAVE_CURL NAMES curl)
+  # IF ((NOT TD_ARM_64) AND (NOT TD_ARM_32))
+  #   ADD_DEFINITIONS(-DTD_LOWA_CURL)
+  #   LINK_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/libcurl/lib)
+  #   ADD_LIBRARY(curl STATIC IMPORTED)
+  #   SET_PROPERTY(TARGET curl PROPERTY IMPORTED_LOCATION ${TD_COMMUNITY_DIR}/deps/libcurl/lib/libcurl.a)
+  #   TARGET_LINK_LIBRARIES(taosdemox curl)
+  # ENDIF ()
+  #
+  # IF (TD_SOMODE_STATIC)
+  #   TARGET_LINK_LIBRARIES(taosdemox taos_static cJson)
+  # ELSE ()
+  #   TARGET_LINK_LIBRARIES(taosdemox taos cJson)
+  # ENDIF ()
+ENDIF ()
+

src/kit/taosdump/CMakeLists.txt

@@ -12,5 +12,15 @@ IF (TD_LINUX)
     TARGET_LINK_LIBRARIES(taosdump taos_static)
   ELSE ()
     TARGET_LINK_LIBRARIES(taosdump taos)
-  ENDIF ()  
+  ENDIF ()
+ENDIF ()
+
+IF (TD_DARWIN)
+  # missing <argp.h> for macosx
+  # ADD_EXECUTABLE(taosdump ${SRC})
+  # IF (TD_SOMODE_STATIC)
+  #   TARGET_LINK_LIBRARIES(taosdump taos_static)
+  # ELSE ()
+  #   TARGET_LINK_LIBRARIES(taosdump taos)
+  # ENDIF ()
 ENDIF ()

src/mnode/CMakeLists.txt

@@ -9,4 +9,15 @@ IF (TD_LINUX)
   AUX_SOURCE_DIRECTORY(src SRC)
 
   ADD_LIBRARY(mnode ${SRC})
-ENDIF ()
\ No newline at end of file
+ENDIF ()
+
+IF (TD_DARWIN)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/dnode/inc)
+
+  INCLUDE_DIRECTORIES(inc)
+  AUX_SOURCE_DIRECTORY(src SRC)
+
+  ADD_LIBRARY(mnode ${SRC})
+ENDIF ()
+

src/mnode/src/mnodeCluster.c

@@ -138,6 +138,14 @@ 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() {
   int32_t numOfClusters = sdbGetNumOfRows(tsClusterSdb);
   if (numOfClusters != 0) return TSDB_CODE_SUCCESS;

src/os/inc/eok.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _eok_h_fd274616_996c_400e_9023_ae70be881fa3_
+#define _eok_h_fd274616_996c_400e_9023_ae70be881fa3_
+
+#include <stdint.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#ifdef __APPLE__
+
+enum EPOLL_EVENTS
+  {
+    EPOLLIN = 0x001,
+#define EPOLLIN EPOLLIN
+    EPOLLPRI = 0x002,
+#define EPOLLPRI EPOLLPRI
+    EPOLLOUT = 0x004,
+#define EPOLLOUT EPOLLOUT
+    EPOLLRDNORM = 0x040,
+#define EPOLLRDNORM EPOLLRDNORM
+    EPOLLRDBAND = 0x080,
+#define EPOLLRDBAND EPOLLRDBAND
+    EPOLLWRNORM = 0x100,
+#define EPOLLWRNORM EPOLLWRNORM
+    EPOLLWRBAND = 0x200,
+#define EPOLLWRBAND EPOLLWRBAND
+    EPOLLMSG = 0x400,
+#define EPOLLMSG EPOLLMSG
+    EPOLLERR = 0x008,
+#define EPOLLERR EPOLLERR
+    EPOLLHUP = 0x010,
+#define EPOLLHUP EPOLLHUP
+    EPOLLRDHUP = 0x2000,
+#define EPOLLRDHUP EPOLLRDHUP
+    EPOLLEXCLUSIVE = 1u << 28,
+#define EPOLLEXCLUSIVE EPOLLEXCLUSIVE
+    EPOLLWAKEUP = 1u << 29,
+#define EPOLLWAKEUP EPOLLWAKEUP
+    EPOLLONESHOT = 1u << 30,
+#define EPOLLONESHOT EPOLLONESHOT
+    EPOLLET = 1u << 31
+#define EPOLLET EPOLLET
+  };
+
+/* Valid opcodes ( "op" parameter ) to issue to epoll_ctl().  */
+#define EPOLL_CTL_ADD 1 /* Add a file descriptor to the interface.  */
+#define EPOLL_CTL_DEL 2 /* Remove a file descriptor from the interface.  */
+#define EPOLL_CTL_MOD 3 /* Change file descriptor epoll_event structure.  */
+
+
+typedef union epoll_data
+{
+  void *ptr;
+  int fd;
+  uint32_t u32;
+  uint64_t u64;
+} epoll_data_t;
+
+struct epoll_event
+{
+  uint32_t events;  /* Epoll events */
+  epoll_data_t data;  /* User data variable */
+};
+
+int epoll_create(int size);
+int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
+int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
+int epoll_close(int epfd);
+
+#endif // __APPLE__
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // _eok_h_fd274616_996c_400e_9023_ae70be881fa3_
+

src/os/inc/osDarwin.h

@@ -75,11 +75,11 @@ extern "C" {
 #define TAOS_OS_FUNC_FILE_SENDIFLE
 
 #define TAOS_OS_FUNC_SEMPHONE
-  #define tsem_t dispatch_semaphore_t
-  int tsem_init(dispatch_semaphore_t *sem, int pshared, unsigned int value);
-  int tsem_wait(dispatch_semaphore_t *sem);
-  int tsem_post(dispatch_semaphore_t *sem);
-  int tsem_destroy(dispatch_semaphore_t *sem);
+  typedef struct tsem_s *tsem_t;
+  int tsem_init(tsem_t *sem, int pshared, unsigned int value);
+  int tsem_wait(tsem_t *sem);
+  int tsem_post(tsem_t *sem);
+  int tsem_destroy(tsem_t *sem);
 
 #define TAOS_OS_FUNC_SOCKET_SETSOCKETOPT
 #define TAOS_OS_FUNC_STRING_STR2INT64
@@ -91,7 +91,7 @@ extern "C" {
 typedef int(*__compar_fn_t)(const void *, const void *);
 
 // for send function in tsocket.c
-#define MSG_NOSIGNAL             0
+// #define MSG_NOSIGNAL             0
 #define SO_NO_CHECK              0x1234
 #define SOL_TCP                  0x1234
 #define TCP_KEEPIDLE             0x1234
@@ -100,6 +100,17 @@ typedef int(*__compar_fn_t)(const void *, const void *);
   #define  PTHREAD_MUTEX_RECURSIVE_NP PTHREAD_MUTEX_RECURSIVE
 #endif
 
+int64_t tsosStr2int64(char *str);
+
+#include "eok.h"
+
+void taos_block_sigalrm(void);
+
+
+
+
+
+
 #ifdef __cplusplus
 }
 #endif

src/os/inc/osDef.h

@@ -90,11 +90,12 @@ extern "C" {
 #ifdef _ISOC11_SOURCE
   #define threadlocal _Thread_local
 #elif defined(__APPLE__)
-  #define threadlocal
+  #define threadlocal __thread
 #elif defined(__GNUC__) && !defined(threadlocal)
   #define threadlocal __thread
 #else
-  #define threadlocal
+  // #define threadlocal
+  #error please follow with the thread-local implementation on the target platform 
 #endif
 
 #ifdef __cplusplus

src/os/src/darwin/darwinEnv.c

@@ -30,3 +30,4 @@ void osInit() {
   strcpy(tsScriptDir, "~/TDengine/cfg");
   strcpy(tsOsName, "Darwin");
 }
+

src/os/src/darwin/darwinSemphone.c

@@ -13,28 +13,269 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+// fail-fast or let-it-crash philosophy
+// https://en.wikipedia.org/wiki/Fail-fast
+// https://stackoverflow.com/questions/4393197/erlangs-let-it-crash-philosophy-applicable-elsewhere
+// experimentally, we follow log-and-crash here
+
 #define _DEFAULT_SOURCE
 #include "os.h"
 
-int tsem_init(dispatch_semaphore_t *sem, int pshared, unsigned int value) {
-  *sem = dispatch_semaphore_create(value);
-  if (*sem == NULL) {
+// #define SEM_USE_PTHREAD
+// #define SEM_USE_POSIX
+#define SEM_USE_SEM
+
+#ifdef SEM_USE_SEM
+#include <mach/mach_init.h>
+#include <mach/mach_error.h>
+#include <mach/semaphore.h>
+#include <mach/task.h>
+
+static pthread_t                 sem_thread;
+static pthread_once_t            sem_once;
+static task_t                    sem_port;
+static volatile int              sem_inited = 0;
+static semaphore_t               sem_exit;
+
+static void* sem_thread_routine(void *arg) {
+  (void)arg;
+  sem_port = mach_task_self();
+  kern_return_t ret = semaphore_create(sem_port, &sem_exit, SYNC_POLICY_FIFO, 0);
+  if (ret != KERN_SUCCESS) {
+    fprintf(stderr, "==%s[%d]%s()==failed to create sem_exit\n", basename(__FILE__), __LINE__, __func__);
+    sem_inited = -1;
+    return NULL;
+  }
+  sem_inited = 1;
+  semaphore_wait(sem_exit);
+  return NULL;
+}
+
+static void once_init(void) {
+  int r = 0;
+  r = pthread_create(&sem_thread, NULL, sem_thread_routine, NULL);
+  if (r) {
+    fprintf(stderr, "==%s[%d]%s()==failed to create thread\n", basename(__FILE__), __LINE__, __func__);
+    return;
+  }
+  while (sem_inited==0) {
+    ;
+  }
+}
+#endif
+
+struct tsem_s {
+#ifdef SEM_USE_PTHREAD
+  pthread_mutex_t              lock;
+  pthread_cond_t               cond;
+  volatile int64_t             val;
+#elif defined(SEM_USE_POSIX)
+  size_t                       id;
+  sem_t                       *sem;
+#elif defined(SEM_USE_SEM)
+  semaphore_t                  sem;
+#else // SEM_USE_PTHREAD
+  dispatch_semaphore_t         sem;
+#endif // SEM_USE_PTHREAD
+
+  volatile unsigned int        valid:1;
+};
+
+int tsem_init(tsem_t *sem, int pshared, unsigned int value) {
+  // fprintf(stderr, "==%s[%d]%s():[%p]==creating\n", basename(__FILE__), __LINE__, __func__, sem);
+  if (*sem) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==already initialized\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+  struct tsem_s *p = (struct tsem_s*)calloc(1, sizeof(*p));
+  if (!p) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==out of memory\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+
+#ifdef SEM_USE_PTHREAD
+  int r = pthread_mutex_init(&p->lock, NULL);
+  do {
+    if (r) break;
+    r = pthread_cond_init(&p->cond, NULL);
+    if (r) {
+      pthread_mutex_destroy(&p->lock);
+      break;
+    }
+    p->val = value;
+  } while (0);
+  if (r) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==not created\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+#elif defined(SEM_USE_POSIX)
+  static size_t tick = 0;
+  do {
+    size_t id = atomic_add_fetch_64(&tick, 1);
+    if (id==SEM_VALUE_MAX) {
+      atomic_store_64(&tick, 0);
+      id = 0;
+    }
+    char name[NAME_MAX-4];
+    snprintf(name, sizeof(name), "/t%ld", id);
+    p->sem = sem_open(name, O_CREAT|O_EXCL, pshared, value);
+    p->id  = id;
+    if (p->sem!=SEM_FAILED) break;
+    int e = errno;
+    if (e==EEXIST) continue;
+    if (e==EINTR) continue;
+    fprintf(stderr, "==%s[%d]%s():[%p]==not created[%d]%s\n", basename(__FILE__), __LINE__, __func__, sem, e, strerror(e));
+    abort();
+  } while (p->sem==SEM_FAILED);
+#elif defined(SEM_USE_SEM)
+  pthread_once(&sem_once, once_init);
+  if (sem_inited!=1) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal resource init failed\n", basename(__FILE__), __LINE__, __func__, sem);
+    errno = ENOMEM;
     return -1;
-  } else {
-    return 0;
   }
+  kern_return_t ret = semaphore_create(sem_port, &p->sem, SYNC_POLICY_FIFO, 0);
+  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
+    abort();
+  }
+#else // SEM_USE_PTHREAD
+  p->sem = dispatch_semaphore_create(value);
+  if (p->sem == NULL) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==not created\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+#endif // SEM_USE_PTHREAD
+
+  p->valid = 1;
+
+  *sem = p;
+
+  return 0;
 }
 
-int tsem_wait(dispatch_semaphore_t *sem) {
-  dispatch_semaphore_wait(*sem, DISPATCH_TIME_FOREVER);
+int tsem_wait(tsem_t *sem) {
+  if (!*sem) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==not initialized\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+  struct tsem_s *p = *sem;
+  if (!p->valid) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+#ifdef SEM_USE_PTHREAD
+  if (pthread_mutex_lock(&p->lock)) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+  p->val -= 1;
+  if (p->val < 0) {
+    if (pthread_cond_wait(&p->cond, &p->lock)) {
+      fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+      abort();
+    }
+  }
+  if (pthread_mutex_unlock(&p->lock)) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
   return 0;
+#elif defined(SEM_USE_POSIX)
+  return sem_wait(p->sem);
+#elif defined(SEM_USE_SEM)
+  return semaphore_wait(p->sem);
+#else // SEM_USE_PTHREAD
+  return dispatch_semaphore_wait(p->sem, DISPATCH_TIME_FOREVER);
+#endif // SEM_USE_PTHREAD
 }
 
-int tsem_post(dispatch_semaphore_t *sem) {
-  dispatch_semaphore_signal(*sem);
+int tsem_post(tsem_t *sem) {
+  if (!*sem) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==not initialized\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+  struct tsem_s *p = *sem;
+  if (!p->valid) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+#ifdef SEM_USE_PTHREAD
+  if (pthread_mutex_lock(&p->lock)) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+  p->val += 1;
+  if (p->val <= 0) {
+    if (pthread_cond_signal(&p->cond)) {
+      fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+      abort();
+    }
+  }
+  if (pthread_mutex_unlock(&p->lock)) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
   return 0;
+#elif defined(SEM_USE_POSIX)
+  return sem_post(p->sem);
+#elif defined(SEM_USE_SEM)
+  return semaphore_signal(p->sem);
+#else // SEM_USE_PTHREAD
+  return dispatch_semaphore_signal(p->sem);
+#endif // SEM_USE_PTHREAD
 }
 
-int tsem_destroy(dispatch_semaphore_t *sem) {
+int tsem_destroy(tsem_t *sem) {
+  // fprintf(stderr, "==%s[%d]%s():[%p]==destroying\n", basename(__FILE__), __LINE__, __func__, sem);
+  if (!*sem) {
+    // fprintf(stderr, "==%s[%d]%s():[%p]==not initialized\n", basename(__FILE__), __LINE__, __func__, sem);
+    // abort();
+    return 0;
+  }
+  struct tsem_s *p = *sem;
+  if (!p->valid) {
+    // fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", basename(__FILE__), __LINE__, __func__, sem);
+    // abort();
+    return 0;
+  }
+#ifdef SEM_USE_PTHREAD
+  if (pthread_mutex_lock(&p->lock)) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+  p->valid = 0;
+  if (pthread_cond_destroy(&p->cond)) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+  if (pthread_mutex_unlock(&p->lock)) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+  if (pthread_mutex_destroy(&p->lock)) {
+    fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem);
+    abort();
+  }
+#elif defined(SEM_USE_POSIX)
+  char name[NAME_MAX-4];
+  snprintf(name, sizeof(name), "/t%ld", p->id);
+  int r = sem_unlink(name);
+  if (r) {
+    int e = errno;
+    fprintf(stderr, "==%s[%d]%s():[%p]==unlink failed[%d]%s\n", basename(__FILE__), __LINE__, __func__, sem, e, strerror(e));
+    abort();
+  }
+#elif defined(SEM_USE_SEM)
+  semaphore_destroy(sem_port, p->sem);
+#else // SEM_USE_PTHREAD
+#endif // SEM_USE_PTHREAD
+
+  p->valid = 0;
+  free(p);
+
+  *sem = NULL;
   return 0;
 }
+

src/os/src/darwin/darwinTimer.c

@@ -13,9 +13,82 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+// fail-fast or let-it-crash philosophy
+// https://en.wikipedia.org/wiki/Fail-fast
+// https://stackoverflow.com/questions/4393197/erlangs-let-it-crash-philosophy-applicable-elsewhere
+// experimentally, we follow log-and-crash here
+
 #define _DEFAULT_SOURCE
 #include "os.h"
 
+#if 1
+#include <sys/event.h>
+
+static void (*timer_callback)(int);
+static int           timer_ms = 0;
+static pthread_t     timer_thread;
+static int           timer_kq = -1;
+static volatile int  timer_stop = 0;
+
+static void* timer_routine(void *arg) {
+  (void)arg;
+
+  int r = 0;
+  struct timespec to = {0};
+  to.tv_sec    = timer_ms / 1000;
+  to.tv_nsec   = (timer_ms % 1000) * 1000000;
+  while (!timer_stop) {
+    struct kevent64_s kev[10] = {0};
+    r = kevent64(timer_kq, NULL, 0, kev, sizeof(kev)/sizeof(kev[0]), 0, &to);
+    if (r!=0) {
+      fprintf(stderr, "==%s[%d]%s()==kevent64 failed\n", basename(__FILE__), __LINE__, __func__);
+      abort();
+    }
+    timer_callback(SIGALRM); // just mock
+  }
+
+  return NULL;
+}
+
+int taosInitTimer(void (*callback)(int), int ms) {
+  int r = 0;
+  timer_ms       = ms;
+  timer_callback = callback;
+
+  timer_kq = kqueue();
+  if (timer_kq==-1) {
+    fprintf(stderr, "==%s[%d]%s()==failed to create timer kq\n", basename(__FILE__), __LINE__, __func__);
+    // since no caller of this func checks the return value for the moment
+    abort();
+  }
+
+  r = pthread_create(&timer_thread, NULL, timer_routine, NULL);
+  if (r) {
+    fprintf(stderr, "==%s[%d]%s()==failed to create timer thread\n", basename(__FILE__), __LINE__, __func__);
+    // since no caller of this func checks the return value for the moment
+    abort();
+  }
+  return 0;
+}
+
+void taosUninitTimer() {
+  int r = 0;
+  timer_stop = 1;
+  r = pthread_join(timer_thread, NULL);
+  if (r) {
+    fprintf(stderr, "==%s[%d]%s()==failed to join timer thread\n", basename(__FILE__), __LINE__, __func__);
+    // since no caller of this func checks the return value for the moment
+    abort();
+  }
+  close(timer_kq);
+  timer_kq = -1;
+}
+
+void taos_block_sigalrm(void) {
+  // we don't know if there's any specific API for SIGALRM to deliver to specific thread
+  // this implementation relies on kqueue rather than SIGALRM
+}
+#else
 int taosInitTimer(void (*callback)(int), int ms) {
   signal(SIGALRM, callback);
 
@@ -34,3 +107,17 @@ void taosUninitTimer() {
   setitimer(ITIMER_REAL, &tv, NULL);
 }
 
+void taos_block_sigalrm(void) {
+  // since SIGALRM has been used
+  // consideration: any better solution?
+  static __thread int already_set = 0;
+  if (!already_set) {
+    sigset_t set;
+    sigemptyset(&set);
+    sigaddset(&set, SIGALRM);
+    pthread_sigmask(SIG_BLOCK, &set, NULL);
+    already_set = 1;
+  }
+}
+#endif
+

src/os/src/detail/CMakeLists.txt

@@ -13,3 +13,4 @@ TARGET_LINK_LIBRARIES(osdetail os)
 IF (TD_ARM_32 OR TD_LINUX_32)
   TARGET_LINK_LIBRARIES(osdetail atomic)
 ENDIF ()
+

src/os/src/detail/osSocket.c

@@ -73,4 +73,4 @@ const char *taosInetNtoa(struct in_addr ipInt) {
   return inet_ntoa(ipInt);
 }
 
-#endif
\ No newline at end of file
+#endif

src/os/src/detail/osTimer.c

@@ -116,6 +116,9 @@ void taosUninitTimer() {
   pthread_sigmask(SIG_BLOCK, &set, NULL);
 */
 void taosMsleep(int mseconds) {
+#ifdef __APPLE__
+  taos_block_sigalrm();
+#endif // __APPLE__
 #if 1
   usleep(mseconds * 1000);
 #else
@@ -136,6 +139,7 @@ void taosMsleep(int mseconds) {
 
 /* pthread_sigmask(SIG_UNBLOCK, &set, NULL); */
 #endif
+
 }
 
-#endif
\ No newline at end of file
+#endif

src/plugins/http/CMakeLists.txt

@@ -8,14 +8,29 @@ INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
 INCLUDE_DIRECTORIES(inc)
 AUX_SOURCE_DIRECTORY(src SRC)
-  
+
 IF (TD_LINUX)
   ADD_LIBRARY(http ${SRC})
   TARGET_LINK_LIBRARIES(http z)
 
   IF (TD_SOMODE_STATIC)
     TARGET_LINK_LIBRARIES(http taos_static)
-  ELSE ()  
+  ELSE ()
+    TARGET_LINK_LIBRARIES(http taos)
+  ENDIF ()
+
+  IF (TD_ADMIN)
+    TARGET_LINK_LIBRARIES(http admin)
+  ENDIF ()
+ENDIF ()
+
+IF (TD_DARWIN)
+  ADD_LIBRARY(http ${SRC})
+  TARGET_LINK_LIBRARIES(http z)
+
+  IF (TD_SOMODE_STATIC)
+    TARGET_LINK_LIBRARIES(http taos_static)
+  ELSE ()
     TARGET_LINK_LIBRARIES(http taos)
   ENDIF ()
 

src/plugins/http/src/httpServer.c

@@ -31,11 +31,36 @@
 
 static bool httpReadData(HttpContext *pContext);
 
+#ifdef __APPLE__
+static int sv_dummy = 0;
+#endif // __APPLE__
+
 static void httpStopThread(HttpThread* pThread) {
   pThread->stop = true;
 
   // signal the thread to stop, try graceful method first,
   // and use pthread_cancel when failed
+#ifdef __APPLE__
+  int sv[2];
+  sv[0] = sv[1] = -1;
+  int r = socketpair(PF_LOCAL, SOCK_STREAM, 0, sv);
+  do {
+    if (r) break;
+    struct epoll_event ev = {0};
+    ev.events = EPOLLIN;
+    ev.data.ptr = &sv_dummy;
+    pThread->stop = true;
+    r = epoll_ctl(pThread->pollFd, EPOLL_CTL_ADD, sv[0], &ev);
+    if (r) break;
+    if (1!=send(sv[1], "1", 1, 0)) {
+      r = -1;
+      break;
+    }
+  } while (0);
+  if (r) {
+    pthread_cancel(pThread->thread);
+  }
+#else // __APPLE__
   struct epoll_event event = { .events = EPOLLIN };
   eventfd_t fd = eventfd(1, 0);
   if (fd == -1) {
@@ -46,13 +71,29 @@ static void httpStopThread(HttpThread* pThread) {
     httpError("%s, failed to call epoll_ctl, will call pthread_cancel instead, which may result in data corruption: %s", pThread->label, strerror(errno));
     pthread_cancel(pThread->thread);
   }
+#endif // __APPLE__
 
   pthread_join(pThread->thread, NULL);
+#ifdef __APPLE__
+  if (sv[0]!=-1) {
+    close(sv[0]);
+    sv[0] = -1;
+  }
+  if (sv[1]!=-1) {
+    close(sv[1]);
+    sv[1] = -1;
+  }
+#else // __APPLE__
   if (fd != -1) {
     close(fd);
   }
+#endif // __APPLE__
 
+#ifdef __APPLE__
+  epoll_close(pThread->pollFd);
+#else
   close(pThread->pollFd);
+#endif
   pthread_mutex_destroy(&(pThread->threadMutex));
 }
 
@@ -96,6 +137,15 @@ static void httpProcessHttpData(void *param) {
     if (fdNum <= 0) continue;
 
     for (int32_t i = 0; i < fdNum; ++i) {
+#ifdef __APPLE__
+      if (events[i].data.ptr == &sv_dummy) {
+        // no need to drain the recv buffer of sv[0]
+        // since there's only one time to send at most 1 byte to sv[0]
+        // btw, pThread->stop shall be already set, thus never reached here
+        httpDebug("if you see this line, there's internal logic error");
+        continue;
+      }
+#endif // __APPLE__
       pContext = httpGetContext(events[i].data.ptr);
       if (pContext == NULL) {
         httpError("context:%p, is already released, close connect", events[i].data.ptr);

src/plugins/monitor/CMakeLists.txt

@@ -8,10 +8,20 @@ AUX_SOURCE_DIRECTORY(./src SRC)
 
 IF (TD_LINUX)
   ADD_LIBRARY(monitor ${SRC})
-  
+
+  IF (TD_SOMODE_STATIC)
+    TARGET_LINK_LIBRARIES(monitor taos_static)
+  ELSE ()
+    TARGET_LINK_LIBRARIES(monitor taos)
+  ENDIF ()
+ENDIF ()
+
+IF (TD_DARWIN)
+  ADD_LIBRARY(monitor ${SRC})
+
   IF (TD_SOMODE_STATIC)
     TARGET_LINK_LIBRARIES(monitor taos_static)
   ELSE ()
     TARGET_LINK_LIBRARIES(monitor taos)
   ENDIF ()
-ENDIF ()  
+ENDIF ()

src/plugins/mqtt/CMakeLists.txt

@@ -18,3 +18,15 @@ IF (TD_LINUX)
     TARGET_LINK_LIBRARIES(mqtt taos)
   ENDIF ()
 ENDIF ()
+
+IF (TD_DARWIN)
+  ADD_LIBRARY(mqtt ${SRC})
+  TARGET_LINK_LIBRARIES(mqtt cJson mqttc)
+
+  IF (TD_SOMODE_STATIC)
+    TARGET_LINK_LIBRARIES(mqtt taos_static)
+  ELSE ()
+    TARGET_LINK_LIBRARIES(mqtt taos)
+  ENDIF ()
+ENDIF ()
+

src/query/CMakeLists.txt

@@ -14,3 +14,8 @@ IF (TD_LINUX)
   TARGET_LINK_LIBRARIES(query m rt)
   ADD_SUBDIRECTORY(tests)
 ENDIF ()
+
+IF (TD_DARWIN)
+  TARGET_LINK_LIBRARIES(query m)
+  ADD_SUBDIRECTORY(tests)
+ENDIF ()

src/rpc/src/rpcTcp.c

@@ -307,7 +307,14 @@ void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *
   int code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
   pthread_attr_destroy(&thattr);
   if (code != 0) {
+#ifdef __APPLE__
+    if (pThreadObj->pollFd!=-1) {
+      epoll_close(pThreadObj->pollFd);
+      pThreadObj->pollFd = -1;
+    }
+#else // __APPLE__
     taosCloseSocket(pThreadObj->pollFd);
+#endif // __APPLE__
     free(pThreadObj);
     terrno = TAOS_SYSTEM_ERROR(errno); 
     tError("%s failed to create TCP read data thread(%s)", label, strerror(errno));
@@ -470,7 +477,10 @@ static void *taosProcessTcpData(void *param) {
   SFdObj            *pFdObj;
   struct epoll_event events[maxEvents];
   SRecvInfo          recvInfo;
- 
+
+#ifdef __APPLE__
+  taos_block_sigalrm();
+#endif // __APPLE__
   while (1) {
     int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, TAOS_EPOLL_WAIT_TIME);
     if (pThreadObj->stop) {
@@ -512,7 +522,14 @@ static void *taosProcessTcpData(void *param) {
     if (pThreadObj->stop) break; 
   }
 
+#ifdef __APPLE__
+  if (pThreadObj->pollFd >=0) {
+    epoll_close(pThreadObj->pollFd);
+    pThreadObj->pollFd = -1;
+  }
+#else // __APPLE__
   if (pThreadObj->pollFd >=0) taosCloseSocket(pThreadObj->pollFd);
+#endif // __APPLE__
 
   while (pThreadObj->pHead) {
     SFdObj *pFdObj = pThreadObj->pHead;

src/rpc/test/CMakeLists.txt

@@ -16,3 +16,17 @@ IF (TD_LINUX)
   ADD_EXECUTABLE(rserver ${SERVER_SRC})
   TARGET_LINK_LIBRARIES(rserver trpc)
 ENDIF ()
+
+IF (TD_DARWIN)
+  LIST(APPEND CLIENT_SRC ./rclient.c)
+  ADD_EXECUTABLE(rclient ${CLIENT_SRC})
+  TARGET_LINK_LIBRARIES(rclient trpc)
+
+  LIST(APPEND SCLIENT_SRC ./rsclient.c)
+  ADD_EXECUTABLE(rsclient ${SCLIENT_SRC})
+  TARGET_LINK_LIBRARIES(rsclient trpc)
+
+  LIST(APPEND SERVER_SRC ./rserver.c)
+  ADD_EXECUTABLE(rserver ${SERVER_SRC})
+  TARGET_LINK_LIBRARIES(rserver trpc)
+ENDIF ()

src/sync/CMakeLists.txt

@@ -3,7 +3,7 @@ PROJECT(TDengine)
 
 INCLUDE_DIRECTORIES(inc)
 AUX_SOURCE_DIRECTORY(src SRC)
-  
+
 IF (TD_LINUX)
   LIST(REMOVE_ITEM SRC src/syncArbitrator.c)
   ADD_LIBRARY(sync ${SRC})
@@ -16,3 +16,16 @@ IF (TD_LINUX)
 
   #ADD_SUBDIRECTORY(test)
 ENDIF ()
+
+IF (TD_DARWIN)
+  LIST(REMOVE_ITEM SRC src/syncArbitrator.c)
+  ADD_LIBRARY(sync ${SRC})
+  TARGET_LINK_LIBRARIES(sync tutil pthread common)
+
+  LIST(APPEND BIN_SRC src/syncArbitrator.c)
+  LIST(APPEND BIN_SRC src/syncTcp.c)
+  ADD_EXECUTABLE(tarbitrator ${BIN_SRC})
+  TARGET_LINK_LIBRARIES(tarbitrator sync common osdetail tutil)
+
+  #ADD_SUBDIRECTORY(test)
+ENDIF ()

src/sync/src/syncRetrieve.c

@@ -14,7 +14,7 @@
  */
 
 #define _DEFAULT_SOURCE
-#include <sys/inotify.h>
+// #include <sys/inotify.h>
 #include "os.h"
 #include "taoserror.h"
 #include "tlog.h"

src/sync/src/syncTcp.c

@@ -233,7 +233,11 @@ static void *syncProcessTcpData(void *param) {
 
   sDebug("%p TCP epoll thread exits", pThread);
 
+#ifdef __APPLE__
+  epoll_close(pThread->pollFd);
+#else // __APPLE__
   close(pThread->pollFd);
+#endif // __APPLE__
   tfree(pThread);
   tfree(buffer);
   return NULL;
@@ -290,7 +294,11 @@ static SThreadObj *syncGetTcpThread(SPoolObj *pPool) {
   pthread_attr_destroy(&thattr);
 
   if (ret != 0) {
+#ifdef __APPLE__
+    epoll_close(pThread->pollFd);
+#else // __APPLE__
     close(pThread->pollFd);
+#endif // __APPLE__
     tfree(pThread);
     return NULL;
   }

src/sync/test/CMakeLists.txt

@@ -13,4 +13,3 @@ IF (TD_LINUX)
   TARGET_LINK_LIBRARIES(syncServer sync trpc common)
 ENDIF ()
 
-

src/tsdb/inc/tsdbMain.h

@@ -233,7 +233,11 @@ typedef struct {
   SMemTable*      mem;
   SMemTable*      imem;
   STsdbFileH*     tsdbFileH;
+#ifdef __APPLE__
+  sem_t          *readyToCommit;
+#else // __APPLE__
   sem_t           readyToCommit;
+#endif // __APPLE__
   pthread_mutex_t mutex;
   bool            repoLocked;
   int32_t         code; // Commit code
@@ -616,4 +620,4 @@ int tsdbScheduleCommit(STsdbRepo *pRepo);
 }
 #endif
 
-#endif
\ No newline at end of file
+#endif

src/tsdb/src/tsdbCommit.c

@@ -166,7 +166,11 @@ static void tsdbEndCommit(STsdbRepo *pRepo, int eno) {
   pRepo->imem = NULL;
   tsdbUnlockRepo(pRepo);
   tsdbUnRefMemTable(pRepo, pIMem);
+#ifdef __APPLE__
+  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,7 +146,11 @@ int tsdbCloseRepo(TSDB_REPO_T *repo, int toCommit) {
 
   if (toCommit) {
     tsdbAsyncCommit(pRepo);
+#ifdef __APPLE__
+    sem_wait(pRepo->readyToCommit);
+#else // __APPLE__
     sem_wait(&(pRepo->readyToCommit));
+#endif // __APPLE__
     terrno = pRepo->code;
   }
   tsdbUnRefMemTable(pRepo, pRepo->mem);
@@ -643,11 +647,21 @@ static STsdbRepo *tsdbNewRepo(char *rootDir, STsdbAppH *pAppH, STsdbCfg *pCfg) {
     goto _err;
   }
 
+#ifdef __APPLE__
+  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;
 
@@ -693,7 +707,11 @@ static void tsdbFreeRepo(STsdbRepo *pRepo) {
     // tsdbFreeMemTable(pRepo->mem);
     // tsdbFreeMemTable(pRepo->imem);
     tfree(pRepo->rootDir);
+#ifdef __APPLE__
+    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,7 +207,11 @@ void *tsdbAllocBytes(STsdbRepo *pRepo, int bytes) {
 int tsdbAsyncCommit(STsdbRepo *pRepo) {
   if (pRepo->mem == NULL) return 0;
 
+#ifdef __APPLE__
+  sem_wait(pRepo->readyToCommit);
+#else // __APPLE__
   sem_wait(&(pRepo->readyToCommit));
+#endif // __APPLE__
 
   ASSERT(pRepo->imem == NULL);
 
@@ -229,8 +233,13 @@ int tsdbSyncCommit(TSDB_REPO_T *repo) {
   STsdbRepo *pRepo = (STsdbRepo *)repo;
 
   tsdbAsyncCommit(pRepo);
+#ifdef __APPLE__
+  sem_wait(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;
@@ -927,4 +936,4 @@ static int tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SDataRow
   }
 
   return 0;
-}
\ No newline at end of file
+}

src/util/CMakeLists.txt

@@ -5,7 +5,7 @@ INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/rpc/inc)
 AUX_SOURCE_DIRECTORY(src SRC)
 ADD_LIBRARY(tutil ${SRC})
 TARGET_LINK_LIBRARIES(tutil pthread osdetail lz4 z)
-  
+
 IF (TD_LINUX)
   TARGET_LINK_LIBRARIES(tutil m rt)
   # ADD_SUBDIRECTORY(tests)
@@ -28,5 +28,6 @@ IF (TD_LINUX)
 ELSEIF (TD_WINDOWS)
   TARGET_LINK_LIBRARIES(tutil iconv regex winmm IPHLPAPI ws2_32 wepoll)
 ELSEIF(TD_DARWIN)
+  TARGET_LINK_LIBRARIES(tutil m)
   TARGET_LINK_LIBRARIES(tutil iconv)
 ENDIF()

src/util/src/tsocket.c

@@ -354,6 +354,8 @@ int32_t taosKeepTcpAlive(SOCKET sockFd) {
     return -1;
   }
 
+#ifndef __APPLE__
+  // all fails on macosx
   int32_t probes = 3;
   if (taosSetSockOpt(sockFd, SOL_TCP, TCP_KEEPCNT, (void *)&probes, sizeof(probes)) < 0) {
     uError("fd:%d setsockopt SO_KEEPCNT failed: %d (%s)", sockFd, errno, strerror(errno));
@@ -374,6 +376,7 @@ int32_t taosKeepTcpAlive(SOCKET sockFd) {
     taosCloseSocket(sockFd);
     return -1;
   }
+#endif // __APPLE__
 
   int32_t nodelay = 1;
   if (taosSetSockOpt(sockFd, IPPROTO_TCP, TCP_NODELAY, (void *)&nodelay, sizeof(nodelay)) < 0) {

src/vnode/CMakeLists.txt

@@ -13,3 +13,9 @@ IF (TD_LINUX)
   ADD_LIBRARY(vnode ${SRC})
   TARGET_LINK_LIBRARIES(vnode tsdb tcq)
 ENDIF ()
+
+IF (TD_DARWIN)
+  ADD_LIBRARY(vnode ${SRC})
+  TARGET_LINK_LIBRARIES(vnode tsdb tcq)
+ENDIF ()
+

src/wal/CMakeLists.txt

@@ -8,4 +8,10 @@ IF (TD_LINUX)
   ADD_LIBRARY(twal ${SRC})
   TARGET_LINK_LIBRARIES(twal tutil common)
   ADD_SUBDIRECTORY(test)
-ENDIF ()  
+ENDIF ()
+
+IF (TD_DARWIN)
+  ADD_LIBRARY(twal ${SRC})
+  TARGET_LINK_LIBRARIES(twal tutil common)
+  ADD_SUBDIRECTORY(test)
+ENDIF ()

src/wal/test/CMakeLists.txt

@@ -10,4 +10,12 @@ IF (TD_LINUX)
 
 ENDIF ()
 
+IF (TD_DARWIN)
+  INCLUDE_DIRECTORIES(../inc)
+
+  LIST(APPEND WALTEST_SRC ./waltest.c)
+  ADD_EXECUTABLE(waltest ${WALTEST_SRC})
+  TARGET_LINK_LIBRARIES(waltest twal osdetail tutil)
+
+ENDIF ()
 

tests/comparisonTest/tdengine/CMakeLists.txt

@@ -5,3 +5,8 @@ IF (TD_LINUX)
   add_executable(tdengineTest tdengineTest.c)
   target_link_libraries(tdengineTest taos_static tutil common pthread)
 ENDIF()
+
+IF (TD_DARWIN)
+  add_executable(tdengineTest tdengineTest.c)
+  target_link_libraries(tdengineTest taos_static tutil common pthread)
+ENDIF()

tests/examples/c/CMakeLists.txt

@@ -5,4 +5,14 @@ IF (TD_LINUX)
   AUX_SOURCE_DIRECTORY(. SRC)
   ADD_EXECUTABLE(demo apitest.c)
   TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread )
+  ADD_EXECUTABLE(epoll epoll.c)
+  TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread )
+ENDIF ()
+IF (TD_DARWIN)
+  INCLUDE_DIRECTORIES(. ${TD_COMMUNITY_DIR}/src/inc ${TD_COMMUNITY_DIR}/src/client/inc  ${TD_COMMUNITY_DIR}/inc)
+  AUX_SOURCE_DIRECTORY(. SRC)
+  ADD_EXECUTABLE(demo demo.c)
+  TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread )
+  ADD_EXECUTABLE(epoll epoll.c)
+  TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread )
 ENDIF ()

tests/examples/c/demo.c

@@ -86,7 +86,7 @@ void Test(TAOS *taos, char *qstr, int index)  {
 
   int i = 0;
   for (i = 0; i < 10; ++i) {
-    sprintf(qstr, "insert into m1 values (%" PRId64 ", %d, %d, %d, %d, %f, %lf, '%s')", 1546300800000 + i * 1000, i, i, i, i*10000000, i*1.0, i*2.0, "hello");
+    sprintf(qstr, "insert into m1 values (%" PRId64 ", %d, %d, %d, %d, %f, %lf, '%s')", (uint64_t)(1546300800000 + i * 1000), i, i, i, i*10000000, i*1.0, i*2.0, "hello");
     printf("qstr: %s\n", qstr);
     
     // note: how do you wanna do if taos_query returns non-NULL

tests/examples/c/epoll.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifdef __APPLE__
+#include "eok.h"
+#else // __APPLE__
+#include <sys/epoll.h>
+#endif // __APPLE__
+#include <sys/types.h>
+#include <sys/time.h>
+#include <sys/socket.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <string.h>
+#include <arpa/inet.h>
+#include <libgen.h>
+
+#define D(fmt, ...) fprintf(stderr, "%s[%d]%s(): " fmt "\n", basename(__FILE__), __LINE__, __func__, ##__VA_ARGS__)
+#define A(statement, fmt, ...) do {                                     \
+  if (statement) break;                                                 \
+  fprintf(stderr, "%s[%d]%s(): assert [%s] failed: %d[%s]: " fmt "\n",  \
+          basename(__FILE__), __LINE__, __func__,                       \
+          #statement, errno, strerror(errno),                           \
+          ##__VA_ARGS__);                                               \
+  abort();                                                              \
+} while (0)
+
+#define E(fmt, ...) do {                                                \
+  fprintf(stderr, "%s[%d]%s(): %d[%s]: " fmt "\n",                      \
+          basename(__FILE__), __LINE__, __func__,                       \
+          errno, strerror(errno),                                       \
+          ##__VA_ARGS__);                                               \
+} while (0)
+
+typedef struct ep_s            ep_t;
+struct ep_s {
+  int                    ep;
+
+  pthread_mutex_t        lock;
+  int                    sv[2];  // 0 for read, 1 for write;
+  pthread_t              thread;
+
+  volatile unsigned int  stopping:1;
+  volatile unsigned int  waiting:1;
+  volatile unsigned int  wakenup:1;
+};
+
+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;
+struct client_s {
+  int                          skt;
+  void (*on_event)(ep_t *ep, struct epoll_event *events, client_t *client);
+  volatile unsigned int        state; // 1: listenning; 2: connected
+};
+
+static void echo_event(ep_t *ep, struct epoll_event *ev, client_t *client);
+
+int main(int argc, char *argv[]) {
+  ep_t* ep = ep_create();
+  A(ep, "failed");
+  int skt = open_connect(6789);
+  if (skt!=-1) {
+    client_t *client = (client_t*)calloc(1, sizeof(*client));
+    if (client) {
+      client->skt = skt;
+      client->on_event = echo_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), "");
+    }
+  }
+  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;
+  while ((linelen = getline(&line, &linecap, stdin)) > 0) {
+    line[strlen(line)-1] = '\0';
+    if (0==strcmp(line, "exit")) break;
+    if (0==strcmp(line, "quit")) break;
+    if (line==strstr(line, "close")) {
+      int fd = 0;
+      sscanf(line, "close %d", &fd);
+      if (fd<=2) {
+        fprintf(stderr, "fd [%d] invalid\n", fd);
+        continue;
+      }
+      A(0==epoll_ctl(ep->ep, EPOLL_CTL_DEL, fd, NULL), "");
+      continue;
+    }
+    if (strlen(line)==0) continue;
+    fprintf(stderr, "unknown cmd:[%s]\n", line);
+  }
+  ep_destroy(ep);
+  D("");
+  return 0;
+}
+
+ep_t* ep_create(void) {
+  ep_t *ep = (ep_t*)calloc(1, sizeof(*ep));
+  A(ep, "out of memory");
+  A(-1!=(ep->ep = epoll_create(1)), "");
+  ep->sv[0] = -1;
+  ep->sv[1] = -1;
+  A(0==socketpair(AF_LOCAL, SOCK_STREAM, 0, ep->sv), "");
+  A(0==pthread_mutex_init(&ep->lock, NULL), "");
+  A(0==pthread_mutex_lock(&ep->lock), "");
+  struct epoll_event ev = {0};
+  ev.events   = EPOLLIN;
+  ev.data.ptr = &ep_dummy;
+  A(0==epoll_ctl(ep->ep, EPOLL_CTL_ADD, ep->sv[0], &ev), "");
+  A(0==pthread_create(&ep->thread, NULL, routine, ep), "");
+  A(0==pthread_mutex_unlock(&ep->lock), "");
+  return ep;
+}
+
+static void ep_destroy(ep_t *ep) {
+  A(ep, "invalid argument");
+  ep->stopping = 1;
+  A(1==send(ep->sv[1], "1", 1, 0), "");
+  A(0==pthread_join(ep->thread, NULL), "");
+  A(0==pthread_mutex_destroy(&ep->lock), "");
+  A(0==close(ep->sv[0]), "");
+  A(0==close(ep->sv[1]), "");
+  A(0==close(ep->ep), "");
+  free(ep);
+}
+
+static void* routine(void* arg) {
+  A(arg, "invalid argument");
+  ep_t *ep = (ep_t*)arg;
+
+  while (!ep->stopping) {
+    struct epoll_event evs[10];
+    memset(evs, 0, sizeof(evs));
+
+    A(0==pthread_mutex_lock(&ep->lock), "");
+    A(ep->waiting==0, "internal logic error");
+    ep->waiting = 1;
+    A(0==pthread_mutex_unlock(&ep->lock), "");
+
+    int r = epoll_wait(ep->ep, evs, sizeof(evs)/sizeof(evs[0]), -1);
+    A(r>0, "indefinite epoll_wait shall not timeout:[%d]", r);
+
+    A(0==pthread_mutex_lock(&ep->lock), "");
+    A(ep->waiting==1, "internal logic error");
+    ep->waiting = 0;
+    A(0==pthread_mutex_unlock(&ep->lock), "");
+
+    for (int i=0; i<r; ++i) {
+      struct epoll_event *ev = evs + i;
+      if (ev->data.ptr == &ep_dummy) {
+        char c = '\0';
+        A(1==recv(ep->sv[0], &c, 1, 0), "internal logic error");
+        A(0==pthread_mutex_lock(&ep->lock), "");
+        ep->wakenup = 0;
+        A(0==pthread_mutex_unlock(&ep->lock), "");
+        D("........");
+        continue;
+      }
+      A(ev->data.ptr, "internal logic error");
+      client_t *client = (client_t*)ev->data.ptr;
+      client->on_event(ep, ev, client);
+      continue;
+    }
+  }
+  return NULL;
+}
+
+static int open_listen(unsigned short port) {
+  int r = 0;
+  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 = bind(skt, (struct sockaddr*)&si, sizeof(si));
+    if (r) {
+      E("bind(%u) failed", port);
+      break;
+    }
+    r = listen(skt, 100);
+    if (r) {
+      E("listen() failed");
+      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("listenning at: %d", ntohs(si.sin_port));
+    return skt;
+  } while (0);
+  close(skt);
+  return -1;
+}
+
+static int open_connect(unsigned short port) {
+  int r = 0;
+  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) {
+        client_t *server = (client_t*)calloc(1, sizeof(*server));
+        if (server) {
+          server->skt = skt;
+          server->on_event = echo_event;
+          server->state = 2;
+          struct epoll_event ev = {0};
+          ev.events   = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP;
+          ev.data.ptr = server;
+          A(0==epoll_ctl(ep->ep, EPOLL_CTL_ADD, skt, &ev), "");
+        }
+      }
+    }
+    if (client->state==2) {
+      char buf[4];
+      int n = recv(client->skt, buf, sizeof(buf)-1, 0);
+      A(n>=0 && n<sizeof(buf), "internal logic error:[%d]", n);
+      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), "");
+    A(0==epoll_ctl(ep->ep, EPOLL_CTL_DEL, client->skt, NULL), "");
+    A(0==pthread_mutex_unlock(&ep->lock), "");
+    close(client->skt);
+    client->skt = -1;
+    client->on_event = NULL;
+    free(client);
+  }
+}
+