/*
* Copyright 2004 The WebRTC Project Authors. All rights reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* in the file PATENTS. All contributing project authors may
* be found in the AUTHORS file in the root of the source tree.
*/
#include "webrtc/base/virtualsocketserver.h"
#include <errno.h>
#include <math.h>
#include <algorithm>
#include <map>
#include <vector>
#include "webrtc/base/common.h"
#include "webrtc/base/logging.h"
#include "webrtc/base/physicalsocketserver.h"
#include "webrtc/base/socketaddresspair.h"
#include "webrtc/base/thread.h"
#include "webrtc/base/timeutils.h"
namespace rtc {
#if defined(WEBRTC_WIN)
const in_addr kInitialNextIPv4 = { {0x01, 0, 0, 0} };
#else
// This value is entirely arbitrary, hence the lack of concern about endianness.
const in_addr kInitialNextIPv4 = { 0x01000000 };
#endif
// Starts at ::2 so as to not cause confusion with ::1.
const in6_addr kInitialNextIPv6 = { { {
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2
} } };
const uint16 kFirstEphemeralPort = 49152;
const uint16 kLastEphemeralPort = 65535;
const uint16 kEphemeralPortCount = kLastEphemeralPort - kFirstEphemeralPort + 1;
const uint32 kDefaultNetworkCapacity = 64 * 1024;
const uint32 kDefaultTcpBufferSize = 32 * 1024;
const uint32 UDP_HEADER_SIZE = 28; // IP + UDP headers
const uint32 TCP_HEADER_SIZE = 40; // IP + TCP headers
const uint32 TCP_MSS = 1400; // Maximum segment size
// Note: The current algorithm doesn't work for sample sizes smaller than this.
const int NUM_SAMPLES = 1000;
enum {
MSG_ID_PACKET,
MSG_ID_CONNECT,
MSG_ID_DISCONNECT,
};
// Packets are passed between sockets as messages. We copy the data just like
// the kernel does.
class Packet : public MessageData {
public:
Packet(const char* data, size_t size, const SocketAddress& from)
: size_(size), consumed_(0), from_(from) {
ASSERT(NULL != data);
data_ = new char[size_];
memcpy(data_, data, size_);
}
virtual ~Packet() {
delete[] data_;
}
const char* data() const { return data_ + consumed_; }
size_t size() const { return size_ - consumed_; }
const SocketAddress& from() const { return from_; }
// Remove the first size bytes from the data.
void Consume(size_t size) {
ASSERT(size + consumed_ < size_);
consumed_ += size;
}
private:
char* data_;
size_t size_, consumed_;
SocketAddress from_;
};
struct MessageAddress : public MessageData {
explicit MessageAddress(const SocketAddress& a) : addr(a) { }
SocketAddress addr;
};
// Implements the socket interface using the virtual network. Packets are
// passed as messages using the message queue of the socket server.
class VirtualSocket : public AsyncSocket, public MessageHandler {
public:
VirtualSocket(VirtualSocketServer* server, int family, int type, bool async)
: server_(server), family_(family), type_(type), async_(async),
state_(CS_CLOSED), error_(0), listen_queue_(NULL),
write_enabled_(false),
network_size_(0), recv_buffer_size_(0), bound_(false), was_any_(false) {
ASSERT((type_ == SOCK_DGRAM) || (type_ == SOCK_STREAM));
ASSERT(async_ || (type_ != SOCK_STREAM)); // We only support async streams
}
virtual ~VirtualSocket() {
Close();
for (RecvBuffer::iterator it = recv_buffer_.begin();
it != recv_buffer_.end(); ++it) {
delete *it;
}
}
virtual SocketAddress GetLocalAddress() const {
return local_addr_;
}
virtual SocketAddress GetRemoteAddress() const {
return remote_addr_;
}
// Used by server sockets to set the local address without binding.
void SetLocalAddress(const SocketAddress& addr) {
local_addr_ = addr;
}
virtual int Bind(const SocketAddress& addr) {
if (!local_addr_.IsNil()) {
error_ = EINVAL;
return -1;
}
local_addr_ = addr;
int result = server_->Bind(this, &local_addr_);
if (result != 0) {
local_addr_.Clear();
error_ = EADDRINUSE;
} else {
bound_ = true;
was_any_ = addr.IsAnyIP();
}
return result;
}
virtual int Connect(const SocketAddress& addr) {
return InitiateConnect(addr, true);
}
virtual int Close() {
if (!local_addr_.IsNil() && bound_) {
// Remove from the binding table.
server_->Unbind(local_addr_, this);
bound_ = false;
}
if (SOCK_STREAM == type_) {
// Cancel pending sockets
if (listen_queue_) {
while (!listen_queue_->empty()) {
SocketAddress addr = listen_queue_->front();
// Disconnect listening socket.
server_->Disconnect(server_->LookupBinding(addr));
listen_queue_->pop_front();
}
delete listen_queue_;
listen_queue_ = NULL;
}
// 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: 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_);
}
// 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) {
ASSERT(NULL != 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;
}
// Clear incoming packets and disconnect messages
if (server_->msg_queue_) {
server_->msg_queue_->Clear(this);
}
}
state_ = CS_CLOSED;
local_addr_.Clear();
remote_addr_.Clear();
return 0;
}
virtual int Send(const void *pv, size_t cb) {
if (CS_CONNECTED != state_) {
error_ = ENOTCONN;
return -1;
}
if (SOCK_DGRAM == type_) {
return SendUdp(pv, cb, remote_addr_);
} else {
return SendTcp(pv, cb);
}
}
virtual int SendTo(const void *pv, size_t cb, const SocketAddress& addr) {
if (SOCK_DGRAM == type_) {
return SendUdp(pv, cb, addr);
} else {
if (CS_CONNECTED != state_) {
error_ = ENOTCONN;
return -1;
}
return SendTcp(pv, cb);
}
}
virtual int Recv(void *pv, size_t cb) {
SocketAddress addr;
return RecvFrom(pv, cb, &addr);
}
virtual int RecvFrom(void *pv, size_t cb, SocketAddress *paddr) {
// If we don't have a packet, then either error or wait for one to arrive.
if (recv_buffer_.empty()) {
if (async_) {
error_ = EAGAIN;
return -1;
}
while (recv_buffer_.empty()) {
Message msg;
server_->msg_queue_->Get(&msg);
server_->msg_queue_->Dispatch(&msg);
}
}
// Return the packet at the front of the queue.
Packet* packet = recv_buffer_.front();
size_t data_read = _min(cb, packet->size());
memcpy(pv, packet->data(), data_read);
*paddr = packet->from();
if (data_read < packet->size()) {
packet->Consume(data_read);
} else {
recv_buffer_.pop_front();
delete packet;
}
if (SOCK_STREAM == type_) {
bool was_full = (recv_buffer_size_ == server_->recv_buffer_capacity_);
recv_buffer_size_ -= data_read;
if (was_full) {
VirtualSocket* sender = server_->LookupBinding(remote_addr_);
ASSERT(NULL != sender);
server_->SendTcp(sender);
}
}
return static_cast<int>(data_read);
}
virtual int Listen(int backlog) {
ASSERT(SOCK_STREAM == type_);
ASSERT(CS_CLOSED == state_);
if (local_addr_.IsNil()) {
error_ = EINVAL;
return -1;
}
ASSERT(NULL == listen_queue_);
listen_queue_ = new ListenQueue;
state_ = CS_CONNECTING;
return 0;
}
virtual VirtualSocket* Accept(SocketAddress *paddr) {
if (NULL == listen_queue_) {
error_ = EINVAL;
return NULL;
}
while (!listen_queue_->empty()) {
VirtualSocket* socket = new VirtualSocket(server_, AF_INET, type_,
async_);
// Set the new local address to the same as this server socket.
socket->SetLocalAddress(local_addr_);
// Sockets made from a socket that 'was Any' need to inherit that.
socket->set_was_any(was_any_);
SocketAddress remote_addr(listen_queue_->front());
int result = socket->InitiateConnect(remote_addr, false);
listen_queue_->pop_front();
if (result != 0) {
delete socket;
continue;
}
socket->CompleteConnect(remote_addr, false);
if (paddr) {
*paddr = remote_addr;
}
return socket;
}
error_ = EWOULDBLOCK;
return NULL;
}
virtual int GetError() const {
return error_;
}
virtual void SetError(int error) {
error_ = error;
}
virtual ConnState GetState() const {
return state_;
}
virtual int GetOption(Option opt, int* value) {
OptionsMap::const_iterator it = options_map_.find(opt);
if (it == options_map_.end()) {
return -1;
}
*value = it->second;
return 0; // 0 is success to emulate getsockopt()
}
virtual int SetOption(Option opt, int value) {
options_map_[opt] = value;
return 0; // 0 is success to emulate setsockopt()
}
virtual int EstimateMTU(uint16* mtu) {
if (CS_CONNECTED != state_)
return ENOTCONN;
else
return 65536;
}
void OnMessage(Message *pmsg) {
if (pmsg->message_id == MSG_ID_PACKET) {
//ASSERT(!local_addr_.IsAny());
ASSERT(NULL != pmsg->pdata);
Packet* packet = static_cast<Packet*>(pmsg->pdata);
recv_buffer_.push_back(packet);
if (async_) {
SignalReadEvent(this);
}
} else if (pmsg->message_id == MSG_ID_CONNECT) {
ASSERT(NULL != pmsg->pdata);
MessageAddress* data = static_cast<MessageAddress*>(pmsg->pdata);
if (listen_queue_ != NULL) {
listen_queue_->push_back(data->addr);
if (async_) {
SignalReadEvent(this);
}
} else if ((SOCK_STREAM == type_) && (CS_CONNECTING == state_)) {
CompleteConnect(data->addr, true);
} else {
LOG(LS_VERBOSE) << "Socket at " << local_addr_ << " is not listening";
server_->Disconnect(server_->LookupBinding(data->addr));
}
delete data;
} else if (pmsg->message_id == MSG_ID_DISCONNECT) {
ASSERT(SOCK_STREAM == type_);
if (CS_CLOSED != state_) {
int error = (CS_CONNECTING == state_) ? ECONNREFUSED : 0;
state_ = CS_CLOSED;
remote_addr_.Clear();
if (async_) {
SignalCloseEvent(this, error);
}
}
} else {
ASSERT(false);
}
}
bool was_any() { return was_any_; }
void set_was_any(bool was_any) { was_any_ = was_any; }
private:
struct NetworkEntry {
size_t size;
uint32 done_time;
};
typedef std::deque<SocketAddress> ListenQueue;
typedef std::deque<NetworkEntry> NetworkQueue;
typedef std::vector<char> SendBuffer;
typedef std::list<Packet*> RecvBuffer;
typedef std::map<Option, int> OptionsMap;
int InitiateConnect(const SocketAddress& addr, bool use_delay) {
if (!remote_addr_.IsNil()) {
error_ = (CS_CONNECTED == state_) ? EISCONN : EINPROGRESS;
return -1;
}
if (local_addr_.IsNil()) {
// If there's no local address set, grab a random one in the correct AF.
int result = 0;
if (addr.ipaddr().family() == AF_INET) {
result = Bind(SocketAddress("0.0.0.0", 0));
} else if (addr.ipaddr().family() == AF_INET6) {
result = Bind(SocketAddress("::", 0));
}
if (result != 0) {
return result;
}
}
if (type_ == SOCK_DGRAM) {
remote_addr_ = addr;
state_ = CS_CONNECTED;
} else {
int result = server_->Connect(this, addr, use_delay);
if (result != 0) {
error_ = EHOSTUNREACH;
return -1;
}
state_ = CS_CONNECTING;
}
return 0;
}
void CompleteConnect(const SocketAddress& addr, bool notify) {
ASSERT(CS_CONNECTING == state_);
remote_addr_ = addr;
state_ = CS_CONNECTED;
server_->AddConnection(remote_addr_, local_addr_, this);
if (async_ && notify) {
SignalConnectEvent(this);
}
}
int SendUdp(const void* pv, size_t cb, const SocketAddress& addr) {
// If we have not been assigned a local port, then get one.
if (local_addr_.IsNil()) {
local_addr_ = EmptySocketAddressWithFamily(addr.ipaddr().family());
int result = server_->Bind(this, &local_addr_);
if (result != 0) {
local_addr_.Clear();
error_ = EADDRINUSE;
return result;
}
}
// Send the data in a message to the appropriate socket.
return server_->SendUdp(this, static_cast<const char*>(pv), cb, addr);
}
int SendTcp(const void* pv, size_t cb) {
size_t capacity = server_->send_buffer_capacity_ - send_buffer_.size();
if (0 == capacity) {
write_enabled_ = true;
error_ = EWOULDBLOCK;
return -1;
}
size_t consumed = _min(cb, capacity);
const char* cpv = static_cast<const char*>(pv);
send_buffer_.insert(send_buffer_.end(), cpv, cpv + consumed);
server_->SendTcp(this);
return static_cast<int>(consumed);
}
VirtualSocketServer* server_;
int family_;
int type_;
bool async_;
ConnState state_;
int error_;
SocketAddress local_addr_;
SocketAddress remote_addr_;
// Pending sockets which can be Accepted
ListenQueue* listen_queue_;
// Data which tcp has buffered for sending
SendBuffer send_buffer_;
bool write_enabled_;
// Critical section to protect the recv_buffer and queue_
CriticalSection crit_;
// Network model that enforces bandwidth and capacity constraints
NetworkQueue network_;
size_t network_size_;
// Data which has been received from the network
RecvBuffer recv_buffer_;
// The amount of data which is in flight or in recv_buffer_
size_t recv_buffer_size_;
// Is this socket bound?
bool bound_;
// When we bind a socket to Any, VSS's Bind gives it another address. For
// dual-stack sockets, we want to distinguish between sockets that were
// explicitly given a particular address and sockets that had one picked
// for them by VSS.
bool was_any_;
// Store the options that are set
OptionsMap options_map_;
friend class VirtualSocketServer;
};
VirtualSocketServer::VirtualSocketServer(SocketServer* ss)
: server_(ss), server_owned_(false), msg_queue_(NULL), stop_on_idle_(false),
network_delay_(Time()), next_ipv4_(kInitialNextIPv4),
next_ipv6_(kInitialNextIPv6), next_port_(kFirstEphemeralPort),
bindings_(new AddressMap()), connections_(new ConnectionMap()),
bandwidth_(0), network_capacity_(kDefaultNetworkCapacity),
send_buffer_capacity_(kDefaultTcpBufferSize),
recv_buffer_capacity_(kDefaultTcpBufferSize),
delay_mean_(0), delay_stddev_(0), delay_samples_(NUM_SAMPLES),
delay_dist_(NULL), drop_prob_(0.0) {
if (!server_) {
server_ = new PhysicalSocketServer();
server_owned_ = true;
}
UpdateDelayDistribution();
}
VirtualSocketServer::~VirtualSocketServer() {
delete bindings_;
delete connections_;
delete delay_dist_;
if (server_owned_) {
delete server_;
}
}
IPAddress VirtualSocketServer::GetNextIP(int family) {
if (family == AF_INET) {
IPAddress next_ip(next_ipv4_);
next_ipv4_.s_addr =
HostToNetwork32(NetworkToHost32(next_ipv4_.s_addr) + 1);
return next_ip;
} else if (family == AF_INET6) {
IPAddress next_ip(next_ipv6_);
uint32* as_ints = reinterpret_cast<uint32*>(&next_ipv6_.s6_addr);
as_ints[3] += 1;
return next_ip;
}
return IPAddress();
}
uint16 VirtualSocketServer::GetNextPort() {
uint16 port = next_port_;
if (next_port_ < kLastEphemeralPort) {
++next_port_;
} else {
next_port_ = kFirstEphemeralPort;
}
return port;
}
Socket* VirtualSocketServer::CreateSocket(int type) {
return CreateSocket(AF_INET, type);
}
Socket* VirtualSocketServer::CreateSocket(int family, int type) {
return CreateSocketInternal(family, type);
}
AsyncSocket* VirtualSocketServer::CreateAsyncSocket(int type) {
return CreateAsyncSocket(AF_INET, type);
}
AsyncSocket* VirtualSocketServer::CreateAsyncSocket(int family, int type) {
return CreateSocketInternal(family, type);
}
VirtualSocket* VirtualSocketServer::CreateSocketInternal(int family, int type) {
return new VirtualSocket(this, family, type, true);
}
void VirtualSocketServer::SetMessageQueue(MessageQueue* msg_queue) {
msg_queue_ = msg_queue;
if (msg_queue_) {
msg_queue_->SignalQueueDestroyed.connect(this,
&VirtualSocketServer::OnMessageQueueDestroyed);
}
}
bool VirtualSocketServer::Wait(int cmsWait, bool process_io) {
ASSERT(msg_queue_ == Thread::Current());
if (stop_on_idle_ && Thread::Current()->empty()) {
return false;
}
return socketserver()->Wait(cmsWait, process_io);
}
void VirtualSocketServer::WakeUp() {
socketserver()->WakeUp();
}
bool VirtualSocketServer::ProcessMessagesUntilIdle() {
ASSERT(msg_queue_ == Thread::Current());
stop_on_idle_ = true;
while (!msg_queue_->empty()) {
Message msg;
if (msg_queue_->Get(&msg, kForever)) {
msg_queue_->Dispatch(&msg);
}
}
stop_on_idle_ = false;
return !msg_queue_->IsQuitting();
}
int VirtualSocketServer::Bind(VirtualSocket* socket,
const SocketAddress& addr) {
ASSERT(NULL != socket);
// Address must be completely specified at this point
ASSERT(!IPIsUnspec(addr.ipaddr()));
ASSERT(addr.port() != 0);
// Normalize the address (turns v6-mapped addresses into v4-addresses).
SocketAddress normalized(addr.ipaddr().Normalized(), addr.port());
AddressMap::value_type entry(normalized, socket);
return bindings_->insert(entry).second ? 0 : -1;
}
int VirtualSocketServer::Bind(VirtualSocket* socket, SocketAddress* addr) {
ASSERT(NULL != socket);
if (IPIsAny(addr->ipaddr())) {
addr->SetIP(GetNextIP(addr->ipaddr().family()));
} else if (!IPIsUnspec(addr->ipaddr())) {
addr->SetIP(addr->ipaddr().Normalized());
} else {
ASSERT(false);
}
if (addr->port() == 0) {
for (int i = 0; i < kEphemeralPortCount; ++i) {
addr->SetPort(GetNextPort());
if (bindings_->find(*addr) == bindings_->end()) {
break;
}
}
}
return Bind(socket, *addr);
}
VirtualSocket* VirtualSocketServer::LookupBinding(const SocketAddress& addr) {
SocketAddress normalized(addr.ipaddr().Normalized(),
addr.port());
AddressMap::iterator it = bindings_->find(normalized);
return (bindings_->end() != it) ? it->second : NULL;
}
int VirtualSocketServer::Unbind(const SocketAddress& addr,
VirtualSocket* socket) {
SocketAddress normalized(addr.ipaddr().Normalized(),
addr.port());
ASSERT((*bindings_)[normalized] == socket);
bindings_->erase(bindings_->find(normalized));
return 0;
}
void VirtualSocketServer::AddConnection(const SocketAddress& local,
const SocketAddress& remote,
VirtualSocket* remote_socket) {
// Add this socket pair to our routing table. This will allow
// multiple clients to connect to the same server address.
SocketAddress local_normalized(local.ipaddr().Normalized(),
local.port());
SocketAddress remote_normalized(remote.ipaddr().Normalized(),
remote.port());
SocketAddressPair address_pair(local_normalized, remote_normalized);
connections_->insert(std::pair<SocketAddressPair,
VirtualSocket*>(address_pair, remote_socket));
}
VirtualSocket* VirtualSocketServer::LookupConnection(
const SocketAddress& local,
const SocketAddress& remote) {
SocketAddress local_normalized(local.ipaddr().Normalized(),
local.port());
SocketAddress remote_normalized(remote.ipaddr().Normalized(),
remote.port());
SocketAddressPair address_pair(local_normalized, remote_normalized);
ConnectionMap::iterator it = connections_->find(address_pair);
return (connections_->end() != it) ? it->second : NULL;
}
void VirtualSocketServer::RemoveConnection(const SocketAddress& local,
const SocketAddress& remote) {
SocketAddress local_normalized(local.ipaddr().Normalized(),
local.port());
SocketAddress remote_normalized(remote.ipaddr().Normalized(),
remote.port());
SocketAddressPair address_pair(local_normalized, remote_normalized);
connections_->erase(address_pair);
}
static double Random() {
return static_cast<double>(rand()) / RAND_MAX;
}
int VirtualSocketServer::Connect(VirtualSocket* socket,
const SocketAddress& remote_addr,
bool use_delay) {
uint32 delay = use_delay ? GetRandomTransitDelay() : 0;
VirtualSocket* remote = LookupBinding(remote_addr);
if (!CanInteractWith(socket, remote)) {
LOG(LS_INFO) << "Address family mismatch between "
<< socket->GetLocalAddress() << " and " << remote_addr;
return -1;
}
if (remote != NULL) {
SocketAddress addr = socket->GetLocalAddress();
msg_queue_->PostDelayed(delay, remote, MSG_ID_CONNECT,
new MessageAddress(addr));
} else {
LOG(LS_INFO) << "No one listening at " << remote_addr;
msg_queue_->PostDelayed(delay, socket, MSG_ID_DISCONNECT);
}
return 0;
}
bool VirtualSocketServer::Disconnect(VirtualSocket* socket) {
if (socket) {
// Remove the mapping.
msg_queue_->Post(socket, MSG_ID_DISCONNECT);
return true;
}
return false;
}
int VirtualSocketServer::SendUdp(VirtualSocket* socket,
const char* data, size_t data_size,
const SocketAddress& remote_addr) {
// See if we want to drop this packet.
if (Random() < drop_prob_) {
LOG(LS_VERBOSE) << "Dropping packet: bad luck";
return static_cast<int>(data_size);
}
VirtualSocket* recipient = LookupBinding(remote_addr);
if (!recipient) {
// Make a fake recipient for address family checking.
scoped_ptr<VirtualSocket> dummy_socket(
CreateSocketInternal(AF_INET, SOCK_DGRAM));
dummy_socket->SetLocalAddress(remote_addr);
if (!CanInteractWith(socket, dummy_socket.get())) {
LOG(LS_VERBOSE) << "Incompatible address families: "
<< socket->GetLocalAddress() << " and " << remote_addr;
return -1;
}
LOG(LS_VERBOSE) << "No one listening at " << remote_addr;
return static_cast<int>(data_size);
}
if (!CanInteractWith(socket, recipient)) {
LOG(LS_VERBOSE) << "Incompatible address families: "
<< socket->GetLocalAddress() << " and " << remote_addr;
return -1;
}
CritScope cs(&socket->crit_);
uint32 cur_time = Time();
PurgeNetworkPackets(socket, 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,
// we know whether we can fit this packet.
//
// NOTE: There are better algorithms for maintaining such a queue (such as
// "Derivative Random Drop"); however, this algorithm is a more accurate
// 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_) {
LOG(LS_VERBOSE) << "Dropping packet: network capacity exceeded";
return static_cast<int>(data_size);
}
AddPacketToNetwork(socket, recipient, cur_time, data, data_size,
UDP_HEADER_SIZE, false);
return static_cast<int>(data_size);
}
void VirtualSocketServer::SendTcp(VirtualSocket* socket) {
// TCP can't send more data than will fill up the receiver's buffer.
// We track the data that is in the buffer plus data in flight using the
// recipient's recv_buffer_size_. Anything beyond that must be stored in the
// sender's buffer. We will trigger the buffered data to be sent when data
// is read from the recv_buffer.
// Lookup the local/remote pair in the connections table.
VirtualSocket* recipient = LookupConnection(socket->local_addr_,
socket->remote_addr_);
if (!recipient) {
LOG(LS_VERBOSE) << "Sending data to no one.";
return;
}
CritScope cs(&socket->crit_);
uint32 cur_time = Time();
PurgeNetworkPackets(socket, cur_time);
while (true) {
size_t available = recv_buffer_capacity_ - recipient->recv_buffer_size_;
size_t max_data_size = _min<size_t>(available, TCP_MSS - TCP_HEADER_SIZE);
size_t data_size = _min(socket->send_buffer_.size(), max_data_size);
if (0 == data_size)
break;
AddPacketToNetwork(socket, recipient, cur_time, &socket->send_buffer_[0],
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);
}
if (socket->write_enabled_
&& (socket->send_buffer_.size() < send_buffer_capacity_)) {
socket->write_enabled_ = false;
socket->SignalWriteEvent(socket);
}
}
void VirtualSocketServer::AddPacketToNetwork(VirtualSocket* sender,
VirtualSocket* recipient,
uint32 cur_time,
const char* data,
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 send_delay = SendDelay(static_cast<uint32>(sender->network_size_));
entry.done_time = cur_time + send_delay;
sender->network_.push_back(entry);
// Find the delay for crossing the many virtual hops of the network.
uint32 transit_delay = GetRandomTransitDelay();
// Post the packet as a message to be delivered (on our own thread)
Packet* p = new Packet(data, data_size, sender->local_addr_);
uint32 ts = TimeAfter(send_delay + transit_delay);
if (ordered) {
// Ensure that new packets arrive after previous ones
// TODO: consider ordering on a per-socket basis, since this
// introduces artifical delay.
ts = TimeMax(ts, network_delay_);
}
msg_queue_->PostAt(ts, recipient, MSG_ID_PACKET, p);
network_delay_ = TimeMax(ts, network_delay_);
}
void VirtualSocketServer::PurgeNetworkPackets(VirtualSocket* socket,
uint32 cur_time) {
while (!socket->network_.empty() &&
(socket->network_.front().done_time <= cur_time)) {
ASSERT(socket->network_size_ >= socket->network_.front().size);
socket->network_size_ -= socket->network_.front().size;
socket->network_.pop_front();
}
}
uint32 VirtualSocketServer::SendDelay(uint32 size) {
if (bandwidth_ == 0)
return 0;
else
return 1000 * size / bandwidth_;
}
#if 0
void PrintFunction(std::vector<std::pair<double, double> >* f) {
return;
double sum = 0;
for (uint32 i = 0; i < f->size(); ++i) {
std::cout << (*f)[i].first << '\t' << (*f)[i].second << std::endl;
sum += (*f)[i].second;
}
if (!f->empty()) {
const double mean = sum / f->size();
double sum_sq_dev = 0;
for (uint32 i = 0; i < f->size(); ++i) {
double dev = (*f)[i].second - mean;
sum_sq_dev += dev * dev;
}
std::cout << "Mean = " << mean << " StdDev = "
<< sqrt(sum_sq_dev / f->size()) << std::endl;
}
}
#endif // <unused>
void VirtualSocketServer::UpdateDelayDistribution() {
Function* dist = CreateDistribution(delay_mean_, delay_stddev_,
delay_samples_);
// We take a lock just to make sure we don't leak memory.
{
CritScope cs(&delay_crit_);
delete delay_dist_;
delay_dist_ = dist;
}
}
static double PI = 4 * atan(1.0);
static double Normal(double x, double mean, double stddev) {
double a = (x - mean) * (x - mean) / (2 * stddev * stddev);
return exp(-a) / (stddev * sqrt(2 * PI));
}
#if 0 // static unused gives a warning
static double Pareto(double x, double min, double k) {
if (x < min)
return 0;
else
return k * std::pow(min, k) / std::pow(x, k+1);
}
#endif
VirtualSocketServer::Function* VirtualSocketServer::CreateDistribution(
uint32 mean, uint32 stddev, uint32 samples) {
Function* f = new Function();
if (0 == stddev) {
f->push_back(Point(mean, 1.0));
} else {
double start = 0;
if (mean >= 4 * static_cast<double>(stddev))
start = mean - 4 * static_cast<double>(stddev);
double end = mean + 4 * static_cast<double>(stddev);
for (uint32 i = 0; i < samples; i++) {
double x = start + (end - start) * i / (samples - 1);
double y = Normal(x, mean, stddev);
f->push_back(Point(x, y));
}
}
return Resample(Invert(Accumulate(f)), 0, 1, samples);
}
uint32 VirtualSocketServer::GetRandomTransitDelay() {
size_t index = rand() % delay_dist_->size();
double delay = (*delay_dist_)[index].second;
//LOG_F(LS_INFO) << "random[" << index << "] = " << delay;
return static_cast<uint32>(delay);
}
struct FunctionDomainCmp {
bool operator()(const VirtualSocketServer::Point& p1,
const VirtualSocketServer::Point& p2) {
return p1.first < p2.first;
}
bool operator()(double v1, const VirtualSocketServer::Point& p2) {
return v1 < p2.first;
}
bool operator()(const VirtualSocketServer::Point& p1, double v2) {
return p1.first < v2;
}
};
VirtualSocketServer::Function* VirtualSocketServer::Accumulate(Function* f) {
ASSERT(f->size() >= 1);
double v = 0;
for (Function::size_type i = 0; i < f->size() - 1; ++i) {
double dx = (*f)[i + 1].first - (*f)[i].first;
double avgy = ((*f)[i + 1].second + (*f)[i].second) / 2;
(*f)[i].second = v;
v = v + dx * avgy;
}
(*f)[f->size()-1].second = v;
return f;
}
VirtualSocketServer::Function* VirtualSocketServer::Invert(Function* f) {
for (Function::size_type i = 0; i < f->size(); ++i)
std::swap((*f)[i].first, (*f)[i].second);
std::sort(f->begin(), f->end(), FunctionDomainCmp());
return f;
}
VirtualSocketServer::Function* VirtualSocketServer::Resample(
Function* f, double x1, double x2, uint32 samples) {
Function* g = new Function();
for (size_t i = 0; i < samples; i++) {
double x = x1 + (x2 - x1) * i / (samples - 1);
double y = Evaluate(f, x);
g->push_back(Point(x, y));
}
delete f;
return g;
}
double VirtualSocketServer::Evaluate(Function* f, double x) {
Function::iterator iter =
std::lower_bound(f->begin(), f->end(), x, FunctionDomainCmp());
if (iter == f->begin()) {
return (*f)[0].second;
} else if (iter == f->end()) {
ASSERT(f->size() >= 1);
return (*f)[f->size() - 1].second;
} else if (iter->first == x) {
return iter->second;
} else {
double x1 = (iter - 1)->first;
double y1 = (iter - 1)->second;
double x2 = iter->first;
double y2 = iter->second;
return y1 + (y2 - y1) * (x - x1) / (x2 - x1);
}
}
bool VirtualSocketServer::CanInteractWith(VirtualSocket* local,
VirtualSocket* remote) {
if (!local || !remote) {
return false;
}
IPAddress local_ip = local->GetLocalAddress().ipaddr();
IPAddress remote_ip = remote->GetLocalAddress().ipaddr();
IPAddress local_normalized = local_ip.Normalized();
IPAddress remote_normalized = remote_ip.Normalized();
// Check if the addresses are the same family after Normalization (turns
// mapped IPv6 address into IPv4 addresses).
// This will stop unmapped V6 addresses from talking to mapped V6 addresses.
if (local_normalized.family() == remote_normalized.family()) {
return true;
}
// If ip1 is IPv4 and ip2 is :: and ip2 is not IPV6_V6ONLY.
int remote_v6_only = 0;
remote->GetOption(Socket::OPT_IPV6_V6ONLY, &remote_v6_only);
if (local_ip.family() == AF_INET && !remote_v6_only && IPIsAny(remote_ip)) {
return true;
}
// Same check, backwards.
int local_v6_only = 0;
local->GetOption(Socket::OPT_IPV6_V6ONLY, &local_v6_only);
if (remote_ip.family() == AF_INET && !local_v6_only && IPIsAny(local_ip)) {
return true;
}
// Check to see if either socket was explicitly bound to IPv6-any.
// These sockets can talk with anyone.
if (local_ip.family() == AF_INET6 && local->was_any()) {
return true;
}
if (remote_ip.family() == AF_INET6 && remote->was_any()) {
return true;
}
return false;
}
} // namespace rtc