concurrent_bag.hpp

Go to the documentation of this file.

#pragma once
#include "iproducer_consumer_collection.hpp"
#include "../generic/dictionary.hpp"
#include "../generic/queue.hpp"
#include "../../threading/interlocked.hpp"
#include "../../as.hpp"
#include "../../lock_guard.hpp"
#include "../../new_ptr.hpp"
#include <atomic>

namespace xtd {
  namespace collections {
    namespace concurrent {
      template<typename type_t>
      class concurrent_bag : public xtd::object, public xtd::collections::concurrent::iproducer_consumer_collection<type_t> {
      public:
        
        using value_type = typename iproducer_consumer_collection<type_t>::value_type;
        using reference = value_type&;
        using const_reference = const value_type&;
        using collection_type = xtd::collections::generic::queue<value_type>;
        using local_storage_type = collection_type*;
        using storage_type = xtd::collections::generic::dictionary<xtd::intptr, xtd::ptr<collection_type>>;

        
        concurrent_bag() = default;
        concurrent_bag(std::initializer_list<type_t> items) {
          for (const auto& item : items)
            add(item);
        }

        concurrent_bag(const xtd::collections::generic::ienumerable<value_type>& collection) {
          for (const auto& item : collection)
            add(item);
        }

        
        
        [[nodiscard]] auto count() const noexcept -> xtd::usize override {return as<xtd::usize>(count_);}
        
        [[nodiscard]] auto is_empty() const noexcept -> bool {return count_ == 0;}

        [[nodiscard]] auto is_read_only() const noexcept -> bool override {return false;}
        
        [[nodiscard]] auto is_synchronized() const noexcept -> bool override {return true;}
        
        
        auto add(const_reference item) -> void override {if (try_add(item) == false) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::not_supported);}
                
        auto copy_to(xtd::array<value_type>& array, xtd::usize array_index) const -> void override {
          if (array_index + count() > array.length()) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::argument_out_of_range);
          static thread_local auto items = xtd::array<value_type> {};
          items = to_array();
          for (auto i = xtd::usize {0}; i < count(); ++i)
            array[array_index + i] = items[i];
        }
        
        [[nodiscard]] auto get_enumerator() const -> xtd::collections::generic::enumerator<value_type> override {
          static thread_local auto items = xtd::array<value_type> {};
          items = to_array();
          return items.get_enumerator();
        }
                
        [[nodiscard]] auto to_array() const -> xtd::array<value_type> override {
          auto result = xtd::collections::generic::list<value_type> {};
          lock_guard_(storages_) {
            for (const auto& [thread_id, storage] : storages_)
              result.add_range(*storage);
          }
          return result.to_array();
        }
        
        [[nodiscard]] auto to_string() const noexcept -> xtd::string override {return xtd::string::format("[{}]", xtd::string::join(", ", self_));}
        
        auto try_add(const_reference item) -> bool override {
          current_local_storage()->enqueue(item);
          xtd::threading::interlocked::increment(count_);
          return true;
        }
        
        auto try_peek(reference item) const -> bool {
          if (current_local_storage()->count()) return current_local_storage()->try_peek(item);
          lock_guard_(storages_) {
            for (auto& [thread_id, storage] : storages_) {
              if (storage->count()) return storage->try_peek(item);
            }
          }
          return false;
        }
        
        auto try_take(reference item) -> bool override {
          if (current_local_storage()->count()) {
            auto result = current_local_storage()->try_dequeue(item);
            if (result) {
              xtd::threading::interlocked::decrement(count_);
              return true;
            }
          }
          lock_guard_(storages_) {
            for (auto& [thread_id, storage] : storages_) {
              if (storage->count()) {
                auto result = storage->try_dequeue(item);
                if (result) {
                  xtd::threading::interlocked::decrement(count_);
                  return true;
                }
              }
            }
          }
          return false;
        }
        
      private:
        auto clear() -> void override {xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::not_supported);}
        [[nodiscard]] auto contains(const_reference item) const noexcept -> bool override {return false;}
        auto remove(const_reference item) -> bool override {return false;}
        [[nodiscard]] auto sync_root() const noexcept -> const object& override {return storages_;}
        
        auto current_local_storage() const -> local_storage_type {
          if (thread_local_storages_.contains_key(instance_id_)) return thread_local_storages_[instance_id_];
          lock_guard_(storages_) {
            auto id = xtd::threading::thread::current_thread().thread_id();
            if (!storages_.contains_key(id)) storages_.add(id, new_ptr<collection_type>());
            auto storage = storages_[id].get();
            thread_local_storages_[instance_id_] = storage;
            return storage;
          }
          return thread_local_storages_[instance_id_];
        }
 
        inline static thread_local xtd::collections::generic::dictionary<xtd::uint64, local_storage_type> thread_local_storages_;
        inline static std::atomic<xtd::uint64> next_id_;
        mutable storage_type storages_;
        xtd::int64 count_ = 0;
        xtd::uint64 instance_id_ = ++next_id_;
      };

      // Deduction guides for xtd::collections::concurrent::concurrent_bag
      // {
      template<typename type_t>
      concurrent_bag(std::initializer_list<type_t>) -> concurrent_bag<type_t>;
      
      template<typename type_t>
      concurrent_bag(const xtd::collections::generic::ienumerable<type_t>&) -> concurrent_bag<type_t>;
      // }
    }
  }
}