src/third_party/libjingle/source/talk/media/sctp/sctpdataengine.cc

   1 /*
   2  * libjingle SCTP
   3  * Copyright 2012 Google Inc, and Robin Seggelmann
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions are met:
   7  *
   8  *  1. Redistributions of source code must retain the above copyright notice,
   9  *     this list of conditions and the following disclaimer.
  10  *  2. Redistributions in binary form must reproduce the above copyright notice,
  11  *     this list of conditions and the following disclaimer in the documentation
  12  *     and/or other materials provided with the distribution.
  13  *  3. The name of the author may not be used to endorse or promote products
  14  *     derived from this software without specific prior written permission.
  15  *
  16  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
  17  * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  18  * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
  19  * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  20  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  21  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
  22  * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
  23  * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
  24  * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
  25  * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26  */
  27
  28 #include "talk/media/sctp/sctpdataengine.h"
  29
  30 #include <stdarg.h>
  31 #include <stdio.h>
  32 #include <sstream>
  33 #include <vector>
  34
  35 #include "talk/base/buffer.h"
  36 #include "talk/base/helpers.h"
  37 #include "talk/base/logging.h"
  38 #include "talk/base/safe_conversions.h"
  39 #include "talk/media/base/codec.h"
  40 #include "talk/media/base/constants.h"
  41 #include "talk/media/base/streamparams.h"
  42 #include "usrsctplib/usrsctp.h"
  43
  44 namespace {
  45 typedef cricket::SctpDataMediaChannel::StreamSet StreamSet;
  46 // Returns a comma-separated, human-readable list of the stream IDs in 's'
  47 std::string ListStreams(const StreamSet& s) {
  48   std::stringstream result;
  49   bool first = true;
  50   for (StreamSet::const_iterator it = s.begin(); it != s.end(); ++it) {
  51     if (!first) {
  52       result << ", " << *it;
  53     } else {
  54       result << *it;
  55       first = false;
  56     }
  57   }
  58   return result.str();
  59 }
  60
  61 // Returns a pipe-separated, human-readable list of the SCTP_STREAM_RESET
  62 // flags in 'flags'
  63 std::string ListFlags(int flags) {
  64   std::stringstream result;
  65   bool first = true;
  66   // Skip past the first 12 chars (strlen("SCTP_STREAM_"))
  67 #define MAKEFLAG(X) { X, #X + 12}
  68   struct flaginfo_t {
  69     int value;
  70     const char* name;
  71   } flaginfo[] = {
  72     MAKEFLAG(SCTP_STREAM_RESET_INCOMING_SSN),
  73     MAKEFLAG(SCTP_STREAM_RESET_OUTGOING_SSN),
  74     MAKEFLAG(SCTP_STREAM_RESET_DENIED),
  75     MAKEFLAG(SCTP_STREAM_RESET_FAILED),
  76     MAKEFLAG(SCTP_STREAM_CHANGE_DENIED)
  77   };
  78 #undef MAKEFLAG
  79   for (int i = 0; i < ARRAY_SIZE(flaginfo); ++i) {
  80     if (flags & flaginfo[i].value) {
  81       if (!first) result << " | ";
  82       result << flaginfo[i].name;
  83       first = false;
  84     }
  85   }
  86   return result.str();
  87 }
  88
  89 // Returns a comma-separated, human-readable list of the integers in 'array'.
  90 // All 'num_elems' of them.
  91 std::string ListArray(const uint16* array, int num_elems) {
  92   std::stringstream result;
  93   for (int i = 0; i < num_elems; ++i) {
  94     if (i) {
  95       result << ", " << array[i];
  96     } else {
  97       result << array[i];
  98     }
  99   }
 100   return result.str();
 101 }
 102 }  // namespace
 103
 104 namespace cricket {
 105 typedef talk_base::ScopedMessageData<SctpInboundPacket> InboundPacketMessage;
 106 typedef talk_base::ScopedMessageData<talk_base::Buffer> OutboundPacketMessage;
 107
 108 // This is the SCTP port to use. It is passed along the wire and the listener
 109 // and connector must be using the same port. It is not related to the ports at
 110 // the IP level. (Corresponds to: sockaddr_conn.sconn_port in usrsctp.h)
 111 //
 112 // TODO(ldixon): Allow port to be set from higher level code.
 113 static const int kSctpDefaultPort = 5001;
 114 // TODO(ldixon): Find where this is defined, and also check is Sctp really
 115 // respects this.
 116 static const size_t kSctpMtu = 1280;
 117
 118 enum {
 119   MSG_SCTPINBOUNDPACKET = 1,   // MessageData is SctpInboundPacket
 120   MSG_SCTPOUTBOUNDPACKET = 2,  // MessageData is talk_base:Buffer
 121 };
 122
 123 struct SctpInboundPacket {
 124   talk_base::Buffer buffer;
 125   ReceiveDataParams params;
 126   // The |flags| parameter is used by SCTP to distinguish notification packets
 127   // from other types of packets.
 128   int flags;
 129 };
 130
 131 // Helper for logging SCTP messages.
 132 static void debug_sctp_printf(const char *format, ...) {
 133   char s[255];
 134   va_list ap;
 135   va_start(ap, format);
 136   vsnprintf(s, sizeof(s), format, ap);
 137   LOG(LS_INFO) << "SCTP: " << s;
 138   va_end(ap);
 139 }
 140
 141 // Get the PPID to use for the terminating fragment of this type.
 142 static SctpDataMediaChannel::PayloadProtocolIdentifier GetPpid(
 143     cricket::DataMessageType type) {
 144   switch (type) {
 145   default:
 146   case cricket::DMT_NONE:
 147     return SctpDataMediaChannel::PPID_NONE;
 148   case cricket::DMT_CONTROL:
 149     return SctpDataMediaChannel::PPID_CONTROL;
 150   case cricket::DMT_BINARY:
 151     return SctpDataMediaChannel::PPID_BINARY_LAST;
 152   case cricket::DMT_TEXT:
 153     return SctpDataMediaChannel::PPID_TEXT_LAST;
 154   };
 155 }
 156
 157 static bool GetDataMediaType(
 158     SctpDataMediaChannel::PayloadProtocolIdentifier ppid,
 159     cricket::DataMessageType *dest) {
 160   ASSERT(dest != NULL);
 161   switch (ppid) {
 162     case SctpDataMediaChannel::PPID_BINARY_PARTIAL:
 163     case SctpDataMediaChannel::PPID_BINARY_LAST:
 164       *dest = cricket::DMT_BINARY;
 165       return true;
 166
 167     case SctpDataMediaChannel::PPID_TEXT_PARTIAL:
 168     case SctpDataMediaChannel::PPID_TEXT_LAST:
 169       *dest = cricket::DMT_TEXT;
 170       return true;
 171
 172     case SctpDataMediaChannel::PPID_CONTROL:
 173       *dest = cricket::DMT_CONTROL;
 174       return true;
 175
 176     case SctpDataMediaChannel::PPID_NONE:
 177       *dest = cricket::DMT_NONE;
 178       return true;
 179
 180     default:
 181       return false;
 182   }
 183 }
 184
 185 // This is the callback usrsctp uses when there's data to send on the network
 186 // that has been wrapped appropriatly for the SCTP protocol.
 187 static int OnSctpOutboundPacket(void* addr, void* data, size_t length,
 188                                 uint8_t tos, uint8_t set_df) {
 189   SctpDataMediaChannel* channel = static_cast<SctpDataMediaChannel*>(addr);
 190   LOG(LS_VERBOSE) << "global OnSctpOutboundPacket():"
 191                   << "addr: " << addr << "; length: " << length
 192                   << "; tos: " << std::hex << static_cast<int>(tos)
 193                   << "; set_df: " << std::hex << static_cast<int>(set_df);
 194   // Note: We have to copy the data; the caller will delete it.
 195   OutboundPacketMessage* msg =
 196       new OutboundPacketMessage(new talk_base::Buffer(data, length));
 197   channel->worker_thread()->Post(channel, MSG_SCTPOUTBOUNDPACKET, msg);
 198   return 0;
 199 }
 200
 201 // This is the callback called from usrsctp when data has been received, after
 202 // a packet has been interpreted and parsed by usrsctp and found to contain
 203 // payload data. It is called by a usrsctp thread. It is assumed this function
 204 // will free the memory used by 'data'.
 205 static int OnSctpInboundPacket(struct socket* sock, union sctp_sockstore addr,
 206                                void* data, size_t length,
 207                                struct sctp_rcvinfo rcv, int flags,
 208                                void* ulp_info) {
 209   SctpDataMediaChannel* channel = static_cast<SctpDataMediaChannel*>(ulp_info);
 210   // Post data to the channel's receiver thread (copying it).
 211   // TODO(ldixon): Unclear if copy is needed as this method is responsible for
 212   // memory cleanup. But this does simplify code.
 213   const SctpDataMediaChannel::PayloadProtocolIdentifier ppid =
 214       static_cast<SctpDataMediaChannel::PayloadProtocolIdentifier>(
 215           talk_base::HostToNetwork32(rcv.rcv_ppid));
 216   cricket::DataMessageType type = cricket::DMT_NONE;
 217   if (!GetDataMediaType(ppid, &type) && !(flags & MSG_NOTIFICATION)) {
 218     // It's neither a notification nor a recognized data packet.  Drop it.
 219     LOG(LS_ERROR) << "Received an unknown PPID " << ppid
 220                   << " on an SCTP packet.  Dropping.";
 221   } else {
 222     SctpInboundPacket* packet = new SctpInboundPacket;
 223     packet->buffer.SetData(data, length);
 224     packet->params.ssrc = rcv.rcv_sid;
 225     packet->params.seq_num = rcv.rcv_ssn;
 226     packet->params.timestamp = rcv.rcv_tsn;
 227     packet->params.type = type;
 228     packet->flags = flags;
 229     // The ownership of |packet| transfers to |msg|.
 230     InboundPacketMessage* msg = new InboundPacketMessage(packet);
 231     channel->worker_thread()->Post(channel, MSG_SCTPINBOUNDPACKET, msg);
 232   }
 233   free(data);
 234   return 1;
 235 }
 236
 237 // Set the initial value of the static SCTP Data Engines reference count.
 238 int SctpDataEngine::usrsctp_engines_count = 0;
 239
 240 SctpDataEngine::SctpDataEngine() {
 241   if (usrsctp_engines_count == 0) {
 242     // First argument is udp_encapsulation_port, which is not releveant for our
 243     // AF_CONN use of sctp.
 244     usrsctp_init(0, cricket::OnSctpOutboundPacket, debug_sctp_printf);
 245
 246     // To turn on/off detailed SCTP debugging. You will also need to have the
 247     // SCTP_DEBUG cpp defines flag.
 248     // usrsctp_sysctl_set_sctp_debug_on(SCTP_DEBUG_ALL);
 249
 250     // TODO(ldixon): Consider turning this on/off.
 251     usrsctp_sysctl_set_sctp_ecn_enable(0);
 252
 253     // TODO(ldixon): Consider turning this on/off.
 254     // This is not needed right now (we don't do dynamic address changes):
 255     // If SCTP Auto-ASCONF is enabled, the peer is informed automatically
 256     // when a new address is added or removed. This feature is enabled by
 257     // default.
 258     // usrsctp_sysctl_set_sctp_auto_asconf(0);
 259
 260     // TODO(ldixon): Consider turning this on/off.
 261     // Add a blackhole sysctl. Setting it to 1 results in no ABORTs
 262     // being sent in response to INITs, setting it to 2 results
 263     // in no ABORTs being sent for received OOTB packets.
 264     // This is similar to the TCP sysctl.
 265     //
 266     // See: http://lakerest.net/pipermail/sctp-coders/2012-January/009438.html
 267     // See: http://svnweb.freebsd.org/base?view=revision&revision=229805
 268     // usrsctp_sysctl_set_sctp_blackhole(2);
 269
 270     // Set the number of default outgoing streams.  This is the number we'll
 271     // send in the SCTP INIT message.  The 'appropriate default' in the
 272     // second paragraph of
 273     // http://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-05#section-6.2
 274     // is cricket::kMaxSctpSid.
 275     usrsctp_sysctl_set_sctp_nr_outgoing_streams_default(
 276         cricket::kMaxSctpSid);
 277   }
 278   usrsctp_engines_count++;
 279
 280   // We don't put in a codec because we don't want one offered when we
 281   // use the hybrid data engine.
 282   // codecs_.push_back(cricket::DataCodec( kGoogleSctpDataCodecId,
 283   // kGoogleSctpDataCodecName, 0));
 284 }
 285
 286 SctpDataEngine::~SctpDataEngine() {
 287   // TODO(ldixon): There is currently a bug in teardown of usrsctp that blocks
 288   // indefintely if a finish call made too soon after close calls. So teardown
 289   // has been skipped. Once the bug is fixed, retest and enable teardown.
 290   // Tracked in webrtc issue 2749.
 291   //
 292   // usrsctp_engines_count--;
 293   // LOG(LS_VERBOSE) << "usrsctp_engines_count:" << usrsctp_engines_count;
 294   // if (usrsctp_engines_count == 0) {
 295   //   if (usrsctp_finish() != 0) {
 296   //     LOG(LS_WARNING) << "usrsctp_finish.";
 297   //   }
 298   // }
 299 }
 300
 301 DataMediaChannel* SctpDataEngine::CreateChannel(
 302     DataChannelType data_channel_type) {
 303   if (data_channel_type != DCT_SCTP) {
 304     return NULL;
 305   }
 306   return new SctpDataMediaChannel(talk_base::Thread::Current());
 307 }
 308
 309 SctpDataMediaChannel::SctpDataMediaChannel(talk_base::Thread* thread)
 310     : worker_thread_(thread),
 311       local_port_(-1),
 312       remote_port_(-1),
 313       sock_(NULL),
 314       sending_(false),
 315       receiving_(false),
 316       debug_name_("SctpDataMediaChannel") {
 317 }
 318
 319 SctpDataMediaChannel::~SctpDataMediaChannel() {
 320   CloseSctpSocket();
 321 }
 322
 323 sockaddr_conn SctpDataMediaChannel::GetSctpSockAddr(int port) {
 324   sockaddr_conn sconn = {0};
 325   sconn.sconn_family = AF_CONN;
 326 #ifdef HAVE_SCONN_LEN
 327   sconn.sconn_len = sizeof(sockaddr_conn);
 328 #endif
 329   // Note: conversion from int to uint16_t happens here.
 330   sconn.sconn_port = talk_base::HostToNetwork16(port);
 331   sconn.sconn_addr = this;
 332   return sconn;
 333 }
 334
 335 bool SctpDataMediaChannel::OpenSctpSocket() {
 336   if (sock_) {
 337     LOG(LS_VERBOSE) << debug_name_
 338                     << "->Ignoring attempt to re-create existing socket.";
 339     return false;
 340   }
 341   sock_ = usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP,
 342                          cricket::OnSctpInboundPacket, NULL, 0, this);
 343   if (!sock_) {
 344     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to create SCTP socket.";
 345     return false;
 346   }
 347
 348   // Make the socket non-blocking. Connect, close, shutdown etc will not block
 349   // the thread waiting for the socket operation to complete.
 350   if (usrsctp_set_non_blocking(sock_, 1) < 0) {
 351     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP to non blocking.";
 352     return false;
 353   }
 354
 355   // This ensures that the usrsctp close call deletes the association. This
 356   // prevents usrsctp from calling OnSctpOutboundPacket with references to
 357   // this class as the address.
 358   linger linger_opt;
 359   linger_opt.l_onoff = 1;
 360   linger_opt.l_linger = 0;
 361   if (usrsctp_setsockopt(sock_, SOL_SOCKET, SO_LINGER, &linger_opt,
 362                          sizeof(linger_opt))) {
 363     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SO_LINGER.";
 364     return false;
 365   }
 366
 367   // Enable stream ID resets.
 368   struct sctp_assoc_value stream_rst;
 369   stream_rst.assoc_id = SCTP_ALL_ASSOC;
 370   stream_rst.assoc_value = 1;
 371   if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_ENABLE_STREAM_RESET,
 372                          &stream_rst, sizeof(stream_rst))) {
 373     LOG_ERRNO(LS_ERROR) << debug_name_
 374                         << "Failed to set SCTP_ENABLE_STREAM_RESET.";
 375     return false;
 376   }
 377
 378   // Nagle.
 379   uint32_t nodelay = 1;
 380   if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_NODELAY, &nodelay,
 381                          sizeof(nodelay))) {
 382     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP_NODELAY.";
 383     return false;
 384   }
 385
 386   // Subscribe to SCTP event notifications.
 387   int event_types[] = {SCTP_ASSOC_CHANGE,
 388                        SCTP_PEER_ADDR_CHANGE,
 389                        SCTP_SEND_FAILED_EVENT,
 390                        SCTP_SENDER_DRY_EVENT,
 391                        SCTP_STREAM_RESET_EVENT};
 392   struct sctp_event event = {0};
 393   event.se_assoc_id = SCTP_ALL_ASSOC;
 394   event.se_on = 1;
 395   for (size_t i = 0; i < ARRAY_SIZE(event_types); i++) {
 396     event.se_type = event_types[i];
 397     if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_EVENT, &event,
 398                            sizeof(event)) < 0) {
 399       LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP_EVENT type: "
 400                           << event.se_type;
 401       return false;
 402     }
 403   }
 404
 405   // Register this class as an address for usrsctp. This is used by SCTP to
 406   // direct the packets received (by the created socket) to this class.
 407   usrsctp_register_address(this);
 408   sending_ = true;
 409   return true;
 410 }
 411
 412 void SctpDataMediaChannel::CloseSctpSocket() {
 413   sending_ = false;
 414   if (sock_) {
 415     // We assume that SO_LINGER option is set to close the association when
 416     // close is called. This means that any pending packets in usrsctp will be
 417     // discarded instead of being sent.
 418     usrsctp_close(sock_);
 419     sock_ = NULL;
 420     usrsctp_deregister_address(this);
 421   }
 422 }
 423
 424 bool SctpDataMediaChannel::Connect() {
 425   LOG(LS_VERBOSE) << debug_name_ << "->Connect().";
 426   if (remote_port_ < 0) {
 427     remote_port_ = kSctpDefaultPort;
 428   }
 429   if (local_port_ < 0) {
 430     local_port_ = kSctpDefaultPort;
 431   }
 432
 433   // If we already have a socket connection, just return.
 434   if (sock_) {
 435     LOG(LS_WARNING) << debug_name_ << "->Connect(): Ignored as socket "
 436                                       "is already established.";
 437     return true;
 438   }
 439
 440   // If no socket (it was closed) try to start it again. This can happen when
 441   // the socket we are connecting to closes, does an sctp shutdown handshake,
 442   // or behaves unexpectedly causing us to perform a CloseSctpSocket.
 443   if (!sock_ && !OpenSctpSocket()) {
 444     return false;
 445   }
 446
 447   // Note: conversion from int to uint16_t happens on assignment.
 448   sockaddr_conn local_sconn = GetSctpSockAddr(local_port_);
 449   if (usrsctp_bind(sock_, reinterpret_cast<sockaddr *>(&local_sconn),
 450                    sizeof(local_sconn)) < 0) {
 451     LOG_ERRNO(LS_ERROR) << debug_name_ << "->Connect(): "
 452                         << ("Failed usrsctp_bind");
 453     CloseSctpSocket();
 454     return false;
 455   }
 456
 457   // Note: conversion from int to uint16_t happens on assignment.
 458   sockaddr_conn remote_sconn = GetSctpSockAddr(remote_port_);
 459   int connect_result = usrsctp_connect(
 460       sock_, reinterpret_cast<sockaddr *>(&remote_sconn), sizeof(remote_sconn));
 461   if (connect_result < 0 && errno != SCTP_EINPROGRESS) {
 462     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed usrsctp_connect. got errno="
 463                         << errno << ", but wanted " << SCTP_EINPROGRESS;
 464     CloseSctpSocket();
 465     return false;
 466   }
 467   return true;
 468 }
 469
 470 void SctpDataMediaChannel::Disconnect() {
 471   // TODO(ldixon): Consider calling |usrsctp_shutdown(sock_, ...)| to do a
 472   // shutdown handshake and remove the association.
 473   CloseSctpSocket();
 474 }
 475
 476 bool SctpDataMediaChannel::SetSend(bool send) {
 477   if (!sending_ && send) {
 478     return Connect();
 479   }
 480   if (sending_ && !send) {
 481     Disconnect();
 482   }
 483   return true;
 484 }
 485
 486 bool SctpDataMediaChannel::SetReceive(bool receive) {
 487   receiving_ = receive;
 488   return true;
 489 }
 490
 491 bool SctpDataMediaChannel::AddSendStream(const StreamParams& stream) {
 492   return AddStream(stream);
 493 }
 494
 495 bool SctpDataMediaChannel::RemoveSendStream(uint32 ssrc) {
 496   return ResetStream(ssrc);
 497 }
 498
 499 bool SctpDataMediaChannel::AddRecvStream(const StreamParams& stream) {
 500   // SCTP DataChannels are always bi-directional and calling AddSendStream will
 501   // enable both sending and receiving on the stream. So AddRecvStream is a
 502   // no-op.
 503   return true;
 504 }
 505
 506 bool SctpDataMediaChannel::RemoveRecvStream(uint32 ssrc) {
 507   // SCTP DataChannels are always bi-directional and calling RemoveSendStream
 508   // will disable both sending and receiving on the stream. So RemoveRecvStream
 509   // is a no-op.
 510   return true;
 511 }
 512
 513 bool SctpDataMediaChannel::SendData(
 514     const SendDataParams& params,
 515     const talk_base::Buffer& payload,
 516     SendDataResult* result) {
 517   if (result) {
 518     // Preset |result| to assume an error.  If SendData succeeds, we'll
 519     // overwrite |*result| once more at the end.
 520     *result = SDR_ERROR;
 521   }
 522
 523   if (!sending_) {
 524     LOG(LS_WARNING) << debug_name_ << "->SendData(...): "
 525                     << "Not sending packet with ssrc=" << params.ssrc
 526                     << " len=" << payload.length() << " before SetSend(true).";
 527     return false;
 528   }
 529
 530   if (params.type != cricket::DMT_CONTROL &&
 531       open_streams_.find(params.ssrc) == open_streams_.end()) {
 532     LOG(LS_WARNING) << debug_name_ << "->SendData(...): "
 533                     << "Not sending data because ssrc is unknown: "
 534                     << params.ssrc;
 535     return false;
 536   }
 537
 538   //
 539   // Send data using SCTP.
 540   ssize_t send_res = 0;  // result from usrsctp_sendv.
 541   struct sctp_sendv_spa spa = {0};
 542   spa.sendv_flags |= SCTP_SEND_SNDINFO_VALID;
 543   spa.sendv_sndinfo.snd_sid = params.ssrc;
 544   spa.sendv_sndinfo.snd_ppid = talk_base::HostToNetwork32(
 545       GetPpid(params.type));
 546
 547   // Ordered implies reliable.
 548   if (!params.ordered) {
 549     spa.sendv_sndinfo.snd_flags |= SCTP_UNORDERED;
 550     if (params.max_rtx_count >= 0 || params.max_rtx_ms == 0) {
 551       spa.sendv_flags |= SCTP_SEND_PRINFO_VALID;
 552       spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_RTX;
 553       spa.sendv_prinfo.pr_value = params.max_rtx_count;
 554     } else {
 555       spa.sendv_flags |= SCTP_SEND_PRINFO_VALID;
 556       spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_TTL;
 557       spa.sendv_prinfo.pr_value = params.max_rtx_ms;
 558     }
 559   }
 560
 561   // We don't fragment.
 562   send_res = usrsctp_sendv(sock_, payload.data(),
 563                            static_cast<size_t>(payload.length()),
 564                            NULL, 0, &spa,
 565                            talk_base::checked_cast<socklen_t>(sizeof(spa)),
 566                            SCTP_SENDV_SPA, 0);
 567   if (send_res < 0) {
 568     if (errno == EWOULDBLOCK) {
 569       *result = SDR_BLOCK;
 570       LOG(LS_INFO) << debug_name_ << "->SendData(...): EWOULDBLOCK returned";
 571     } else {
 572       LOG_ERRNO(LS_ERROR) << "ERROR:" << debug_name_
 573                           << "->SendData(...): "
 574                           << " usrsctp_sendv: ";
 575     }
 576     return false;
 577   }
 578   if (result) {
 579     // Only way out now is success.
 580     *result = SDR_SUCCESS;
 581   }
 582   return true;
 583 }
 584
 585 // Called by network interface when a packet has been received.
 586 void SctpDataMediaChannel::OnPacketReceived(
 587     talk_base::Buffer* packet, const talk_base::PacketTime& packet_time) {
 588   LOG(LS_VERBOSE) << debug_name_ << "->OnPacketReceived(...): " << " length="
 589                   << packet->length() << ", sending: " << sending_;
 590   // Only give receiving packets to usrsctp after if connected. This enables two
 591   // peers to each make a connect call, but for them not to receive an INIT
 592   // packet before they have called connect; least the last receiver of the INIT
 593   // packet will have called connect, and a connection will be established.
 594   if (sending_) {
 595     // Pass received packet to SCTP stack. Once processed by usrsctp, the data
 596     // will be will be given to the global OnSctpInboundData, and then,
 597     // marshalled by a Post and handled with OnMessage.
 598     usrsctp_conninput(this, packet->data(), packet->length(), 0);
 599   } else {
 600     // TODO(ldixon): Consider caching the packet for very slightly better
 601     // reliability.
 602   }
 603 }
 604
 605 void SctpDataMediaChannel::OnInboundPacketFromSctpToChannel(
 606     SctpInboundPacket* packet) {
 607   LOG(LS_VERBOSE) << debug_name_ << "->OnInboundPacketFromSctpToChannel(...): "
 608                   << "Received SCTP data:"
 609                   << " ssrc=" << packet->params.ssrc
 610                   << " notification: " << (packet->flags & MSG_NOTIFICATION)
 611                   << " length=" << packet->buffer.length();
 612   // Sending a packet with data == NULL (no data) is SCTPs "close the
 613   // connection" message. This sets sock_ = NULL;
 614   if (!packet->buffer.length() || !packet->buffer.data()) {
 615     LOG(LS_INFO) << debug_name_ << "->OnInboundPacketFromSctpToChannel(...): "
 616                                    "No data, closing.";
 617     return;
 618   }
 619   if (packet->flags & MSG_NOTIFICATION) {
 620     OnNotificationFromSctp(&packet->buffer);
 621   } else {
 622     OnDataFromSctpToChannel(packet->params, &packet->buffer);
 623   }
 624 }
 625
 626 void SctpDataMediaChannel::OnDataFromSctpToChannel(
 627     const ReceiveDataParams& params, talk_base::Buffer* buffer) {
 628   if (receiving_) {
 629     LOG(LS_VERBOSE) << debug_name_ << "->OnDataFromSctpToChannel(...): "
 630                     << "Posting with length: " << buffer->length()
 631                     << " on stream " << params.ssrc;
 632     // Reports all received messages to upper layers, no matter whether the sid
 633     // is known.
 634     SignalDataReceived(params, buffer->data(), buffer->length());
 635   } else {
 636     LOG(LS_WARNING) << debug_name_ << "->OnDataFromSctpToChannel(...): "
 637                     << "Not receiving packet with sid=" << params.ssrc
 638                     << " len=" <<  buffer->length()
 639                     << " before SetReceive(true).";
 640   }
 641 }
 642
 643 bool SctpDataMediaChannel::AddStream(const StreamParams& stream) {
 644   if (!stream.has_ssrcs()) {
 645     return false;
 646   }
 647
 648   const uint32 ssrc = stream.first_ssrc();
 649   if (open_streams_.find(ssrc) != open_streams_.end()) {
 650     LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
 651                     << "Not adding data stream '" << stream.id
 652                     << "' with ssrc=" << ssrc
 653                     << " because stream is already open.";
 654     return false;
 655   } else if (queued_reset_streams_.find(ssrc) != queued_reset_streams_.end()
 656              || sent_reset_streams_.find(ssrc) != sent_reset_streams_.end()) {
 657     LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
 658                     << "Not adding data stream '" << stream.id
 659                     << "' with ssrc=" << ssrc
 660                     << " because stream is still closing.";
 661     return false;
 662   }
 663
 664   open_streams_.insert(ssrc);
 665   return true;
 666 }
 667
 668 bool SctpDataMediaChannel::ResetStream(uint32 ssrc) {
 669   // We typically get this called twice for the same stream, once each for
 670   // Send and Recv.
 671   StreamSet::iterator found = open_streams_.find(ssrc);
 672
 673   if (found == open_streams_.end()) {
 674     LOG(LS_VERBOSE) << debug_name_ << "->ResetStream(" << ssrc << "): "
 675                     << "stream not found.";
 676     return false;
 677   } else {
 678     LOG(LS_VERBOSE) << debug_name_ << "->ResetStream(" << ssrc << "): "
 679                     << "Removing and queuing RE-CONFIG chunk.";
 680     open_streams_.erase(found);
 681   }
 682
 683   // SCTP won't let you have more than one stream reset pending at a time, but
 684   // you can close multiple streams in a single reset.  So, we keep an internal
 685   // queue of streams-to-reset, and send them as one reset message in
 686   // SendQueuedStreamResets().
 687   queued_reset_streams_.insert(ssrc);
 688
 689   // Signal our stream-reset logic that it should try to send now, if it can.
 690   SendQueuedStreamResets();
 691
 692   // The stream will actually get removed when we get the acknowledgment.
 693   return true;
 694 }
 695
 696 void SctpDataMediaChannel::OnNotificationFromSctp(talk_base::Buffer* buffer) {
 697   const sctp_notification& notification =
 698       reinterpret_cast<const sctp_notification&>(*buffer->data());
 699   ASSERT(notification.sn_header.sn_length == buffer->length());
 700
 701   // TODO(ldixon): handle notifications appropriately.
 702   switch (notification.sn_header.sn_type) {
 703     case SCTP_ASSOC_CHANGE:
 704       LOG(LS_VERBOSE) << "SCTP_ASSOC_CHANGE";
 705       OnNotificationAssocChange(notification.sn_assoc_change);
 706       break;
 707     case SCTP_REMOTE_ERROR:
 708       LOG(LS_INFO) << "SCTP_REMOTE_ERROR";
 709       break;
 710     case SCTP_SHUTDOWN_EVENT:
 711       LOG(LS_INFO) << "SCTP_SHUTDOWN_EVENT";
 712       break;
 713     case SCTP_ADAPTATION_INDICATION:
 714       LOG(LS_INFO) << "SCTP_ADAPTATION_INDICATION";
 715       break;
 716     case SCTP_PARTIAL_DELIVERY_EVENT:
 717       LOG(LS_INFO) << "SCTP_PARTIAL_DELIVERY_EVENT";
 718       break;
 719     case SCTP_AUTHENTICATION_EVENT:
 720       LOG(LS_INFO) << "SCTP_AUTHENTICATION_EVENT";
 721       break;
 722     case SCTP_SENDER_DRY_EVENT:
 723       LOG(LS_VERBOSE) << "SCTP_SENDER_DRY_EVENT";
 724       SignalReadyToSend(true);
 725       break;
 726     // TODO(ldixon): Unblock after congestion.
 727     case SCTP_NOTIFICATIONS_STOPPED_EVENT:
 728       LOG(LS_INFO) << "SCTP_NOTIFICATIONS_STOPPED_EVENT";
 729       break;
 730     case SCTP_SEND_FAILED_EVENT:
 731       LOG(LS_INFO) << "SCTP_SEND_FAILED_EVENT";
 732       break;
 733     case SCTP_STREAM_RESET_EVENT:
 734       OnStreamResetEvent(&notification.sn_strreset_event);
 735       break;
 736     case SCTP_ASSOC_RESET_EVENT:
 737       LOG(LS_INFO) << "SCTP_ASSOC_RESET_EVENT";
 738       break;
 739     case SCTP_STREAM_CHANGE_EVENT:
 740       LOG(LS_INFO) << "SCTP_STREAM_CHANGE_EVENT";
 741       // An acknowledgment we get after our stream resets have gone through,
 742       // if they've failed.  We log the message, but don't react -- we don't
 743       // keep around the last-transmitted set of SSIDs we wanted to close for
 744       // error recovery.  It doesn't seem likely to occur, and if so, likely
 745       // harmless within the lifetime of a single SCTP association.
 746       break;
 747     default:
 748       LOG(LS_WARNING) << "Unknown SCTP event: "
 749                       << notification.sn_header.sn_type;
 750       break;
 751   }
 752 }
 753
 754 void SctpDataMediaChannel::OnNotificationAssocChange(
 755     const sctp_assoc_change& change) {
 756   switch (change.sac_state) {
 757     case SCTP_COMM_UP:
 758       LOG(LS_VERBOSE) << "Association change SCTP_COMM_UP";
 759       break;
 760     case SCTP_COMM_LOST:
 761       LOG(LS_INFO) << "Association change SCTP_COMM_LOST";
 762       break;
 763     case SCTP_RESTART:
 764       LOG(LS_INFO) << "Association change SCTP_RESTART";
 765       break;
 766     case SCTP_SHUTDOWN_COMP:
 767       LOG(LS_INFO) << "Association change SCTP_SHUTDOWN_COMP";
 768       break;
 769     case SCTP_CANT_STR_ASSOC:
 770       LOG(LS_INFO) << "Association change SCTP_CANT_STR_ASSOC";
 771       break;
 772     default:
 773       LOG(LS_INFO) << "Association change UNKNOWN";
 774       break;
 775   }
 776 }
 777
 778 void SctpDataMediaChannel::OnStreamResetEvent(
 779     const struct sctp_stream_reset_event* evt) {
 780   // A stream reset always involves two RE-CONFIG chunks for us -- we always
 781   // simultaneously reset a sid's sequence number in both directions.  The
 782   // requesting side transmits a RE-CONFIG chunk and waits for the peer to send
 783   // one back.  Both sides get this SCTP_STREAM_RESET_EVENT when they receive
 784   // RE-CONFIGs.
 785   const int num_ssrcs = (evt->strreset_length - sizeof(*evt)) /
 786       sizeof(evt->strreset_stream_list[0]);
 787   LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
 788                   << "): Flags = 0x"
 789                   << std::hex << evt->strreset_flags << " ("
 790                   << ListFlags(evt->strreset_flags) << ")";
 791   LOG(LS_VERBOSE) << "Assoc = " << evt->strreset_assoc_id << ", Streams = ["
 792                   << ListArray(evt->strreset_stream_list, num_ssrcs)
 793                   << "], Open: ["
 794                   << ListStreams(open_streams_) << "], Q'd: ["
 795                   << ListStreams(queued_reset_streams_) << "], Sent: ["
 796                   << ListStreams(sent_reset_streams_) << "]";
 797   bool local_stream_reset_acknowledged = false;
 798
 799   // If both sides try to reset some streams at the same time (even if they're
 800   // disjoint sets), we can get reset failures.
 801   if (evt->strreset_flags & SCTP_STREAM_RESET_FAILED) {
 802     // OK, just try again.  The stream IDs sent over when the RESET_FAILED flag
 803     // is set seem to be garbage values.  Ignore them.
 804     queued_reset_streams_.insert(
 805         sent_reset_streams_.begin(),
 806         sent_reset_streams_.end());
 807     sent_reset_streams_.clear();
 808     local_stream_reset_acknowledged = true;
 809
 810   } else if (evt->strreset_flags & SCTP_STREAM_RESET_INCOMING_SSN) {
 811     // Each side gets an event for each direction of a stream.  That is,
 812     // closing sid k will make each side receive INCOMING and OUTGOING reset
 813     // events for k.  As per RFC6525, Section 5, paragraph 2, each side will
 814     // get an INCOMING event first.
 815     for (int i = 0; i < num_ssrcs; i++) {
 816       const int stream_id = evt->strreset_stream_list[i];
 817
 818       // See if this stream ID was closed by our peer or ourselves.
 819       StreamSet::iterator it = sent_reset_streams_.find(stream_id);
 820
 821       // The reset was requested locally.
 822       if (it != sent_reset_streams_.end()) {
 823         LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
 824                         << "): local sid " << stream_id << " acknowledged.";
 825         local_stream_reset_acknowledged = true;
 826         sent_reset_streams_.erase(it);
 827
 828       } else if ((it = open_streams_.find(stream_id))
 829                  != open_streams_.end()) {
 830         // The peer requested the reset.
 831         LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
 832                         << "): closing sid " << stream_id;
 833         open_streams_.erase(it);
 834         SignalStreamClosed(stream_id);
 835
 836       } else if ((it = queued_reset_streams_.find(stream_id))
 837                  != queued_reset_streams_.end()) {
 838         // The peer requested the reset, but there was a local reset
 839         // queued.
 840         LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
 841                         << "): double-sided close for sid " << stream_id;
 842         // Both sides want the stream closed, and the peer got to send the
 843         // RE-CONFIG first.  Treat it like the local Remove(Send|Recv)Stream
 844         // finished quickly.
 845         queued_reset_streams_.erase(it);
 846
 847       } else {
 848         // This stream is unknown.  Sometimes this can be from an
 849         // RESET_FAILED-related retransmit.
 850         LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
 851                         << "): Unknown sid " << stream_id;
 852       }
 853     }
 854   }
 855
 856   if (local_stream_reset_acknowledged) {
 857     // This message acknowledges the last stream-reset request we sent out
 858     // (only one can be outstanding at a time).  Send out the next one.
 859     SendQueuedStreamResets();
 860   }
 861 }
 862
 863 // Puts the specified |param| from the codec identified by |id| into |dest|
 864 // and returns true.  Or returns false if it wasn't there, leaving |dest|
 865 // untouched.
 866 static bool GetCodecIntParameter(const std::vector<DataCodec>& codecs,
 867                                  int id, const std::string& name,
 868                                  const std::string& param, int* dest) {
 869   std::string value;
 870   Codec match_pattern;
 871   match_pattern.id = id;
 872   match_pattern.name = name;
 873   for (size_t i = 0; i < codecs.size(); ++i) {
 874     if (codecs[i].Matches(match_pattern)) {
 875       if (codecs[i].GetParam(param, &value)) {
 876         *dest = talk_base::FromString<int>(value);
 877         return true;
 878       }
 879     }
 880   }
 881   return false;
 882 }
 883
 884 bool SctpDataMediaChannel::SetSendCodecs(const std::vector<DataCodec>& codecs) {
 885   return GetCodecIntParameter(
 886       codecs, kGoogleSctpDataCodecId, kGoogleSctpDataCodecName, kCodecParamPort,
 887       &remote_port_);
 888 }
 889
 890 bool SctpDataMediaChannel::SetRecvCodecs(const std::vector<DataCodec>& codecs) {
 891   return GetCodecIntParameter(
 892       codecs, kGoogleSctpDataCodecId, kGoogleSctpDataCodecName, kCodecParamPort,
 893       &local_port_);
 894 }
 895
 896 void SctpDataMediaChannel::OnPacketFromSctpToNetwork(
 897     talk_base::Buffer* buffer) {
 898   if (buffer->length() > kSctpMtu) {
 899     LOG(LS_ERROR) << debug_name_ << "->OnPacketFromSctpToNetwork(...): "
 900                   << "SCTP seems to have made a packet that is bigger "
 901                      "than its official MTU.";
 902   }
 903   MediaChannel::SendPacket(buffer);
 904 }
 905
 906 bool SctpDataMediaChannel::SendQueuedStreamResets() {
 907   if (!sent_reset_streams_.empty() || queued_reset_streams_.empty())
 908     return true;
 909
 910   LOG(LS_VERBOSE) << "SendQueuedStreamResets[" << debug_name_ << "]: Sending ["
 911                   << ListStreams(queued_reset_streams_) << "], Open: ["
 912                   << ListStreams(open_streams_) << "], Sent: ["
 913                   << ListStreams(sent_reset_streams_) << "]";
 914
 915   const size_t num_streams = queued_reset_streams_.size();
 916   const size_t num_bytes = sizeof(struct sctp_reset_streams)
 917     + (num_streams * sizeof(uint16));
 918
 919   std::vector<uint8> reset_stream_buf(num_bytes, 0);
 920   struct sctp_reset_streams* resetp = reinterpret_cast<sctp_reset_streams*>(
 921       &reset_stream_buf[0]);
 922   resetp->srs_assoc_id = SCTP_ALL_ASSOC;
 923   resetp->srs_flags = SCTP_STREAM_RESET_INCOMING | SCTP_STREAM_RESET_OUTGOING;
 924   resetp->srs_number_streams = talk_base::checked_cast<uint16_t>(num_streams);
 925   int result_idx = 0;
 926   for (StreamSet::iterator it = queued_reset_streams_.begin();
 927        it != queued_reset_streams_.end(); ++it) {
 928     resetp->srs_stream_list[result_idx++] = *it;
 929   }
 930
 931   int ret = usrsctp_setsockopt(
 932       sock_, IPPROTO_SCTP, SCTP_RESET_STREAMS, resetp,
 933       talk_base::checked_cast<socklen_t>(reset_stream_buf.size()));
 934   if (ret < 0) {
 935     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to send a stream reset for "
 936                         << num_streams << " streams";
 937     return false;
 938   }
 939
 940   // sent_reset_streams_ is empty, and all the queued_reset_streams_ go into
 941   // it now.
 942   queued_reset_streams_.swap(sent_reset_streams_);
 943   return true;
 944 }
 945
 946 void SctpDataMediaChannel::OnMessage(talk_base::Message* msg) {
 947   switch (msg->message_id) {
 948     case MSG_SCTPINBOUNDPACKET: {
 949       talk_base::scoped_ptr<InboundPacketMessage> pdata(
 950           static_cast<InboundPacketMessage*>(msg->pdata));
 951       OnInboundPacketFromSctpToChannel(pdata->data().get());
 952       break;
 953     }
 954     case MSG_SCTPOUTBOUNDPACKET: {
 955       talk_base::scoped_ptr<OutboundPacketMessage> pdata(
 956           static_cast<OutboundPacketMessage*>(msg->pdata));
 957       OnPacketFromSctpToNetwork(pdata->data().get());
 958       break;
 959     }
 960   }
 961 }
 962 }  // namespace cricket