ThreadingExperiments/CompetingProbabilities.h at master · jjhegedus/ThreadingExperiments · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
#pragma once

#include "pch.h"
#include <vector>
#include <functional>
#include <utility>
#include <chrono>
#include <condition_variable>
#include <algorithm>
#include <thread>
#include <memory>
#include <mutex>

#include "TypeUtilities.h"


using namespace std::chrono;
using namespace std::chrono_literals;

namespace ndtech {

  template<typename ItemType>
	struct CompetingProbability {
    std::vector<ItemType> m_items;
    using ProbabilityProcessingFunctionType = std::function<ItemType(ItemType)>;
    CompetingProbability::ProbabilityProcessingFunctionType m_ProbabilityProcessingFunction;
    bool m_readyForNextProbability = false;

    CompetingProbability(std::function<ItemType(ItemType)> ProbabilityProcessingFunction) :
      m_ProbabilityProcessingFunction(ProbabilityProcessingFunction) {

    }

    void Process() {
      std::cout << "Process Probability" << this << std::endl;
      for (auto item : m_items) {
        std::cout << "item = " << item << std::endl;
        item.m_readyForNextProbability = true;
      }
    }

  };

  template<typename ItemType>
	struct CompetingProbabilitiesProcessing {

    CompetingProbabilitiesProcessing(std::vector<CompetingProbability<ItemType>> Probabilitys) :
      m_Probabilities(Probabilitys) {
      m_thread = std::thread{ &CompetingProbabilitiesProcessing::Run, this };
    }

    CompetingProbabilitiesProcessing(std::thread thread)
      :m_thread(std::move(thread)) {
    }

    void Run() {
      while (!m_done) {

        for (CompetingProbability<ItemType> Probability : m_Probabilities) {

          { // to scope the lock
            std::lock_guard<std::mutex> guard(m_itemsMutex);
            Probability.Process();
          }

        }

      }

    }

    void AddItem(ItemType item, std::size_t ProbabilityNumber) {

      std::lock_guard<std::mutex> guard(m_itemsMutex);
      m_Probabilities[ProbabilityNumber].m_items.push_back(item);

    }

    void Join() {
      this->m_done = true;
      m_thread.join();
    }

  private:
    std::thread                                                                         m_thread;
    std::vector<ndtech::CompetingProbability<ItemType>>                           m_Probabilities;
    std::mutex                                                                          m_itemsMutex;
    bool                                                                                m_done = false;
  };

  template <typename ItemType>
  CompetingProbabilitiesProcessing<ItemType> CreateCompetingProbabilitiesProcessing(std::vector<CompetingProbability<ItemType>> Probabilitys) {
    return CompetingProbabilitiesProcessing(Probabilitys);
  }

}