Adding ability to simulate EWOULDBLOCK/SignalReadyToSend.

webrtc/base/virtualsocketserver.cc

 /*
  *  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.
  */
 
@@ skipping 47 matching lines @@
   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);
+    RTC_DCHECK(NULL != data);
     data_ = new char[size_];
     memcpy(data_, data, size_);
   }
 
   ~Packet() override {
     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_);
+    RTC_DCHECK(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;
 };
 
 VirtualSocket::VirtualSocket(VirtualSocketServer* server,
                              int family,
                              int type,
                              bool async)
     : server_(server),
       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
+  RTC_DCHECK((type_ == SOCK_DGRAM) || (type_ == SOCK_STREAM));
+  RTC_DCHECK(async_ ||
+             (type_ != SOCK_STREAM));  // We only support async streams
+  server->SignalReadyToSend.connect(this,
+                                    &VirtualSocket::OnSocketServerReadyToSend);
 }
 
 VirtualSocket::~VirtualSocket() {
   Close();
 
   for (RecvBuffer::iterator it = recv_buffer_.begin(); it != recv_buffer_.end();
        ++it) {
     delete *it;
   }
 }
@@ skipping 77 matching lines @@
       // 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);
+      RTC_DCHECK(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);
@@ skipping 77 matching lines @@
   } 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);
+      RTC_DCHECK(NULL != sender);
       server_->SendTcp(sender);
     }
   }
 
   return static_cast<int>(data_read);
 }
 
 int VirtualSocket::Listen(int backlog) {
-  ASSERT(SOCK_STREAM == type_);
-  ASSERT(CS_CLOSED == state_);
+  RTC_DCHECK(SOCK_STREAM == type_);
+  RTC_DCHECK(CS_CLOSED == state_);
   if (local_addr_.IsNil()) {
     error_ = EINVAL;
     return -1;
   }
-  ASSERT(NULL == listen_queue_);
+  RTC_DCHECK(NULL == listen_queue_);
   listen_queue_ = new ListenQueue;
   state_ = CS_CONNECTING;
   return 0;
 }
 
 VirtualSocket* VirtualSocket::Accept(SocketAddress* paddr) {
   if (NULL == listen_queue_) {
     error_ = EINVAL;
     return NULL;
   }
@@ skipping 49 matching lines @@
 
 int VirtualSocket::EstimateMTU(uint16_t* mtu) {
   if (CS_CONNECTED != state_)
     return ENOTCONN;
   else
     return 65536;
 }
 
 void VirtualSocket::OnMessage(Message* pmsg) {
   if (pmsg->message_id == MSG_ID_PACKET) {
-    // ASSERT(!local_addr_.IsAnyIP());
-    ASSERT(NULL != pmsg->pdata);
+    // RTC_DCHECK(!local_addr_.IsAnyIP());

[skvlad, 2016/08/26 22:39:04]

Do you happen to know why this is commented out? Should we perhaps delete the
line instead?

[Taylor Brandstetter, 2016/08/26 23:01:03]

It appears this line has been commented out from the very beginning, and I can't
find an explanation. I'll just remove it.

+    RTC_DCHECK(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);
+    RTC_DCHECK(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_);
+    RTC_DCHECK(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 if (pmsg->message_id == MSG_ID_ADDRESS_BOUND) {
     SignalAddressReady(this, GetLocalAddress());
   } else {
-    ASSERT(false);
+    RTC_DCHECK(false);
   }
 }
 
 int VirtualSocket::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.
@@ skipping 15 matching lines @@
     if (result != 0) {
       error_ = EHOSTUNREACH;
       return -1;
     }
     state_ = CS_CONNECTING;
   }
   return 0;
 }
 
 void VirtualSocket::CompleteConnect(const SocketAddress& addr, bool notify) {
-  ASSERT(CS_CONNECTING == state_);
+  RTC_DCHECK(CS_CONNECTING == state_);
   remote_addr_ = addr;
   state_ = CS_CONNECTED;
   server_->AddConnection(remote_addr_, local_addr_, this);
   if (async_ && notify) {
     SignalConnectEvent(this);
   }
 }
 
 int VirtualSocket::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 VirtualSocket::SendTcp(const void* pv, size_t cb) {
   size_t capacity = server_->send_buffer_capacity_ - send_buffer_.size();
   if (0 == capacity) {
-    write_enabled_ = true;
+    ready_to_send_ = false;
     error_ = EWOULDBLOCK;
     return -1;
   }
   size_t consumed = std::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);
 }
 
+void VirtualSocket::OnSocketServerReadyToSend() {
+  if (ready_to_send_) {
+    // This socket didn't encounter EWOULDBLOCK, so there's nothing to do.
+    return;
+  }
+  if (type_ == SOCK_DGRAM) {
+    ready_to_send_ = true;
+    SignalWriteEvent(this);
+  } else {
+    RTC_DCHECK(type_ == SOCK_STREAM);
+    // This will attempt to empty the full send buffer, and will fire
+    // SignalWriteEvent if successful.
+    server_->SendTcp(this);
+  }
+}
+
 VirtualSocketServer::VirtualSocketServer(SocketServer* ss)
     : server_(ss),
       server_owned_(false),
       msg_queue_(NULL),
       stop_on_idle_(false),
       network_delay_(0),
       next_ipv4_(kInitialNextIPv4),
       next_ipv6_(kInitialNextIPv6),
       next_port_(kFirstEphemeralPort),
       bindings_(new AddressMap()),
@@ skipping 41 matching lines @@
 uint16_t VirtualSocketServer::GetNextPort() {
   uint16_t port = next_port_;
   if (next_port_ < kLastEphemeralPort) {
     ++next_port_;
   } else {
     next_port_ = kFirstEphemeralPort;
   }
   return port;
 }
 
+void VirtualSocketServer::SetSendingBlocked(bool blocked) {
+  if (blocked == sending_blocked_) {
+    // Unchanged; nothing to do.
+    return;
+  }
+  sending_blocked_ = blocked;
+  if (!sending_blocked_) {
+    // Sending was blocked, but is now unblocked. This signal gives sockets a
+    // chance to fire SignalWriteEvent, and for TCP, send buffered data.
+    SignalReadyToSend();
+  }
+}
+
 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);
@@ skipping 11 matching lines @@
 
 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());
+  RTC_DCHECK(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());
+  RTC_DCHECK(msg_queue_ == Thread::Current());
   stop_on_idle_ = true;
   while (!msg_queue_->empty()) {
     Message msg;
     if (msg_queue_->Get(&msg, Thread::kForever)) {
       msg_queue_->Dispatch(&msg);
     }
   }
   stop_on_idle_ = false;
   return !msg_queue_->IsQuitting();
 }
@@ skipping 13 matching lines @@
   socket->SignalCloseEvent(socket, 0);
 
   // Trigger the remote connection's close event.
   socket->Close();
 
   return true;
 }
 
 int VirtualSocketServer::Bind(VirtualSocket* socket,
                               const SocketAddress& addr) {
-  ASSERT(NULL != socket);
+  RTC_DCHECK(NULL != socket);
   // Address must be completely specified at this point
-  ASSERT(!IPIsUnspec(addr.ipaddr()));
-  ASSERT(addr.port() != 0);
+  RTC_DCHECK(!IPIsUnspec(addr.ipaddr()));
+  RTC_DCHECK(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);
+  RTC_DCHECK(NULL != socket);
 
   if (!IPIsUnspec(addr->ipaddr())) {
     addr->SetIP(addr->ipaddr().Normalized());
   } else {
-    ASSERT(false);
+    RTC_DCHECK(false);
   }
 
   if (addr->port() == 0) {
     for (int i = 0; i < kEphemeralPortCount; ++i) {
       addr->SetPort(GetNextPort());
       if (bindings_->find(*addr) == bindings_->end()) {
         break;
       }
     }
   }
@@ skipping 20 matching lines @@
     return LookupBinding(sock_addr);
   }
 
   return nullptr;
 }
 
 int VirtualSocketServer::Unbind(const SocketAddress& addr,
                                 VirtualSocket* socket) {
   SocketAddress normalized(addr.ipaddr().Normalized(),
                            addr.port());
-  ASSERT((*bindings_)[normalized] == socket);
+  RTC_DCHECK((*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(),
@@ skipping 60 matching lines @@
     // Remove the mapping.
     msg_queue_->PostDelayed(RTC_FROM_HERE, delay, socket, MSG_ID_DISCONNECT);
     return true;
   }
   return false;
 }
 
 int VirtualSocketServer::SendUdp(VirtualSocket* socket,
                                  const char* data, size_t data_size,
                                  const SocketAddress& remote_addr) {
+  if (sending_blocked_) {
+    CritScope cs(&socket->crit_);
+    socket->ready_to_send_ = false;
+    socket->error_ = EWOULDBLOCK;
+    return -1;
+  }
+
   // 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.
     std::unique_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_);
+  {
+    CritScope cs(&socket->crit_);
 
-  int64_t cur_time = TimeMillis();
-  PurgeNetworkPackets(socket, cur_time);
+    int64_t cur_time = TimeMillis();
+    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.
+    // 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";
+    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);
   }
-
-  AddPacketToNetwork(socket, recipient, cur_time, data, data_size,
-                     UDP_HEADER_SIZE, false);
-
-  return static_cast<int>(data_size);
 }
 
 void VirtualSocketServer::SendTcp(VirtualSocket* socket) {
+  if (sending_blocked_) {
+    // Eventually the socket's buffer will fill and VirtualSocket::SendTcp will
+    // set EWOULDBLOCK.
+    return;
+  }
+
   // 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) {
@@ skipping 22 matching lines @@
     // 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;
+  if (!socket->ready_to_send_ &&
+      (socket->send_buffer_.size() < send_buffer_capacity_)) {
+    socket->ready_to_send_ = true;
     socket->SignalWriteEvent(socket);
   }
 }
 
 void VirtualSocketServer::AddPacketToNetwork(VirtualSocket* sender,
                                              VirtualSocket* recipient,
                                              int64_t cur_time,
                                              const char* data,
                                              size_t data_size,
                                              size_t header_size,
@@ skipping 29 matching lines @@
     ts = std::max(ts, network_delay_);
   }
   msg_queue_->PostAt(RTC_FROM_HERE, ts, recipient, MSG_ID_PACKET, p);
   network_delay_ = std::max(ts, network_delay_);
 }
 
 void VirtualSocketServer::PurgeNetworkPackets(VirtualSocket* socket,
                                               int64_t cur_time) {
   while (!socket->network_.empty() &&
          (socket->network_.front().done_time <= cur_time)) {
-    ASSERT(socket->network_size_ >= socket->network_.front().size);
+    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;
   else
     return 1000 * size / bandwidth_;
@@ skipping 84 matching lines @@
   }
   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);
+  RTC_DCHECK(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;
 }
@@ skipping 21 matching lines @@
   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);
+    RTC_DCHECK(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);
   }
@@ skipping 52 matching lines @@
 void VirtualSocketServer::SetDefaultRoute(const IPAddress& from_addr) {
   RTC_DCHECK(!IPIsAny(from_addr));
   if (from_addr.family() == AF_INET) {
     default_route_v4_ = from_addr;
   } else if (from_addr.family() == AF_INET6) {
     default_route_v6_ = from_addr;
   }
 }
 
 }  // namespace rtc