Delete friendship between VirtualSocket and VirtualSocketServer
Bug: webrtc:11567
Change-Id: I07d01b9eed798a69ba798e899b2bae57409ce332
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/207181
Reviewed-by: Tommi <tommi@webrtc.org>
Commit-Queue: Niels Moller <nisse@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#33276}
diff --git a/rtc_base/virtual_socket_server.cc b/rtc_base/virtual_socket_server.cc
index 708ee14..5b0d3af 100644
--- a/rtc_base/virtual_socket_server.cc
+++ b/rtc_base/virtual_socket_server.cc
@@ -149,9 +149,6 @@
} else {
bound_ = true;
was_any_ = addr.IsAnyIP();
- // Post a message here such that test case could have chance to
- // process the local address. (i.e. SetAlternativeLocalAddress).
- server_->msg_queue_->Post(RTC_FROM_HERE, this, MSG_ID_ADDRESS_BOUND);
}
return result;
}
@@ -174,7 +171,7 @@
SocketAddress addr = listen_queue_->front();
// Disconnect listening socket.
- server_->Disconnect(server_->LookupBinding(addr));
+ server_->Disconnect(addr);
listen_queue_->pop_front();
}
delete listen_queue_;
@@ -182,51 +179,14 @@
}
// Disconnect stream sockets
if (CS_CONNECTED == state_) {
- // Disconnect remote socket, check if it is a child of a server socket.
- VirtualSocket* socket =
- server_->LookupConnection(local_addr_, remote_addr_);
- if (!socket) {
- // Not a server socket child, then see if it is bound.
- // TODO(tbd): If this is indeed a server socket that has no
- // children this will cause the server socket to be
- // closed. This might lead to unexpected results, how to fix this?
- socket = server_->LookupBinding(remote_addr_);
- }
- server_->Disconnect(socket);
-
- // Remove mapping for both directions.
- server_->RemoveConnection(remote_addr_, local_addr_);
- server_->RemoveConnection(local_addr_, remote_addr_);
+ server_->Disconnect(local_addr_, remote_addr_);
}
// Cancel potential connects
- MessageList msgs;
- if (server_->msg_queue_) {
- server_->msg_queue_->Clear(this, MSG_ID_CONNECT, &msgs);
- }
- for (MessageList::iterator it = msgs.begin(); it != msgs.end(); ++it) {
- RTC_DCHECK(nullptr != it->pdata);
- MessageAddress* data = static_cast<MessageAddress*>(it->pdata);
-
- // Lookup remote side.
- VirtualSocket* socket =
- server_->LookupConnection(local_addr_, data->addr);
- if (socket) {
- // Server socket, remote side is a socket retreived by
- // accept. Accepted sockets are not bound so we will not
- // find it by looking in the bindings table.
- server_->Disconnect(socket);
- server_->RemoveConnection(local_addr_, data->addr);
- } else {
- server_->Disconnect(server_->LookupBinding(data->addr));
- }
- delete data;
- }
+ server_->CancelConnects(this);
}
// Clear incoming packets and disconnect messages
- if (server_->msg_queue_) {
- server_->msg_queue_->Clear(this);
- }
+ server_->Clear(this);
state_ = CS_CLOSED;
local_addr_.Clear();
@@ -279,9 +239,7 @@
return -1;
}
while (recv_buffer_.empty()) {
- Message msg;
- server_->msg_queue_->Get(&msg);
- server_->msg_queue_->Dispatch(&msg);
+ server_->ProcessOneMessage();
}
}
@@ -301,18 +259,14 @@
// To behave like a real socket, SignalReadEvent should fire in the next
// message loop pass if there's still data buffered.
if (!recv_buffer_.empty()) {
- // Clear the message so it doesn't end up posted multiple times.
- server_->msg_queue_->Clear(this, MSG_ID_SIGNALREADEVENT);
- server_->msg_queue_->Post(RTC_FROM_HERE, this, MSG_ID_SIGNALREADEVENT);
+ server_->PostSignalReadEvent(this);
}
if (SOCK_STREAM == type_) {
- bool was_full = (recv_buffer_size_ == server_->recv_buffer_capacity_);
+ bool was_full = (recv_buffer_size_ == server_->recv_buffer_capacity());
recv_buffer_size_ -= data_read;
if (was_full) {
- VirtualSocket* sender = server_->LookupBinding(remote_addr_);
- RTC_DCHECK(nullptr != sender);
- server_->SendTcp(sender);
+ server_->SendTcp(remote_addr_);
}
}
@@ -410,7 +364,7 @@
} else {
RTC_LOG(LS_VERBOSE) << "Socket at " << local_addr_.ToString()
<< " is not listening";
- server_->Disconnect(server_->LookupBinding(data->addr));
+ server_->Disconnect(data->addr);
}
delete data;
} else if (pmsg->message_id == MSG_ID_DISCONNECT) {
@@ -494,7 +448,7 @@
}
int VirtualSocket::SendTcp(const void* pv, size_t cb) {
- size_t capacity = server_->send_buffer_capacity_ - send_buffer_.size();
+ size_t capacity = server_->send_buffer_capacity() - send_buffer_.size();
if (0 == capacity) {
ready_to_send_ = false;
error_ = EWOULDBLOCK;
@@ -523,6 +477,67 @@
}
}
+void VirtualSocket::SetToBlocked() {
+ CritScope cs(&crit_);
+ ready_to_send_ = false;
+ error_ = EWOULDBLOCK;
+}
+
+void VirtualSocket::UpdateRecv(size_t data_size) {
+ recv_buffer_size_ += data_size;
+}
+
+void VirtualSocket::UpdateSend(size_t data_size) {
+ size_t new_buffer_size = send_buffer_.size() - data_size;
+ // Avoid undefined access beyond the last element of the vector.
+ // This only happens when new_buffer_size is 0.
+ if (data_size < send_buffer_.size()) {
+ // memmove is required for potentially overlapping source/destination.
+ memmove(&send_buffer_[0], &send_buffer_[data_size], new_buffer_size);
+ }
+ send_buffer_.resize(new_buffer_size);
+}
+
+void VirtualSocket::MaybeSignalWriteEvent(size_t capacity) {
+ if (!ready_to_send_ && (send_buffer_.size() < capacity)) {
+ ready_to_send_ = true;
+ SignalWriteEvent(this);
+ }
+}
+
+uint32_t VirtualSocket::AddPacket(int64_t cur_time, size_t packet_size) {
+ network_size_ += packet_size;
+ uint32_t send_delay =
+ server_->SendDelay(static_cast<uint32_t>(network_size_));
+
+ NetworkEntry entry;
+ entry.size = packet_size;
+ entry.done_time = cur_time + send_delay;
+ network_.push_back(entry);
+
+ return send_delay;
+}
+
+int64_t VirtualSocket::UpdateOrderedDelivery(int64_t ts) {
+ // Ensure that new packets arrive after previous ones
+ ts = std::max(ts, last_delivery_time_);
+ // A socket should not have both ordered and unordered delivery, so its last
+ // delivery time only needs to be updated when it has ordered delivery.
+ last_delivery_time_ = ts;
+ return ts;
+}
+
+size_t VirtualSocket::PurgeNetworkPackets(int64_t cur_time) {
+ CritScope cs(&crit_);
+
+ while (!network_.empty() && (network_.front().done_time <= cur_time)) {
+ RTC_DCHECK(network_size_ >= network_.front().size);
+ network_size_ -= network_.front().size;
+ network_.pop_front();
+ }
+ return network_size_;
+}
+
VirtualSocketServer::VirtualSocketServer() : VirtualSocketServer(nullptr) {}
VirtualSocketServer::VirtualSocketServer(ThreadProcessingFakeClock* fake_clock)
@@ -596,9 +611,7 @@
}
VirtualSocket* VirtualSocketServer::CreateSocketInternal(int family, int type) {
- VirtualSocket* socket = new VirtualSocket(this, family, type, true);
- SignalSocketCreated(socket);
- return socket;
+ return new VirtualSocket(this, family, type, true);
}
void VirtualSocketServer::SetMessageQueue(Thread* msg_queue) {
@@ -683,7 +696,13 @@
SocketAddress normalized(addr.ipaddr().Normalized(), addr.port());
AddressMap::value_type entry(normalized, socket);
- return bindings_->insert(entry).second ? 0 : -1;
+ if (bindings_->insert(entry).second) {
+ // Post a message here such that test case could have chance to
+ // process the local address. (i.e. SetAlternativeLocalAddress).
+ msg_queue_->Post(RTC_FROM_HERE, socket, MSG_ID_ADDRESS_BOUND);
+ return 0;
+ }
+ return -1;
}
int VirtualSocketServer::Bind(VirtualSocket* socket, SocketAddress* addr) {
@@ -814,15 +833,79 @@
return false;
}
+bool VirtualSocketServer::Disconnect(const SocketAddress& addr) {
+ return Disconnect(LookupBinding(addr));
+}
+
+bool VirtualSocketServer::Disconnect(const SocketAddress& local_addr,
+ const SocketAddress& remote_addr) {
+ // Disconnect remote socket, check if it is a child of a server socket.
+ VirtualSocket* socket = LookupConnection(local_addr, remote_addr);
+ if (!socket) {
+ // Not a server socket child, then see if it is bound.
+ // TODO(tbd): If this is indeed a server socket that has no
+ // children this will cause the server socket to be
+ // closed. This might lead to unexpected results, how to fix this?
+ socket = LookupBinding(remote_addr);
+ }
+ Disconnect(socket);
+
+ // Remove mapping for both directions.
+ RemoveConnection(remote_addr, local_addr);
+ RemoveConnection(local_addr, remote_addr);
+ return socket != nullptr;
+}
+
+void VirtualSocketServer::CancelConnects(VirtualSocket* socket) {
+ MessageList msgs;
+ if (msg_queue_) {
+ msg_queue_->Clear(socket, MSG_ID_CONNECT, &msgs);
+ }
+ for (MessageList::iterator it = msgs.begin(); it != msgs.end(); ++it) {
+ RTC_DCHECK(nullptr != it->pdata);
+ MessageAddress* data = static_cast<MessageAddress*>(it->pdata);
+ SocketAddress local_addr = socket->GetLocalAddress();
+ // Lookup remote side.
+ VirtualSocket* socket = LookupConnection(local_addr, data->addr);
+ if (socket) {
+ // Server socket, remote side is a socket retreived by
+ // accept. Accepted sockets are not bound so we will not
+ // find it by looking in the bindings table.
+ Disconnect(socket);
+ RemoveConnection(local_addr, data->addr);
+ } else {
+ Disconnect(data->addr);
+ }
+ delete data;
+ }
+}
+
+void VirtualSocketServer::Clear(VirtualSocket* socket) {
+ // Clear incoming packets and disconnect messages
+ if (msg_queue_) {
+ msg_queue_->Clear(socket);
+ }
+}
+
+void VirtualSocketServer::ProcessOneMessage() {
+ Message msg;
+ msg_queue_->Get(&msg);
+ msg_queue_->Dispatch(&msg);
+}
+
+void VirtualSocketServer::PostSignalReadEvent(VirtualSocket* socket) {
+ // Clear the message so it doesn't end up posted multiple times.
+ msg_queue_->Clear(socket, MSG_ID_SIGNALREADEVENT);
+ msg_queue_->Post(RTC_FROM_HERE, socket, MSG_ID_SIGNALREADEVENT);
+}
+
int VirtualSocketServer::SendUdp(VirtualSocket* socket,
const char* data,
size_t data_size,
const SocketAddress& remote_addr) {
++sent_packets_;
if (sending_blocked_) {
- CritScope cs(&socket->crit_);
- socket->ready_to_send_ = false;
- socket->error_ = EWOULDBLOCK;
+ socket->SetToBlocked();
return -1;
}
@@ -856,10 +939,8 @@
}
{
- CritScope cs(&socket->crit_);
-
int64_t cur_time = TimeMillis();
- PurgeNetworkPackets(socket, cur_time);
+ size_t network_size = socket->PurgeNetworkPackets(cur_time);
// Determine whether we have enough bandwidth to accept this packet. To do
// this, we need to update the send queue. Once we know it's current size,
@@ -870,7 +951,7 @@
// simulation of what a normal network would do.
size_t packet_size = data_size + UDP_HEADER_SIZE;
- if (socket->network_size_ + packet_size > network_capacity_) {
+ if (network_size + packet_size > network_capacity_) {
RTC_LOG(LS_VERBOSE) << "Dropping packet: network capacity exceeded";
return static_cast<int>(data_size);
}
@@ -898,45 +979,36 @@
// Lookup the local/remote pair in the connections table.
VirtualSocket* recipient =
- LookupConnection(socket->local_addr_, socket->remote_addr_);
+ LookupConnection(socket->GetLocalAddress(), socket->GetRemoteAddress());
if (!recipient) {
RTC_LOG(LS_VERBOSE) << "Sending data to no one.";
return;
}
- CritScope cs(&socket->crit_);
-
int64_t cur_time = TimeMillis();
- PurgeNetworkPackets(socket, cur_time);
+ socket->PurgeNetworkPackets(cur_time);
while (true) {
- size_t available = recv_buffer_capacity_ - recipient->recv_buffer_size_;
+ size_t available = recv_buffer_capacity_ - recipient->recv_buffer_size();
size_t max_data_size =
std::min<size_t>(available, TCP_MSS - TCP_HEADER_SIZE);
- size_t data_size = std::min(socket->send_buffer_.size(), max_data_size);
+ size_t data_size = std::min(socket->send_buffer_size(), max_data_size);
if (0 == data_size)
break;
- AddPacketToNetwork(socket, recipient, cur_time, &socket->send_buffer_[0],
+ AddPacketToNetwork(socket, recipient, cur_time, socket->send_buffer_data(),
data_size, TCP_HEADER_SIZE, true);
- recipient->recv_buffer_size_ += data_size;
-
- size_t new_buffer_size = socket->send_buffer_.size() - data_size;
- // Avoid undefined access beyond the last element of the vector.
- // This only happens when new_buffer_size is 0.
- if (data_size < socket->send_buffer_.size()) {
- // memmove is required for potentially overlapping source/destination.
- memmove(&socket->send_buffer_[0], &socket->send_buffer_[data_size],
- new_buffer_size);
- }
- socket->send_buffer_.resize(new_buffer_size);
+ recipient->UpdateRecv(data_size);
+ socket->UpdateSend(data_size);
}
- if (!socket->ready_to_send_ &&
- (socket->send_buffer_.size() < send_buffer_capacity_)) {
- socket->ready_to_send_ = true;
- socket->SignalWriteEvent(socket);
- }
+ socket->MaybeSignalWriteEvent(send_buffer_capacity_);
+}
+
+void VirtualSocketServer::SendTcp(const SocketAddress& addr) {
+ VirtualSocket* sender = LookupBinding(addr);
+ RTC_DCHECK(nullptr != sender);
+ SendTcp(sender);
}
void VirtualSocketServer::AddPacketToNetwork(VirtualSocket* sender,
@@ -946,13 +1018,7 @@
size_t data_size,
size_t header_size,
bool ordered) {
- VirtualSocket::NetworkEntry entry;
- entry.size = data_size + header_size;
-
- sender->network_size_ += entry.size;
- uint32_t send_delay = SendDelay(static_cast<uint32_t>(sender->network_size_));
- entry.done_time = cur_time + send_delay;
- sender->network_.push_back(entry);
+ uint32_t send_delay = sender->AddPacket(cur_time, data_size + header_size);
// Find the delay for crossing the many virtual hops of the network.
uint32_t transit_delay = GetTransitDelay(sender);
@@ -960,7 +1026,7 @@
// When the incoming packet is from a binding of the any address, translate it
// to the default route here such that the recipient will see the default
// route.
- SocketAddress sender_addr = sender->local_addr_;
+ SocketAddress sender_addr = sender->GetLocalAddress();
IPAddress default_ip = GetDefaultRoute(sender_addr.ipaddr().family());
if (sender_addr.IsAnyIP() && !IPIsUnspec(default_ip)) {
sender_addr.SetIP(default_ip);
@@ -971,25 +1037,11 @@
int64_t ts = TimeAfter(send_delay + transit_delay);
if (ordered) {
- // Ensure that new packets arrive after previous ones
- ts = std::max(ts, sender->last_delivery_time_);
- // A socket should not have both ordered and unordered delivery, so its last
- // delivery time only needs to be updated when it has ordered delivery.
- sender->last_delivery_time_ = ts;
+ ts = sender->UpdateOrderedDelivery(ts);
}
msg_queue_->PostAt(RTC_FROM_HERE, ts, recipient, MSG_ID_PACKET, p);
}
-void VirtualSocketServer::PurgeNetworkPackets(VirtualSocket* socket,
- int64_t cur_time) {
- while (!socket->network_.empty() &&
- (socket->network_.front().done_time <= cur_time)) {
- RTC_DCHECK(socket->network_size_ >= socket->network_.front().size);
- socket->network_size_ -= socket->network_.front().size;
- socket->network_.pop_front();
- }
-}
-
uint32_t VirtualSocketServer::SendDelay(uint32_t size) {
if (bandwidth_ == 0)
return 0;