Skip to content

Commit 0d54f3f

Browse files
eduarduseldorfeduarduseldorf
authored
RB-423 removing unnecessary data structures (#95)
* feat(rtbot): data and control queues max amount enforced * fix(rtbot): removing unnecessary data structures * feat(rtbot): operator == implementation * fix(rtbot): wrong formating
1 parent 72f551a commit 0d54f3f

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

67 files changed

+2461
-1341
lines changed

‎libs/api/include/rtbot/Program.h‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -114,14 +114,14 @@ class Program {
114114

115115
// Message processing
116116
ProgramMsgBatch receive(const Message<NumberData>& msg, const std::string& port_id = "i1") {
117-
send_to_entry(msg, port_id);
117+
send_to_entry(msg, port_id, false);
118118
ProgramMsgBatch result = collect_outputs(false);
119119
clear_all_outputs();
120120
return result;
121121
}
122122

123123
ProgramMsgBatch receive_debug(const Message<NumberData>& msg, const std::string& port_id = "i1") {
124-
send_to_entry(msg, port_id);
124+
send_to_entry(msg, port_id, true);
125125
ProgramMsgBatch result = collect_outputs(true);
126126
clear_all_outputs();
127127
return result;
@@ -229,10 +229,10 @@ class Program {
229229
throw runtime_error("Could not resolve operator ID: " + id);
230230
}
231231

232-
void send_to_entry(const Message<NumberData>& msg, const std::string& port_id) {
232+
void send_to_entry(const Message<NumberData>& msg, const std::string& port_id, bool debug=false) {
233233
auto port_info = OperatorJson::parse_port_name(port_id);
234234
operators_[entry_operator_id_]->receive_data(create_message<NumberData>(msg.time, msg.data), port_info.index);
235-
operators_[entry_operator_id_]->execute();
235+
operators_[entry_operator_id_]->execute(debug);
236236
}
237237

238238
ProgramMsgBatch collect_outputs(bool debug_mode = false) {
@@ -255,7 +255,7 @@ class Program {
255255
// In debug mode, collect all ports
256256
if (debug_mode) {
257257
for (size_t i = 0; i < op->num_output_ports(); i++) {
258-
const auto& queue = op->get_output_queue(i);
258+
const auto& queue = op->get_debug_output_queue(i);
259259
if (!queue.empty()) {
260260
PortMsgBatch port_msgs;
261261
for (const auto& msg : queue) {

‎libs/core/include/rtbot/Buffer.h‎

Lines changed: 78 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -72,11 +72,38 @@ class Buffer : public Operator {
7272
return std::sqrt(variance());
7373
}
7474

75-
Bytes collect() override {
76-
Bytes bytes = Operator::collect();
75+
bool equals(const Buffer& other) const {
76+
77+
if (window_size_ != other.window_size_) return false;
78+
79+
if (buffer_.size() != other.buffer_.size()) return false;
80+
81+
auto it1 = buffer_.begin();
82+
auto it2 = other.buffer_.begin();
83+
84+
for (; it1 != buffer_.end() && it2 != other.buffer_.end(); ++it1, ++it2) {
85+
const auto& msg1 = *it1;
86+
const auto& msg2 = *it2;
7787

78-
bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&window_size_),
79-
reinterpret_cast<const uint8_t*>(&window_size_) + sizeof(window_size_));
88+
if (msg1 && msg2) {
89+
if (msg1->time != msg2->time) return false;
90+
if (msg1->hash() != msg2->hash()) return false;
91+
} else return false;
92+
}
93+
94+
if constexpr (Features::TRACK_SUM) {
95+
if (StateSerializer::hash_double(sum_) != StateSerializer::hash_double(other.sum_)) return false;
96+
}
97+
98+
if constexpr (Features::TRACK_VARIANCE) {
99+
if (StateSerializer::hash_double(M2_) != StateSerializer::hash_double(other.M2_)) return false;
100+
}
101+
102+
return Operator::equals(other);
103+
}
104+
105+
Bytes collect() override {
106+
Bytes bytes = Operator::collect();
80107

81108
size_t buffer_size = buffer_.size();
82109
bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&buffer_size),
@@ -90,65 +117,68 @@ class Buffer : public Operator {
90117
bytes.insert(bytes.end(), msg_bytes.begin(), msg_bytes.end());
91118
}
92119

93-
if constexpr (Features::TRACK_SUM) {
94-
bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&sum_),
95-
reinterpret_cast<const uint8_t*>(&sum_) + sizeof(sum_));
96-
}
97-
98-
if constexpr (Features::TRACK_VARIANCE) {
99-
bytes.insert(bytes.end(), reinterpret_cast<const uint8_t*>(&M2_),
100-
reinterpret_cast<const uint8_t*>(&M2_) + sizeof(M2_));
101-
}
102-
103120
return bytes;
104121
}
105122

106123
void restore(Bytes::const_iterator& it) override {
107-
Operator::restore(it);
108-
109-
window_size_ = *reinterpret_cast<const size_t*>(&(*it));
110-
it += sizeof(size_t);
124+
// Call base restore first
125+
Operator::restore(it);
111126

112-
size_t buffer_size = *reinterpret_cast<const size_t*>(&(*it));
113-
it += sizeof(size_t);
127+
// ---- Read buffer_size safely ----
128+
size_t buffer_size;
129+
std::memcpy(&buffer_size, &(*it), sizeof(buffer_size));
130+
it += sizeof(buffer_size);
114131

132+
// ---- Deserialize buffer ----
115133
buffer_.clear();
116134
for (size_t i = 0; i < buffer_size; ++i) {
117-
size_t msg_size = *reinterpret_cast<const size_t*>(&(*it));
118-
it += sizeof(size_t);
119-
120-
Bytes msg_bytes(it, it + msg_size);
121-
buffer_.push_back(
122-
std::unique_ptr<Message<T>>(dynamic_cast<Message<T>*>(BaseMessage::deserialize(msg_bytes).release())));
123-
it += msg_size;
135+
// Read size of each message
136+
size_t msg_size;
137+
std::memcpy(&msg_size, &(*it), sizeof(msg_size));
138+
it += sizeof(msg_size);
139+
140+
// Extract message bytes
141+
Bytes msg_bytes(it, it + msg_size);
142+
143+
// Deserialize message and cast to derived type
144+
buffer_.push_back(
145+
std::unique_ptr<Message<T>>(
146+
dynamic_cast<Message<T>*>(BaseMessage::deserialize(msg_bytes).release())
147+
)
148+
);
149+
150+
it += msg_size;
124151
}
125152

153+
// ---- Optional statistics ----
126154
if constexpr (Features::TRACK_SUM) {
127-
sum_ = *reinterpret_cast<const double*>(&(*it));
128-
it += sizeof(double);
155+
sum_ = 0.0;
156+
if (!buffer_.empty()) {
157+
// First pass: compute sum
158+
for (const auto& msg : buffer_) {
159+
sum_ += msg->data.value;
160+
}
161+
}
129162
}
130163

131164
if constexpr (Features::TRACK_VARIANCE) {
132-
M2_ = *reinterpret_cast<const double*>(&(*it));
133-
it += sizeof(double);
134-
135-
// Recompute statistics from buffer to ensure consistency
136-
sum_ = 0.0;
137-
M2_ = 0.0;
138-
139-
if (!buffer_.empty()) {
140-
// First pass: compute mean
141-
for (const auto& msg : buffer_) {
142-
sum_ += msg->data.value;
165+
// Recompute statistics from buffer to ensure consistency
166+
sum_ = 0.0;
167+
M2_ = 0.0;
168+
169+
if (!buffer_.empty()) {
170+
// First pass: compute sum
171+
for (const auto& msg : buffer_) {
172+
sum_ += msg->data.value;
173+
}
174+
175+
// Second pass: compute M2
176+
double mean = sum_ / buffer_.size();
177+
for (const auto& msg : buffer_) {
178+
double delta = msg->data.value - mean;
179+
M2_ += delta * delta;
180+
}
143181
}
144-
145-
// Second pass: compute M2
146-
double mean = sum_ / buffer_.size();
147-
for (const auto& msg : buffer_) {
148-
double delta = msg->data.value - mean;
149-
M2_ += delta * delta;
150-
}
151-
}
152182
}
153183
}
154184

‎libs/core/include/rtbot/Demultiplexer.h‎

Lines changed: 43 additions & 126 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,11 @@ class Demultiplexer : public Operator {
2020
}
2121

2222
// Add single data input port with type T
23-
add_data_port<T>();
24-
data_time_tracker_ = std::set<timestamp_t>();
23+
add_data_port<T>();
2524

2625
// Add corresponding control ports (always boolean)
2726
for (size_t i = 0; i < num_ports; ++i) {
28-
add_control_port<BooleanData>();
29-
control_time_tracker_[i] = std::map<timestamp_t, bool>();
27+
add_control_port<BooleanData>();
3028
}
3129

3230
// Add output ports (same type as input)
@@ -37,143 +35,62 @@ class Demultiplexer : public Operator {
3735

3836
std::string type_name() const override { return "Demultiplexer"; }
3937

40-
size_t get_num_ports() const { return control_time_tracker_.size(); }
38+
size_t get_num_ports() const { return num_control_ports(); }
4139

42-
Bytes collect() override {
43-
Bytes bytes = Operator::collect(); // First collect base state
44-
45-
// Serialize data time tracker
46-
StateSerializer::serialize_timestamp_set(bytes, data_time_tracker_);
47-
48-
// Serialize control time tracker
49-
StateSerializer::serialize_port_control_map(bytes, control_time_tracker_);
50-
51-
return bytes;
40+
bool equals(const Demultiplexer& other) const {
41+
return Operator::equals(other);
5242
}
53-
54-
void restore(Bytes::const_iterator& it) override {
55-
// First restore base state
56-
Operator::restore(it);
57-
58-
// Clear current state
59-
data_time_tracker_.clear();
60-
control_time_tracker_.clear();
61-
62-
// Restore data time tracker
63-
StateSerializer::deserialize_timestamp_set(it, data_time_tracker_);
64-
65-
// Restore control time tracker
66-
StateSerializer::deserialize_port_control_map(it, control_time_tracker_);
67-
68-
// Validate control port count
69-
StateSerializer::validate_port_count(control_time_tracker_.size(), num_control_ports(), "Control");
43+
44+
bool operator==(const Demultiplexer& other) const {
45+
return equals(other);
7046
}
7147

72-
void reset() override {
73-
Operator::reset();
74-
data_time_tracker_.clear();
75-
control_time_tracker_.clear();
48+
bool operator!=(const Demultiplexer& other) const {
49+
return !(*this == other);
7650
}
7751

78-
void receive_data(std::unique_ptr<BaseMessage> msg, size_t port_index) override {
79-
auto time = msg->time;
80-
Operator::receive_data(std::move(msg), port_index);
81-
82-
data_time_tracker_.insert(time);
83-
}
52+
protected:
8453

85-
void receive_control(std::unique_ptr<BaseMessage> msg, size_t port_index) override {
86-
if (port_index >= num_control_ports()) {
87-
throw std::runtime_error("Invalid control port index");
88-
}
89-
90-
auto* ctrl_msg = dynamic_cast<const Message<BooleanData>*>(msg.get());
91-
if (!ctrl_msg) {
92-
throw std::runtime_error("Invalid control message type");
93-
}
94-
95-
// Update control tracker
96-
control_time_tracker_[port_index][ctrl_msg->time] = ctrl_msg->data.value;
97-
98-
// Add message to queue
99-
get_control_queue(port_index).push_back(std::move(msg));
100-
control_ports_with_new_data_.insert(port_index);
101-
}
102-
103-
protected:
10454
void process_data() override {
105-
while (true) {
106-
// Find oldest common control timestamp
107-
auto common_control_time = TimestampTracker::find_oldest_common_time(control_time_tracker_);
108-
if (!common_control_time) {
109-
break;
110-
}
111-
112-
// Clean up any old input data messages
113-
auto& data_queue = get_data_queue(0);
114-
while (!data_queue.empty()) {
115-
auto* msg = dynamic_cast<const Message<T>*>(data_queue.front().get());
116-
if (msg && msg->time < *common_control_time) {
117-
data_time_tracker_.erase(msg->time);
118-
data_queue.pop_front();
119-
} else {
120-
break;
121-
}
122-
}
123-
124-
// Look for matching data message
125-
bool message_found = false;
126-
if (!data_queue.empty()) {
127-
auto* msg = dynamic_cast<const Message<T>*>(data_queue.front().get());
128-
if (msg && msg->time == *common_control_time) {
129-
// Get active control ports
130-
std::vector<size_t> active_ports;
131-
for (size_t i = 0; i < num_control_ports(); ++i) {
132-
if (control_time_tracker_[i].at(*common_control_time)) {
133-
active_ports.push_back(i);
134-
}
55+
while(true) {
56+
57+
bool is_any_control_empty;
58+
bool are_controls_sync;
59+
do {
60+
is_any_control_empty = false;
61+
are_controls_sync = sync_control_inputs();
62+
for (int i=0; i < num_control_ports(); i++) {
63+
if (get_control_queue(i).empty()) {
64+
is_any_control_empty = true;
65+
break;
13566
}
67+
}
68+
} while (!are_controls_sync && !is_any_control_empty );
13669

137-
// Route message to all active ports
138-
for (size_t port : active_ports) {
139-
get_output_queue(port).push_back(data_queue.front()->clone());
140-
}
141-
142-
data_time_tracker_.erase(msg->time);
143-
data_queue.pop_front();
144-
message_found = true;
145-
}
146-
}
147-
148-
clean_up_control_messages(*common_control_time);
70+
if (!are_controls_sync) return;
14971

150-
if (!message_found) {
151-
break;
152-
}
153-
}
154-
}
155-
156-
private:
157-
void clean_up_control_messages(timestamp_t time) {
158-
for (auto& [port, tracker] : control_time_tracker_) {
159-
tracker.erase(time);
160-
}
161-
162-
for (size_t port = 0; port < num_control_ports(); ++port) {
163-
auto& queue = get_control_queue(port);
164-
while (!queue.empty()) {
165-
auto* msg = dynamic_cast<const Message<BooleanData>*>(queue.front().get());
166-
if (msg && msg->time <= time) {
167-
queue.pop_front();
168-
} else {
169-
break;
72+
auto& data_queue = get_data_queue(0);
73+
if (data_queue.empty()) return;
74+
auto* msg = dynamic_cast<const Message<T>*>(data_queue.front().get());
75+
auto* ctrl_msg = dynamic_cast<const Message<BooleanData>*>(get_control_queue(0).front().get());
76+
if (msg && ctrl_msg && msg->time == ctrl_msg->time) {
77+
for (int i = 0; i < num_control_ports(); i++) {
78+
ctrl_msg = dynamic_cast<const Message<BooleanData>*>(get_control_queue(i).front().get());
79+
if (ctrl_msg->data.value) {
80+
get_output_queue(i).push_back(data_queue.front()->clone());
81+
}
82+
get_control_queue(i).pop_front();
17083
}
84+
data_queue.pop_front();
85+
} else if (msg && ctrl_msg && msg->time < ctrl_msg->time) {
86+
data_queue.pop_front();
87+
} else if (msg && ctrl_msg && ctrl_msg->time < msg->time) {
88+
for (int i = 0; i < num_control_ports(); i++)
89+
get_control_queue(i).pop_front();
90+
17191
}
17292
}
17393
}
174-
175-
std::set<timestamp_t> data_time_tracker_;
176-
std::map<size_t, std::map<timestamp_t, bool>> control_time_tracker_;
17794
};
17895

17996
// Factory functions for common configurations using PortType

0 commit comments

Comments
 (0)