Skip to content

Commit 046acb4

Browse files
tannebergertanneberger
committed
clean up
1 parent a723f07 commit 046acb4

File tree

15 files changed

+134
-112
lines changed

15 files changed

+134
-112
lines changed

‎examples/CMakeLists.txt‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,4 +7,3 @@ add_subdirectory(count)
77
add_subdirectory(ports)
88
add_subdirectory(hello)
99
add_subdirectory(power_train)
10-
add_subdirectory(multiport_mutation)

‎examples/multiport_mutation/CMakeLists.txt‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,2 @@
1-
add_executable(mutation_multiports main.cc)
2-
target_link_libraries(mutation_multiports reactor-cpp)
3-
add_dependencies(examples mutation_multiports)
1+
add_executable(mutation_multiports main.cc) target_link_libraries(mutation_multiports reactor - cpp)
2+
add_dependencies(examples mutation_multiports)

‎examples/multiport_mutation/consumer.hh‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ class Consumer final : public Reactor { // NOLINT
1515
std::size_t index_ = 0;
1616

1717
void reaction_1(const Input<unsigned>& in) const {
18-
std::cout << "consumer: " << index_ << " received value:" << *in.get() << '\n';
18+
// std::cout << "consumer: " << index_ << " received value:" << *in.get() << '\n';
1919
}
2020

2121
friend Consumer;
@@ -34,9 +34,7 @@ public:
3434

3535
Input<unsigned> in{"in", this}; // NOLINT
3636

37-
void assemble() override {
38-
handle.declare_trigger(&in);
39-
}
37+
void assemble() override { handle.declare_trigger(&in); }
4038
};
4139

4240
#endif // CONSUMER_HH

‎examples/multiport_mutation/load_balancer.hh‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,16 +16,15 @@ class LoadBalancer final : public Reactor { // NOLINT
1616
// reaction bodies
1717
static void reaction_1(const Input<unsigned>& inbound, Output<unsigned>& scale_bank,
1818
Multiport<Output<unsigned>>& outbound) {
19-
if (std::rand() % 15 == 0) { // NOLINT
19+
if (std::rand() % 15 == 0) { // NOLINT
2020
scale_bank.set(std::rand() % 20 + 1); // NOLINT
2121
}
2222
const unsigned sel = std::rand() % outbound.size(); // NOLINT
23-
std::cout << "Sending out to:" << sel << '\n';
23+
// std::cout << "Sending out to:" << sel << '\n';
2424
outbound[sel].set(inbound.get());
2525
outbound[std::min(4ul, outbound.size() - 1)].set(inbound.get());
2626
}
2727

28-
2928
friend LoadBalancer;
3029
};
3130

@@ -44,10 +43,10 @@ public:
4443
}
4544
~LoadBalancer() override = default;
4645

47-
ModifableMultiport<Output<unsigned>> out{"out", this}; // NOLINT
46+
ModifableMultiport<Output<unsigned>> out{"out", this}; // NOLINT
4847
std::size_t out_size_ = 0;
49-
Input<unsigned> inbound{"inbound", this}; // NOLINT
50-
Output<unsigned> scale_bank{"scale_bank", this}; // NOLINT
48+
Input<unsigned> inbound{"inbound", this}; // NOLINT
49+
Output<unsigned> scale_bank{"scale_bank", this}; // NOLINT
5150

5251
void assemble() override {
5352
std::cout << "assemble LoadBalancer\n";

‎examples/multiport_mutation/main.cc‎

Lines changed: 5 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,9 @@
44
#include <reactor-cpp/mutations/bank.hh>
55
#include <reactor-cpp/mutations/connection.hh>
66

7-
#include "./multiport_to_bank.hh"
87
#include "./consumer.hh"
98
#include "./load_balancer.hh"
9+
#include "./multiport_to_bank.hh"
1010
#include "./producer.hh"
1111
#include <reactor-cpp/reactor-cpp.hh>
1212

@@ -19,7 +19,6 @@ class Deployment final : public Reactor { // NOLINT
1919
Reaction scale_bank{"scale_bank", 1, this,
2020
[this]() { this->_inner.reaction_1(this->scale, this->consumers_, load_balancer_->out); }};
2121

22-
2322
class Inner : public MutableScope {
2423
int state = 0;
2524

@@ -35,15 +34,9 @@ class Deployment final : public Reactor { // NOLINT
3534
return std::make_unique<Consumer>(_lf_inst_name, env, index);
3635
};
3736

38-
std::function get_input_port = [](const std::unique_ptr<Consumer>& consumer) {
39-
return &consumer->in;
40-
};
41-
auto rescale = std::make_shared<ResizeMultiportToBank<unsigned, Consumer>>(
42-
&load_balancer,
43-
&reactor_bank,
44-
get_input_port,
45-
lambda,
46-
new_size);
37+
std::function get_input_port = [](const std::unique_ptr<Consumer>& consumer) { return &consumer->in; };
38+
auto rescale = std::make_shared<ResizeMultiportToBank<unsigned, Consumer>>(&load_balancer, &reactor_bank,
39+
get_input_port, lambda, new_size);
4740

4841
add_to_transaction(rescale);
4942

@@ -53,7 +46,7 @@ class Deployment final : public Reactor { // NOLINT
5346
friend LoadBalancer;
5447
};
5548

56-
Inner _inner;
49+
Inner _inner;
5750

5851
public:
5952
Deployment(const std::string& name, Environment* env)

‎examples/multiport_mutation/multiport_to_bank.hh‎

Lines changed: 55 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,12 @@
55
#ifndef MULTIPORT_TO_BANK_HH
66
#define MULTIPORT_TO_BANK_HH
77

8-
#include <reactor-cpp/mutations.hh>
98
#include <reactor-cpp/multiport.hh>
10-
#include <reactor-cpp/port.hh>
11-
#include <reactor-cpp/mutations/multiport.hh>
9+
#include <reactor-cpp/mutations.hh>
1210
#include <reactor-cpp/mutations/bank.hh>
1311
#include <reactor-cpp/mutations/connection.hh>
12+
#include <reactor-cpp/mutations/multiport.hh>
13+
#include <reactor-cpp/port.hh>
1414
#include <reactor-cpp/reactor.hh>
1515

1616
#include "../../lib/mutations/bank.cc"
@@ -21,75 +21,71 @@
2121

2222
namespace reactor {
2323

24-
template<class PortType, class ReactorType>
25-
class ResizeMultiportToBank : public Mutation {
26-
ModifableMultiport<Output<PortType>>* multiport_;
27-
std::vector<std::unique_ptr<ReactorType>>* bank_;
28-
std::function<Input<PortType>*(const std::unique_ptr<ReactorType>&)> get_input_port_;
29-
std::function<std::unique_ptr<ReactorType>(Environment* env, std::size_t index)> create_lambda_;
30-
std::size_t new_size_ = 0;
31-
public:
32-
ResizeMultiportToBank(ModifableMultiport<Output<PortType>>* multiport,
33-
std::vector<std::unique_ptr<ReactorType>>* bank,
34-
std::function<Input<PortType>*(const std::unique_ptr<ReactorType>&)> get_input_port,
35-
std::function<std::unique_ptr<ReactorType>(Environment* env, std::size_t index)> create_lambda,
36-
std::size_t new_size) :
37-
multiport_(multiport), bank_(bank), get_input_port_(get_input_port), create_lambda_(create_lambda), new_size_(new_size) {}
38-
39-
~ResizeMultiportToBank() = default;
40-
auto run() -> MutationResult {
41-
if (multiport_->size() != bank_->size()) {
42-
return NotMatchingBankSize;
43-
}
44-
auto old_size = multiport_->size();
45-
46-
if (new_size_ > old_size) {
47-
// TODO: this is an assumption
48-
auto change_multiport_size =
49-
std::make_shared<MutationChangeOutputMultiportSize<unsigned>>(multiport_, new_size_);
24+
template <class PortType, class ReactorType> class ResizeMultiportToBank : public Mutation {
25+
ModifableMultiport<Output<PortType>>* multiport_;
26+
std::vector<std::unique_ptr<ReactorType>>* bank_;
27+
std::function<Input<PortType>*(const std::unique_ptr<ReactorType>&)> get_input_port_;
28+
std::function<std::unique_ptr<ReactorType>(Environment* env, std::size_t index)> create_lambda_;
29+
std::size_t new_size_ = 0;
5030

51-
change_multiport_size->run();
31+
public:
32+
ResizeMultiportToBank(ModifableMultiport<Output<PortType>>* multiport,
33+
std::vector<std::unique_ptr<ReactorType>>* bank,
34+
std::function<Input<PortType>*(const std::unique_ptr<ReactorType>&)> get_input_port,
35+
std::function<std::unique_ptr<ReactorType>(Environment* env, std::size_t index)> create_lambda,
36+
std::size_t new_size)
37+
: multiport_(multiport)
38+
, bank_(bank)
39+
, get_input_port_(get_input_port)
40+
, create_lambda_(create_lambda)
41+
, new_size_(new_size) {}
5242

53-
auto change_bank_size = std::make_shared<MutationChangeBankSize<std::unique_ptr<ReactorType>>>(
54-
bank_, (*bank_)[0]->environment(), new_size_, create_lambda_);
43+
~ResizeMultiportToBank() = default;
44+
auto run() -> MutationResult {
45+
if (multiport_->size() != bank_->size()) {
46+
return NotMatchingBankSize;
47+
}
48+
auto old_size = multiport_->size();
5549

56-
change_bank_size->run();
50+
if (new_size_ > old_size) {
51+
auto change_multiport_size = std::make_shared<MutationChangeOutputMultiportSize<unsigned>>(multiport_, new_size_);
5752

58-
for (auto i = old_size; i < new_size_; i++) {
59-
auto add_conn = std::make_shared<MutationAddConnection<Output<PortType>, Input<PortType>>>(
60-
&(*multiport_)[i], get_input_port_((*bank_)[i]), (*bank_)[0]->environment(), true);
53+
change_multiport_size->run();
6154

62-
add_conn->run();
63-
}
64-
} else if (new_size_ < old_size) {
65-
for (auto i = old_size - 1; i >= new_size_; i--) {
66-
auto add_conn = std::make_shared<MutationAddConnection<Output<PortType>, Input<PortType>>>(
67-
&(*multiport_)[i], get_input_port_((*bank_)[i]), (*bank_)[0]->environment(), false);
55+
auto change_bank_size = std::make_shared<MutationChangeBankSize<std::unique_ptr<ReactorType>>>(
56+
bank_, (*bank_)[0]->environment(), new_size_, create_lambda_);
6857

69-
add_conn->run();
70-
}
58+
change_bank_size->run();
7159

72-
auto change_multiport_size =
73-
std::make_shared<MutationChangeOutputMultiportSize<unsigned>>(multiport_, new_size_);
60+
for (auto i = old_size; i < new_size_; i++) {
61+
auto add_conn = std::make_shared<MutationAddConnection<Output<PortType>, Input<PortType>>>(
62+
&(*multiport_)[i], get_input_port_((*bank_)[i]), (*bank_)[0]->environment(), true);
7463

75-
change_multiport_size->run();
64+
add_conn->run();
65+
}
66+
} else if (new_size_ < old_size) {
67+
for (auto i = old_size - 1; i >= new_size_; i--) {
68+
auto add_conn = std::make_shared<MutationAddConnection<Output<PortType>, Input<PortType>>>(
69+
&(*multiport_)[i], get_input_port_((*bank_)[i]), (*bank_)[0]->environment(), false);
7670

77-
auto change_bank_size = std::make_shared<MutationChangeBankSize<std::unique_ptr<ReactorType>>>(
78-
bank_, (*bank_)[0]->environment(), new_size_, create_lambda_);
71+
add_conn->run();
72+
}
7973

80-
change_bank_size->run();
81-
}
74+
auto change_multiport_size = std::make_shared<MutationChangeOutputMultiportSize<unsigned>>(multiport_, new_size_);
8275

76+
change_multiport_size->run();
8377

84-
return Success;
85-
}
78+
auto change_bank_size = std::make_shared<MutationChangeBankSize<std::unique_ptr<ReactorType>>>(
79+
bank_, (*bank_)[0]->environment(), new_size_, create_lambda_);
8680

87-
auto rollback() -> MutationResult {
88-
return Success;
89-
}
90-
};
91-
}
81+
change_bank_size->run();
82+
}
9283

84+
return Success;
85+
}
9386

87+
auto rollback() -> MutationResult { return Success; }
88+
};
89+
} // namespace reactor
9490

95-
#endif //MULTIPORT_TO_BANK_HH
91+
#endif // MULTIPORT_TO_BANK_HH

‎examples/multiport_mutation/producer.hh‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ private:
1515
unsigned int counter_ = 0;
1616

1717
void reaction_1([[maybe_unused]] Output<unsigned>& out) {
18-
std::cout << "producing value:" << counter_ << "\n";
18+
// std::cout << "producing value:" << counter_ << "\n";
1919
out.set(counter_++);
2020
}
2121

‎include/reactor-cpp/connection.hh‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -143,8 +143,8 @@ public:
143143
};
144144
}
145145

146-
auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
147-
const std::function<bool(void)>& abort_waiting) -> bool override {
146+
auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, const std::function<bool(void)>& abort_waiting)
147+
-> bool override {
148148
reactor_assert(lock.owns_lock());
149149
log_.debug() << "downstream tries to acquire tag " << tag;
150150

@@ -210,8 +210,8 @@ public:
210210
};
211211
}
212212

213-
auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
214-
const std::function<bool(void)>& abort_waiting) -> bool override {
213+
auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, const std::function<bool(void)>& abort_waiting)
214+
-> bool override {
215215
// Since this is a delayed connection, we can go back in time and need to
216216
// acquire the latest upstream tag that can create an event at the given
217217
// tag. We also need to consider that given a delay d and a tag g=(t, n),
@@ -240,8 +240,8 @@ public:
240240
};
241241
}
242242

243-
auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
244-
const std::function<bool(void)>& abort_waiting) -> bool override {
243+
auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, const std::function<bool(void)>& abort_waiting)
244+
-> bool override {
245245
this->log_.debug() << "downstream tries to acquire tag " << tag;
246246
return PhysicalTimeBarrier::acquire_tag(tag, lock, this->environment()->scheduler(), abort_waiting);
247247
}

‎include/reactor-cpp/reactor.hh‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,12 +31,16 @@ private:
3131

3232
void register_action(BaseAction* action);
3333
void register_input(BasePort* port);
34-
void unregister_input(BasePort* port);
3534
void register_output(BasePort* port);
36-
void unregister_output(BasePort* port);
3735
void register_reaction(Reaction* reaction);
3836
void register_reactor(Reactor* reactor);
3937

38+
void unregister_action(BaseAction* action);
39+
void unregister_input(BasePort* port);
40+
void unregister_output(BasePort* port);
41+
void unregister_reaction(Reaction* reaction);
42+
void unregister_reactor(Reactor* reactor);
43+
4044
public:
4145
Reactor(const std::string& name, Reactor* container);
4246
Reactor(const std::string& name, Environment* environment);

‎include/reactor-cpp/statistics.hh‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,12 +54,17 @@ public:
5454
static void increment_reactions() { increment(reactions_); }
5555
static void increment_actions() { increment(actions_); }
5656
static void increment_ports() { increment(ports_); }
57+
5758
static void increment_processed_events() { increment(processed_events_); }
5859
static void increment_processed_reactions() { increment(processed_reactions_); }
5960
static void increment_triggered_actions() { increment(triggered_actions_); }
6061
static void increment_set_ports() { increment(set_ports_); }
6162
static void increment_scheduled_actions() { increment(scheduled_actions_); }
6263

64+
static void decrement_reactor_instances() { decrement(reactor_instances_); }
65+
static void decrement_connections() { decrement(connections_); }
66+
static void decrement_reactions() { decrement(reactions_); }
67+
static void decrement_actions() { decrement(actions_); }
6368
static void decrement_ports() { decrement(ports_); }
6469

6570
static auto reactor_instances() { return reactor_instances_.load(std::memory_order_acquire); }

0 commit comments

Comments
 (0)