Browse Source

DEV: atomic strength for queues

master
parent
commit
d9da3917a9
6 changed files with 138 additions and 40 deletions
  1. +23
    -28
      include/cont/deque.h
  2. +5
    -5
      include/cont/edeque.h
  3. +16
    -6
      include/core/ring_iterator.h
  4. +42
    -0
      include/tbx.h
  5. +1
    -1
      test/Makefile
  6. +51
    -0
      test/tests/deque.cpp

+ 23
- 28
include/cont/deque.h View File

@@ -86,13 +86,18 @@ class deque {
constexpr deque () noexcept : constexpr deque () noexcept :
data_{}, data_{},
f{data_.data(), N}, f{data_.data(), N},
r{data_.data()} { }
r{data_.data()} {
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_release);
}
//! fill contructor //! fill contructor
constexpr deque(const Data_t& value) noexcept { constexpr deque(const Data_t& value) noexcept {
data_.fill(value); data_.fill(value);
f = iterator(data_.data(), N); f = iterator(data_.data(), N);
r = iterator(data_.data(), N); r = iterator(data_.data(), N);
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_release);
} }
//! Initializer list contructor //! Initializer list contructor
@@ -100,7 +105,10 @@ class deque {
constexpr deque(It&& ...it) noexcept : constexpr deque(It&& ...it) noexcept :
data_{{std::forward<It>(it)...}}, data_{{std::forward<It>(it)...}},
f(data_.data(), N), f(data_.data(), N),
r(data_.data(), sizeof...(It)) { }
r(data_.data(), sizeof...(It)) {
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_release);
}
deque(const deque&) = delete; //!< No copies deque(const deque&) = delete; //!< No copies
deque& operator= (const deque&) = delete; //!< No copy assignments deque& operator= (const deque&) = delete; //!< No copy assignments
@@ -132,10 +140,10 @@ class deque {
public: public:
//! \return The size of the deque. The items currently in queue. //! \return The size of the deque. The items currently in queue.
constexpr size_t size() noexcept { constexpr size_t size() noexcept {
return full() ? N: (r - f) -1;
return r - (f +1);
} }
constexpr size_t size() const noexcept { constexpr size_t size() const noexcept {
return full() ? N: (r - f) -1;
return r - (f +1);
} }
//! \return The maximum size of the deque. The items the queue can hold. //! \return The maximum size of the deque. The items the queue can hold.
constexpr size_t max_size() noexcept { return N; } constexpr size_t max_size() noexcept { return N; }
@@ -144,11 +152,8 @@ class deque {
//! \return True if the deque is empty //! \return True if the deque is empty
constexpr bool empty() noexcept { return size() == 0 ? true : false; } constexpr bool empty() noexcept { return size() == 0 ? true : false; }
//! \return True if the deque is full //! \return True if the deque is full
constexpr bool full() noexcept {
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_acquire);
return (r == f) ? true : false;
}
constexpr bool full() noexcept { return size() == N ? true : false; }
//! @} //! @}
//! \name Member access //! \name Member access
@@ -166,36 +171,26 @@ class deque {
//! \param it The item to push //! \param it The item to push
constexpr void push_front (const Data_t& it) noexcept { constexpr void push_front (const Data_t& it) noexcept {
if (full()) return; if (full()) return;
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_acquire);
*f-- = it;
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_release);
*f = it;
--f; // keep this separate for thread safety
}
//! \brief Push an item in the back of the deque
//! \param it The item to push
constexpr void push_back (const Data_t& it) noexcept {
if (full()) return;
*r = it;
++r; // keep this separate for thread safety
} }
//! \brief Extract an item from the front of the deque and remove it from the deque //! \brief Extract an item from the front of the deque and remove it from the deque
//! \param it The item to push //! \param it The item to push
constexpr Data_t pop_front () noexcept { constexpr Data_t pop_front () noexcept {
if (empty()) return Data_t{}; if (empty()) return Data_t{};
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_acquire);
return *++f; return *++f;
} }
//! \brief Push an item in the back of the deque
//! \param it The item to push
constexpr void push_back (const Data_t& it) noexcept {
if (full()) return;
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_acquire);
*r++ = it;
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_release);
}
//! \brief Extract an item from the back of the deque and remove it from the deque //! \brief Extract an item from the back of the deque and remove it from the deque
//! \param it The item to push //! \param it The item to push
constexpr Data_t pop_back () noexcept { constexpr Data_t pop_back () noexcept {
if (empty()) return Data_t{}; if (empty()) return Data_t{};
if constexpr (SemiAtomic)
std::atomic_thread_fence(std::memory_order_acquire);
return *--r; return *--r;
} }


+ 5
- 5
include/cont/edeque.h View File

@@ -191,21 +191,21 @@ class edeque : public deque<Data_t, N, SemiAtomic> {
} }
//! @} //! @}
//! \name Base class overwrites
//! \name Base class uses and overwrites
//! @{ //! @{
void push_front (const Data_t& it) noexcept { void push_front (const Data_t& it) noexcept {
base_type::push_front(it); base_type::push_front(it);
check_trigger_push_async_(it); check_trigger_push_async_(it);
} }
void push_back (const Data_t& it) noexcept {
base_type::push_back(it);
check_trigger_push_async_(it);
}
Data_t pop_front () noexcept { Data_t pop_front () noexcept {
Data_t t = base_type::pop_front(); Data_t t = base_type::pop_front();
check_trigger_pop_async_(t); check_trigger_pop_async_(t);
return t; return t;
} }
void push_back (const Data_t& it) noexcept {
base_type::push_back(it);
check_trigger_push_async_(it);
}
Data_t pop_back () noexcept { Data_t pop_back () noexcept {
Data_t t = base_type::pop_back(); Data_t t = base_type::pop_back();
check_trigger_pop_async_(t); check_trigger_pop_async_(t);


+ 16
- 6
include/core/ring_iterator.h View File

@@ -293,9 +293,14 @@ class ring_iterator<Iter_t, N, true> {
return *this; return *this;
} }
constexpr ring_iterator operator++(int) noexcept { constexpr ring_iterator operator++(int) noexcept {
ring_iterator it = *this;
this->operator ++();
return it;
ring_iterator ret = *this;
Iter_t itnew, it = iter_.load(std::memory_order_acquire);
do {
itnew = it;
if (static_cast<size_t>(++itnew - base_) >= N)
itnew = base_;
} while (!iter_.compare_exchange_weak(it, itnew, std::memory_order_acq_rel));
return ret;
} }
//! @} //! @}
@@ -312,9 +317,14 @@ class ring_iterator<Iter_t, N, true> {
return *this; return *this;
} }
constexpr ring_iterator operator--(int) noexcept { constexpr ring_iterator operator--(int) noexcept {
ring_iterator it = *this;
this->operator --();
return it;
ring_iterator ret = *this;
Iter_t itnew, it = iter_.load(std::memory_order_acquire);
do {
itnew = it;
if (--itnew < base_)
itnew = base_ + N -1;
} while (!iter_.compare_exchange_weak(it, itnew, std::memory_order_acq_rel));
return ret;
} }
//! @} //! @}


+ 42
- 0
include/tbx.h View File

@@ -0,0 +1,42 @@
/*!
* \file tbx.h
* \brief
* Main tbx header
*
* \copyright Copyright (C) 2021 Christos Choutouridis <christos@choutouridis.net>
*
* <dl class=\"section copyright\"><dt>License</dt><dd>
* The MIT License (MIT)
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
* </dd></dl>
*/
#ifndef TBX_H_
#define TBX_H_
#include <com/sequencer.h>
#include <cont/deque.h>
#include <cont/edeque.h>
#include <cont/queue.h>
#include <cont/equeue.h>
#include <drv/cli_device.h>
#endif /* TBX_H_ */

+ 1
- 1
test/Makefile View File

@@ -188,7 +188,7 @@ $(BUILD_DIR)/$(TARGET): $(OBJ)
@mkdir -p $(@D) @mkdir -p $(@D)
@echo Linking to target: $(TARGET) @echo Linking to target: $(TARGET)
$(DOCKER) $(CXX) $(LDFLAGS) $(MAP_FLAG) -o $(@D)/$(TARGET) $(OBJ) $(DOCKER) $(CXX) $(LDFLAGS) $(MAP_FLAG) -o $(@D)/$(TARGET) $(OBJ)
$(DOCKER) $(ODUMP) -h -S $(BUILD_DIR)/$(TARGET) > $(BUILD_DIR)/$(basename $(TARGET)).list
# $(DOCKER) $(ODUMP) -h -S $(BUILD_DIR)/$(TARGET) > $(BUILD_DIR)/$(basename $(TARGET)).list
# $(DOCKER) $(OCOPY) -O ihex $(BUILD_DIR)/$(TARGET) $(BUILD_DIR)/$(basename $(TARGET)).hex # $(DOCKER) $(OCOPY) -O ihex $(BUILD_DIR)/$(TARGET) $(BUILD_DIR)/$(basename $(TARGET)).hex
@echo @echo
@echo Print size information @echo Print size information


+ 51
- 0
test/tests/deque.cpp View File

@@ -34,6 +34,14 @@
#include <array> #include <array>
#include <type_traits> #include <type_traits>
#include <cstring>
#ifndef WIN_TRHEADS
#include <mutex>
#include <thread>
#else
#include <mingw.thread.h>
#include <mingw.mutex.h>
#endif
namespace Tdeque { namespace Tdeque {
using namespace tbx; using namespace tbx;
@@ -415,4 +423,47 @@ namespace Tdeque {
EXPECT_EQ(9, check_it); // run through all EXPECT_EQ(9, check_it); // run through all
} }
TEST(Tdeque, race) {
constexpr size_t N = 1000000;
deque<int, N, true> q;
int result[N];
auto push_front = [&](){
for (size_t i=1 ; i<=N ; ++i) q.push_front(i);
};
auto push_back = [&](){
for (size_t i=1 ; i<=N ; ++i) q.push_back(i);
};
auto pop_front = [&](){
for (size_t i=0 ; i<N ; ) {
result[i] = q.pop_front();
if (result[i] != int{})
++i;
}
};
auto pop_back = [&](){
for (size_t i=0 ; i<N ; ) {
result[i] = q.pop_back();
if (result[i] != int{})
++i;
}
};
std::memset(result, 0, sizeof result);
std::thread th1 (push_front);
std::thread th2 (pop_back);
th1.join();
th2.join();
for (size_t i=0 ; i<N ; ++i)
EXPECT_EQ (result[i], (int)i+1);
std::memset(result, 0, sizeof result);
std::thread th3 (push_back);
std::thread th4 (pop_front);
th3.join();
th4.join();
for (size_t i=0 ; i<N ; ++i)
EXPECT_EQ (result[i], (int)i+1);
}
} }

Loading…
Cancel
Save