Replace readable with receiving where receiving means receiving anything (stun ping, response or da…

webrtc/p2p/base/port.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 764 matching lines @@
 //
 // Connection
 //
 
 Connection::Connection(Port* port,
                        size_t index,
                        const Candidate& remote_candidate)
     : port_(port),
       local_candidate_index_(index),
       remote_candidate_(remote_candidate),
-      read_state_(STATE_READ_INIT),
       write_state_(STATE_WRITE_INIT),
+      receiving_(false),
       connected_(true),
       pruned_(false),
       use_candidate_attr_(false),
       nominated_(false),
       remote_ice_mode_(ICEMODE_FULL),
       requests_(port->thread()),
       rtt_(DEFAULT_RTT),
+      time_created_(rtc::Time()),
       last_ping_sent_(0),
       last_ping_received_(0),
       last_data_received_(0),
       last_ping_response_received_(0),
       recv_rate_tracker_(100u, 10u),
       send_rate_tracker_(100u, 10u),
       sent_packets_discarded_(0),
       sent_packets_total_(0),
       reported_(false),
-      state_(STATE_WAITING) {
+      state_(STATE_WAITING),
+      receiving_timeout_(WEAK_CONNECTION_RECEIVE_TIMEOUT) {
   // All of our connections start in WAITING state.
   // TODO(mallinath) - Start connections from STATE_FROZEN.
   // Wire up to send stun packets
   requests_.SignalSendPacket.connect(this, &Connection::OnSendStunPacket);
   LOG_J(LS_INFO, this) << "Connection created";
 }
 
 Connection::~Connection() {
 }
 
@@ skipping 20 matching lines @@
       g = remote_candidate_.priority();
       d = local_candidate().priority();
     }
     priority = std::min(g, d);
     priority = priority << 32;
     priority += 2 * std::max(g, d) + (g > d ? 1 : 0);
   }
   return priority;
 }
 
-void Connection::set_read_state(ReadState value) {
-  ReadState old_value = read_state_;
-  read_state_ = value;
-  if (value != old_value) {
-    LOG_J(LS_VERBOSE, this) << "set_read_state";
-    SignalStateChange(this);
-    CheckTimeout();
-  }
-}
-
 void Connection::set_write_state(WriteState value) {
   WriteState old_value = write_state_;
   write_state_ = value;
   if (value != old_value) {
     LOG_J(LS_VERBOSE, this) << "set_write_state from: " << old_value << " to "
                             << value;
     SignalStateChange(this);
     CheckTimeout();
   }
 }
 
+void Connection::set_receiving(bool value) {
+  if (value != receiving_) {
+    LOG_J(LS_VERBOSE, this) << "set_receiving to " << value;
+    receiving_ = value;
+    SignalStateChange(this);
+    CheckTimeout();
+  }
+}
+
 void Connection::set_state(State state) {
   State old_state = state_;
   state_ = state;
   if (state != old_state) {
     LOG_J(LS_VERBOSE, this) << "set_state";
   }
 }
 
 void Connection::set_connected(bool value) {
   bool old_value = connected_;
@@ skipping 20 matching lines @@
   }
 }
 
 void Connection::OnReadPacket(
   const char* data, size_t size, const rtc::PacketTime& packet_time) {
   rtc::scoped_ptr<IceMessage> msg;
   std::string remote_ufrag;
   const rtc::SocketAddress& addr(remote_candidate_.address());
   if (!port_->GetStunMessage(data, size, addr, msg.accept(), &remote_ufrag)) {
     // The packet did not parse as a valid STUN message
+    // This is a data packet, pass it along.
+    set_receiving(true);
+    last_data_received_ = rtc::Time();
+    recv_rate_tracker_.AddSamples(size);
+    SignalReadPacket(this, data, size, packet_time);
 
-    // If this connection is readable, then pass along the packet.
-    if (read_state_ == STATE_READABLE) {
-      // readable means data from this address is acceptable
-      // Send it on!
-      last_data_received_ = rtc::Time();
-      recv_rate_tracker_.AddSamples(size);
-      SignalReadPacket(this, data, size, packet_time);
-
-      // If timed out sending writability checks, start up again
-      if (!pruned_ && (write_state_ == STATE_WRITE_TIMEOUT)) {
-        LOG(LS_WARNING) << "Received a data packet on a timed-out Connection. "
-                        << "Resetting state to STATE_WRITE_INIT.";
-        set_write_state(STATE_WRITE_INIT);
-      }
-    } else {
-      // Not readable means the remote address hasn't sent a valid
-      // binding request yet.
-
-      LOG_J(LS_WARNING, this)
-        << "Received non-STUN packet from an unreadable connection.";
+    // If timed out sending writability checks, start up again
+    if (!pruned_ && (write_state_ == STATE_WRITE_TIMEOUT)) {
+      LOG(LS_WARNING) << "Received a data packet on a timed-out Connection. "
+                      << "Resetting state to STATE_WRITE_INIT.";
+      set_write_state(STATE_WRITE_INIT);

[pthatcher1, 2015/09/18 21:51:33]

Interesting.  This basically means "ping every connection for which we are
receiving data".  That's kind of like "unprune if we receive data".  We should
remember to address this in your second CL when we change the definition of
"pruned".

[honghaiz3, 2015/09/18 22:03:02]

We need to un-prune sometimes. We also need this in other cases. 

     }
   } else if (!msg) {
     // The packet was STUN, but failed a check and was handled internally.
   } else {
     // The packet is STUN and passed the Port checks.
     // Perform our own checks to ensure this packet is valid.
     // If this is a STUN request, then update the readable bit and respond.
     // If this is a STUN response, then update the writable bit.
     // Log at LS_INFO if we receive a ping on an unwritable connection.
     rtc::LoggingSeverity sev = (!writable() ? rtc::LS_INFO : rtc::LS_VERBOSE);
@@ skipping 49 matching lines @@
           requests_.CheckResponse(msg.get());
         }
         // Otherwise silently discard the response message.
         break;
 
       // Remote end point sent an STUN indication instead of regular
       // binding request. In this case |last_ping_received_| will be updated.
       // Otherwise we can mark connection to read timeout. No response will be
       // sent in this scenario.
       case STUN_BINDING_INDICATION:
-        if (read_state_ == STATE_READABLE) {
-          ReceivedPing();
-        } else {
-          LOG_J(LS_WARNING, this) << "Received STUN binding indication "
-                                  << "from an unreadable connection.";
-        }
+        ReceivedPing();
         break;
 
       default:
         ASSERT(false);
         break;
     }
   }
 }
 
 void Connection::OnReadyToSend() {
   if (write_state_ == STATE_WRITABLE) {
     SignalReadyToSend(this);
   }
 }
 
 void Connection::Prune() {
   if (!pruned_) {
     LOG_J(LS_VERBOSE, this) << "Connection pruned";
     pruned_ = true;
     requests_.Clear();
     set_write_state(STATE_WRITE_TIMEOUT);
   }
 }
 
 void Connection::Destroy() {
   LOG_J(LS_VERBOSE, this) << "Connection destroyed";
-  set_read_state(STATE_READ_TIMEOUT);
-  set_write_state(STATE_WRITE_TIMEOUT);
+  port_->thread()->Post(this, MSG_DELETE);
 }
 
 void Connection::PrintPingsSinceLastResponse(std::string* s, size_t max) {
   std::ostringstream oss;
   oss << std::boolalpha;
   if (pings_since_last_response_.size() > max) {
     for (size_t i = 0; i < max; i++) {
       const SentPing& ping = pings_since_last_response_[i];
       oss << rtc::hex_encode(ping.id) << " ";
     }
@@ skipping 43 matching lines @@
                          << " ping failures and "
                          << now - pings_since_last_response_[0].sent_time
                          << " ms without a response,"
                          << " ms since last received ping="
                          << now - last_ping_received_
                          << " ms since last received data="
                          << now - last_data_received_
                          << " rtt=" << rtt;
     set_write_state(STATE_WRITE_UNRELIABLE);
   }
-
   if ((write_state_ == STATE_WRITE_UNRELIABLE ||
        write_state_ == STATE_WRITE_INIT) &&
       TooLongWithoutResponse(pings_since_last_response_,
                              CONNECTION_WRITE_TIMEOUT,
                              now)) {
     LOG_J(LS_INFO, this) << "Timed out after "
                          << now - pings_since_last_response_[0].sent_time
                          << " ms without a response"
                          << ", rtt=" << rtt;
     set_write_state(STATE_WRITE_TIMEOUT);
   }
+
+  // Check the receiving state.
+  uint32 last_recv_time = last_received();
+  bool receiving = now <= last_recv_time + receiving_timeout_;
+  set_receiving(receiving);
+  if (receiving) {
+    return;
+  }
+  // If this connection has never received anything, use the connection
+  // creating time as the starting time for connection timeout.
+  if (now > std::max(last_recv_time, time_created_) +
+      DEAD_CONNECTION_RECEIVE_TIMEOUT) {
+    // Delete self.
+    Destroy();
+  }

[pthatcher1, 2015/09/18 21:51:33]

Doesn't CheckTimeout() handle this?  I think we can leave this part off (and
time_created_ also).

[honghaiz3, 2015/09/18 22:03:02]

Done.

 }
 
 void Connection::Ping(uint32 now) {
   last_ping_sent_ = now;
   ConnectionRequest *req = new ConnectionRequest(this);
   pings_since_last_response_.push_back(SentPing(req->id(), now));
   LOG_J(LS_VERBOSE, this) << "Sending STUN ping "
                           << ", id=" << rtc::hex_encode(req->id());
   requests_.Send(req);
   state_ = STATE_INPROGRESS;
 }
 
 void Connection::ReceivedPing() {
+  set_receiving(true);
   last_ping_received_ = rtc::Time();
-  set_read_state(STATE_READABLE);
 }
 
 void Connection::ReceivedPingResponse() {
   // We've already validated that this is a STUN binding response with
   // the correct local and remote username for this connection.
   // So if we're not already, become writable. We may be bringing a pruned
   // connection back to life, but if we don't really want it, we can always
   // prune it again.
+  set_receiving(true);
   set_write_state(STATE_WRITABLE);
   set_state(STATE_SUCCEEDED);
   pings_since_last_response_.clear();
   last_ping_response_received_ = rtc::Time();
 }
 
 std::string Connection::ToDebugId() const {
   std::stringstream ss;
   ss << std::hex << this;
   return ss.str();
 }
 
 std::string Connection::ToString() const {
   const char CONNECT_STATE_ABBREV[2] = {
     '-',  // not connected (false)
     'C',  // connected (true)
   };
-  const char READ_STATE_ABBREV[3] = {
-    '-',  // STATE_READ_INIT
-    'R',  // STATE_READABLE
-    'x',  // STATE_READ_TIMEOUT
+  const char RECEIVE_STATE_ABBREV[2] = {
+    '-',  // not receiving (false)
+    'R',  // receiving (true)
   };
   const char WRITE_STATE_ABBREV[4] = {
     'W',  // STATE_WRITABLE
     'w',  // STATE_WRITE_UNRELIABLE
     '-',  // STATE_WRITE_INIT
     'x',  // STATE_WRITE_TIMEOUT
   };
   const std::string ICESTATE[4] = {
     "W",  // STATE_WAITING
     "I",  // STATE_INPROGRESS
     "S",  // STATE_SUCCEEDED
     "F"   // STATE_FAILED
   };
   const Candidate& local = local_candidate();
   const Candidate& remote = remote_candidate();
   std::stringstream ss;
   ss << "Conn[" << ToDebugId()
      << ":" << port_->content_name()
      << ":" << local.id() << ":" << local.component()
      << ":" << local.generation()
      << ":" << local.type() << ":" << local.protocol()
      << ":" << local.address().ToSensitiveString()
      << "->" << remote.id() << ":" << remote.component()
      << ":" << remote.priority()
      << ":" << remote.type() << ":"
      << remote.protocol() << ":" << remote.address().ToSensitiveString() << "|"
      << CONNECT_STATE_ABBREV[connected()]
-     << READ_STATE_ABBREV[read_state()]
+     << RECEIVE_STATE_ABBREV[receiving()]
      << WRITE_STATE_ABBREV[write_state()]
      << ICESTATE[state()] << "|"
      << priority() << "|";
   if (rtt_ < DEFAULT_RTT) {
     ss << rtt_ << "]";
   } else {
     ss << "-]";
   }
   return ss.str();
 }
 
 std::string Connection::ToSensitiveString() const {
   return ToString();
 }
 
 void Connection::OnConnectionRequestResponse(ConnectionRequest* request,
                                              StunMessage* response) {
   // Log at LS_INFO if we receive a ping response on an unwritable
   // connection.
   rtc::LoggingSeverity sev = !writable() ? rtc::LS_INFO : rtc::LS_VERBOSE;
 
   uint32 rtt = request->Elapsed();
 
   ReceivedPingResponse();
-  if (remote_ice_mode_ == ICEMODE_LITE) {
-    // A ice-lite end point never initiates ping requests. This will allow
-    // us to move to STATE_READABLE without an incoming ping request.
-    set_read_state(STATE_READABLE);
-  }
 
   if (LOG_CHECK_LEVEL_V(sev)) {
     bool use_candidate = (
         response->GetByteString(STUN_ATTR_USE_CANDIDATE) != nullptr);
     std::string pings;
     PrintPingsSinceLastResponse(&pings, 5);
     LOG_JV(sev, this) << "Received STUN ping response"
                       << ", id=" << rtc::hex_encode(request->id())
                       << ", code=0"  // Makes logging easier to parse.
                       << ", rtt=" << rtt
@@ skipping 47 matching lines @@
 void Connection::OnConnectionRequestSent(ConnectionRequest* request) {
   // Log at LS_INFO if we send a ping on an unwritable connection.
   rtc::LoggingSeverity sev = !writable() ? rtc::LS_INFO : rtc::LS_VERBOSE;
   bool use_candidate = use_candidate_attr();
   LOG_JV(sev, this) << "Sent STUN ping"
                     << ", id=" << rtc::hex_encode(request->id())
                     << ", use_candidate=" << use_candidate;
 }
 
 void Connection::CheckTimeout() {
-  // If both read and write have timed out or read has never initialized, then
-  // this connection can contribute no more to p2p socket unless at some later
-  // date readability were to come back.  However, we gave readability a long
-  // time to timeout, so at this point, it seems fair to get rid of this
-  // connection.
-  if ((read_state_ == STATE_READ_TIMEOUT ||
-       read_state_ == STATE_READ_INIT) &&
-      write_state_ == STATE_WRITE_TIMEOUT) {
+  // If write has timed out and it is not receiving, remove the connection.
+  if (!receiving_ && write_state_ == STATE_WRITE_TIMEOUT) {
     port_->thread()->Post(this, MSG_DELETE);
   }
 }
 
 void Connection::HandleRoleConflictFromPeer() {
   port_->SignalRoleConflict(port_);
 }
 
 void Connection::MaybeSetRemoteIceCredentials(const std::string& ice_ufrag,
                                               const std::string& ice_pwd) {
@@ skipping 135 matching lines @@
     ASSERT(sent < 0);
     error_ = port_->GetError();
     sent_packets_discarded_++;
   } else {
     send_rate_tracker_.AddSamples(sent);
   }
   return sent;
 }
 
 }  // namespace cricket