(try 2) Added send-thresholding and fixed text packet dumping

talk/media/sctp/sctpdataengine.cc

 /*
  * libjingle
  * Copyright 2012 Google Inc. and Robin Seggelmann
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are met:
  *
  *  1. Redistributions of source code must retain the above copyright notice,
  *     this list of conditions and the following disclaimer.
  *  2. Redistributions in binary form must reproduce the above copyright notice,
@@ skipping 91 matching lines @@
 }  // namespace
 
 namespace cricket {
 typedef rtc::ScopedMessageData<SctpInboundPacket> InboundPacketMessage;
 typedef rtc::ScopedMessageData<rtc::Buffer> OutboundPacketMessage;
 
 // The biggest SCTP packet.  Starting from a 'safe' wire MTU value of 1280,
 // take off 80 bytes for DTLS/TURN/TCP/IP overhead.
 static const size_t kSctpMtu = 1200;
 
+// The size of the SCTP association send buffer.  256kB, the usrsctp default.
+static const int kSendBufferSize = 262144;
 enum {
   MSG_SCTPINBOUNDPACKET = 1,   // MessageData is SctpInboundPacket
   MSG_SCTPOUTBOUNDPACKET = 2,  // MessageData is rtc:Buffer
 };
 
 struct SctpInboundPacket {
   rtc::Buffer buffer;
   ReceiveDataParams params;
   // The |flags| parameter is used by SCTP to distinguish notification packets
   // from other types of packets.
@@ skipping 48 matching lines @@
     case SctpDataMediaChannel::PPID_NONE:
       *dest = cricket::DMT_NONE;
       return true;
 
     default:
       return false;
   }
 }
 
 // Log the packet in text2pcap format, if log level is at LS_VERBOSE.
-static void VerboseLogPacket(void *addr, size_t length, int direction) {
+static void VerboseLogPacket(void *data, size_t length, int direction) {
   if (LOG_CHECK_LEVEL(LS_VERBOSE) && length > 0) {
     char *dump_buf;
     if ((dump_buf = usrsctp_dumppacket(
-             addr, length, direction)) != NULL) {
+             data, length, direction)) != NULL) {
       LOG(LS_VERBOSE) << dump_buf;
       usrsctp_freedumpbuffer(dump_buf);
     }
   }
 }
 
 // This is the callback usrsctp uses when there's data to send on the network
 // that has been wrapped appropriatly for the SCTP protocol.
 static int OnSctpOutboundPacket(void* addr, void* data, size_t length,
                                 uint8_t tos, uint8_t set_df) {
@@ skipping 56 matching lines @@
     // AF_CONN use of sctp.
     usrsctp_init(0, cricket::OnSctpOutboundPacket, debug_sctp_printf);
 
     // To turn on/off detailed SCTP debugging. You will also need to have the
     // SCTP_DEBUG cpp defines flag.
     // usrsctp_sysctl_set_sctp_debug_on(SCTP_DEBUG_ALL);
 
     // TODO(ldixon): Consider turning this on/off.
     usrsctp_sysctl_set_sctp_ecn_enable(0);
 
+    // This is harmless, but we should find out when the library default
+    // changes.
+    int send_size = usrsctp_sysctl_get_sctp_sendspace();
+    if (send_size != kSendBufferSize) {
+      LOG(LS_ERROR) << "Got different send size than expected: " << send_size;
+    }
+
     // TODO(ldixon): Consider turning this on/off.
     // This is not needed right now (we don't do dynamic address changes):
     // If SCTP Auto-ASCONF is enabled, the peer is informed automatically
     // when a new address is added or removed. This feature is enabled by
     // default.
     // usrsctp_sysctl_set_sctp_auto_asconf(0);
 
     // TODO(ldixon): Consider turning this on/off.
     // Add a blackhole sysctl. Setting it to 1 results in no ABORTs
     // being sent in response to INITs, setting it to 2 results
@@ skipping 37 matching lines @@
 }
 
 DataMediaChannel* SctpDataEngine::CreateChannel(
     DataChannelType data_channel_type) {
   if (data_channel_type != DCT_SCTP) {
     return NULL;
   }
   return new SctpDataMediaChannel(rtc::Thread::Current());
 }
 
+// static
+SctpDataMediaChannel* SctpDataEngine::GetChannelFromSocket(
+    struct socket* sock) {
+  struct sockaddr* addrs = nullptr;
+  int naddrs = usrsctp_getladdrs(sock, 0, &addrs);
+  if (naddrs <= 0 || addrs[0].sa_family != AF_CONN) {
+    return nullptr;
+  }
+  // usrsctp_getladdrs() returns the addresses bound to this socket, which
+  // contains the SctpDataMediaChannel* as sconn_addr.  Read the pointer,
+  // then free the list of addresses once we have the pointer.  We only open
+  // AF_CONN sockets, and they should all have the sconn_addr set to the
+  // pointer that created them, so [0] is as good as any other.
+  struct sockaddr_conn* sconn =
+      reinterpret_cast<struct sockaddr_conn*>(&addrs[0]);
+  SctpDataMediaChannel* channel =
+      reinterpret_cast<SctpDataMediaChannel*>(sconn->sconn_addr);
+  usrsctp_freeladdrs(addrs);
+
+  return channel;
+}
+
+// static
+int SctpDataEngine::SendThresholdCallback(struct socket* sock,
+                                          uint32_t sb_free) {
+  // Fired on our I/O thread.  SctpDataMediaChannel::OnPacketReceived() gets
+  // a packet containing acknowledgments, which goes into usrsctp_conninput,
+  // and then back here.
+  SctpDataMediaChannel* channel = GetChannelFromSocket(sock);
+  if (!channel) {
+    LOG(LS_ERROR) << "SendThresholdCallback: Failed to get channel for socket "
+                  << sock;
+    return 0;
+  }
+  channel->OnSendThresholdCallback();
+  return 0;
+}
+
 SctpDataMediaChannel::SctpDataMediaChannel(rtc::Thread* thread)
     : worker_thread_(thread),
       local_port_(kSctpDefaultPort),
       remote_port_(kSctpDefaultPort),
       sock_(NULL),
       sending_(false),
       receiving_(false),
       debug_name_("SctpDataMediaChannel") {
 }
 
 SctpDataMediaChannel::~SctpDataMediaChannel() {
   CloseSctpSocket();
 }
 
+void SctpDataMediaChannel::OnSendThresholdCallback() {
+  DCHECK(rtc::Thread::Current() == worker_thread_);
+  SignalReadyToSend(true);
+}
+
 sockaddr_conn SctpDataMediaChannel::GetSctpSockAddr(int port) {
   sockaddr_conn sconn = {0};
   sconn.sconn_family = AF_CONN;
 #ifdef HAVE_SCONN_LEN
   sconn.sconn_len = sizeof(sockaddr_conn);
 #endif
   // Note: conversion from int to uint16_t happens here.
   sconn.sconn_port = rtc::HostToNetwork16(port);
   sconn.sconn_addr = this;
   return sconn;
 }
 
 bool SctpDataMediaChannel::OpenSctpSocket() {
   if (sock_) {
     LOG(LS_VERBOSE) << debug_name_
                     << "->Ignoring attempt to re-create existing socket.";
     return false;
   }
+
+  // If kSendBufferSize isn't reflective of reality, we log an error, but we
+  // still have to do something reasonable here.  Look up what the buffer's
+  // real size is and set our threshold to something reasonable.
+  const static int kSendThreshold = usrsctp_sysctl_get_sctp_sendspace() / 2;
+
   sock_ = usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP,
-                         cricket::OnSctpInboundPacket, NULL, 0, this);
+                         cricket::OnSctpInboundPacket,
+                         &SctpDataEngine::SendThresholdCallback,
+                         kSendThreshold, this);
   if (!sock_) {
     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to create SCTP socket.";
     return false;
   }
 
   // Make the socket non-blocking. Connect, close, shutdown etc will not block
   // the thread waiting for the socket operation to complete.
   if (usrsctp_set_non_blocking(sock_, 1) < 0) {
     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP to non blocking.";
     return false;
@@ skipping 24 matching lines @@
 
   // Nagle.
   uint32_t nodelay = 1;
   if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_NODELAY, &nodelay,
                          sizeof(nodelay))) {
     LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP_NODELAY.";
     return false;
   }
 
   // Disable MTU discovery
-  struct sctp_paddrparams params = {{0}};
+  sctp_paddrparams params = {{0}};
   params.spp_assoc_id = 0;
   params.spp_flags = SPP_PMTUD_DISABLE;
   params.spp_pathmtu = kSctpMtu;
   if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_PEER_ADDR_PARAMS, &params,
       sizeof(params))) {
     LOG_ERRNO(LS_ERROR) << debug_name_
                         << "Failed to set SCTP_PEER_ADDR_PARAMS.";
     return false;
   }
 
@@ skipping 184 matching lines @@
   if (result) {
     // Only way out now is success.
     *result = SDR_SUCCESS;
   }
   return true;
 }
 
 // Called by network interface when a packet has been received.
 void SctpDataMediaChannel::OnPacketReceived(
     rtc::Buffer* packet, const rtc::PacketTime& packet_time) {
+  DCHECK(rtc::Thread::Current() == worker_thread_);
   LOG(LS_VERBOSE) << debug_name_ << "->OnPacketReceived(...): "
                   << " length=" << packet->size() << ", sending: " << sending_;
   // Only give receiving packets to usrsctp after if connected. This enables two
   // peers to each make a connect call, but for them not to receive an INIT
   // packet before they have called connect; least the last receiver of the INIT
   // packet will have called connect, and a connection will be established.
   if (sending_) {
     // Pass received packet to SCTP stack. Once processed by usrsctp, the data
     // will be will be given to the global OnSctpInboundData, and then,
     // marshalled by a Post and handled with OnMessage.
-
     VerboseLogPacket(packet->data(), packet->size(), SCTP_DUMP_INBOUND);
     usrsctp_conninput(this, packet->data(), packet->size(), 0);
   } else {
     // TODO(ldixon): Consider caching the packet for very slightly better
     // reliability.
   }
 }
 
 void SctpDataMediaChannel::OnInboundPacketFromSctpToChannel(
     SctpInboundPacket* packet) {
@@ skipping 275 matching lines @@
 }
 
 bool SctpDataMediaChannel::SetRecvCodecs(const std::vector<DataCodec>& codecs) {
   return GetCodecIntParameter(
       codecs, kGoogleSctpDataCodecId, kGoogleSctpDataCodecName, kCodecParamPort,
       &local_port_);
 }
 
 void SctpDataMediaChannel::OnPacketFromSctpToNetwork(
     rtc::Buffer* buffer) {
-  if (buffer->size() > kSctpMtu) {
+  // usrsctp seems to interpret the MTU we give it strangely -- it seems to
+  // give us back packets bigger than that MTU, if only by a fixed amount.
+  // This is that amount that we've observed.
+  const int kSctpOverhead = 76;
+  if (buffer->size() > (kSctpOverhead + kSctpMtu)) {
     LOG(LS_ERROR) << debug_name_ << "->OnPacketFromSctpToNetwork(...): "
                   << "SCTP seems to have made a packet that is bigger "
-                     "than its official MTU.";
+                  << "than its official MTU: " << buffer->size()
+                  << " vs max of " << kSctpMtu
+                  << " even after adding " << kSctpOverhead
+                  << " extra SCTP overhead";
   }
   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: ["
@@ skipping 41 matching lines @@
     }
     case MSG_SCTPOUTBOUNDPACKET: {
       rtc::scoped_ptr<OutboundPacketMessage> pdata(
           static_cast<OutboundPacketMessage*>(msg->pdata));
       OnPacketFromSctpToNetwork(pdata->data().get());
       break;
     }
   }
 }
 }  // namespace cricket