semaphore test

include/os/osSemaphore.h

@@ -33,6 +33,17 @@ int tsem_timewait(tsem_t *sim, int64_t milis);
 int tsem_post(tsem_t *sem);
 int tsem_destroy(tsem_t *sem);
 
+#elif defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
+#include <windows.h>
+
+#define tsem_t HANDLE
+
+int tsem_init(tsem_t *sem, int pshared, unsigned int value);
+int tsem_wait(tsem_t *sem);
+int tsem_timewait(tsem_t *sim, int64_t milis);
+int tsem_post(tsem_t *sem);
+int tsem_destroy(tsem_t *sem);
+
 #else
 
 #define tsem_t       sem_t

source/os/src/osSemaphore.c

@@ -68,37 +68,34 @@ int32_t taosGetAppName(char* name, int32_t* len) {
 }
 
 int32_t tsem_wait(tsem_t* sem) {
-  int ret = 0;
-  do {
-    ret = sem_wait(sem);
-  } while (ret != 0 && errno == EINTR);
-  return ret;
+  return WaitForSingleObject(*sem, INFINITE);
 }
 
-int32_t tsem_timewait(tsem_t* sem, int64_t ms) {
-  struct timespec ts;
-  taosClockGetTime(0, &ts);
+int32_t tsem_timewait(tsem_t* sem, int64_t timeout_ms) {
+  DWORD result = WaitForSingleObject(*sem, timeout_ms);
+  if (result == WAIT_OBJECT_0) {
+    return 0;  // Semaphore acquired
+  } else if (result == WAIT_TIMEOUT) {
+    return -1;  // Timeout reached
+  } else {
+    return result;
+  }
+}
 
-  ts.tv_nsec += ms * 1000000;
-  ts.tv_sec += ts.tv_nsec / 1000000000;
-  ts.tv_nsec %= 1000000000;
-  int rc;
-  while ((rc = sem_timedwait(sem, &ts)) == -1 && errno == EINTR) continue;
-  return rc;
-  /* This should have timed out */
-  // ASSERT(errno == ETIMEDOUT);
-  // ASSERT(rc != 0);
-  // GetSystemTimeAsFileTime(&ft_after);
-  // // We specified a non-zero wait. Time must advance.
-  // if (ft_before.dwLowDateTime == ft_after.dwLowDateTime && ft_before.dwHighDateTime == ft_after.dwHighDateTime)
-  //   {
-  //     printf("nanoseconds: %d, rc: %d, code:0x%x. before filetime: %d, %d; after filetime: %d, %d\n",
-  //         nanosecs, rc, errno,
-  //         (int)ft_before.dwLowDateTime, (int)ft_before.dwHighDateTime,
-  //         (int)ft_after.dwLowDateTime, (int)ft_after.dwHighDateTime);
-  //     printf("time must advance during sem_timedwait.");
-  //     return 1;
-  //   }
+// Inter-process sharing is not currently supported. The pshared parameter is invalid.
+int tsem_init(tsem_t* sem, int pshared, unsigned int value) {
+  *sem = CreateSemaphore(NULL, value, LONG_MAX, NULL);
+  return (*sem != NULL) ? 0 : -1;
+}
+
+int tsem_post(tsem_t* sem) {
+  if (ReleaseSemaphore(*sem, 1, NULL)) return 0;
+  return -1;
+}
+
+int tsem_destroy(tsem_t* sem) {
+  if (CloseHandle(*sem)) return 0;
+  return -1;
 }
 
 #elif defined(_TD_DARWIN_64)

source/os/test/CMakeLists.txt

@@ -77,4 +77,11 @@ target_link_libraries(osAtomicTests os util gtest_main)
 add_test(
     NAME osAtomicTests
     COMMAND osAtomicTests
-)
+)
+
+add_executable(osSemaphoreTests "osSemaphoreTests.cpp")
+target_link_libraries(osSemaphoreTests os util gtest_main)
+add_test(
+    NAME osSemaphoreTests
+    COMMAND osSemaphoreTests
+)

source/os/test/osSemaphoreTests.cpp

@@ -0,0 +1,107 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <xsren@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/>.
+ */
+
+#include <gtest/gtest.h>
+#include <thread>
+#include "os.h"
+#include "tlog.h"
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wwrite-strings"
+#pragma GCC diagnostic ignored "-Wunused-function"
+#pragma GCC diagnostic ignored "-Wunused-variable"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+#pragma GCC diagnostic ignored "-Wformat"
+#pragma GCC diagnostic ignored "-Wint-to-pointer-cast"
+#pragma GCC diagnostic ignored "-Wpointer-arith"
+
+TEST(osSemaphoreTests, InitAndDestroy) {
+  tsem_t sem;
+  int    result = tsem_init(&sem, 0, 1);
+  EXPECT_EQ(result, 0);
+
+  result = tsem_destroy(&sem);
+  EXPECT_EQ(result, 0);
+}
+
+TEST(osSemaphoreTests, Destroy) {
+  tsem_t sem;
+  int    result = tsem_init(&sem, 0, 1);
+  EXPECT_EQ(result, 0);
+
+  result = tsem_destroy(&sem);
+  EXPECT_EQ(result, 0);
+  result = tsem_destroy(&sem);
+  EXPECT_NE(result, 0);
+}
+
+// TEST(osSemaphoreTests, Wait) {
+//   tsem_t sem;
+//   tsem_init(&sem, 0, 0);
+//   ASSERT_EQ(tsem_wait(&sem), -1);
+//   tsem_destroy(&sem);
+// }
+
+TEST(osSemaphoreTests, WaitTime0) {
+  tsem_t sem;
+  tsem_init(&sem, 0, 0);
+  EXPECT_NE(tsem_timewait(&sem, 1000), 0);
+  tsem_destroy(&sem);
+}
+
+TEST(osSemaphoreTests, WaitTime1) {
+  tsem_t sem;
+  tsem_init(&sem, 0, 1);
+  EXPECT_EQ(tsem_timewait(&sem, 1000), 0);
+  EXPECT_NE(tsem_timewait(&sem, 1000), 0);
+  tsem_destroy(&sem);
+}
+
+
+TEST(osSemaphoreTests, WaitAndPost) {
+  tsem_t sem;
+  int    result = tsem_init(&sem, 0, 0);
+  EXPECT_EQ(result, 0);
+
+  std::thread([&sem]() {
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    tsem_post(&sem);
+  }).detach();
+
+  result = tsem_wait(&sem);
+  EXPECT_EQ(result, 0);
+
+  result = tsem_destroy(&sem);
+  EXPECT_EQ(result, 0);
+}
+
+
+TEST(osSemaphoreTests, TimedWait) {
+  tsem_t sem;
+  int    result = tsem_init(&sem, 0, 0);
+  EXPECT_EQ(result, 0);
+
+  std::thread([&sem]() {
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    tsem_post(&sem);
+  }).detach();
+
+  result = tsem_timewait(&sem, 1000);
+  EXPECT_EQ(result, 0);
+
+  result = tsem_destroy(&sem);
+  EXPECT_EQ(result, 0);
+}