src/third_party/webrtc/base/virtualsocketserver.cc

   1 /*
   2  *  Copyright 2004 The WebRTC Project Authors. All rights reserved.
   3  *
   4  *  Use of this source code is governed by a BSD-style license
   5  *  that can be found in the LICENSE file in the root of the source
   6  *  tree. An additional intellectual property rights grant can be found
   7  *  in the file PATENTS.  All contributing project authors may
   8  *  be found in the AUTHORS file in the root of the source tree.
   9  */
  10
  11 #include "webrtc/base/virtualsocketserver.h"
  12
  13 #include <errno.h>
  14 #include <math.h>
  15
  16 #include <algorithm>
  17 #include <map>
  18 #include <vector>
  19
  20 #include "webrtc/base/common.h"
  21 #include "webrtc/base/logging.h"
  22 #include "webrtc/base/physicalsocketserver.h"
  23 #include "webrtc/base/socketaddresspair.h"
  24 #include "webrtc/base/thread.h"
  25 #include "webrtc/base/timeutils.h"
  26
  27 namespace rtc {
  28 #if defined(WEBRTC_WIN)
  29 const in_addr kInitialNextIPv4 = { {0x01, 0, 0, 0} };
  30 #else
  31 // This value is entirely arbitrary, hence the lack of concern about endianness.
  32 const in_addr kInitialNextIPv4 = { 0x01000000 };
  33 #endif
  34 // Starts at ::2 so as to not cause confusion with ::1.
  35 const in6_addr kInitialNextIPv6 = { { {
  36       0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2
  37     } } };
  38
  39 const uint16 kFirstEphemeralPort = 49152;
  40 const uint16 kLastEphemeralPort = 65535;
  41 const uint16 kEphemeralPortCount = kLastEphemeralPort - kFirstEphemeralPort + 1;
  42 const uint32 kDefaultNetworkCapacity = 64 * 1024;
  43 const uint32 kDefaultTcpBufferSize = 32 * 1024;
  44
  45 const uint32 UDP_HEADER_SIZE = 28;  // IP + UDP headers
  46 const uint32 TCP_HEADER_SIZE = 40;  // IP + TCP headers
  47 const uint32 TCP_MSS = 1400;  // Maximum segment size
  48
  49 // Note: The current algorithm doesn't work for sample sizes smaller than this.
  50 const int NUM_SAMPLES = 1000;
  51
  52 enum {
  53   MSG_ID_PACKET,
  54   MSG_ID_CONNECT,
  55   MSG_ID_DISCONNECT,
  56 };
  57
  58 // Packets are passed between sockets as messages.  We copy the data just like
  59 // the kernel does.
  60 class Packet : public MessageData {
  61  public:
  62   Packet(const char* data, size_t size, const SocketAddress& from)
  63         : size_(size), consumed_(0), from_(from) {
  64     ASSERT(NULL != data);
  65     data_ = new char[size_];
  66     memcpy(data_, data, size_);
  67   }
  68
  69   virtual ~Packet() {
  70     delete[] data_;
  71   }
  72
  73   const char* data() const { return data_ + consumed_; }
  74   size_t size() const { return size_ - consumed_; }
  75   const SocketAddress& from() const { return from_; }
  76
  77   // Remove the first size bytes from the data.
  78   void Consume(size_t size) {
  79     ASSERT(size + consumed_ < size_);
  80     consumed_ += size;
  81   }
  82
  83  private:
  84   char* data_;
  85   size_t size_, consumed_;
  86   SocketAddress from_;
  87 };
  88
  89 struct MessageAddress : public MessageData {
  90   explicit MessageAddress(const SocketAddress& a) : addr(a) { }
  91   SocketAddress addr;
  92 };
  93
  94 // Implements the socket interface using the virtual network.  Packets are
  95 // passed as messages using the message queue of the socket server.
  96 class VirtualSocket : public AsyncSocket, public MessageHandler {
  97  public:
  98   VirtualSocket(VirtualSocketServer* server, int family, int type, bool async)
  99       : server_(server), family_(family), type_(type), async_(async),
 100         state_(CS_CLOSED), error_(0), listen_queue_(NULL),
 101         write_enabled_(false),
 102         network_size_(0), recv_buffer_size_(0), bound_(false), was_any_(false) {
 103     ASSERT((type_ == SOCK_DGRAM) || (type_ == SOCK_STREAM));
 104     ASSERT(async_ || (type_ != SOCK_STREAM));  // We only support async streams
 105   }
 106
 107   virtual ~VirtualSocket() {
 108     Close();
 109
 110     for (RecvBuffer::iterator it = recv_buffer_.begin();
 111          it != recv_buffer_.end(); ++it) {
 112       delete *it;
 113     }
 114   }
 115
 116   virtual SocketAddress GetLocalAddress() const {
 117     return local_addr_;
 118   }
 119
 120   virtual SocketAddress GetRemoteAddress() const {
 121     return remote_addr_;
 122   }
 123
 124   // Used by server sockets to set the local address without binding.
 125   void SetLocalAddress(const SocketAddress& addr) {
 126     local_addr_ = addr;
 127   }
 128
 129   virtual int Bind(const SocketAddress& addr) {
 130     if (!local_addr_.IsNil()) {
 131       error_ = EINVAL;
 132       return -1;
 133     }
 134     local_addr_ = addr;
 135     int result = server_->Bind(this, &local_addr_);
 136     if (result != 0) {
 137       local_addr_.Clear();
 138       error_ = EADDRINUSE;
 139     } else {
 140       bound_ = true;
 141       was_any_ = addr.IsAnyIP();
 142     }
 143     return result;
 144   }
 145
 146   virtual int Connect(const SocketAddress& addr) {
 147     return InitiateConnect(addr, true);
 148   }
 149
 150   virtual int Close() {
 151     if (!local_addr_.IsNil() && bound_) {
 152       // Remove from the binding table.
 153       server_->Unbind(local_addr_, this);
 154       bound_ = false;
 155     }
 156
 157     if (SOCK_STREAM == type_) {
 158       // Cancel pending sockets
 159       if (listen_queue_) {
 160         while (!listen_queue_->empty()) {
 161           SocketAddress addr = listen_queue_->front();
 162
 163           // Disconnect listening socket.
 164           server_->Disconnect(server_->LookupBinding(addr));
 165           listen_queue_->pop_front();
 166         }
 167         delete listen_queue_;
 168         listen_queue_ = NULL;
 169       }
 170       // Disconnect stream sockets
 171       if (CS_CONNECTED == state_) {
 172         // Disconnect remote socket, check if it is a child of a server socket.
 173         VirtualSocket* socket =
 174             server_->LookupConnection(local_addr_, remote_addr_);
 175         if (!socket) {
 176           // Not a server socket child, then see if it is bound.
 177           // TODO: If this is indeed a server socket that has no
 178           // children this will cause the server socket to be
 179           // closed. This might lead to unexpected results, how to fix this?
 180           socket = server_->LookupBinding(remote_addr_);
 181         }
 182         server_->Disconnect(socket);
 183
 184         // Remove mapping for both directions.
 185         server_->RemoveConnection(remote_addr_, local_addr_);
 186         server_->RemoveConnection(local_addr_, remote_addr_);
 187       }
 188       // Cancel potential connects
 189       MessageList msgs;
 190       if (server_->msg_queue_) {
 191         server_->msg_queue_->Clear(this, MSG_ID_CONNECT, &msgs);
 192       }
 193       for (MessageList::iterator it = msgs.begin(); it != msgs.end(); ++it) {
 194         ASSERT(NULL != it->pdata);
 195         MessageAddress* data = static_cast<MessageAddress*>(it->pdata);
 196
 197         // Lookup remote side.
 198         VirtualSocket* socket = server_->LookupConnection(local_addr_,
 199                                                           data->addr);
 200         if (socket) {
 201           // Server socket, remote side is a socket retreived by
 202           // accept. Accepted sockets are not bound so we will not
 203           // find it by looking in the bindings table.
 204           server_->Disconnect(socket);
 205           server_->RemoveConnection(local_addr_, data->addr);
 206         } else {
 207           server_->Disconnect(server_->LookupBinding(data->addr));
 208         }
 209         delete data;
 210       }
 211       // Clear incoming packets and disconnect messages
 212       if (server_->msg_queue_) {
 213         server_->msg_queue_->Clear(this);
 214       }
 215     }
 216
 217     state_ = CS_CLOSED;
 218     local_addr_.Clear();
 219     remote_addr_.Clear();
 220     return 0;
 221   }
 222
 223   virtual int Send(const void *pv, size_t cb) {
 224     if (CS_CONNECTED != state_) {
 225       error_ = ENOTCONN;
 226       return -1;
 227     }
 228     if (SOCK_DGRAM == type_) {
 229       return SendUdp(pv, cb, remote_addr_);
 230     } else {
 231       return SendTcp(pv, cb);
 232     }
 233   }
 234
 235   virtual int SendTo(const void *pv, size_t cb, const SocketAddress& addr) {
 236     if (SOCK_DGRAM == type_) {
 237       return SendUdp(pv, cb, addr);
 238     } else {
 239       if (CS_CONNECTED != state_) {
 240         error_ = ENOTCONN;
 241         return -1;
 242       }
 243       return SendTcp(pv, cb);
 244     }
 245   }
 246
 247   virtual int Recv(void *pv, size_t cb) {
 248     SocketAddress addr;
 249     return RecvFrom(pv, cb, &addr);
 250   }
 251
 252   virtual int RecvFrom(void *pv, size_t cb, SocketAddress *paddr) {
 253     // If we don't have a packet, then either error or wait for one to arrive.
 254     if (recv_buffer_.empty()) {
 255       if (async_) {
 256         error_ = EAGAIN;
 257         return -1;
 258       }
 259       while (recv_buffer_.empty()) {
 260         Message msg;
 261         server_->msg_queue_->Get(&msg);
 262         server_->msg_queue_->Dispatch(&msg);
 263       }
 264     }
 265
 266     // Return the packet at the front of the queue.
 267     Packet* packet = recv_buffer_.front();
 268     size_t data_read = _min(cb, packet->size());
 269     memcpy(pv, packet->data(), data_read);
 270     *paddr = packet->from();
 271
 272     if (data_read < packet->size()) {
 273       packet->Consume(data_read);
 274     } else {
 275       recv_buffer_.pop_front();
 276       delete packet;
 277     }
 278
 279     if (SOCK_STREAM == type_) {
 280       bool was_full = (recv_buffer_size_ == server_->recv_buffer_capacity_);
 281       recv_buffer_size_ -= data_read;
 282       if (was_full) {
 283         VirtualSocket* sender = server_->LookupBinding(remote_addr_);
 284         ASSERT(NULL != sender);
 285         server_->SendTcp(sender);
 286       }
 287     }
 288
 289     return static_cast<int>(data_read);
 290   }
 291
 292   virtual int Listen(int backlog) {
 293     ASSERT(SOCK_STREAM == type_);
 294     ASSERT(CS_CLOSED == state_);
 295     if (local_addr_.IsNil()) {
 296       error_ = EINVAL;
 297       return -1;
 298     }
 299     ASSERT(NULL == listen_queue_);
 300     listen_queue_ = new ListenQueue;
 301     state_ = CS_CONNECTING;
 302     return 0;
 303   }
 304
 305   virtual VirtualSocket* Accept(SocketAddress *paddr) {
 306     if (NULL == listen_queue_) {
 307       error_ = EINVAL;
 308       return NULL;
 309     }
 310     while (!listen_queue_->empty()) {
 311       VirtualSocket* socket = new VirtualSocket(server_, AF_INET, type_,
 312                                                 async_);
 313
 314       // Set the new local address to the same as this server socket.
 315       socket->SetLocalAddress(local_addr_);
 316       // Sockets made from a socket that 'was Any' need to inherit that.
 317       socket->set_was_any(was_any_);
 318       SocketAddress remote_addr(listen_queue_->front());
 319       int result = socket->InitiateConnect(remote_addr, false);
 320       listen_queue_->pop_front();
 321       if (result != 0) {
 322         delete socket;
 323         continue;
 324       }
 325       socket->CompleteConnect(remote_addr, false);
 326       if (paddr) {
 327         *paddr = remote_addr;
 328       }
 329       return socket;
 330     }
 331     error_ = EWOULDBLOCK;
 332     return NULL;
 333   }
 334
 335   virtual int GetError() const {
 336     return error_;
 337   }
 338
 339   virtual void SetError(int error) {
 340     error_ = error;
 341   }
 342
 343   virtual ConnState GetState() const {
 344     return state_;
 345   }
 346
 347   virtual int GetOption(Option opt, int* value) {
 348     OptionsMap::const_iterator it = options_map_.find(opt);
 349     if (it == options_map_.end()) {
 350       return -1;
 351     }
 352     *value = it->second;
 353     return 0;  // 0 is success to emulate getsockopt()
 354   }
 355
 356   virtual int SetOption(Option opt, int value) {
 357     options_map_[opt] = value;
 358     return 0;  // 0 is success to emulate setsockopt()
 359   }
 360
 361   virtual int EstimateMTU(uint16* mtu) {
 362     if (CS_CONNECTED != state_)
 363       return ENOTCONN;
 364     else
 365       return 65536;
 366   }
 367
 368   void OnMessage(Message *pmsg) {
 369     if (pmsg->message_id == MSG_ID_PACKET) {
 370       //ASSERT(!local_addr_.IsAny());
 371       ASSERT(NULL != pmsg->pdata);
 372       Packet* packet = static_cast<Packet*>(pmsg->pdata);
 373
 374       recv_buffer_.push_back(packet);
 375
 376       if (async_) {
 377         SignalReadEvent(this);
 378       }
 379     } else if (pmsg->message_id == MSG_ID_CONNECT) {
 380       ASSERT(NULL != pmsg->pdata);
 381       MessageAddress* data = static_cast<MessageAddress*>(pmsg->pdata);
 382       if (listen_queue_ != NULL) {
 383         listen_queue_->push_back(data->addr);
 384         if (async_) {
 385           SignalReadEvent(this);
 386         }
 387       } else if ((SOCK_STREAM == type_) && (CS_CONNECTING == state_)) {
 388         CompleteConnect(data->addr, true);
 389       } else {
 390         LOG(LS_VERBOSE) << "Socket at " << local_addr_ << " is not listening";
 391         server_->Disconnect(server_->LookupBinding(data->addr));
 392       }
 393       delete data;
 394     } else if (pmsg->message_id == MSG_ID_DISCONNECT) {
 395       ASSERT(SOCK_STREAM == type_);
 396       if (CS_CLOSED != state_) {
 397         int error = (CS_CONNECTING == state_) ? ECONNREFUSED : 0;
 398         state_ = CS_CLOSED;
 399         remote_addr_.Clear();
 400         if (async_) {
 401           SignalCloseEvent(this, error);
 402         }
 403       }
 404     } else {
 405       ASSERT(false);
 406     }
 407   }
 408
 409   bool was_any() { return was_any_; }
 410   void set_was_any(bool was_any) { was_any_ = was_any; }
 411
 412  private:
 413   struct NetworkEntry {
 414     size_t size;
 415     uint32 done_time;
 416   };
 417
 418   typedef std::deque<SocketAddress> ListenQueue;
 419   typedef std::deque<NetworkEntry> NetworkQueue;
 420   typedef std::vector<char> SendBuffer;
 421   typedef std::list<Packet*> RecvBuffer;
 422   typedef std::map<Option, int> OptionsMap;
 423
 424   int InitiateConnect(const SocketAddress& addr, bool use_delay) {
 425     if (!remote_addr_.IsNil()) {
 426       error_ = (CS_CONNECTED == state_) ? EISCONN : EINPROGRESS;
 427       return -1;
 428     }
 429     if (local_addr_.IsNil()) {
 430       // If there's no local address set, grab a random one in the correct AF.
 431       int result = 0;
 432       if (addr.ipaddr().family() == AF_INET) {
 433         result = Bind(SocketAddress("0.0.0.0", 0));
 434       } else if (addr.ipaddr().family() == AF_INET6) {
 435         result = Bind(SocketAddress("::", 0));
 436       }
 437       if (result != 0) {
 438         return result;
 439       }
 440     }
 441     if (type_ == SOCK_DGRAM) {
 442       remote_addr_ = addr;
 443       state_ = CS_CONNECTED;
 444     } else {
 445       int result = server_->Connect(this, addr, use_delay);
 446       if (result != 0) {
 447         error_ = EHOSTUNREACH;
 448         return -1;
 449       }
 450       state_ = CS_CONNECTING;
 451     }
 452     return 0;
 453   }
 454
 455   void CompleteConnect(const SocketAddress& addr, bool notify) {
 456     ASSERT(CS_CONNECTING == state_);
 457     remote_addr_ = addr;
 458     state_ = CS_CONNECTED;
 459     server_->AddConnection(remote_addr_, local_addr_, this);
 460     if (async_ && notify) {
 461       SignalConnectEvent(this);
 462     }
 463   }
 464
 465   int SendUdp(const void* pv, size_t cb, const SocketAddress& addr) {
 466     // If we have not been assigned a local port, then get one.
 467     if (local_addr_.IsNil()) {
 468       local_addr_ = EmptySocketAddressWithFamily(addr.ipaddr().family());
 469       int result = server_->Bind(this, &local_addr_);
 470       if (result != 0) {
 471         local_addr_.Clear();
 472         error_ = EADDRINUSE;
 473         return result;
 474       }
 475     }
 476
 477     // Send the data in a message to the appropriate socket.
 478     return server_->SendUdp(this, static_cast<const char*>(pv), cb, addr);
 479   }
 480
 481   int SendTcp(const void* pv, size_t cb) {
 482     size_t capacity = server_->send_buffer_capacity_ - send_buffer_.size();
 483     if (0 == capacity) {
 484       write_enabled_ = true;
 485       error_ = EWOULDBLOCK;
 486       return -1;
 487     }
 488     size_t consumed = _min(cb, capacity);
 489     const char* cpv = static_cast<const char*>(pv);
 490     send_buffer_.insert(send_buffer_.end(), cpv, cpv + consumed);
 491     server_->SendTcp(this);
 492     return static_cast<int>(consumed);
 493   }
 494
 495   VirtualSocketServer* server_;
 496   int family_;
 497   int type_;
 498   bool async_;
 499   ConnState state_;
 500   int error_;
 501   SocketAddress local_addr_;
 502   SocketAddress remote_addr_;
 503
 504   // Pending sockets which can be Accepted
 505   ListenQueue* listen_queue_;
 506
 507   // Data which tcp has buffered for sending
 508   SendBuffer send_buffer_;
 509   bool write_enabled_;
 510
 511   // Critical section to protect the recv_buffer and queue_
 512   CriticalSection crit_;
 513
 514   // Network model that enforces bandwidth and capacity constraints
 515   NetworkQueue network_;
 516   size_t network_size_;
 517
 518   // Data which has been received from the network
 519   RecvBuffer recv_buffer_;
 520   // The amount of data which is in flight or in recv_buffer_
 521   size_t recv_buffer_size_;
 522
 523   // Is this socket bound?
 524   bool bound_;
 525
 526   // When we bind a socket to Any, VSS's Bind gives it another address. For
 527   // dual-stack sockets, we want to distinguish between sockets that were
 528   // explicitly given a particular address and sockets that had one picked
 529   // for them by VSS.
 530   bool was_any_;
 531
 532   // Store the options that are set
 533   OptionsMap options_map_;
 534
 535   friend class VirtualSocketServer;
 536 };
 537
 538 VirtualSocketServer::VirtualSocketServer(SocketServer* ss)
 539     : server_(ss), server_owned_(false), msg_queue_(NULL), stop_on_idle_(false),
 540       network_delay_(Time()), next_ipv4_(kInitialNextIPv4),
 541       next_ipv6_(kInitialNextIPv6), next_port_(kFirstEphemeralPort),
 542       bindings_(new AddressMap()), connections_(new ConnectionMap()),
 543       bandwidth_(0), network_capacity_(kDefaultNetworkCapacity),
 544       send_buffer_capacity_(kDefaultTcpBufferSize),
 545       recv_buffer_capacity_(kDefaultTcpBufferSize),
 546       delay_mean_(0), delay_stddev_(0), delay_samples_(NUM_SAMPLES),
 547       delay_dist_(NULL), drop_prob_(0.0) {
 548   if (!server_) {
 549     server_ = new PhysicalSocketServer();
 550     server_owned_ = true;
 551   }
 552   UpdateDelayDistribution();
 553 }
 554
 555 VirtualSocketServer::~VirtualSocketServer() {
 556   delete bindings_;
 557   delete connections_;
 558   delete delay_dist_;
 559   if (server_owned_) {
 560     delete server_;
 561   }
 562 }
 563
 564 IPAddress VirtualSocketServer::GetNextIP(int family) {
 565   if (family == AF_INET) {
 566     IPAddress next_ip(next_ipv4_);
 567     next_ipv4_.s_addr =
 568         HostToNetwork32(NetworkToHost32(next_ipv4_.s_addr) + 1);
 569     return next_ip;
 570   } else if (family == AF_INET6) {
 571     IPAddress next_ip(next_ipv6_);
 572     uint32* as_ints = reinterpret_cast<uint32*>(&next_ipv6_.s6_addr);
 573     as_ints[3] += 1;
 574     return next_ip;
 575   }
 576   return IPAddress();
 577 }
 578
 579 uint16 VirtualSocketServer::GetNextPort() {
 580   uint16 port = next_port_;
 581   if (next_port_ < kLastEphemeralPort) {
 582     ++next_port_;
 583   } else {
 584     next_port_ = kFirstEphemeralPort;
 585   }
 586   return port;
 587 }
 588
 589 Socket* VirtualSocketServer::CreateSocket(int type) {
 590   return CreateSocket(AF_INET, type);
 591 }
 592
 593 Socket* VirtualSocketServer::CreateSocket(int family, int type) {
 594   return CreateSocketInternal(family, type);
 595 }
 596
 597 AsyncSocket* VirtualSocketServer::CreateAsyncSocket(int type) {
 598   return CreateAsyncSocket(AF_INET, type);
 599 }
 600
 601 AsyncSocket* VirtualSocketServer::CreateAsyncSocket(int family, int type) {
 602   return CreateSocketInternal(family, type);
 603 }
 604
 605 VirtualSocket* VirtualSocketServer::CreateSocketInternal(int family, int type) {
 606   return new VirtualSocket(this, family, type, true);
 607 }
 608
 609 void VirtualSocketServer::SetMessageQueue(MessageQueue* msg_queue) {
 610   msg_queue_ = msg_queue;
 611   if (msg_queue_) {
 612     msg_queue_->SignalQueueDestroyed.connect(this,
 613         &VirtualSocketServer::OnMessageQueueDestroyed);
 614   }
 615 }
 616
 617 bool VirtualSocketServer::Wait(int cmsWait, bool process_io) {
 618   ASSERT(msg_queue_ == Thread::Current());
 619   if (stop_on_idle_ && Thread::Current()->empty()) {
 620     return false;
 621   }
 622   return socketserver()->Wait(cmsWait, process_io);
 623 }
 624
 625 void VirtualSocketServer::WakeUp() {
 626   socketserver()->WakeUp();
 627 }
 628
 629 bool VirtualSocketServer::ProcessMessagesUntilIdle() {
 630   ASSERT(msg_queue_ == Thread::Current());
 631   stop_on_idle_ = true;
 632   while (!msg_queue_->empty()) {
 633     Message msg;
 634     if (msg_queue_->Get(&msg, kForever)) {
 635       msg_queue_->Dispatch(&msg);
 636     }
 637   }
 638   stop_on_idle_ = false;
 639   return !msg_queue_->IsQuitting();
 640 }
 641
 642 void VirtualSocketServer::SetNextPortForTesting(uint16 port) {
 643   next_port_ = port;
 644 }
 645
 646 int VirtualSocketServer::Bind(VirtualSocket* socket,
 647                               const SocketAddress& addr) {
 648   ASSERT(NULL != socket);
 649   // Address must be completely specified at this point
 650   ASSERT(!IPIsUnspec(addr.ipaddr()));
 651   ASSERT(addr.port() != 0);
 652
 653   // Normalize the address (turns v6-mapped addresses into v4-addresses).
 654   SocketAddress normalized(addr.ipaddr().Normalized(), addr.port());
 655
 656   AddressMap::value_type entry(normalized, socket);
 657   return bindings_->insert(entry).second ? 0 : -1;
 658 }
 659
 660 int VirtualSocketServer::Bind(VirtualSocket* socket, SocketAddress* addr) {
 661   ASSERT(NULL != socket);
 662
 663   if (IPIsAny(addr->ipaddr())) {
 664     addr->SetIP(GetNextIP(addr->ipaddr().family()));
 665   } else if (!IPIsUnspec(addr->ipaddr())) {
 666     addr->SetIP(addr->ipaddr().Normalized());
 667   } else {
 668     ASSERT(false);
 669   }
 670
 671   if (addr->port() == 0) {
 672     for (int i = 0; i < kEphemeralPortCount; ++i) {
 673       addr->SetPort(GetNextPort());
 674       if (bindings_->find(*addr) == bindings_->end()) {
 675         break;
 676       }
 677     }
 678   }
 679
 680   return Bind(socket, *addr);
 681 }
 682
 683 VirtualSocket* VirtualSocketServer::LookupBinding(const SocketAddress& addr) {
 684   SocketAddress normalized(addr.ipaddr().Normalized(),
 685                            addr.port());
 686   AddressMap::iterator it = bindings_->find(normalized);
 687   return (bindings_->end() != it) ? it->second : NULL;
 688 }
 689
 690 int VirtualSocketServer::Unbind(const SocketAddress& addr,
 691                                 VirtualSocket* socket) {
 692   SocketAddress normalized(addr.ipaddr().Normalized(),
 693                            addr.port());
 694   ASSERT((*bindings_)[normalized] == socket);
 695   bindings_->erase(bindings_->find(normalized));
 696   return 0;
 697 }
 698
 699 void VirtualSocketServer::AddConnection(const SocketAddress& local,
 700                                         const SocketAddress& remote,
 701                                         VirtualSocket* remote_socket) {
 702   // Add this socket pair to our routing table. This will allow
 703   // multiple clients to connect to the same server address.
 704   SocketAddress local_normalized(local.ipaddr().Normalized(),
 705                                  local.port());
 706   SocketAddress remote_normalized(remote.ipaddr().Normalized(),
 707                                   remote.port());
 708   SocketAddressPair address_pair(local_normalized, remote_normalized);
 709   connections_->insert(std::pair<SocketAddressPair,
 710                        VirtualSocket*>(address_pair, remote_socket));
 711 }
 712
 713 VirtualSocket* VirtualSocketServer::LookupConnection(
 714     const SocketAddress& local,
 715     const SocketAddress& remote) {
 716   SocketAddress local_normalized(local.ipaddr().Normalized(),
 717                                  local.port());
 718   SocketAddress remote_normalized(remote.ipaddr().Normalized(),
 719                                   remote.port());
 720   SocketAddressPair address_pair(local_normalized, remote_normalized);
 721   ConnectionMap::iterator it = connections_->find(address_pair);
 722   return (connections_->end() != it) ? it->second : NULL;
 723 }
 724
 725 void VirtualSocketServer::RemoveConnection(const SocketAddress& local,
 726                                            const SocketAddress& remote) {
 727   SocketAddress local_normalized(local.ipaddr().Normalized(),
 728                                 local.port());
 729   SocketAddress remote_normalized(remote.ipaddr().Normalized(),
 730                                  remote.port());
 731   SocketAddressPair address_pair(local_normalized, remote_normalized);
 732   connections_->erase(address_pair);
 733 }
 734
 735 static double Random() {
 736   return static_cast<double>(rand()) / RAND_MAX;
 737 }
 738
 739 int VirtualSocketServer::Connect(VirtualSocket* socket,
 740                                  const SocketAddress& remote_addr,
 741                                  bool use_delay) {
 742   uint32 delay = use_delay ? GetRandomTransitDelay() : 0;
 743   VirtualSocket* remote = LookupBinding(remote_addr);
 744   if (!CanInteractWith(socket, remote)) {
 745     LOG(LS_INFO) << "Address family mismatch between "
 746                  << socket->GetLocalAddress() << " and " << remote_addr;
 747     return -1;
 748   }
 749   if (remote != NULL) {
 750     SocketAddress addr = socket->GetLocalAddress();
 751     msg_queue_->PostDelayed(delay, remote, MSG_ID_CONNECT,
 752                             new MessageAddress(addr));
 753   } else {
 754     LOG(LS_INFO) << "No one listening at " << remote_addr;
 755     msg_queue_->PostDelayed(delay, socket, MSG_ID_DISCONNECT);
 756   }
 757   return 0;
 758 }
 759
 760 bool VirtualSocketServer::Disconnect(VirtualSocket* socket) {
 761   if (socket) {
 762     // Remove the mapping.
 763     msg_queue_->Post(socket, MSG_ID_DISCONNECT);
 764     return true;
 765   }
 766   return false;
 767 }
 768
 769 int VirtualSocketServer::SendUdp(VirtualSocket* socket,
 770                                  const char* data, size_t data_size,
 771                                  const SocketAddress& remote_addr) {
 772   // See if we want to drop this packet.
 773   if (Random() < drop_prob_) {
 774     LOG(LS_VERBOSE) << "Dropping packet: bad luck";
 775     return static_cast<int>(data_size);
 776   }
 777
 778   VirtualSocket* recipient = LookupBinding(remote_addr);
 779   if (!recipient) {
 780     // Make a fake recipient for address family checking.
 781     scoped_ptr<VirtualSocket> dummy_socket(
 782         CreateSocketInternal(AF_INET, SOCK_DGRAM));
 783     dummy_socket->SetLocalAddress(remote_addr);
 784     if (!CanInteractWith(socket, dummy_socket.get())) {
 785       LOG(LS_VERBOSE) << "Incompatible address families: "
 786                       << socket->GetLocalAddress() << " and " << remote_addr;
 787       return -1;
 788     }
 789     LOG(LS_VERBOSE) << "No one listening at " << remote_addr;
 790     return static_cast<int>(data_size);
 791   }
 792
 793   if (!CanInteractWith(socket, recipient)) {
 794     LOG(LS_VERBOSE) << "Incompatible address families: "
 795                     << socket->GetLocalAddress() << " and " << remote_addr;
 796     return -1;
 797   }
 798
 799   CritScope cs(&socket->crit_);
 800
 801   uint32 cur_time = Time();
 802   PurgeNetworkPackets(socket, cur_time);
 803
 804   // Determine whether we have enough bandwidth to accept this packet.  To do
 805   // this, we need to update the send queue.  Once we know it's current size,
 806   // we know whether we can fit this packet.
 807   //
 808   // NOTE: There are better algorithms for maintaining such a queue (such as
 809   // "Derivative Random Drop"); however, this algorithm is a more accurate
 810   // simulation of what a normal network would do.
 811
 812   size_t packet_size = data_size + UDP_HEADER_SIZE;
 813   if (socket->network_size_ + packet_size > network_capacity_) {
 814     LOG(LS_VERBOSE) << "Dropping packet: network capacity exceeded";
 815     return static_cast<int>(data_size);
 816   }
 817
 818   AddPacketToNetwork(socket, recipient, cur_time, data, data_size,
 819                      UDP_HEADER_SIZE, false);
 820
 821   return static_cast<int>(data_size);
 822 }
 823
 824 void VirtualSocketServer::SendTcp(VirtualSocket* socket) {
 825   // TCP can't send more data than will fill up the receiver's buffer.
 826   // We track the data that is in the buffer plus data in flight using the
 827   // recipient's recv_buffer_size_.  Anything beyond that must be stored in the
 828   // sender's buffer.  We will trigger the buffered data to be sent when data
 829   // is read from the recv_buffer.
 830
 831   // Lookup the local/remote pair in the connections table.
 832   VirtualSocket* recipient = LookupConnection(socket->local_addr_,
 833                                               socket->remote_addr_);
 834   if (!recipient) {
 835     LOG(LS_VERBOSE) << "Sending data to no one.";
 836     return;
 837   }
 838
 839   CritScope cs(&socket->crit_);
 840
 841   uint32 cur_time = Time();
 842   PurgeNetworkPackets(socket, cur_time);
 843
 844   while (true) {
 845     size_t available = recv_buffer_capacity_ - recipient->recv_buffer_size_;
 846     size_t max_data_size = _min<size_t>(available, TCP_MSS - TCP_HEADER_SIZE);
 847     size_t data_size = _min(socket->send_buffer_.size(), max_data_size);
 848     if (0 == data_size)
 849       break;
 850
 851     AddPacketToNetwork(socket, recipient, cur_time, &socket->send_buffer_[0],
 852                        data_size, TCP_HEADER_SIZE, true);
 853     recipient->recv_buffer_size_ += data_size;
 854
 855     size_t new_buffer_size = socket->send_buffer_.size() - data_size;
 856     // Avoid undefined access beyond the last element of the vector.
 857     // This only happens when new_buffer_size is 0.
 858     if (data_size < socket->send_buffer_.size()) {
 859       // memmove is required for potentially overlapping source/destination.
 860       memmove(&socket->send_buffer_[0], &socket->send_buffer_[data_size],
 861               new_buffer_size);
 862     }
 863     socket->send_buffer_.resize(new_buffer_size);
 864   }
 865
 866   if (socket->write_enabled_
 867       && (socket->send_buffer_.size() < send_buffer_capacity_)) {
 868     socket->write_enabled_ = false;
 869     socket->SignalWriteEvent(socket);
 870   }
 871 }
 872
 873 void VirtualSocketServer::AddPacketToNetwork(VirtualSocket* sender,
 874                                              VirtualSocket* recipient,
 875                                              uint32 cur_time,
 876                                              const char* data,
 877                                              size_t data_size,
 878                                              size_t header_size,
 879                                              bool ordered) {
 880   VirtualSocket::NetworkEntry entry;
 881   entry.size = data_size + header_size;
 882
 883   sender->network_size_ += entry.size;
 884   uint32 send_delay = SendDelay(static_cast<uint32>(sender->network_size_));
 885   entry.done_time = cur_time + send_delay;
 886   sender->network_.push_back(entry);
 887
 888   // Find the delay for crossing the many virtual hops of the network.
 889   uint32 transit_delay = GetRandomTransitDelay();
 890
 891   // Post the packet as a message to be delivered (on our own thread)
 892   Packet* p = new Packet(data, data_size, sender->local_addr_);
 893   uint32 ts = TimeAfter(send_delay + transit_delay);
 894   if (ordered) {
 895     // Ensure that new packets arrive after previous ones
 896     // TODO: consider ordering on a per-socket basis, since this
 897     // introduces artifical delay.
 898     ts = TimeMax(ts, network_delay_);
 899   }
 900   msg_queue_->PostAt(ts, recipient, MSG_ID_PACKET, p);
 901   network_delay_ = TimeMax(ts, network_delay_);
 902 }
 903
 904 void VirtualSocketServer::PurgeNetworkPackets(VirtualSocket* socket,
 905                                               uint32 cur_time) {
 906   while (!socket->network_.empty() &&
 907          (socket->network_.front().done_time <= cur_time)) {
 908     ASSERT(socket->network_size_ >= socket->network_.front().size);
 909     socket->network_size_ -= socket->network_.front().size;
 910     socket->network_.pop_front();
 911   }
 912 }
 913
 914 uint32 VirtualSocketServer::SendDelay(uint32 size) {
 915   if (bandwidth_ == 0)
 916     return 0;
 917   else
 918     return 1000 * size / bandwidth_;
 919 }
 920
 921 #if 0
 922 void PrintFunction(std::vector<std::pair<double, double> >* f) {
 923   return;
 924   double sum = 0;
 925   for (uint32 i = 0; i < f->size(); ++i) {
 926     std::cout << (*f)[i].first << '\t' << (*f)[i].second << std::endl;
 927     sum += (*f)[i].second;
 928   }
 929   if (!f->empty()) {
 930     const double mean = sum / f->size();
 931     double sum_sq_dev = 0;
 932     for (uint32 i = 0; i < f->size(); ++i) {
 933       double dev = (*f)[i].second - mean;
 934       sum_sq_dev += dev * dev;
 935     }
 936     std::cout << "Mean = " << mean << " StdDev = "
 937               << sqrt(sum_sq_dev / f->size()) << std::endl;
 938   }
 939 }
 940 #endif  // <unused>
 941
 942 void VirtualSocketServer::UpdateDelayDistribution() {
 943   Function* dist = CreateDistribution(delay_mean_, delay_stddev_,
 944                                       delay_samples_);
 945   // We take a lock just to make sure we don't leak memory.
 946   {
 947     CritScope cs(&delay_crit_);
 948     delete delay_dist_;
 949     delay_dist_ = dist;
 950   }
 951 }
 952
 953 static double PI = 4 * atan(1.0);
 954
 955 static double Normal(double x, double mean, double stddev) {
 956   double a = (x - mean) * (x - mean) / (2 * stddev * stddev);
 957   return exp(-a) / (stddev * sqrt(2 * PI));
 958 }
 959
 960 #if 0  // static unused gives a warning
 961 static double Pareto(double x, double min, double k) {
 962   if (x < min)
 963     return 0;
 964   else
 965     return k * std::pow(min, k) / std::pow(x, k+1);
 966 }
 967 #endif
 968
 969 VirtualSocketServer::Function* VirtualSocketServer::CreateDistribution(
 970     uint32 mean, uint32 stddev, uint32 samples) {
 971   Function* f = new Function();
 972
 973   if (0 == stddev) {
 974     f->push_back(Point(mean, 1.0));
 975   } else {
 976     double start = 0;
 977     if (mean >= 4 * static_cast<double>(stddev))
 978       start = mean - 4 * static_cast<double>(stddev);
 979     double end = mean + 4 * static_cast<double>(stddev);
 980
 981     for (uint32 i = 0; i < samples; i++) {
 982       double x = start + (end - start) * i / (samples - 1);
 983       double y = Normal(x, mean, stddev);
 984       f->push_back(Point(x, y));
 985     }
 986   }
 987   return Resample(Invert(Accumulate(f)), 0, 1, samples);
 988 }
 989
 990 uint32 VirtualSocketServer::GetRandomTransitDelay() {
 991   size_t index = rand() % delay_dist_->size();
 992   double delay = (*delay_dist_)[index].second;
 993   //LOG_F(LS_INFO) << "random[" << index << "] = " << delay;
 994   return static_cast<uint32>(delay);
 995 }
 996
 997 struct FunctionDomainCmp {
 998   bool operator()(const VirtualSocketServer::Point& p1,
 999                    const VirtualSocketServer::Point& p2) {
1000     return p1.first < p2.first;
1001   }
1002   bool operator()(double v1, const VirtualSocketServer::Point& p2) {
1003     return v1 < p2.first;
1004   }
1005   bool operator()(const VirtualSocketServer::Point& p1, double v2) {
1006     return p1.first < v2;
1007   }
1008 };
1009
1010 VirtualSocketServer::Function* VirtualSocketServer::Accumulate(Function* f) {
1011   ASSERT(f->size() >= 1);
1012   double v = 0;
1013   for (Function::size_type i = 0; i < f->size() - 1; ++i) {
1014     double dx = (*f)[i + 1].first - (*f)[i].first;
1015     double avgy = ((*f)[i + 1].second + (*f)[i].second) / 2;
1016     (*f)[i].second = v;
1017     v = v + dx * avgy;
1018   }
1019   (*f)[f->size()-1].second = v;
1020   return f;
1021 }
1022
1023 VirtualSocketServer::Function* VirtualSocketServer::Invert(Function* f) {
1024   for (Function::size_type i = 0; i < f->size(); ++i)
1025     std::swap((*f)[i].first, (*f)[i].second);
1026
1027   std::sort(f->begin(), f->end(), FunctionDomainCmp());
1028   return f;
1029 }
1030
1031 VirtualSocketServer::Function* VirtualSocketServer::Resample(
1032     Function* f, double x1, double x2, uint32 samples) {
1033   Function* g = new Function();
1034
1035   for (size_t i = 0; i < samples; i++) {
1036     double x = x1 + (x2 - x1) * i / (samples - 1);
1037     double y = Evaluate(f, x);
1038     g->push_back(Point(x, y));
1039   }
1040
1041   delete f;
1042   return g;
1043 }
1044
1045 double VirtualSocketServer::Evaluate(Function* f, double x) {
1046   Function::iterator iter =
1047       std::lower_bound(f->begin(), f->end(), x, FunctionDomainCmp());
1048   if (iter == f->begin()) {
1049     return (*f)[0].second;
1050   } else if (iter == f->end()) {
1051     ASSERT(f->size() >= 1);
1052     return (*f)[f->size() - 1].second;
1053   } else if (iter->first == x) {
1054     return iter->second;
1055   } else {
1056     double x1 = (iter - 1)->first;
1057     double y1 = (iter - 1)->second;
1058     double x2 = iter->first;
1059     double y2 = iter->second;
1060     return y1 + (y2 - y1) * (x - x1) / (x2 - x1);
1061   }
1062 }
1063
1064 bool VirtualSocketServer::CanInteractWith(VirtualSocket* local,
1065                                           VirtualSocket* remote) {
1066   if (!local || !remote) {
1067     return false;
1068   }
1069   IPAddress local_ip = local->GetLocalAddress().ipaddr();
1070   IPAddress remote_ip = remote->GetLocalAddress().ipaddr();
1071   IPAddress local_normalized = local_ip.Normalized();
1072   IPAddress remote_normalized = remote_ip.Normalized();
1073   // Check if the addresses are the same family after Normalization (turns
1074   // mapped IPv6 address into IPv4 addresses).
1075   // This will stop unmapped V6 addresses from talking to mapped V6 addresses.
1076   if (local_normalized.family() == remote_normalized.family()) {
1077     return true;
1078   }
1079
1080   // If ip1 is IPv4 and ip2 is :: and ip2 is not IPV6_V6ONLY.
1081   int remote_v6_only = 0;
1082   remote->GetOption(Socket::OPT_IPV6_V6ONLY, &remote_v6_only);
1083   if (local_ip.family() == AF_INET && !remote_v6_only && IPIsAny(remote_ip)) {
1084     return true;
1085   }
1086   // Same check, backwards.
1087   int local_v6_only = 0;
1088   local->GetOption(Socket::OPT_IPV6_V6ONLY, &local_v6_only);
1089   if (remote_ip.family() == AF_INET && !local_v6_only && IPIsAny(local_ip)) {
1090     return true;
1091   }
1092
1093   // Check to see if either socket was explicitly bound to IPv6-any.
1094   // These sockets can talk with anyone.
1095   if (local_ip.family() == AF_INET6 && local->was_any()) {
1096     return true;
1097   }
1098   if (remote_ip.family() == AF_INET6 && remote->was_any()) {
1099     return true;
1100   }
1101
1102   return false;
1103 }
1104
1105 }  // namespace rtc