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
|
// Copyright 2016 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "mojo/core/ports/port.h"
namespace mojo {
namespace core {
namespace ports {
// Used by std::{push,pop}_heap functions
inline bool operator<(const mozilla::UniquePtr<Event>& a,
const mozilla::UniquePtr<Event>& b) {
return a->control_sequence_num() > b->control_sequence_num();
}
Port::Port(uint64_t next_sequence_num_to_send,
uint64_t next_sequence_num_to_receive)
: state(kUninitialized),
pending_merge_peer(false),
next_control_sequence_num_to_send(kInitialSequenceNum),
next_control_sequence_num_to_receive(kInitialSequenceNum),
next_sequence_num_to_send(next_sequence_num_to_send),
last_sequence_num_acknowledged(next_sequence_num_to_send - 1),
sequence_num_acknowledge_interval(0),
last_sequence_num_to_receive(0),
sequence_num_to_acknowledge(0),
message_queue(next_sequence_num_to_receive),
remove_proxy_on_last_message(false),
peer_closed(false),
peer_lost_unexpectedly(false) {}
Port::~Port() = default;
bool Port::IsNextEvent(const NodeName& from_node, const Event& event) {
if (from_node != prev_node_name) {
return false;
}
if (event.from_port() != prev_port_name) {
return false;
}
DCHECK_GE(event.control_sequence_num(), next_control_sequence_num_to_receive);
return event.control_sequence_num() == next_control_sequence_num_to_receive;
}
void Port::NextEvent(NodeName* from_node, ScopedEvent* event) {
auto it = control_event_queues_.find({prev_node_name, prev_port_name});
if (it == control_event_queues_.end()) {
return;
}
auto& msg_queue = it->second;
// There must always be one entry since we delete the queue after processing
// the last element.
DCHECK_GE(msg_queue.size(), 1lu);
if (msg_queue[0]->control_sequence_num() !=
next_control_sequence_num_to_receive) {
return;
}
std::pop_heap(msg_queue.begin(), msg_queue.end());
*from_node = prev_node_name;
*event = std::move(msg_queue.back());
msg_queue.pop_back();
if (msg_queue.size() == 0) {
control_event_queues_.erase(it);
}
}
void Port::BufferEvent(const NodeName& from_node, ScopedEvent event) {
DCHECK(!IsNextEvent(from_node, *event));
auto& event_heap = control_event_queues_[{from_node, event->from_port()}];
event_heap.emplace_back(std::move(event));
std::push_heap(event_heap.begin(), event_heap.end());
}
void Port::TakePendingMessages(
std::vector<mozilla::UniquePtr<UserMessageEvent>>& messages) {
for (auto& node_queue_pair : control_event_queues_) {
auto& events = node_queue_pair.second;
for (auto& event : events) {
if (event->type() != Event::Type::kUserMessage) continue;
messages.emplace_back(Event::Cast<UserMessageEvent>(&event));
}
}
control_event_queues_.clear();
}
} // namespace ports
} // namespace core
} // namespace mojo
|
