coffee/cpp-ipc
1
0
Fork 0
mirror of https://github.com/mutouyun/cpp-ipc.git synced 2025-12-06 16:56:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

upd: [concur] adjust and optimize the comments, as well as some other details.

Browse Source
This commit is contained in:
mutouyun 2023-04-02 21:00:55 +08:00
parent 43021c1aba
commit 1565ae0eaf
2 changed files with 64 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
include/libconcur/concurrent.h
View File

@ -8,6 +8,7 @@
#include <type_traits> #include <type_traits>
#include <atomic> #include <atomic>
#include <array>
#include <cstdint> #include <cstdint>
#include "libimp/span.h" #include "libimp/span.h"
@ -17,7 +18,7 @@
LIBCONCUR_NAMESPACE_BEG_ LIBCONCUR_NAMESPACE_BEG_
/// \brief The queue index type. /// \typedef The queue index type.
using index_t = std::uint32_t; using index_t = std::uint32_t;
/// \brief Multiplicity of the relationship. /// \brief Multiplicity of the relationship.
@ -31,7 +32,12 @@ class multi {};
/// \brief Transmission mode /// \brief Transmission mode
namespace trans { namespace trans {
class unicast {}; /// \brief In this transmission mode, the message transmission is similar to a queue.
/// When the receiving side is not timely enough, the sending side will be unable to write data.
class unicast {};
/// \brief In this transmission mode, the message will be perceived by each receiver.
/// When the receiving side is not timely enough, the sending side will overwrite unread data.
class broadcast {}; class broadcast {};
} // namespace trans } // namespace trans
@ -56,7 +62,7 @@ using is_context = decltype(typename std::enable_if<std::is_convertible<decltype
*/ */
template <typename C, typename = is_context<C>> template <typename C, typename = is_context<C>>
constexpr index_t trunc_index(C const &ctx, index_t idx) noexcept { constexpr index_t trunc_index(C const &ctx, index_t idx) noexcept {
/// \remark `circ_size == 2^N` => `idx & (circ_size - 1)` // `circ_size == 2^N` => `idx & (circ_size - 1)`
return ctx.valid() ? (idx % ctx.circ_size) : 0; return ctx.valid() ? (idx % ctx.circ_size) : 0;
} }
@ -81,8 +87,8 @@ template <>
struct producer<trans::unicast, relation::single> { struct producer<trans::unicast, relation::single> {
struct context_impl { struct context_impl {
/// \brief Write index. index_t w_idx {0}; ///< write index
alignas(cache_line_size) index_t w_idx {0}; private: padding<decltype(w_idx)> ___;
}; };
template <typename T, typename C, typename U, template <typename T, typename C, typename U,
@ -93,14 +99,14 @@ struct producer<trans::unicast, relation::single> {
auto w_cur = trunc_index(ctx, w_idx); auto w_cur = trunc_index(ctx, w_idx);
auto &elem = elems[w_cur]; auto &elem = elems[w_cur];
auto f_ct = elem.get_flag(); auto f_ct = elem.get_flag();
/// \remark Verify index. // Verify index.
if ((f_ct != state::invalid_value) && if ((f_ct != state::invalid_value) &&
(f_ct != w_idx)) { (f_ct != w_idx)) {
return false; // full return false; // full
} }
/// \remark Get a valid index and iterate backwards. // Get a valid index and iterate backwards.
ctx.w_idx += 1; ctx.w_idx += 1;
/// \remark Set data & flag. // Set data & flag.
elem.set_data(std::forward<U>(src)); elem.set_data(std::forward<U>(src));
elem.set_flag(static_cast<index_t>(~w_idx)); elem.set_flag(static_cast<index_t>(~w_idx));
return true; return true;
@ -112,8 +118,8 @@ template <>
struct producer<trans::unicast, relation::multi> { struct producer<trans::unicast, relation::multi> {
struct context_impl { struct context_impl {
/// \brief Write index. std::atomic<index_t> w_idx {0}; ///< write index
alignas(cache_line_size) std::atomic<index_t> w_idx {0}; private: padding<decltype(w_idx)> ___;
}; };
template <typename T, typename C, typename U, template <typename T, typename C, typename U,
@ -125,16 +131,16 @@ struct producer<trans::unicast, relation::multi> {
auto w_cur = trunc_index(ctx, w_idx); auto w_cur = trunc_index(ctx, w_idx);
auto &elem = elems[w_cur]; auto &elem = elems[w_cur];
auto f_ct = elem.get_flag(); auto f_ct = elem.get_flag();
/// \remark Verify index. // Verify index.
if ((f_ct != state::invalid_value) && if ((f_ct != state::invalid_value) &&
(f_ct != w_idx)) { (f_ct != w_idx)) {
return false; // full return false; // full
} }
/// \remark Get a valid index and iterate backwards. // Get a valid index and iterate backwards.
if (!ctx.w_idx.compare_exchange_weak(w_idx, w_idx + 1, std::memory_order_acq_rel)) { if (!ctx.w_idx.compare_exchange_weak(w_idx, w_idx + 1, std::memory_order_acq_rel)) {
continue; continue;
} }
/// \remark Set data & flag. // Set data & flag.
elem.set_data(std::forward<U>(src)); elem.set_data(std::forward<U>(src));
elem.set_flag(static_cast<index_t>(~w_idx)); elem.set_flag(static_cast<index_t>(~w_idx));
return true; return true;
@ -147,8 +153,8 @@ template <>
struct consumer<trans::unicast, relation::single> { struct consumer<trans::unicast, relation::single> {
struct context_impl { struct context_impl {
/// \brief Read index. index_t r_idx {0}; ///< read index
alignas(cache_line_size) index_t r_idx {0}; private: padding<decltype(r_idx)> ___;
}; };
template <typename T, typename C, typename U, template <typename T, typename C, typename U,
@ -159,15 +165,15 @@ struct consumer<trans::unicast, relation::single> {
auto r_cur = trunc_index(ctx, r_idx); auto r_cur = trunc_index(ctx, r_idx);
auto &elem = elems[r_cur]; auto &elem = elems[r_cur];
auto f_ct = elem.get_flag(); auto f_ct = elem.get_flag();
/// \remark Verify index. // Verify index.
if (f_ct != static_cast<index_t>(~r_idx)) { if (f_ct != static_cast<index_t>(~r_idx)) {
return false; // empty return false; // empty
} }
/// \remark Get a valid index and iterate backwards. // Get a valid index and iterate backwards.
ctx.r_idx += 1; ctx.r_idx += 1;
/// \remark Get data & set flag. // Get data & set flag.
des = elem.get_data(); des = LIBCONCUR::get(elem);
elem.set_flag(r_idx + static_cast<index_t>(elems.size())); elem.set_flag(r_idx + static_cast<index_t>(elems.size())/*avoid overflow*/);
return true; return true;
} }
}; };
@ -177,8 +183,8 @@ template <>
struct consumer<trans::unicast, relation::multi> { struct consumer<trans::unicast, relation::multi> {
struct context_impl { struct context_impl {
/// \brief Read index. std::atomic<index_t> r_idx {0}; ///< read index
alignas(cache_line_size) std::atomic<index_t> r_idx {0}; private: padding<decltype(r_idx)> ___;
}; };
template <typename T, typename C, typename U, template <typename T, typename C, typename U,
@ -190,17 +196,17 @@ struct consumer<trans::unicast, relation::multi> {
auto r_cur = trunc_index(ctx, r_idx); auto r_cur = trunc_index(ctx, r_idx);
auto &elem = elems[r_cur]; auto &elem = elems[r_cur];
auto f_ct = elem.get_flag(); auto f_ct = elem.get_flag();
/// \remark Verify index. // Verify index.
if (f_ct != static_cast<index_t>(~r_idx)) { if (f_ct != static_cast<index_t>(~r_idx)) {
return false; // empty return false; // empty
} }
/// \remark Get a valid index and iterate backwards. // Get a valid index and iterate backwards.
if (!ctx.r_idx.compare_exchange_weak(r_idx, r_idx + 1, std::memory_order_acq_rel)) { if (!ctx.r_idx.compare_exchange_weak(r_idx, r_idx + 1, std::memory_order_acq_rel)) {
continue; continue;
} }
/// \remark Get data & set flag. // Get data & set flag.
des = elem.get_data(); des = LIBCONCUR::get(elem);
elem.set_flag(r_idx + static_cast<index_t>(elems.size())); elem.set_flag(r_idx + static_cast<index_t>(elems.size())/*avoid overflow*/);
return true; return true;
} }
} }
@ -220,12 +226,8 @@ template <>
struct producer<trans::broadcast, relation::multi> { struct producer<trans::broadcast, relation::multi> {
}; };
/// \brief Single-read consumer model implementation.
template <>
struct consumer<trans::broadcast, relation::single> {
};
/// \brief Multi-read consumer model implementation. /// \brief Multi-read consumer model implementation.
// Single-read consumer model is not required.
template <> template <>
struct consumer<trans::broadcast, relation::multi> { struct consumer<trans::broadcast, relation::multi> {
}; };
@ -254,7 +256,7 @@ struct prod_cons : producer<TransModT, ProdModT>
: circ_size(static_cast<index_t>(elems.size())) {} : circ_size(static_cast<index_t>(elems.size())) {}
constexpr bool valid() const noexcept { constexpr bool valid() const noexcept {
/// \remark circ_size must be a power of two. // circ_size must be a power of two.
return (circ_size > 1) && ((circ_size & (circ_size - 1)) == 0); return (circ_size > 1) && ((circ_size & (circ_size - 1)) == 0);
} }
}; };

37
include/libconcur/element.h
View File

@ -8,32 +8,50 @@
#include <atomic> #include <atomic>
#include <utility> #include <utility>
#include <limits> // std::numeric_limits
#include <cstdint> #include <cstdint>
#include <exception> #include <exception>
#include "libimp/detect_plat.h" #include "libimp/detect_plat.h"
#include "libimp/log.h" #include "libimp/log.h"
#include "libimp/byte.h"
#include "libimp/generic.h"
#include "libconcur/def.h" #include "libconcur/def.h"
LIBCONCUR_NAMESPACE_BEG_ LIBCONCUR_NAMESPACE_BEG_
namespace state { namespace state {
/// \brief The state flag type for the queue element. /// \typedef The state flag type for the queue element.
using flag_t = std::uint64_t; using flag_t = std::uint64_t;
enum : flag_t { enum : flag_t {
/// \brief The invalid state value. /// \brief The invalid state value.
invalid_value = (std::numeric_limits<flag_t>::max)(), invalid_value = ~flag_t(0),
}; };
} // namespace state } // namespace state
/// \brief Define the padding type.
template <typename T>
using padding = std::array<::LIBIMP::byte, (cache_line_size - sizeof(T))>;
/**
* \class template <typename T> element
* \brief User-defined type element wrapper.
* Wrapper for wrapping user-defined types as elements.
*
* @tparam T - User-defined type.
*/
template <typename T> template <typename T>
class element { class element {
template <typename E>
friend auto get(E &&elem) noexcept;
/// \brief Committed flag. /// \brief Committed flag.
alignas(cache_line_size) std::atomic<state::flag_t> f_ct_; std::atomic<state::flag_t> f_ct_;
padding<decltype(f_ct_)> ___;
/// \brief The user data segment. /// \brief The user data segment.
T data_; T data_;
@ -42,6 +60,8 @@ class element {
element &operator=(element const &) = delete; element &operator=(element const &) = delete;
public: public:
using value_type = T;
template <typename... A> template <typename... A>
element(A &&... args) element(A &&... args)
noexcept(noexcept(T{std::forward<A>(args)...})) noexcept(noexcept(T{std::forward<A>(args)...}))
@ -60,10 +80,6 @@ public:
} }
} }
T &&get_data() noexcept {
return std::move(data_);
}
void set_flag(state::flag_t flag) noexcept { void set_flag(state::flag_t flag) noexcept {
f_ct_.store(flag, std::memory_order_release); f_ct_.store(flag, std::memory_order_release);
} }
@ -73,4 +89,9 @@ public:
} }
}; };
template <typename E>
auto get(E &&elem) noexcept {
return static_cast<imp::copy_cvref_t<E, typename std::decay<E>::type::value_type>>(elem.data_);
}
LIBCONCUR_NAMESPACE_END_ LIBCONCUR_NAMESPACE_END_