This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 7d0df2ea47 [INLONG-8861][SDK] Add mutex utils for dataproxy cpp sdk
(#8863)
7d0df2ea47 is described below
commit 7d0df2ea4735c46934dcca3cde24e2761c2a1a32
Author: doleyzi <[email protected]>
AuthorDate: Sat Sep 9 18:05:02 2023 +0800
[INLONG-8861][SDK] Add mutex utils for dataproxy cpp sdk (#8863)
Co-authored-by: doleyzi <[email protected]>
---
.../dataproxy-sdk-cpp/src/utils/noncopyable.h | 35 ++++++
.../dataproxy-sdk-cpp/src/utils/read_write_mutex.h | 133 +++++++++++++++++++++
.../dataproxy-sdk-cpp/src/utils/singleton.h | 56 +++++++++
3 files changed, 224 insertions(+)
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/noncopyable.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/noncopyable.h
new file mode 100644
index 0000000000..50b57fdc56
--- /dev/null
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/noncopyable.h
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#ifndef INLONG_SDK_NONCOPYABLE_H
+#define INLONG_SDK_NONCOPYABLE_H
+
+namespace inlong {
+class noncopyable {
+public:
+ noncopyable(const noncopyable &) = delete;
+ void operator=(const noncopyable &) = delete;
+
+protected:
+ noncopyable() = default;
+ virtual ~noncopyable() = default;
+};
+} // namespace inlong
+
+#endif // INLONG_SDK_NONCOPYABLE_H
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/read_write_mutex.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/read_write_mutex.h
new file mode 100644
index 0000000000..44bb7471fe
--- /dev/null
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/read_write_mutex.h
@@ -0,0 +1,133 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#ifndef INLONG_SDK_READ_WRITE_MUTEX_H
+#define INLONG_SDK_READ_WRITE_MUTEX_H
+
+#include <condition_variable>
+#include <mutex>
+
+namespace dataproxy_sdk {
+// wirte operation add lock：unique_read_lock<read_write_mutex> lock( rwmutex );
+// read operation add lock：unique_write_lock<read_write_mutex> lock(rwmutex);
+
+class read_write_mutex {
+public:
+ read_write_mutex() = default;
+ ~read_write_mutex() = default;
+
+ read_write_mutex(const read_write_mutex &) = delete;
+ read_write_mutex &operator=(const read_write_mutex &) = delete;
+
+ read_write_mutex(const read_write_mutex &&) = delete;
+ read_write_mutex &operator=(const read_write_mutex &&) = delete;
+
+ void lock_read() {
+ std::unique_lock<std::mutex> lock(m_mutex);
+ m_cond_read.wait(lock, [this]() -> bool { return m_write_count == 0; });
+ ++m_read_count;
+ }
+
+ void unlock_read() {
+ std::unique_lock<std::mutex> lock(m_mutex);
+ if (--m_read_count == 0 && m_write_count > 0) {
+ m_cond_write.notify_one();
+ }
+ }
+
+ void lock_write() {
+ std::unique_lock<std::mutex> lock(m_mutex);
+ ++m_write_count;
+ m_cond_write.wait(
+ lock, [this]() -> bool { return m_read_count == 0 && !m_writing; });
+ m_writing = true;
+ }
+
+ void unlock_write() {
+ std::unique_lock<std::mutex> lock(m_mutex);
+ if (--m_write_count == 0) {
+ m_cond_read.notify_all();
+ } else {
+ m_cond_write.notify_one();
+ }
+ m_writing = false;
+ }
+
+private:
+ volatile size_t m_read_count = 0;
+ volatile size_t m_write_count = 0;
+ volatile bool m_writing = false;
+ mutable std::mutex m_mutex; // KEYPOINT: add mutable
+ std::condition_variable m_cond_read;
+ std::condition_variable m_cond_write;
+};
+
+template <typename _ReadWriteLock> class unique_read_lock {
+public:
+ explicit unique_read_lock(_ReadWriteLock &rwLock) : m_ptr_rw_lock(&rwLock) {
+ m_ptr_rw_lock->lock_read();
+ }
+
+ ~unique_read_lock() {
+ if (m_ptr_rw_lock) {
+ m_ptr_rw_lock->unlock_read();
+ }
+ }
+
+ unique_read_lock() = delete;
+ unique_read_lock(const unique_read_lock &) = delete;
+ unique_read_lock &operator=(const unique_read_lock &) = delete;
+ unique_read_lock(const unique_read_lock &&) = delete;
+ unique_read_lock &operator=(const unique_read_lock &&) = delete;
+
+private:
+ _ReadWriteLock *m_ptr_rw_lock = nullptr;
+};
+
+template <typename _ReadWriteLock> class unique_write_lock {
+public:
+ explicit unique_write_lock(_ReadWriteLock &rwLock) : m_ptr_rw_lock(&rwLock) {
+ m_ptr_rw_lock->lock_write();
+ }
+
+ ~unique_write_lock() {
+ if (m_ptr_rw_lock) {
+ m_ptr_rw_lock->unlock_write();
+ }
+ }
+
+ unique_write_lock() = delete;
+ unique_write_lock(const unique_write_lock &) = delete;
+ unique_write_lock &operator=(const unique_write_lock &) = delete;
+ unique_write_lock(const unique_write_lock &&) = delete;
+ unique_write_lock &operator=(const unique_write_lock &&) = delete;
+
+ void unlock() {
+ if (m_ptr_rw_lock) {
+ m_ptr_rw_lock->unlock_write();
+ }
+ }
+
+private:
+ _ReadWriteLock *m_ptr_rw_lock = nullptr;
+};
+
+} // namespace dataproxy_sdk
+
+#endif // INLONG_SDK_READ_WRITE_MUTEX_H
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/singleton.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/singleton.h
new file mode 100644
index 0000000000..66956c504d
--- /dev/null
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/singleton.h
@@ -0,0 +1,56 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#ifndef INLONG_SDK_SINGLETON_H
+#define INLONG_SDK_SINGLETON_H
+
+#include <assert.h>
+#include <mutex>
+#include <thread>
+
+#include "noncopyable.h"
+
+namespace inlong {
+template <typename T> class Singleton : noncopyable {
+private:
+ static std::once_flag once_;
+ static T *value_;
+
+public:
+ Singleton() = delete;
+ ~Singleton() = delete;
+
+ static T &instance() {
+ std::call_once(once_, Singleton::init);
+ assert(value_ != nullptr);
+
+ return *value_;
+ }
+
+private:
+ static void init() { value_ = new T(); }
+};
+
+template <typename T> std::once_flag Singleton<T>::once_;
+
+template <typename T> T *Singleton<T>::value_ = nullptr;
+
+} // namespace inlong
+
+#endif // INLONG_SDK_SINGLETON_H
\ No newline at end of file
