Update talk to 59039880.

R=mallinath@webrtc.org

Review URL: https://webrtc-codereview.appspot.com/6569004

git-svn-id: http://webrtc.googlecode.com/svn/trunk/talk@5339 4adac7df-926f-26a2-2b94-8c16560cd09d

1 files changed, 284 insertions, 73 deletions

diff --git a/media/sctp/sctpdataengine.cc b/media/sctp/sctpdataengine.cc
index 653273b..5f18d4f 100644
--- a/media/sctp/sctpdataengine.cc
+++ b/media/sctp/sctpdataengine.cc
@@ -29,6 +29,7 @@
 
 #include <stdarg.h>
 #include <stdio.h>
+#include <sstream>
 #include <vector>
 
 #include "talk/app/webrtc/datachannelinterface.h"
@@ -41,7 +42,69 @@
 #include "talk/media/sctp/sctputils.h"
 #include "usrsctplib/usrsctp.h"
 
+namespace {
+typedef cricket::SctpDataMediaChannel::StreamSet StreamSet;
+// Returns a comma-separated, human-readable list of the stream IDs in 's'
+std::string ListStreams(const StreamSet& s) {
+  std::stringstream result;
+  bool first = true;
+  for (StreamSet::iterator it = s.begin(); it != s.end(); ++it) {
+    if (!first) {
+      result << ", " << *it;
+    } else {
+      result << *it;
+      first = false;
+    }
+  }
+  return result.str();
+}
+
+// Returns a pipe-separated, human-readable list of the SCTP_STREAM_RESET
+// flags in 'flags'
+std::string ListFlags(int flags) {
+  std::stringstream result;
+  bool first = true;
+  // Skip past the first 12 chars (strlen("SCTP_STREAM_"))
+#define MAKEFLAG(X) { X, #X + 12}
+  struct flaginfo_t {
+    int value;
+    const char* name;
+  } flaginfo[] = {
+    MAKEFLAG(SCTP_STREAM_RESET_INCOMING_SSN),
+    MAKEFLAG(SCTP_STREAM_RESET_OUTGOING_SSN),
+    MAKEFLAG(SCTP_STREAM_RESET_DENIED),
+    MAKEFLAG(SCTP_STREAM_RESET_FAILED),
+    MAKEFLAG(SCTP_STREAM_CHANGE_DENIED)
+  };
+#undef MAKEFLAG
+  for (int i = 0; i < ARRAY_SIZE(flaginfo); ++i) {
+    if (flags & flaginfo[i].value) {
+      if (!first) result << " | ";
+      result << flaginfo[i].name;
+      first = false;
+    }
+  }
+  return result.str();
+}
+
+// Returns a comma-separated, human-readable list of the integers in 'array'.
+// All 'num_elems' of them.
+std::string ListArray(const uint16* array, int num_elems) {
+  std::stringstream result;
+  for (int i = 0; i < num_elems; ++i) {
+    if (i) {
+      result << ", " << array[i];
+    } else {
+      result << array[i];
+    }
+  }
+  return result.str();
+}
+}  // namespace
+
 namespace cricket {
+typedef talk_base::ScopedMessageData<SctpInboundPacket> InboundPacketMessage;
+typedef talk_base::ScopedMessageData<talk_base::Buffer> OutboundPacketMessage;
 
 // This is the SCTP port to use. It is passed along the wire and the listener
 // and connector must be using the same port. It is not related to the ports at
@@ -130,9 +193,9 @@ static int OnSctpOutboundPacket(void* addr, void* data, size_t length,
                   << "; tos: " << std::hex << static_cast<int>(tos)
                   << "; set_df: " << std::hex << static_cast<int>(set_df);
   // Note: We have to copy the data; the caller will delete it.
-  talk_base::Buffer* buffer = new talk_base::Buffer(data, length);
-  channel->worker_thread()->Post(channel, MSG_SCTPOUTBOUNDPACKET,
-                                 talk_base::WrapMessageData(buffer));
+  OutboundPacketMessage* msg =
+      new OutboundPacketMessage(new talk_base::Buffer(data, length));
+  channel->worker_thread()->Post(channel, MSG_SCTPOUTBOUNDPACKET, msg);
   return 0;
 }
 
@@ -164,8 +227,9 @@ static int OnSctpInboundPacket(struct socket* sock, union sctp_sockstore addr,
     packet->params.timestamp = rcv.rcv_tsn;
     packet->params.type = type;
     packet->flags = flags;
-    channel->worker_thread()->Post(channel, MSG_SCTPINBOUNDPACKET,
-                                   talk_base::WrapMessageData(packet));
+    // The ownership of |packet| transfers to |msg|.
+    InboundPacketMessage* msg = new InboundPacketMessage(packet);
+    channel->worker_thread()->Post(channel, MSG_SCTPINBOUNDPACKET, msg);
   }
   free(data);
   return 1;
@@ -300,6 +364,18 @@ bool SctpDataMediaChannel::OpenSctpSocket() {
     return false;
   }
 
+  // Enable stream ID resets.
+  struct sctp_assoc_value stream_rst;
+  stream_rst.assoc_id = SCTP_ALL_ASSOC;
+  stream_rst.assoc_value = 1;
+  if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_ENABLE_STREAM_RESET,
+                         &stream_rst, sizeof(stream_rst))) {
+    LOG_ERRNO(LS_ERROR) << debug_name_
+                        << "Failed to set SCTP_ENABLE_STREAM_RESET.";
+    return false;
+  }
+
+  // Nagle.
   uint32_t nodelay = 1;
   if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_NODELAY, &nodelay,
                          sizeof(nodelay))) {
@@ -311,7 +387,8 @@ bool SctpDataMediaChannel::OpenSctpSocket() {
   int event_types[] = {SCTP_ASSOC_CHANGE,
                        SCTP_PEER_ADDR_CHANGE,
                        SCTP_SEND_FAILED_EVENT,
-                       SCTP_SENDER_DRY_EVENT};
+                       SCTP_SENDER_DRY_EVENT,
+                       SCTP_STREAM_RESET_EVENT};
   struct sctp_event event = {0};
   event.se_assoc_id = SCTP_ALL_ASSOC;
   event.se_on = 1;
@@ -412,60 +489,19 @@ bool SctpDataMediaChannel::SetReceive(bool receive) {
 }
 
 bool SctpDataMediaChannel::AddSendStream(const StreamParams& stream) {
-  if (!stream.has_ssrcs()) {
-    return false;
-  }
-
-  StreamParams found_stream;
-  // TODO(lally): Consider keeping this sorted.
-  if (GetStreamBySsrc(streams_, stream.first_ssrc(), &found_stream)) {
-    LOG(LS_WARNING) << debug_name_ << "->AddSendStream(...): "
-                    << "Not adding data send stream '" << stream.id
-                    << "' with ssrc=" << stream.first_ssrc()
-                    << " because stream already exists.";
-    return false;
-  }
-
-  streams_.push_back(stream);
-  return true;
+  return AddStream(stream);
 }
 
 bool SctpDataMediaChannel::RemoveSendStream(uint32 ssrc) {
-  StreamParams found_stream;
-  if (!GetStreamBySsrc(streams_, ssrc, &found_stream)) {
-    return false;
-  }
-
-  RemoveStreamBySsrc(&streams_, ssrc);
-  return true;
+  return ResetStream(ssrc);
 }
 
-// Note: expects exactly one ssrc.  If none are given, it will fail.  If more
-// than one are given, it will use the first.
 bool SctpDataMediaChannel::AddRecvStream(const StreamParams& stream) {
-  if (!stream.has_ssrcs()) {
-    return false;
-  }
-
-  StreamParams found_stream;
-  if (GetStreamBySsrc(streams_, stream.first_ssrc(), &found_stream)) {
-    LOG(LS_WARNING) << debug_name_ << "->AddRecvStream(...): "
-                    << "Not adding data recv stream '" << stream.id
-                    << "' with ssrc=" << stream.first_ssrc()
-                    << " because stream already exists.";
-    return false;
-  }
-
-  streams_.push_back(stream);
-  LOG(LS_VERBOSE) << debug_name_ << "->AddRecvStream(...): "
-                  << "Added data recv stream '" << stream.id
-                  << "' with ssrc=" << stream.first_ssrc();
-  return true;
+  return AddStream(stream);
 }
 
 bool SctpDataMediaChannel::RemoveRecvStream(uint32 ssrc) {
-  RemoveStreamBySsrc(&streams_, ssrc);
-  return true;
+  return ResetStream(ssrc);
 }
 
 bool SctpDataMediaChannel::SendData(
@@ -485,9 +521,8 @@ bool SctpDataMediaChannel::SendData(
     return false;
   }
 
-  StreamParams found_stream;
   if (params.type != cricket::DMT_CONTROL &&
-      !GetStreamBySsrc(streams_, params.ssrc, &found_stream)) {
+      open_streams_.find(params.ssrc) == open_streams_.end()) {
     LOG(LS_WARNING) << debug_name_ << "->SendData(...): "
                     << "Not sending data because ssrc is unknown: "
                     << params.ssrc;
@@ -584,8 +619,7 @@ void SctpDataMediaChannel::OnInboundPacketFromSctpToChannel(
 
 void SctpDataMediaChannel::OnDataFromSctpToChannel(
     const ReceiveDataParams& params, talk_base::Buffer* buffer) {
-  StreamParams found_stream;
-  if (!GetStreamBySsrc(streams_, params.ssrc, &found_stream)) {
+  if (open_streams_.find(params.ssrc) == open_streams_.end()) {
     if (params.type == DMT_CONTROL) {
       std::string label;
       webrtc::DataChannelInit config;
@@ -596,8 +630,7 @@ void SctpDataMediaChannel::OnDataFromSctpToChannel(
         SignalNewStreamReceived(label, config);
 
         // Add the stream immediately.
-        cricket::StreamParams sparams =
-            cricket::StreamParams::CreateLegacy(params.ssrc);
+        StreamParams sparams = StreamParams::CreateLegacy(params.ssrc);
         AddSendStream(sparams);
         AddRecvStream(sparams);
       } else {
@@ -623,6 +656,61 @@ void SctpDataMediaChannel::OnDataFromSctpToChannel(
   }
 }
 
+bool SctpDataMediaChannel::AddStream(const StreamParams& stream) {
+  if (!stream.has_ssrcs()) {
+    return false;
+  }
+
+  const uint32 ssrc = stream.first_ssrc();
+  if (open_streams_.find(ssrc) != open_streams_.end()) {
+    // We usually get an AddSendStream and an AddRecvStream for each stream, so
+    // this is really unlikely to be a useful warning message.
+    LOG(LS_VERBOSE) << debug_name_ << "->Add(Send|Recv)Stream(...): "
+                    << "Not adding data stream '" << stream.id
+                    << "' with ssrc=" << ssrc
+                    << " because stream is already open.";
+    return false;
+  } else if (queued_reset_streams_.find(ssrc) != queued_reset_streams_.end()
+             || sent_reset_streams_.find(ssrc) != sent_reset_streams_.end()) {
+    LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
+                    << "Not adding data stream '" << stream.id
+                    << "' with ssrc=" << ssrc
+                    << " because stream is still closing.";
+    return false;
+  }
+
+  open_streams_.insert(ssrc);
+  return true;
+}
+
+bool SctpDataMediaChannel::ResetStream(uint32 ssrc) {
+  // We typically get this called twice for the same stream, once each for
+  // Send and Recv.
+  StreamSet::iterator found = open_streams_.find(ssrc);
+
+  if (found == open_streams_.end()) {
+    LOG(LS_VERBOSE) << debug_name_ << "->ResetStream(" << ssrc << "): "
+                    << "stream not found.";
+    return false;
+  } else {
+    LOG(LS_VERBOSE) << debug_name_ << "->ResetStream(" << ssrc << "): "
+                    << "Removing and queuing RE-CONFIG chunk.";
+    open_streams_.erase(found);
+  }
+
+  // SCTP won't let you have more than one stream reset pending at a time, but
+  // you can close multiple streams in a single reset.  So, we keep an internal
+  // queue of streams-to-reset, and send them as one reset message in
+  // SendQueuedStreamResets().
+  queued_reset_streams_.insert(ssrc);
+
+  // Signal our stream-reset logic that it should try to send now, if it can.
+  SendQueuedStreamResets();
+
+  // The stream will actually get removed when we get the acknowledgment.
+  return true;
+}
+
 void SctpDataMediaChannel::OnNotificationFromSctp(talk_base::Buffer* buffer) {
   const sctp_notification& notification =
       reinterpret_cast<const sctp_notification&>(*buffer->data());
@@ -641,7 +729,7 @@ void SctpDataMediaChannel::OnNotificationFromSctp(talk_base::Buffer* buffer) {
       LOG(LS_INFO) << "SCTP_SHUTDOWN_EVENT";
       break;
     case SCTP_ADAPTATION_INDICATION:
-      LOG(LS_INFO) << "SCTP_ADAPTATION_INIDICATION";
+      LOG(LS_INFO) << "SCTP_ADAPTATION_INDICATION";
       break;
     case SCTP_PARTIAL_DELIVERY_EVENT:
       LOG(LS_INFO) << "SCTP_PARTIAL_DELIVERY_EVENT";
@@ -650,7 +738,7 @@ void SctpDataMediaChannel::OnNotificationFromSctp(talk_base::Buffer* buffer) {
       LOG(LS_INFO) << "SCTP_AUTHENTICATION_EVENT";
       break;
     case SCTP_SENDER_DRY_EVENT:
-      LOG(LS_INFO) << "SCTP_SENDER_DRY_EVENT";
+      LOG(LS_VERBOSE) << "SCTP_SENDER_DRY_EVENT";
       SignalReadyToSend(true);
       break;
     // TODO(ldixon): Unblock after congestion.
@@ -661,15 +749,18 @@ void SctpDataMediaChannel::OnNotificationFromSctp(talk_base::Buffer* buffer) {
       LOG(LS_INFO) << "SCTP_SEND_FAILED_EVENT";
       break;
     case SCTP_STREAM_RESET_EVENT:
-      LOG(LS_INFO) << "SCTP_STREAM_RESET_EVENT";
-      // TODO(ldixon): Notify up to channel that stream resent has happened,
-      // and write unit test for this case.
+      OnStreamResetEvent(&notification.sn_strreset_event);
       break;
     case SCTP_ASSOC_RESET_EVENT:
       LOG(LS_INFO) << "SCTP_ASSOC_RESET_EVENT";
       break;
     case SCTP_STREAM_CHANGE_EVENT:
       LOG(LS_INFO) << "SCTP_STREAM_CHANGE_EVENT";
+      // An acknowledgment we get after our stream resets have gone through,
+      // if they've failed.  We log the message, but don't react -- we don't
+      // keep around the last-transmitted set of SSIDs we wanted to close for
+      // error recovery.  It doesn't seem likely to occur, and if so, likely
+      // harmless within the lifetime of a single SCTP association.
       break;
     default:
       LOG(LS_WARNING) << "Unknown SCTP event: "
@@ -702,6 +793,91 @@ void SctpDataMediaChannel::OnNotificationAssocChange(
   }
 }
 
+void SctpDataMediaChannel::OnStreamResetEvent(
+    const struct sctp_stream_reset_event* evt) {
+  // A stream reset always involves two RE-CONFIG chunks for us -- we always
+  // simultaneously reset a sid's sequence number in both directions.  The
+  // requesting side transmits a RE-CONFIG chunk and waits for the peer to send
+  // one back.  Both sides get this SCTP_STREAM_RESET_EVENT when they receive
+  // RE-CONFIGs.
+  const int num_ssrcs = (evt->strreset_length - sizeof(*evt)) /
+      sizeof(evt->strreset_stream_list[0]);
+  LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
+                  << "): Flags = 0x"
+                  << std::hex << evt->strreset_flags << " ("
+                  << ListFlags(evt->strreset_flags) << ")";
+  LOG(LS_VERBOSE) << "Assoc = " << evt->strreset_assoc_id << ", Streams = ["
+                  << ListArray(evt->strreset_stream_list, num_ssrcs)
+                  << "], Open: ["
+                  << ListStreams(open_streams_) << "], Q'd: ["
+                  << ListStreams(queued_reset_streams_) << "], Sent: ["
+                  << ListStreams(sent_reset_streams_) << "]";
+  bool local_stream_reset_acknowledged = false;
+
+  // If both sides try to reset some streams at the same time (even if they're
+  // disjoint sets), we can get reset failures.
+  if (evt->strreset_flags & SCTP_STREAM_RESET_FAILED) {
+    // OK, just try again.  The stream IDs sent over when the RESET_FAILED flag
+    // is set seem to be garbage values.  Ignore them.
+    queued_reset_streams_.insert(
+        sent_reset_streams_.begin(),
+        sent_reset_streams_.end());
+    sent_reset_streams_.clear();
+    local_stream_reset_acknowledged = true;
+
+  } else if (evt->strreset_flags & SCTP_STREAM_RESET_INCOMING_SSN) {
+    // Each side gets an event for each direction of a stream.  That is,
+    // closing sid k will make each side receive INCOMING and OUTGOING reset
+    // events for k.  As per RFC6525, Section 5, paragraph 2, each side will
+    // get an INCOMING event first.
+    for (int i = 0; i < num_ssrcs; i++) {
+      const int stream_id = evt->strreset_stream_list[i];
+
+      // See if this stream ID was closed by our peer or ourselves.
+      StreamSet::iterator it = sent_reset_streams_.find(stream_id);
+
+      // The reset was requested locally.
+      if (it != sent_reset_streams_.end()) {
+        LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
+                        << "): local sid " << stream_id << " acknowledged.";
+        local_stream_reset_acknowledged = true;
+        sent_reset_streams_.erase(it);
+
+      } else if ((it = open_streams_.find(stream_id))
+                 != open_streams_.end()) {
+        // The peer requested the reset.
+        LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
+                        << "): closing sid " << stream_id;
+        open_streams_.erase(it);
+        SignalStreamClosed(stream_id);
+
+      } else if ((it = queued_reset_streams_.find(stream_id))
+                 != queued_reset_streams_.end()) {
+        // The peer requested the reset, but there was a local reset
+        // queued.
+        LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
+                        << "): double-sided close for sid " << stream_id;
+        // Both sides want the stream closed, and the peer got to send the
+        // RE-CONFIG first.  Treat it like the local Remove(Send|Recv)Stream
+        // finished quickly.
+        queued_reset_streams_.erase(it);
+
+      } else {
+        // This stream is unknown.  Sometimes this can be from an
+        // RESET_FAILED-related retransmit.
+        LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
+                        << "): Unknown sid " << stream_id;
+      }
+    }
+  }
+
+  if (local_stream_reset_acknowledged) {
+    // This message acknowledges the last stream-reset request we sent out
+    // (only one can be outstanding at a time).  Send out the next one.
+    SendQueuedStreamResets();
+  }
+}
+
 // Puts the specified |param| from the codec identified by |id| into |dest|
 // and returns true.  Or returns false if it wasn't there, leaving |dest|
 // untouched.
@@ -739,28 +915,63 @@ void SctpDataMediaChannel::OnPacketFromSctpToNetwork(
     talk_base::Buffer* buffer) {
   if (buffer->length() > kSctpMtu) {
     LOG(LS_ERROR) << debug_name_ << "->OnPacketFromSctpToNetwork(...): "
-                  << "SCTP seems to have made a poacket that is bigger "
+                  << "SCTP seems to have made a packet that is bigger "
                      "than its official MTU.";
   }
   MediaChannel::SendPacket(buffer);
 }
 
+bool SctpDataMediaChannel::SendQueuedStreamResets() {
+  if (!sent_reset_streams_.empty() || queued_reset_streams_.empty())
+    return true;
+
+  LOG(LS_VERBOSE) << "SendQueuedStreamResets[" << debug_name_ << "]: Sending ["
+                  << ListStreams(queued_reset_streams_) << "], Open: ["
+                  << ListStreams(open_streams_) << "], Sent: ["
+                  << ListStreams(sent_reset_streams_) << "]";
+
+  const size_t num_streams = queued_reset_streams_.size();
+  const size_t num_bytes = sizeof(struct sctp_reset_streams)
+    + (num_streams * sizeof(uint16));
+
+  std::vector<uint8> reset_stream_buf(num_bytes, 0);
+  struct sctp_reset_streams* resetp = reinterpret_cast<sctp_reset_streams*>(
+      &reset_stream_buf[0]);
+  resetp->srs_assoc_id = SCTP_ALL_ASSOC;
+  resetp->srs_flags = SCTP_STREAM_RESET_INCOMING | SCTP_STREAM_RESET_OUTGOING;
+  resetp->srs_number_streams = num_streams;
+  int result_idx = 0;
+  for (StreamSet::iterator it = queued_reset_streams_.begin();
+       it != queued_reset_streams_.end(); ++it) {
+    resetp->srs_stream_list[result_idx++] = *it;
+  }
+
+  int ret = usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_RESET_STREAMS, resetp,
+                               reset_stream_buf.size());
+  if (ret < 0) {
+    LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to send a stream reset for "
+                        << num_streams << " streams";
+    return false;
+  }
+
+  // sent_reset_streams_ is empty, and all the queued_reset_streams_ go into
+  // it now.
+  queued_reset_streams_.swap(sent_reset_streams_);
+  return true;
+}
+
 void SctpDataMediaChannel::OnMessage(talk_base::Message* msg) {
   switch (msg->message_id) {
     case MSG_SCTPINBOUNDPACKET: {
-      SctpInboundPacket* packet =
-          static_cast<talk_base::TypedMessageData<SctpInboundPacket*>*>(
-              msg->pdata)->data();
-      OnInboundPacketFromSctpToChannel(packet);
-      delete packet;
+      talk_base::scoped_ptr<InboundPacketMessage> pdata(
+          static_cast<InboundPacketMessage*>(msg->pdata));
+      OnInboundPacketFromSctpToChannel(pdata->data().get());
       break;
     }
     case MSG_SCTPOUTBOUNDPACKET: {
-      talk_base::Buffer* buffer =
-          static_cast<talk_base::TypedMessageData<talk_base::Buffer*>*>(
-              msg->pdata)->data();
-      OnPacketFromSctpToNetwork(buffer);
-      delete buffer;
+      talk_base::scoped_ptr<OutboundPacketMessage> pdata(
+          static_cast<OutboundPacketMessage*>(msg->pdata));
+      OnPacketFromSctpToNetwork(pdata->data().get());
       break;
     }
   }