Merge pull request #63 from mutouyun/develop

Develop

Author: mutouyun

.travis.yml

@@ -13,7 +13,7 @@ if(NOT MSVC)
     set(CMAKE_CXX_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -O2")
 endif()
 
-if (MSVC AND LIBIPC_USE_STATIC_CRT)
+if (MSVC)
     set(CompilerFlags
         CMAKE_CXX_FLAGS
         CMAKE_CXX_FLAGS_DEBUG
@@ -22,9 +22,17 @@ if (MSVC AND LIBIPC_USE_STATIC_CRT)
         CMAKE_C_FLAGS_DEBUG
         CMAKE_C_FLAGS_RELEASE
         )
-    foreach(CompilerFlag ${CompilerFlags})
-        string(REPLACE "/MD" "/MT" ${CompilerFlag} "${${CompilerFlag}}")
-    endforeach()
+    if (LIBIPC_USE_STATIC_CRT)
+        foreach(CompilerFlag ${CompilerFlags})
+            string(REPLACE "/MD" "/MT" ${CompilerFlag} "${${CompilerFlag}}")
+            string(REPLACE "/MDd" "/MTd" ${CompilerFlag} "${${CompilerFlag}}")
+        endforeach()
+    else()
+        foreach(CompilerFlag ${CompilerFlags})
+            string(REPLACE "/MT" "/MD" ${CompilerFlag} "${${CompilerFlag}}")
+            string(REPLACE "/MTd" "/MDd" ${CompilerFlag} "${${CompilerFlag}}")
+        endforeach()
+    endif()
 endif()
 
 set(LIBRARY_OUTPUT_PATH    ${CMAKE_BINARY_DIR}/bin)
@@ -50,6 +58,7 @@ endif()
 if (LIBIPC_BUILD_DEMOS)
     add_subdirectory(demo/chat)
     add_subdirectory(demo/msg_que)
+    add_subdirectory(demo/send_recv)
 endif()
 
 install(

CMakeLists.txt

@@ -20,8 +20,8 @@ constexpr char const mode_r__[] = "r";
 constexpr std::size_t const min_sz = 128;
 constexpr std::size_t const max_sz = 1024 * 16;
 
-std::atomic<bool> is_quit__{ false };
-std::atomic<std::size_t> size_counter__{ 0 };
+std::atomic<bool> is_quit__ {false};
+std::atomic<std::size_t> size_counter__ {0};
 
 using msg_que_t = ipc::chan<ipc::relat::single, ipc::relat::multi, ipc::trans::broadcast>;
 
@@ -127,10 +127,10 @@ int main(int argc, char ** argv) {
     ::signal(SIGHUP  , exit);
 #endif
 
-    if (std::string{ argv[1] } == mode_s__) {
+    std::string mode {argv[1]};
+    if (mode == mode_s__) {
         do_send();
-    }
-    else if (std::string{ argv[1] } == mode_r__) {
+    } else if (mode == mode_r__) {
         do_recv();
     }
     return 0;

demo/msg_que/main.cpp

@@ -0,0 +1,11 @@
+project(send_recv)
+
+include_directories(
+    ${LIBIPC_PROJECT_DIR}/3rdparty)
+
+file(GLOB SRC_FILES ./*.cpp)
+file(GLOB HEAD_FILES ./*.h)
+
+add_executable(${PROJECT_NAME} ${SRC_FILES} ${HEAD_FILES})
+
+target_link_libraries(${PROJECT_NAME} ipc)

demo/send_recv/CMakeLists.txt

@@ -0,0 +1,72 @@
+
+#include <signal.h>
+
+#include <iostream>
+#include <string>
+#include <thread>
+#include <chrono>
+#include <atomic>
+
+#include "libipc/ipc.h"
+
+namespace {
+
+std::atomic<bool> is_quit__ {false};
+ipc::channel *ipc__ = nullptr;
+
+void do_send(int size, int interval) {
+    ipc::channel ipc {"ipc", ipc::sender};
+    ipc__ = &ipc;
+    std::string buffer(size, 'A');
+    while (!is_quit__.load(std::memory_order_acquire)) {
+        std::cout << "send size: " << buffer.size() + 1 << "\n";
+        ipc.send(buffer, 0/*tm*/);
+        std::this_thread::sleep_for(std::chrono::milliseconds(interval));
+    }
+}
+
+void do_recv(int interval) {
+    ipc::channel ipc {"ipc", ipc::receiver};
+    ipc__ = &ipc;
+    while (!is_quit__.load(std::memory_order_acquire)) {
+        ipc::buff_t recv;
+        for (int k = 1; recv.empty(); ++k) {
+            std::cout << "recv waiting... " << k << "\n";
+            recv = ipc.recv(interval);
+            if (is_quit__.load(std::memory_order_acquire)) return;
+        }
+        std::cout << "recv size: " << recv.size() << "\n";
+    }
+}
+
+} // namespace
+
+int main(int argc, char ** argv) {
+    if (argc < 3) return -1;
+
+    auto exit = [](int) {
+        is_quit__.store(true, std::memory_order_release);
+        if (ipc__ != nullptr) ipc__->disconnect();
+    };
+    ::signal(SIGINT  , exit);
+    ::signal(SIGABRT , exit);
+    ::signal(SIGSEGV , exit);
+    ::signal(SIGTERM , exit);
+#if defined(WIN64) || defined(_WIN64) || defined(__WIN64__) || \
+    defined(WIN32) || defined(_WIN32) || defined(__WIN32__) || defined(__NT__) || \
+    defined(WINCE) || defined(_WIN32_WCE)
+    ::signal(SIGBREAK, exit);
+#else
+    ::signal(SIGHUP  , exit);
+#endif
+
+    std::string mode {argv[1]};
+    if (mode == "send") {
+        if (argc < 4) return -1;
+        do_send(std::stoi(argv[2]) /*size*/, 
+                std::stoi(argv[3]) /*interval*/);
+    } else if (mode == "recv") {
+        do_recv(std::stoi(argv[2]) /*interval*/);
+    }
+    return 0;
+}

demo/send_recv/main.cpp

@@ -0,0 +1,39 @@
+#pragma once
+
+#include <cstdint>  // std::uint64_t
+
+#include "libipc/export.h"
+#include "libipc/def.h"
+#include "libipc/mutex.h"
+
+namespace ipc {
+namespace sync {
+
+class IPC_EXPORT condition {
+    condition(condition const &) = delete;
+    condition &operator=(condition const &) = delete;
+
+public:
+    condition();
+    explicit condition(char const *name);
+    ~condition();
+
+    void const *native() const noexcept;
+    void *native() noexcept;
+
+    bool valid() const noexcept;
+
+    bool open(char const *name) noexcept;
+    void close() noexcept;
+
+    bool wait(ipc::sync::mutex &mtx, std::uint64_t tm = ipc::invalid_value) noexcept;
+    bool notify() noexcept;
+    bool broadcast() noexcept;
+
+private:
+    class condition_;
+    condition_* p_;
+};
+
+} // namespace sync
+} // namespace ipc

include/libipc/condition.h

@@ -25,13 +25,16 @@ using uint_t = typename uint<N>::type;
 
 // constants
 
+enum : std::uint32_t {
+    invalid_value   = (std::numeric_limits<std::uint32_t>::max)(),
+    default_timeout = 100, // ms
+};
+
 enum : std::size_t {
-    invalid_value   = (std::numeric_limits<std::size_t>::max)(),
     data_length     = 64,
     large_msg_limit = data_length,
     large_msg_align = 1024,
     large_msg_cache = 32,
-    default_timeout = 100 // ms
 };
 
 enum class relat { // multiplicity of the relationship

include/libipc/def.h

@@ -27,12 +27,12 @@ struct IPC_EXPORT chan_impl {
     static char const * name(ipc::handle_t h);
 
     static std::size_t recv_count(ipc::handle_t h);
-    static bool wait_for_recv(ipc::handle_t h, std::size_t r_count, std::size_t tm);
+    static bool wait_for_recv(ipc::handle_t h, std::size_t r_count, std::uint64_t tm);
 
-    static bool   send(ipc::handle_t h, void const * data, std::size_t size, std::size_t tm);
-    static buff_t recv(ipc::handle_t h, std::size_t tm);
+    static bool   send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm);
+    static buff_t recv(ipc::handle_t h, std::uint64_t tm);
 
-    static bool   try_send(ipc::handle_t h, void const * data, std::size_t size, std::size_t tm);
+    static bool   try_send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm);
     static buff_t try_recv(ipc::handle_t h);
 };
 
@@ -120,41 +120,41 @@ class chan_wrapper {
         return detail_t::recv_count(h_);
     }
 
-    bool wait_for_recv(std::size_t r_count, std::size_t tm = invalid_value) const {
+    bool wait_for_recv(std::size_t r_count, std::uint64_t tm = invalid_value) const {
         return detail_t::wait_for_recv(h_, r_count, tm);
     }
 
-    static bool wait_for_recv(char const * name, std::size_t r_count, std::size_t tm = invalid_value) {
+    static bool wait_for_recv(char const * name, std::size_t r_count, std::uint64_t tm = invalid_value) {
         return chan_wrapper(name).wait_for_recv(r_count, tm);
     }
 
     /**
      * If timeout, this function would call 'force_push' to send the data forcibly.
     */
-    bool send(void const * data, std::size_t size, std::size_t tm = default_timeout) {
+    bool send(void const * data, std::size_t size, std::uint64_t tm = default_timeout) {
         return detail_t::send(h_, data, size, tm);
     }
-    bool send(buff_t const & buff, std::size_t tm = default_timeout) {
+    bool send(buff_t const & buff, std::uint64_t tm = default_timeout) {
         return this->send(buff.data(), buff.size(), tm);
     }
-    bool send(std::string const & str, std::size_t tm = default_timeout) {
+    bool send(std::string const & str, std::uint64_t tm = default_timeout) {
         return this->send(str.c_str(), str.size() + 1, tm);
     }
 
     /**
      * If timeout, this function would just return false.
     */
-    bool try_send(void const * data, std::size_t size, std::size_t tm = default_timeout) {
+    bool try_send(void const * data, std::size_t size, std::uint64_t tm = default_timeout) {
         return detail_t::try_send(h_, data, size, tm);
     }
-    bool try_send(buff_t const & buff, std::size_t tm = default_timeout) {
+    bool try_send(buff_t const & buff, std::uint64_t tm = default_timeout) {
         return this->try_send(buff.data(), buff.size(), tm);
     }
-    bool try_send(std::string const & str, std::size_t tm = default_timeout) {
+    bool try_send(std::string const & str, std::uint64_t tm = default_timeout) {
         return this->try_send(str.c_str(), str.size() + 1, tm);
     }
 
-    buff_t recv(std::size_t tm = invalid_value) {
+    buff_t recv(std::uint64_t tm = invalid_value) {
         return detail_t::recv(h_, tm);
     }
 

include/libipc/ipc.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include <cstdint>  // std::uint64_t
+#include <system_error>
+
+#include "libipc/export.h"
+#include "libipc/def.h"
+
+namespace ipc {
+namespace sync {
+
+class IPC_EXPORT mutex {
+    mutex(mutex const &) = delete;
+    mutex &operator=(mutex const &) = delete;
+
+public:
+    mutex();
+    explicit mutex(char const *name);
+    ~mutex();
+
+    void const *native() const noexcept;
+    void *native() noexcept;
+
+    bool valid() const noexcept;
+
+    bool open(char const *name) noexcept;
+    void close() noexcept;
+
+    bool lock(std::uint64_t tm = ipc::invalid_value) noexcept;
+    bool try_lock() noexcept(false); // std::system_error
+    bool unlock() noexcept;
+
+private:
+    class mutex_;
+    mutex_* p_;
+};
+
+} // namespace sync
+} // namespace ipc

include/libipc/mutex.h

@@ -72,7 +72,7 @@ inline void yield(K& k) noexcept {
     ++k;
 }
 
-template <std::size_t N = 4096, typename K, typename F>
+template <std::size_t N = 32, typename K, typename F>
 inline void sleep(K& k, F&& f) {
     if (k < static_cast<K>(N)) {
         std::this_thread::yield();
@@ -84,7 +84,7 @@ inline void sleep(K& k, F&& f) {
     ++k;
 }
 
-template <std::size_t N = 4096, typename K>
+template <std::size_t N = 32, typename K>
 inline void sleep(K& k) {
     sleep<N>(k, [] {
         std::this_thread::sleep_for(std::chrono::milliseconds(1));