Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(3)

Side by Side Diff: webrtc/media/sctp/sctptransport.cc

Issue 2564333002: Reland of: Separating SCTP code from BaseChannel/MediaChannel. (Closed)
Patch Set: Various cleanup. Created 3 years, 12 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
OLDNEW
1 /* 1 /*
2 * Copyright (c) 2012 The WebRTC project authors. All Rights Reserved. 2 * Copyright (c) 2012 The WebRTC project authors. All Rights Reserved.
3 * 3 *
4 * Use of this source code is governed by a BSD-style license 4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source 5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found 6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may 7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree. 8 * be found in the AUTHORS file in the root of the source tree.
9 */ 9 */
10 10
11 #include "webrtc/media/sctp/sctpdataengine.h" 11 #include "webrtc/media/sctp/sctptransport.h"
12 12
13 #include <stdarg.h> 13 #include <stdarg.h>
14 #include <stdio.h> 14 #include <stdio.h>
15 15
16 #include <memory> 16 #include <memory>
17 #include <sstream> 17 #include <sstream>
18 #include <vector>
19 18
20 #include "usrsctplib/usrsctp.h" 19 #include "usrsctplib/usrsctp.h"
21 #include "webrtc/base/arraysize.h" 20 #include "webrtc/base/arraysize.h"
22 #include "webrtc/base/copyonwritebuffer.h" 21 #include "webrtc/base/copyonwritebuffer.h"
23 #include "webrtc/base/criticalsection.h" 22 #include "webrtc/base/criticalsection.h"
24 #include "webrtc/base/helpers.h" 23 #include "webrtc/base/helpers.h"
25 #include "webrtc/base/logging.h" 24 #include "webrtc/base/logging.h"
26 #include "webrtc/base/safe_conversions.h" 25 #include "webrtc/base/safe_conversions.h"
26 #include "webrtc/base/trace_event.h"
27 #include "webrtc/media/base/codec.h" 27 #include "webrtc/media/base/codec.h"
28 #include "webrtc/media/base/mediaconstants.h" 28 #include "webrtc/media/base/mediaconstants.h"
29 #include "webrtc/media/base/rtputils.h" // For IsRtpPacket
29 #include "webrtc/media/base/streamparams.h" 30 #include "webrtc/media/base/streamparams.h"
30 31
31 namespace cricket { 32 namespace {
33
32 // The biggest SCTP packet. Starting from a 'safe' wire MTU value of 1280, 34 // The biggest SCTP packet. Starting from a 'safe' wire MTU value of 1280,
33 // take off 80 bytes for DTLS/TURN/TCP/IP overhead. 35 // take off 80 bytes for DTLS/TURN/TCP/IP overhead.
34 static constexpr size_t kSctpMtu = 1200; 36 static constexpr size_t kSctpMtu = 1200;
35 37
36 // The size of the SCTP association send buffer. 256kB, the usrsctp default. 38 // The size of the SCTP association send buffer. 256kB, the usrsctp default.
37 static constexpr int kSendBufferSize = 262144; 39 static constexpr int kSendBufferSize = 262144;
38 40
39 struct SctpInboundPacket {
40 rtc::CopyOnWriteBuffer buffer;
41 ReceiveDataParams params;
42 // The |flags| parameter is used by SCTP to distinguish notification packets
43 // from other types of packets.
44 int flags;
45 };
46
47 namespace {
48 // Set the initial value of the static SCTP Data Engines reference count. 41 // Set the initial value of the static SCTP Data Engines reference count.
49 int g_usrsctp_usage_count = 0; 42 int g_usrsctp_usage_count = 0;
50 rtc::GlobalLockPod g_usrsctp_lock_; 43 rtc::GlobalLockPod g_usrsctp_lock_;
51 44
52 typedef SctpDataMediaChannel::StreamSet StreamSet; 45 // DataMessageType is used for the SCTP "Payload Protocol Identifier", as
46 // defined in http://tools.ietf.org/html/rfc4960#section-14.4
47 //
48 // For the list of IANA approved values see:
49 // http://www.iana.org/assignments/sctp-parameters/sctp-parameters.xml
50 // The value is not used by SCTP itself. It indicates the protocol running
51 // on top of SCTP.
52 enum PayloadProtocolIdentifier {
53 PPID_NONE = 0, // No protocol is specified.
54 // Matches the PPIDs in mozilla source and
55 // https://datatracker.ietf.org/doc/draft-ietf-rtcweb-data-protocol Sec. 9
56 // They're not yet assigned by IANA.
57 PPID_CONTROL = 50,
58 PPID_BINARY_PARTIAL = 52,
59 PPID_BINARY_LAST = 53,
60 PPID_TEXT_PARTIAL = 54,
61 PPID_TEXT_LAST = 51
62 };
63
64 // Some ERRNO values get re-#defined to WSA* equivalents in some talk/
65 // headers. We save the original ones in an enum.
66 enum PreservedErrno {
67 SCTP_EINPROGRESS = EINPROGRESS,
68 SCTP_EWOULDBLOCK = EWOULDBLOCK
69 };
70
71 typedef std::set<uint32_t> StreamSet;
53 72
54 // Returns a comma-separated, human-readable list of the stream IDs in 's' 73 // Returns a comma-separated, human-readable list of the stream IDs in 's'
55 std::string ListStreams(const StreamSet& s) { 74 std::string ListStreams(const StreamSet& s) {
56 std::stringstream result; 75 std::stringstream result;
57 bool first = true; 76 bool first = true;
58 for (StreamSet::const_iterator it = s.begin(); it != s.end(); ++it) { 77 for (StreamSet::const_iterator it = s.begin(); it != s.end(); ++it) {
59 if (!first) { 78 if (!first) {
60 result << ", " << *it; 79 result << ", " << *it;
61 } else { 80 } else {
62 result << *it; 81 result << *it;
63 first = false; 82 first = false;
64 } 83 }
65 } 84 }
66 return result.str(); 85 return result.str();
67 } 86 }
68 87
69 // Returns a pipe-separated, human-readable list of the SCTP_STREAM_RESET 88 // Returns a pipe-separated, human-readable list of the SCTP_STREAM_RESET
70 // flags in 'flags' 89 // flags in 'flags'
71 std::string ListFlags(int flags) { 90 std::string ListFlags(int flags) {
72 std::stringstream result; 91 std::stringstream result;
73 bool first = true; 92 bool first = true;
74 // Skip past the first 12 chars (strlen("SCTP_STREAM_")) 93 // Skip past the first 12 chars (strlen("SCTP_STREAM_"))
75 #define MAKEFLAG(X) { X, #X + 12} 94 #define MAKEFLAG(X) \
95 { X, #X + 12 }
76 struct flaginfo_t { 96 struct flaginfo_t {
77 int value; 97 int value;
78 const char* name; 98 const char* name;
79 } flaginfo[] = { 99 } flaginfo[] = {MAKEFLAG(SCTP_STREAM_RESET_INCOMING_SSN),
80 MAKEFLAG(SCTP_STREAM_RESET_INCOMING_SSN), 100 MAKEFLAG(SCTP_STREAM_RESET_OUTGOING_SSN),
81 MAKEFLAG(SCTP_STREAM_RESET_OUTGOING_SSN), 101 MAKEFLAG(SCTP_STREAM_RESET_DENIED),
82 MAKEFLAG(SCTP_STREAM_RESET_DENIED), 102 MAKEFLAG(SCTP_STREAM_RESET_FAILED),
83 MAKEFLAG(SCTP_STREAM_RESET_FAILED), 103 MAKEFLAG(SCTP_STREAM_CHANGE_DENIED)};
84 MAKEFLAG(SCTP_STREAM_CHANGE_DENIED)
85 };
86 #undef MAKEFLAG 104 #undef MAKEFLAG
87 for (uint32_t i = 0; i < arraysize(flaginfo); ++i) { 105 for (uint32_t i = 0; i < arraysize(flaginfo); ++i) {
88 if (flags & flaginfo[i].value) { 106 if (flags & flaginfo[i].value) {
89 if (!first) result << " | "; 107 if (!first)
108 result << " | ";
90 result << flaginfo[i].name; 109 result << flaginfo[i].name;
91 first = false; 110 first = false;
92 } 111 }
93 } 112 }
94 return result.str(); 113 return result.str();
95 } 114 }
96 115
97 // Returns a comma-separated, human-readable list of the integers in 'array'. 116 // Returns a comma-separated, human-readable list of the integers in 'array'.
98 // All 'num_elems' of them. 117 // All 'num_elems' of them.
99 std::string ListArray(const uint16_t* array, int num_elems) { 118 std::string ListArray(const uint16_t* array, int num_elems) {
100 std::stringstream result; 119 std::stringstream result;
101 for (int i = 0; i < num_elems; ++i) { 120 for (int i = 0; i < num_elems; ++i) {
102 if (i) { 121 if (i) {
103 result << ", " << array[i]; 122 result << ", " << array[i];
104 } else { 123 } else {
105 result << array[i]; 124 result << array[i];
106 } 125 }
107 } 126 }
108 return result.str(); 127 return result.str();
109 } 128 }
110 129
111 typedef rtc::ScopedMessageData<SctpInboundPacket> InboundPacketMessage;
112 typedef rtc::ScopedMessageData<rtc::CopyOnWriteBuffer> OutboundPacketMessage;
113
114 enum {
115 MSG_SCTPINBOUNDPACKET = 1, // MessageData is SctpInboundPacket
116 MSG_SCTPOUTBOUNDPACKET = 2, // MessageData is rtc:Buffer
117 };
118
119 // Helper for logging SCTP messages. 130 // Helper for logging SCTP messages.
120 void DebugSctpPrintf(const char* format, ...) { 131 void DebugSctpPrintf(const char* format, ...) {
121 #if RTC_DCHECK_IS_ON 132 #if RTC_DCHECK_IS_ON
122 char s[255]; 133 char s[255];
123 va_list ap; 134 va_list ap;
124 va_start(ap, format); 135 va_start(ap, format);
125 vsnprintf(s, sizeof(s), format, ap); 136 vsnprintf(s, sizeof(s), format, ap);
126 LOG(LS_INFO) << "SCTP: " << s; 137 LOG(LS_INFO) << "SCTP: " << s;
127 va_end(ap); 138 va_end(ap);
128 #endif 139 #endif
129 } 140 }
130 141
131 // Get the PPID to use for the terminating fragment of this type. 142 // Get the PPID to use for the terminating fragment of this type.
132 SctpDataMediaChannel::PayloadProtocolIdentifier GetPpid(DataMessageType type) { 143 PayloadProtocolIdentifier GetPpid(cricket::DataMessageType type) {
133 switch (type) { 144 switch (type) {
134 default: 145 default:
135 case DMT_NONE: 146 case cricket::DMT_NONE:
136 return SctpDataMediaChannel::PPID_NONE; 147 return PPID_NONE;
137 case DMT_CONTROL: 148 case cricket::DMT_CONTROL:
138 return SctpDataMediaChannel::PPID_CONTROL; 149 return PPID_CONTROL;
139 case DMT_BINARY: 150 case cricket::DMT_BINARY:
140 return SctpDataMediaChannel::PPID_BINARY_LAST; 151 return PPID_BINARY_LAST;
141 case DMT_TEXT: 152 case cricket::DMT_TEXT:
142 return SctpDataMediaChannel::PPID_TEXT_LAST; 153 return PPID_TEXT_LAST;
143 }; 154 }
144 } 155 }
145 156
146 bool GetDataMediaType(SctpDataMediaChannel::PayloadProtocolIdentifier ppid, 157 bool GetDataMediaType(PayloadProtocolIdentifier ppid,
147 DataMessageType* dest) { 158 cricket::DataMessageType* dest) {
148 ASSERT(dest != NULL); 159 RTC_DCHECK(dest != NULL);
149 switch (ppid) { 160 switch (ppid) {
150 case SctpDataMediaChannel::PPID_BINARY_PARTIAL: 161 case PPID_BINARY_PARTIAL:
151 case SctpDataMediaChannel::PPID_BINARY_LAST: 162 case PPID_BINARY_LAST:
152 *dest = DMT_BINARY; 163 *dest = cricket::DMT_BINARY;
153 return true; 164 return true;
154 165
155 case SctpDataMediaChannel::PPID_TEXT_PARTIAL: 166 case PPID_TEXT_PARTIAL:
156 case SctpDataMediaChannel::PPID_TEXT_LAST: 167 case PPID_TEXT_LAST:
157 *dest = DMT_TEXT; 168 *dest = cricket::DMT_TEXT;
158 return true; 169 return true;
159 170
160 case SctpDataMediaChannel::PPID_CONTROL: 171 case PPID_CONTROL:
161 *dest = DMT_CONTROL; 172 *dest = cricket::DMT_CONTROL;
162 return true; 173 return true;
163 174
164 case SctpDataMediaChannel::PPID_NONE: 175 case PPID_NONE:
165 *dest = DMT_NONE; 176 *dest = cricket::DMT_NONE;
166 return true; 177 return true;
167 178
168 default: 179 default:
169 return false; 180 return false;
170 } 181 }
171 } 182 }
172 183
173 // Log the packet in text2pcap format, if log level is at LS_VERBOSE. 184 // Log the packet in text2pcap format, if log level is at LS_VERBOSE.
174 void VerboseLogPacket(const void* data, size_t length, int direction) { 185 void VerboseLogPacket(const void* data, size_t length, int direction) {
175 if (LOG_CHECK_LEVEL(LS_VERBOSE) && length > 0) { 186 if (LOG_CHECK_LEVEL(LS_VERBOSE) && length > 0) {
176 char *dump_buf; 187 char* dump_buf;
177 // Some downstream project uses an older version of usrsctp that expects 188 // Some downstream project uses an older version of usrsctp that expects
178 // a non-const "void*" as first parameter when dumping the packet, so we 189 // a non-const "void*" as first parameter when dumping the packet, so we
179 // need to cast the const away here to avoid a compiler error. 190 // need to cast the const away here to avoid a compiler error.
180 if ((dump_buf = usrsctp_dumppacket( 191 if ((dump_buf = usrsctp_dumppacket(const_cast<void*>(data), length,
181 const_cast<void*>(data), length, direction)) != NULL) { 192 direction)) != NULL) {
182 LOG(LS_VERBOSE) << dump_buf; 193 LOG(LS_VERBOSE) << dump_buf;
183 usrsctp_freedumpbuffer(dump_buf); 194 usrsctp_freedumpbuffer(dump_buf);
184 } 195 }
185 } 196 }
186 } 197 }
187 198
188 // This is the callback usrsctp uses when there's data to send on the network 199 } // namespace
189 // that has been wrapped appropriatly for the SCTP protocol. 200
190 int OnSctpOutboundPacket(void* addr, 201 namespace cricket {
191 void* data, 202
192 size_t length, 203 // Handles global init/deinit, and mapping from usrsctp callbacks to
193 uint8_t tos, 204 // SctpTransport calls.
194 uint8_t set_df) { 205 class SctpTransport::UsrSctpWrapper {
195 SctpDataMediaChannel* channel = static_cast<SctpDataMediaChannel*>(addr); 206 public:
196 LOG(LS_VERBOSE) << "global OnSctpOutboundPacket():" 207 static void InitializeUsrSctp() {
197 << "addr: " << addr << "; length: " << length 208 LOG(LS_INFO) << __FUNCTION__;
198 << "; tos: " << std::hex << static_cast<int>(tos) 209 // First argument is udp_encapsulation_port, which is not releveant for our
199 << "; set_df: " << std::hex << static_cast<int>(set_df); 210 // AF_CONN use of sctp.
200 211 usrsctp_init(0, &UsrSctpWrapper::OnSctpOutboundPacket, &DebugSctpPrintf);
201 VerboseLogPacket(data, length, SCTP_DUMP_OUTBOUND); 212
202 // Note: We have to copy the data; the caller will delete it. 213 // To turn on/off detailed SCTP debugging. You will also need to have the
203 auto* msg = new OutboundPacketMessage( 214 // SCTP_DEBUG cpp defines flag.
204 new rtc::CopyOnWriteBuffer(reinterpret_cast<uint8_t*>(data), length)); 215 // usrsctp_sysctl_set_sctp_debug_on(SCTP_DEBUG_ALL);
205 channel->worker_thread()->Post(RTC_FROM_HERE, channel, MSG_SCTPOUTBOUNDPACKET, 216
206 msg); 217 // TODO(ldixon): Consider turning this on/off.
207 return 0; 218 usrsctp_sysctl_set_sctp_ecn_enable(0);
208 } 219
209 220 // This is harmless, but we should find out when the library default
210 // This is the callback called from usrsctp when data has been received, after 221 // changes.
211 // a packet has been interpreted and parsed by usrsctp and found to contain 222 int send_size = usrsctp_sysctl_get_sctp_sendspace();
212 // payload data. It is called by a usrsctp thread. It is assumed this function 223 if (send_size != kSendBufferSize) {
213 // will free the memory used by 'data'. 224 LOG(LS_ERROR) << "Got different send size than expected: " << send_size;
214 int OnSctpInboundPacket(struct socket* sock, 225 }
215 union sctp_sockstore addr, 226
216 void* data, 227 // TODO(ldixon): Consider turning this on/off.
217 size_t length, 228 // This is not needed right now (we don't do dynamic address changes):
218 struct sctp_rcvinfo rcv, 229 // If SCTP Auto-ASCONF is enabled, the peer is informed automatically
219 int flags, 230 // when a new address is added or removed. This feature is enabled by
220 void* ulp_info) { 231 // default.
221 SctpDataMediaChannel* channel = static_cast<SctpDataMediaChannel*>(ulp_info); 232 // usrsctp_sysctl_set_sctp_auto_asconf(0);
222 // Post data to the channel's receiver thread (copying it). 233
223 // TODO(ldixon): Unclear if copy is needed as this method is responsible for 234 // TODO(ldixon): Consider turning this on/off.
224 // memory cleanup. But this does simplify code. 235 // Add a blackhole sysctl. Setting it to 1 results in no ABORTs
225 const SctpDataMediaChannel::PayloadProtocolIdentifier ppid = 236 // being sent in response to INITs, setting it to 2 results
226 static_cast<SctpDataMediaChannel::PayloadProtocolIdentifier>( 237 // in no ABORTs being sent for received OOTB packets.
227 rtc::HostToNetwork32(rcv.rcv_ppid)); 238 // This is similar to the TCP sysctl.
228 DataMessageType type = DMT_NONE; 239 //
229 if (!GetDataMediaType(ppid, &type) && !(flags & MSG_NOTIFICATION)) { 240 // See: http://lakerest.net/pipermail/sctp-coders/2012-January/009438.html
230 // It's neither a notification nor a recognized data packet. Drop it. 241 // See: http://svnweb.freebsd.org/base?view=revision&revision=229805
231 LOG(LS_ERROR) << "Received an unknown PPID " << ppid 242 // usrsctp_sysctl_set_sctp_blackhole(2);
232 << " on an SCTP packet. Dropping."; 243
244 // Set the number of default outgoing streams. This is the number we'll
245 // send in the SCTP INIT message.
246 usrsctp_sysctl_set_sctp_nr_outgoing_streams_default(kMaxSctpStreams);
247 }
248
249 static void UninitializeUsrSctp() {
250 LOG(LS_INFO) << __FUNCTION__;
251 // usrsctp_finish() may fail if it's called too soon after the transports
252 // are
253 // closed. Wait and try again until it succeeds for up to 3 seconds.
254 for (size_t i = 0; i < 300; ++i) {
255 if (usrsctp_finish() == 0) {
256 return;
257 }
258
259 rtc::Thread::SleepMs(10);
260 }
261 LOG(LS_ERROR) << "Failed to shutdown usrsctp.";
262 }
263
264 static void IncrementUsrSctpUsageCount() {
265 rtc::GlobalLockScope lock(&g_usrsctp_lock_);
266 if (!g_usrsctp_usage_count) {
267 InitializeUsrSctp();
268 }
269 ++g_usrsctp_usage_count;
270 }
271
272 static void DecrementUsrSctpUsageCount() {
273 rtc::GlobalLockScope lock(&g_usrsctp_lock_);
274 --g_usrsctp_usage_count;
275 if (!g_usrsctp_usage_count) {
276 UninitializeUsrSctp();
277 }
278 }
279
280 // This is the callback usrsctp uses when there's data to send on the network
281 // that has been wrapped appropriatly for the SCTP protocol.
282 static int OnSctpOutboundPacket(void* addr,
283 void* data,
284 size_t length,
285 uint8_t tos,
286 uint8_t set_df) {
287 SctpTransport* transport = static_cast<SctpTransport*>(addr);
288 LOG(LS_VERBOSE) << "global OnSctpOutboundPacket():"
289 << "addr: " << addr << "; length: " << length
290 << "; tos: " << std::hex << static_cast<int>(tos)
291 << "; set_df: " << std::hex << static_cast<int>(set_df);
292
293 VerboseLogPacket(data, length, SCTP_DUMP_OUTBOUND);
294 // Note: We have to copy the data; the caller will delete it.
295 rtc::CopyOnWriteBuffer buf(reinterpret_cast<uint8_t*>(data), length);
296 // TODO(deadbeef): Why do we need an AsyncInvoke here? We're already on the
297 // right thread and don't need to unwind the stack.
298 transport->invoker_.AsyncInvoke<void>(
299 RTC_FROM_HERE, transport->network_thread_,
300 rtc::Bind(&SctpTransport::OnPacketFromSctpToNetwork, transport, buf));
301 return 0;
302 }
303
304 // This is the callback called from usrsctp when data has been received, after
305 // a packet has been interpreted and parsed by usrsctp and found to contain
306 // payload data. It is called by a usrsctp thread. It is assumed this function
307 // will free the memory used by 'data'.
308 static int OnSctpInboundPacket(struct socket* sock,
309 union sctp_sockstore addr,
310 void* data,
311 size_t length,
312 struct sctp_rcvinfo rcv,
313 int flags,
314 void* ulp_info) {
315 SctpTransport* transport = static_cast<SctpTransport*>(ulp_info);
316 // Post data to the transport's receiver thread (copying it).
317 // TODO(ldixon): Unclear if copy is needed as this method is responsible for
318 // memory cleanup. But this does simplify code.
319 const PayloadProtocolIdentifier ppid =
320 static_cast<PayloadProtocolIdentifier>(
321 rtc::HostToNetwork32(rcv.rcv_ppid));
322 DataMessageType type = DMT_NONE;
323 if (!GetDataMediaType(ppid, &type) && !(flags & MSG_NOTIFICATION)) {
324 // It's neither a notification nor a recognized data packet. Drop it.
325 LOG(LS_ERROR) << "Received an unknown PPID " << ppid
326 << " on an SCTP packet. Dropping.";
327 } else {
328 rtc::CopyOnWriteBuffer buffer;
329 ReceiveDataParams params;
330 buffer.SetData(reinterpret_cast<uint8_t*>(data), length);
331 params.ssrc = rcv.rcv_sid;
332 params.seq_num = rcv.rcv_ssn;
333 params.timestamp = rcv.rcv_tsn;
334 params.type = type;
335 // The ownership of the packet transfers to |invoker_|. Using
336 // CopyOnWriteBuffer is the most convenient way to do this.
337 transport->invoker_.AsyncInvoke<void>(
338 RTC_FROM_HERE, transport->network_thread_,
339 rtc::Bind(&SctpTransport::OnInboundPacketFromSctpToChannel, transport,
340 buffer, params, flags));
341 }
342 free(data);
343 return 1;
344 }
345
346 static SctpTransport* GetTransportFromSocket(struct socket* sock) {
347 struct sockaddr* addrs = nullptr;
348 int naddrs = usrsctp_getladdrs(sock, 0, &addrs);
349 if (naddrs <= 0 || addrs[0].sa_family != AF_CONN) {
350 return nullptr;
351 }
352 // usrsctp_getladdrs() returns the addresses bound to this socket, which
353 // contains the SctpTransport* as sconn_addr. Read the pointer,
354 // then free the list of addresses once we have the pointer. We only open
355 // AF_CONN sockets, and they should all have the sconn_addr set to the
356 // pointer that created them, so [0] is as good as any other.
357 struct sockaddr_conn* sconn =
358 reinterpret_cast<struct sockaddr_conn*>(&addrs[0]);
359 SctpTransport* transport =
360 reinterpret_cast<SctpTransport*>(sconn->sconn_addr);
361 usrsctp_freeladdrs(addrs);
362
363 return transport;
364 }
365
366 static int SendThresholdCallback(struct socket* sock, uint32_t sb_free) {
367 // Fired on our I/O thread. SctpTransport::OnPacketReceived() gets
368 // a packet containing acknowledgments, which goes into usrsctp_conninput,
369 // and then back here.
370 SctpTransport* transport = GetTransportFromSocket(sock);
371 if (!transport) {
372 LOG(LS_ERROR)
373 << "SendThresholdCallback: Failed to get transport for socket "
374 << sock;
375 return 0;
376 }
377 transport->OnSendThresholdCallback();
378 return 0;
379 }
380 };
381
382 SctpTransport::SctpTransport(rtc::Thread* network_thread,
383 TransportChannel* channel)
384 : network_thread_(network_thread),
385 transport_channel_(channel),
386 was_ever_writable_(channel->writable()) {
387 RTC_DCHECK(network_thread_);
388 RTC_DCHECK(transport_channel_);
389 ConnectTransportChannelSignals();
390 }
391
392 SctpTransport::~SctpTransport() {
393 // Close abruptly; no reset procedure.
394 CloseSctpSocket();
395 }
396
397 void SctpTransport::SetTransportChannel(cricket::TransportChannel* channel) {
398 RTC_DCHECK(channel);
399 DisconnectTransportChannelSignals();
400 transport_channel_ = channel;
401 ConnectTransportChannelSignals();
402 if (!was_ever_writable_ && channel->writable()) {
403 was_ever_writable_ = true;
404 // New channel is writable, now we can start the SCTP connection if Start
405 // was called already.
406 if (started_) {
407 RTC_DCHECK(!sock_);
408 Connect();
409 }
410 }
411 }
412
413 bool SctpTransport::Start(int local_sctp_port, int remote_sctp_port) {
414 if (local_sctp_port == -1) {
415 local_sctp_port = kSctpDefaultPort;
416 }
417 if (remote_sctp_port == -1) {
418 remote_sctp_port = kSctpDefaultPort;
419 }
420 if (started_) {
421 if (local_sctp_port != local_port_ || remote_sctp_port != remote_port_) {
422 LOG(LS_ERROR) << "Can't change SCTP port after SCTP association formed.";
423 return false;
424 }
425 return true;
426 }
427 local_port_ = local_sctp_port;
428 remote_port_ = remote_sctp_port;
429 started_ = true;
430 RTC_DCHECK(!sock_);
431 // Only try to connect if the DTLS channel has been writable before
432 // (indicating that the DTLS handshake is complete).
433 if (was_ever_writable_) {
434 return Connect();
435 }
436 return true;
437 }
438
439 bool SctpTransport::OpenStream(int sid) {
440 if (sid > kMaxSctpSid) {
441 LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
442 << "Not adding data stream "
443 << "with sid=" << sid << " because sid is too high.";
444 return false;
445 } else if (open_streams_.find(sid) != open_streams_.end()) {
446 LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
447 << "Not adding data stream "
448 << "with sid=" << sid << " because stream is already open.";
449 return false;
450 } else if (queued_reset_streams_.find(sid) != queued_reset_streams_.end() ||
451 sent_reset_streams_.find(sid) != sent_reset_streams_.end()) {
452 LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
453 << "Not adding data stream "
454 << " with sid=" << sid
455 << " because stream is still closing.";
456 return false;
457 }
458
459 open_streams_.insert(sid);
460 return true;
461 }
462
463 bool SctpTransport::ResetStream(int sid) {
464 StreamSet::iterator found = open_streams_.find(sid);
465 if (found == open_streams_.end()) {
466 LOG(LS_WARNING) << debug_name_ << "->ResetStream(" << sid << "): "
467 << "stream not found.";
468 return false;
233 } else { 469 } else {
234 SctpInboundPacket* packet = new SctpInboundPacket; 470 LOG(LS_VERBOSE) << debug_name_ << "->ResetStream(" << sid << "): "
235 packet->buffer.SetData(reinterpret_cast<uint8_t*>(data), length); 471 << "Removing and queuing RE-CONFIG chunk.";
236 packet->params.ssrc = rcv.rcv_sid; 472 open_streams_.erase(found);
237 packet->params.seq_num = rcv.rcv_ssn; 473 }
238 packet->params.timestamp = rcv.rcv_tsn; 474
239 packet->params.type = type; 475 // SCTP won't let you have more than one stream reset pending at a time, but
240 packet->flags = flags; 476 // you can close multiple streams in a single reset. So, we keep an internal
241 // The ownership of |packet| transfers to |msg|. 477 // queue of streams-to-reset, and send them as one reset message in
242 InboundPacketMessage* msg = new InboundPacketMessage(packet); 478 // SendQueuedStreamResets().
243 channel->worker_thread()->Post(RTC_FROM_HERE, channel, 479 queued_reset_streams_.insert(sid);
244 MSG_SCTPINBOUNDPACKET, msg); 480
245 } 481 // Signal our stream-reset logic that it should try to send now, if it can.
246 free(data); 482 SendQueuedStreamResets();
247 return 1; 483
248 } 484 // The stream will actually get removed when we get the acknowledgment.
249 485 return true;
250 void InitializeUsrSctp() { 486 }
251 LOG(LS_INFO) << __FUNCTION__; 487
252 // First argument is udp_encapsulation_port, which is not releveant for our 488 bool SctpTransport::SendData(const SendDataParams& params,
253 // AF_CONN use of sctp. 489 const rtc::CopyOnWriteBuffer& payload,
254 usrsctp_init(0, &OnSctpOutboundPacket, &DebugSctpPrintf); 490 SendDataResult* result) {
255 491 if (result) {
256 // To turn on/off detailed SCTP debugging. You will also need to have the 492 // Preset |result| to assume an error. If SendData succeeds, we'll
257 // SCTP_DEBUG cpp defines flag. 493 // overwrite |*result| once more at the end.
258 // usrsctp_sysctl_set_sctp_debug_on(SCTP_DEBUG_ALL); 494 *result = SDR_ERROR;
259 495 }
260 // TODO(ldixon): Consider turning this on/off. 496
261 usrsctp_sysctl_set_sctp_ecn_enable(0); 497 if (!sock_) {
262 498 LOG(LS_WARNING) << debug_name_ << "->SendData(...): "
263 // This is harmless, but we should find out when the library default 499 << "Not sending packet with sid=" << params.ssrc
264 // changes. 500 << " len=" << payload.size() << " before Start().";
265 int send_size = usrsctp_sysctl_get_sctp_sendspace(); 501 return false;
266 if (send_size != kSendBufferSize) { 502 }
267 LOG(LS_ERROR) << "Got different send size than expected: " << send_size; 503
268 } 504 if (params.type != DMT_CONTROL &&
269 505 open_streams_.find(params.ssrc) == open_streams_.end()) {
270 // TODO(ldixon): Consider turning this on/off. 506 LOG(LS_WARNING) << debug_name_ << "->SendData(...): "
271 // This is not needed right now (we don't do dynamic address changes): 507 << "Not sending data because sid is unknown: "
272 // If SCTP Auto-ASCONF is enabled, the peer is informed automatically 508 << params.ssrc;
273 // when a new address is added or removed. This feature is enabled by 509 return false;
274 // default. 510 }
275 // usrsctp_sysctl_set_sctp_auto_asconf(0); 511
276 512 // Send data using SCTP.
277 // TODO(ldixon): Consider turning this on/off. 513 ssize_t send_res = 0; // result from usrsctp_sendv.
278 // Add a blackhole sysctl. Setting it to 1 results in no ABORTs 514 struct sctp_sendv_spa spa = {0};
279 // being sent in response to INITs, setting it to 2 results 515 spa.sendv_flags |= SCTP_SEND_SNDINFO_VALID;
280 // in no ABORTs being sent for received OOTB packets. 516 spa.sendv_sndinfo.snd_sid = params.ssrc;
281 // This is similar to the TCP sysctl. 517 spa.sendv_sndinfo.snd_ppid = rtc::HostToNetwork32(GetPpid(params.type));
282 // 518
283 // See: http://lakerest.net/pipermail/sctp-coders/2012-January/009438.html 519 // Ordered implies reliable.
284 // See: http://svnweb.freebsd.org/base?view=revision&revision=229805 520 if (!params.ordered) {
285 // usrsctp_sysctl_set_sctp_blackhole(2); 521 spa.sendv_sndinfo.snd_flags |= SCTP_UNORDERED;
286 522 if (params.max_rtx_count >= 0 || params.max_rtx_ms == 0) {
287 // Set the number of default outgoing streams. This is the number we'll 523 spa.sendv_flags |= SCTP_SEND_PRINFO_VALID;
288 // send in the SCTP INIT message. 524 spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_RTX;
289 usrsctp_sysctl_set_sctp_nr_outgoing_streams_default(kMaxSctpStreams); 525 spa.sendv_prinfo.pr_value = params.max_rtx_count;
290 } 526 } else {
291 527 spa.sendv_flags |= SCTP_SEND_PRINFO_VALID;
292 void UninitializeUsrSctp() { 528 spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_TTL;
293 LOG(LS_INFO) << __FUNCTION__; 529 spa.sendv_prinfo.pr_value = params.max_rtx_ms;
294 // usrsctp_finish() may fail if it's called too soon after the channels are 530 }
295 // closed. Wait and try again until it succeeds for up to 3 seconds. 531 }
296 for (size_t i = 0; i < 300; ++i) { 532
297 if (usrsctp_finish() == 0) { 533 // We don't fragment.
298 return; 534 send_res = usrsctp_sendv(
299 } 535 sock_, payload.data(), static_cast<size_t>(payload.size()), NULL, 0, &spa,
300 536 rtc::checked_cast<socklen_t>(sizeof(spa)), SCTP_SENDV_SPA, 0);
301 rtc::Thread::SleepMs(10); 537 if (send_res < 0) {
302 } 538 if (errno == SCTP_EWOULDBLOCK) {
303 LOG(LS_ERROR) << "Failed to shutdown usrsctp."; 539 *result = SDR_BLOCK;
304 } 540 ready_to_send_data_ = false;
305 541 LOG(LS_INFO) << debug_name_ << "->SendData(...): EWOULDBLOCK returned";
306 void IncrementUsrSctpUsageCount() { 542 } else {
307 rtc::GlobalLockScope lock(&g_usrsctp_lock_); 543 LOG_ERRNO(LS_ERROR) << "ERROR:" << debug_name_ << "->SendData(...): "
308 if (!g_usrsctp_usage_count) { 544 << " usrsctp_sendv: ";
309 InitializeUsrSctp(); 545 }
310 } 546 return false;
311 ++g_usrsctp_usage_count; 547 }
312 } 548 if (result) {
313 549 // Only way out now is success.
314 void DecrementUsrSctpUsageCount() { 550 *result = SDR_SUCCESS;
315 rtc::GlobalLockScope lock(&g_usrsctp_lock_); 551 }
316 --g_usrsctp_usage_count; 552 return true;
317 if (!g_usrsctp_usage_count) { 553 }
318 UninitializeUsrSctp(); 554
319 } 555 bool SctpTransport::ReadyToSendData() {
320 } 556 return ready_to_send_data_;
321 557 }
322 DataCodec GetSctpDataCodec() { 558
323 DataCodec codec(kGoogleSctpDataCodecPlType, kGoogleSctpDataCodecName); 559 void SctpTransport::ConnectTransportChannelSignals() {
324 codec.SetParam(kCodecParamPort, kSctpDefaultPort); 560 transport_channel_->SignalWritableState.connect(
325 return codec; 561 this, &SctpTransport::OnWritableState);
326 } 562 transport_channel_->SignalReadPacket.connect(this,
327 563 &SctpTransport::OnPacketRead);
328 } // namespace 564 }
329 565
330 SctpDataEngine::SctpDataEngine() : codecs_(1, GetSctpDataCodec()) {} 566 void SctpTransport::DisconnectTransportChannelSignals() {
331 567 transport_channel_->SignalWritableState.disconnect(this);
332 SctpDataEngine::~SctpDataEngine() {} 568 transport_channel_->SignalReadPacket.disconnect(this);
333 569 }
334 // Called on the worker thread. 570
335 DataMediaChannel* SctpDataEngine::CreateChannel( 571 bool SctpTransport::Connect() {
336 DataChannelType data_channel_type, 572 LOG(LS_VERBOSE) << debug_name_ << "->Connect().";
337 const MediaConfig& config) { 573
338 if (data_channel_type != DCT_SCTP) { 574 // If we already have a socket connection (which shouldn't ever happen), just
339 return NULL; 575 // return.
340 } 576 RTC_DCHECK(!sock_);
341 return new SctpDataMediaChannel(rtc::Thread::Current(), config);
342 }
343
344 // static
345 SctpDataMediaChannel* SctpDataMediaChannel::GetChannelFromSocket(
346 struct socket* sock) {
347 struct sockaddr* addrs = nullptr;
348 int naddrs = usrsctp_getladdrs(sock, 0, &addrs);
349 if (naddrs <= 0 || addrs[0].sa_family != AF_CONN) {
350 return nullptr;
351 }
352 // usrsctp_getladdrs() returns the addresses bound to this socket, which
353 // contains the SctpDataMediaChannel* as sconn_addr. Read the pointer,
354 // then free the list of addresses once we have the pointer. We only open
355 // AF_CONN sockets, and they should all have the sconn_addr set to the
356 // pointer that created them, so [0] is as good as any other.
357 struct sockaddr_conn* sconn =
358 reinterpret_cast<struct sockaddr_conn*>(&addrs[0]);
359 SctpDataMediaChannel* channel =
360 reinterpret_cast<SctpDataMediaChannel*>(sconn->sconn_addr);
361 usrsctp_freeladdrs(addrs);
362
363 return channel;
364 }
365
366 // static
367 int SctpDataMediaChannel::SendThresholdCallback(struct socket* sock,
368 uint32_t sb_free) {
369 // Fired on our I/O thread. SctpDataMediaChannel::OnPacketReceived() gets
370 // a packet containing acknowledgments, which goes into usrsctp_conninput,
371 // and then back here.
372 SctpDataMediaChannel* channel = GetChannelFromSocket(sock);
373 if (!channel) {
374 LOG(LS_ERROR) << "SendThresholdCallback: Failed to get channel for socket "
375 << sock;
376 return 0;
377 }
378 channel->OnSendThresholdCallback();
379 return 0;
380 }
381
382 SctpDataMediaChannel::SctpDataMediaChannel(rtc::Thread* thread,
383 const MediaConfig& config)
384 : DataMediaChannel(config),
385 worker_thread_(thread),
386 local_port_(kSctpDefaultPort),
387 remote_port_(kSctpDefaultPort),
388 sock_(NULL),
389 sending_(false),
390 receiving_(false),
391 debug_name_("SctpDataMediaChannel") {}
392
393 SctpDataMediaChannel::~SctpDataMediaChannel() {
394 CloseSctpSocket();
395 }
396
397 void SctpDataMediaChannel::OnSendThresholdCallback() {
398 RTC_DCHECK(rtc::Thread::Current() == worker_thread_);
399 SignalReadyToSend(true);
400 }
401
402 sockaddr_conn SctpDataMediaChannel::GetSctpSockAddr(int port) {
403 sockaddr_conn sconn = {0};
404 sconn.sconn_family = AF_CONN;
405 #ifdef HAVE_SCONN_LEN
406 sconn.sconn_len = sizeof(sockaddr_conn);
407 #endif
408 // Note: conversion from int to uint16_t happens here.
409 sconn.sconn_port = rtc::HostToNetwork16(port);
410 sconn.sconn_addr = this;
411 return sconn;
412 }
413
414 bool SctpDataMediaChannel::OpenSctpSocket() {
415 if (sock_) { 577 if (sock_) {
416 LOG(LS_VERBOSE) << debug_name_ 578 LOG(LS_ERROR) << debug_name_ << "->Connect(): Ignored as socket "
579 "is already established.";
580 return true;
581 }
582
583 // If no socket (it was closed) try to start it again. This can happen when
584 // the socket we are connecting to closes, does an sctp shutdown handshake,
585 // or behaves unexpectedly causing us to perform a CloseSctpSocket.
586 if (!OpenSctpSocket()) {
587 return false;
588 }
589
590 // Note: conversion from int to uint16_t happens on assignment.
591 sockaddr_conn local_sconn = GetSctpSockAddr(local_port_);
592 if (usrsctp_bind(sock_, reinterpret_cast<sockaddr*>(&local_sconn),
593 sizeof(local_sconn)) < 0) {
594 LOG_ERRNO(LS_ERROR) << debug_name_
595 << "->Connect(): " << ("Failed usrsctp_bind");
596 CloseSctpSocket();
597 return false;
598 }
599
600 // Note: conversion from int to uint16_t happens on assignment.
601 sockaddr_conn remote_sconn = GetSctpSockAddr(remote_port_);
602 int connect_result = usrsctp_connect(
603 sock_, reinterpret_cast<sockaddr*>(&remote_sconn), sizeof(remote_sconn));
604 if (connect_result < 0 && errno != SCTP_EINPROGRESS) {
605 LOG_ERRNO(LS_ERROR) << debug_name_
606 << "Failed usrsctp_connect. got errno=" << errno
607 << ", but wanted " << SCTP_EINPROGRESS;
608 CloseSctpSocket();
609 return false;
610 }
611 // Set the MTU and disable MTU discovery.
612 // We can only do this after usrsctp_connect or it has no effect.
613 sctp_paddrparams params = {{0}};
614 memcpy(&params.spp_address, &remote_sconn, sizeof(remote_sconn));
615 params.spp_flags = SPP_PMTUD_DISABLE;
616 params.spp_pathmtu = kSctpMtu;
617 if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_PEER_ADDR_PARAMS, &params,
618 sizeof(params))) {
619 LOG_ERRNO(LS_ERROR) << debug_name_
620 << "Failed to set SCTP_PEER_ADDR_PARAMS.";
621 }
622 // Since this is a fresh SCTP association, we'll always start out with empty
623 // queues, so "ReadyToSendData" should be true.
624 SetReadyToSendData();
625 return true;
626 }
627
628 bool SctpTransport::OpenSctpSocket() {
629 if (sock_) {
630 LOG(LS_WARNING) << debug_name_
417 << "->Ignoring attempt to re-create existing socket."; 631 << "->Ignoring attempt to re-create existing socket.";
418 return false; 632 return false;
419 } 633 }
420 634
421 IncrementUsrSctpUsageCount(); 635 UsrSctpWrapper::IncrementUsrSctpUsageCount();
422 636
423 // If kSendBufferSize isn't reflective of reality, we log an error, but we 637 // If kSendBufferSize isn't reflective of reality, we log an error, but we
424 // still have to do something reasonable here. Look up what the buffer's 638 // still have to do something reasonable here. Look up what the buffer's
425 // real size is and set our threshold to something reasonable. 639 // real size is and set our threshold to something reasonable.
426 const static int kSendThreshold = usrsctp_sysctl_get_sctp_sendspace() / 2; 640 static const int kSendThreshold = usrsctp_sysctl_get_sctp_sendspace() / 2;
427 641
428 sock_ = usrsctp_socket( 642 sock_ = usrsctp_socket(
429 AF_CONN, SOCK_STREAM, IPPROTO_SCTP, OnSctpInboundPacket, 643 AF_CONN, SOCK_STREAM, IPPROTO_SCTP, &UsrSctpWrapper::OnSctpInboundPacket,
430 &SctpDataMediaChannel::SendThresholdCallback, kSendThreshold, this); 644 &UsrSctpWrapper::SendThresholdCallback, kSendThreshold, this);
431 if (!sock_) { 645 if (!sock_) {
432 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to create SCTP socket."; 646 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to create SCTP socket.";
433 DecrementUsrSctpUsageCount(); 647 UsrSctpWrapper::DecrementUsrSctpUsageCount();
434 return false; 648 return false;
435 } 649 }
436 650
651 if (!ConfigureSctpSocket()) {
652 usrsctp_close(sock_);
653 sock_ = nullptr;
654 UsrSctpWrapper::DecrementUsrSctpUsageCount();
655 return false;
656 }
657 // Register this class as an address for usrsctp. This is used by SCTP to
658 // direct the packets received (by the created socket) to this class.
659 usrsctp_register_address(this);
660 return true;
661 }
662
663 bool SctpTransport::ConfigureSctpSocket() {
664 RTC_DCHECK(sock_);
437 // Make the socket non-blocking. Connect, close, shutdown etc will not block 665 // Make the socket non-blocking. Connect, close, shutdown etc will not block
438 // the thread waiting for the socket operation to complete. 666 // the thread waiting for the socket operation to complete.
439 if (usrsctp_set_non_blocking(sock_, 1) < 0) { 667 if (usrsctp_set_non_blocking(sock_, 1) < 0) {
440 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP to non blocking."; 668 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP to non blocking.";
441 return false; 669 return false;
442 } 670 }
443 671
444 // This ensures that the usrsctp close call deletes the association. This 672 // This ensures that the usrsctp close call deletes the association. This
445 // prevents usrsctp from calling OnSctpOutboundPacket with references to 673 // prevents usrsctp from calling OnSctpOutboundPacket with references to
446 // this class as the address. 674 // this class as the address.
(...skipping 19 matching lines...) Expand all
466 694
467 // Nagle. 695 // Nagle.
468 uint32_t nodelay = 1; 696 uint32_t nodelay = 1;
469 if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_NODELAY, &nodelay, 697 if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_NODELAY, &nodelay,
470 sizeof(nodelay))) { 698 sizeof(nodelay))) {
471 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP_NODELAY."; 699 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP_NODELAY.";
472 return false; 700 return false;
473 } 701 }
474 702
475 // Subscribe to SCTP event notifications. 703 // Subscribe to SCTP event notifications.
476 int event_types[] = {SCTP_ASSOC_CHANGE, 704 int event_types[] = {SCTP_ASSOC_CHANGE, SCTP_PEER_ADDR_CHANGE,
477 SCTP_PEER_ADDR_CHANGE, 705 SCTP_SEND_FAILED_EVENT, SCTP_SENDER_DRY_EVENT,
478 SCTP_SEND_FAILED_EVENT,
479 SCTP_SENDER_DRY_EVENT,
480 SCTP_STREAM_RESET_EVENT}; 706 SCTP_STREAM_RESET_EVENT};
481 struct sctp_event event = {0}; 707 struct sctp_event event = {0};
482 event.se_assoc_id = SCTP_ALL_ASSOC; 708 event.se_assoc_id = SCTP_ALL_ASSOC;
483 event.se_on = 1; 709 event.se_on = 1;
484 for (size_t i = 0; i < arraysize(event_types); i++) { 710 for (size_t i = 0; i < arraysize(event_types); i++) {
485 event.se_type = event_types[i]; 711 event.se_type = event_types[i];
486 if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_EVENT, &event, 712 if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_EVENT, &event,
487 sizeof(event)) < 0) { 713 sizeof(event)) < 0) {
488 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to set SCTP_EVENT type: " 714 LOG_ERRNO(LS_ERROR) << debug_name_
489 << event.se_type; 715 << "Failed to set SCTP_EVENT type: " << event.se_type;
490 return false; 716 return false;
491 } 717 }
492 } 718 }
493
494 // Register this class as an address for usrsctp. This is used by SCTP to
495 // direct the packets received (by the created socket) to this class.
496 usrsctp_register_address(this);
497 sending_ = true;
498 return true; 719 return true;
499 } 720 }
500 721
501 void SctpDataMediaChannel::CloseSctpSocket() { 722 void SctpTransport::CloseSctpSocket() {
502 sending_ = false;
503 if (sock_) { 723 if (sock_) {
504 // We assume that SO_LINGER option is set to close the association when 724 // We assume that SO_LINGER option is set to close the association when
505 // close is called. This means that any pending packets in usrsctp will be 725 // close is called. This means that any pending packets in usrsctp will be
506 // discarded instead of being sent. 726 // discarded instead of being sent.
507 usrsctp_close(sock_); 727 usrsctp_close(sock_);
508 sock_ = NULL; 728 sock_ = nullptr;
509 usrsctp_deregister_address(this); 729 usrsctp_deregister_address(this);
510 730 UsrSctpWrapper::DecrementUsrSctpUsageCount();
511 DecrementUsrSctpUsageCount(); 731 ready_to_send_data_ = false;
512 } 732 }
513 } 733 }
514 734
515 bool SctpDataMediaChannel::Connect() { 735 bool SctpTransport::SendQueuedStreamResets() {
516 LOG(LS_VERBOSE) << debug_name_ << "->Connect()."; 736 if (!sent_reset_streams_.empty() || queued_reset_streams_.empty()) {
517
518 // If we already have a socket connection, just return.
519 if (sock_) {
520 LOG(LS_WARNING) << debug_name_ << "->Connect(): Ignored as socket "
521 "is already established.";
522 return true; 737 return true;
523 } 738 }
524 739
525 // If no socket (it was closed) try to start it again. This can happen when 740 LOG(LS_VERBOSE) << "SendQueuedStreamResets[" << debug_name_ << "]: Sending ["
526 // the socket we are connecting to closes, does an sctp shutdown handshake, 741 << ListStreams(queued_reset_streams_) << "], Open: ["
527 // or behaves unexpectedly causing us to perform a CloseSctpSocket. 742 << ListStreams(open_streams_) << "], Sent: ["
528 if (!sock_ && !OpenSctpSocket()) { 743 << ListStreams(sent_reset_streams_) << "]";
744
745 const size_t num_streams = queued_reset_streams_.size();
746 const size_t num_bytes =
747 sizeof(struct sctp_reset_streams) + (num_streams * sizeof(uint16_t));
748
749 std::vector<uint8_t> reset_stream_buf(num_bytes, 0);
750 struct sctp_reset_streams* resetp =
751 reinterpret_cast<sctp_reset_streams*>(&reset_stream_buf[0]);
752 resetp->srs_assoc_id = SCTP_ALL_ASSOC;
753 resetp->srs_flags = SCTP_STREAM_RESET_INCOMING | SCTP_STREAM_RESET_OUTGOING;
754 resetp->srs_number_streams = rtc::checked_cast<uint16_t>(num_streams);
755 int result_idx = 0;
756 for (StreamSet::iterator it = queued_reset_streams_.begin();
757 it != queued_reset_streams_.end(); ++it) {
758 resetp->srs_stream_list[result_idx++] = *it;
759 }
760
761 int ret =
762 usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_RESET_STREAMS, resetp,
763 rtc::checked_cast<socklen_t>(reset_stream_buf.size()));
764 if (ret < 0) {
765 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to send a stream reset for "
766 << num_streams << " streams";
529 return false; 767 return false;
530 } 768 }
531 769
532 // Note: conversion from int to uint16_t happens on assignment. 770 // sent_reset_streams_ is empty, and all the queued_reset_streams_ go into
533 sockaddr_conn local_sconn = GetSctpSockAddr(local_port_); 771 // it now.
534 if (usrsctp_bind(sock_, reinterpret_cast<sockaddr *>(&local_sconn), 772 queued_reset_streams_.swap(sent_reset_streams_);
535 sizeof(local_sconn)) < 0) {
536 LOG_ERRNO(LS_ERROR) << debug_name_ << "->Connect(): "
537 << ("Failed usrsctp_bind");
538 CloseSctpSocket();
539 return false;
540 }
541
542 // Note: conversion from int to uint16_t happens on assignment.
543 sockaddr_conn remote_sconn = GetSctpSockAddr(remote_port_);
544 int connect_result = usrsctp_connect(
545 sock_, reinterpret_cast<sockaddr *>(&remote_sconn), sizeof(remote_sconn));
546 if (connect_result < 0 && errno != SCTP_EINPROGRESS) {
547 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed usrsctp_connect. got errno="
548 << errno << ", but wanted " << SCTP_EINPROGRESS;
549 CloseSctpSocket();
550 return false;
551 }
552 // Set the MTU and disable MTU discovery.
553 // We can only do this after usrsctp_connect or it has no effect.
554 sctp_paddrparams params = {{0}};
555 memcpy(&params.spp_address, &remote_sconn, sizeof(remote_sconn));
556 params.spp_flags = SPP_PMTUD_DISABLE;
557 params.spp_pathmtu = kSctpMtu;
558 if (usrsctp_setsockopt(sock_, IPPROTO_SCTP, SCTP_PEER_ADDR_PARAMS, &params,
559 sizeof(params))) {
560 LOG_ERRNO(LS_ERROR) << debug_name_
561 << "Failed to set SCTP_PEER_ADDR_PARAMS.";
562 }
563 return true; 773 return true;
564 } 774 }
565 775
566 void SctpDataMediaChannel::Disconnect() { 776 void SctpTransport::SetReadyToSendData() {
567 // TODO(ldixon): Consider calling |usrsctp_shutdown(sock_, ...)| to do a 777 if (!ready_to_send_data_) {
568 // shutdown handshake and remove the association. 778 ready_to_send_data_ = true;
569 CloseSctpSocket(); 779 SignalReadyToSendData();
780 }
570 } 781 }
571 782
572 bool SctpDataMediaChannel::SetSend(bool send) { 783 void SctpTransport::OnWritableState(rtc::PacketTransportInterface* transport) {
573 if (!sending_ && send) { 784 RTC_DCHECK(network_thread_->IsCurrent());
574 return Connect(); 785 RTC_DCHECK_EQ(transport_channel_, transport);
575 } 786 if (!was_ever_writable_ && transport->writable()) {
576 if (sending_ && !send) { 787 was_ever_writable_ = true;
577 Disconnect(); 788 if (started_) {
578 } 789 Connect();
579 return true;
580 }
581
582 bool SctpDataMediaChannel::SetReceive(bool receive) {
583 receiving_ = receive;
584 return true;
585 }
586
587 bool SctpDataMediaChannel::SetSendParameters(const DataSendParameters& params) {
588 return SetSendCodecs(params.codecs);
589 }
590
591 bool SctpDataMediaChannel::SetRecvParameters(const DataRecvParameters& params) {
592 return SetRecvCodecs(params.codecs);
593 }
594
595 bool SctpDataMediaChannel::AddSendStream(const StreamParams& stream) {
596 return AddStream(stream);
597 }
598
599 bool SctpDataMediaChannel::RemoveSendStream(uint32_t ssrc) {
600 return ResetStream(ssrc);
601 }
602
603 bool SctpDataMediaChannel::AddRecvStream(const StreamParams& stream) {
604 // SCTP DataChannels are always bi-directional and calling AddSendStream will
605 // enable both sending and receiving on the stream. So AddRecvStream is a
606 // no-op.
607 return true;
608 }
609
610 bool SctpDataMediaChannel::RemoveRecvStream(uint32_t ssrc) {
611 // SCTP DataChannels are always bi-directional and calling RemoveSendStream
612 // will disable both sending and receiving on the stream. So RemoveRecvStream
613 // is a no-op.
614 return true;
615 }
616
617 bool SctpDataMediaChannel::SendData(
618 const SendDataParams& params,
619 const rtc::CopyOnWriteBuffer& payload,
620 SendDataResult* result) {
621 if (result) {
622 // Preset |result| to assume an error. If SendData succeeds, we'll
623 // overwrite |*result| once more at the end.
624 *result = SDR_ERROR;
625 }
626
627 if (!sending_) {
628 LOG(LS_WARNING) << debug_name_ << "->SendData(...): "
629 << "Not sending packet with ssrc=" << params.ssrc
630 << " len=" << payload.size() << " before SetSend(true).";
631 return false;
632 }
633
634 if (params.type != DMT_CONTROL &&
635 open_streams_.find(params.ssrc) == open_streams_.end()) {
636 LOG(LS_WARNING) << debug_name_ << "->SendData(...): "
637 << "Not sending data because ssrc is unknown: "
638 << params.ssrc;
639 return false;
640 }
641
642 //
643 // Send data using SCTP.
644 ssize_t send_res = 0; // result from usrsctp_sendv.
645 struct sctp_sendv_spa spa = {0};
646 spa.sendv_flags |= SCTP_SEND_SNDINFO_VALID;
647 spa.sendv_sndinfo.snd_sid = params.ssrc;
648 spa.sendv_sndinfo.snd_ppid = rtc::HostToNetwork32(
649 GetPpid(params.type));
650
651 // Ordered implies reliable.
652 if (!params.ordered) {
653 spa.sendv_sndinfo.snd_flags |= SCTP_UNORDERED;
654 if (params.max_rtx_count >= 0 || params.max_rtx_ms == 0) {
655 spa.sendv_flags |= SCTP_SEND_PRINFO_VALID;
656 spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_RTX;
657 spa.sendv_prinfo.pr_value = params.max_rtx_count;
658 } else {
659 spa.sendv_flags |= SCTP_SEND_PRINFO_VALID;
660 spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_TTL;
661 spa.sendv_prinfo.pr_value = params.max_rtx_ms;
662 } 790 }
663 } 791 }
664
665 // We don't fragment.
666 send_res = usrsctp_sendv(
667 sock_, payload.data(), static_cast<size_t>(payload.size()), NULL, 0, &spa,
668 rtc::checked_cast<socklen_t>(sizeof(spa)), SCTP_SENDV_SPA, 0);
669 if (send_res < 0) {
670 if (errno == SCTP_EWOULDBLOCK) {
671 *result = SDR_BLOCK;
672 LOG(LS_INFO) << debug_name_ << "->SendData(...): EWOULDBLOCK returned";
673 } else {
674 LOG_ERRNO(LS_ERROR) << "ERROR:" << debug_name_
675 << "->SendData(...): "
676 << " usrsctp_sendv: ";
677 }
678 return false;
679 }
680 if (result) {
681 // Only way out now is success.
682 *result = SDR_SUCCESS;
683 }
684 return true;
685 } 792 }
686 793
687 // Called by network interface when a packet has been received. 794 // Called by network interface when a packet has been received.
688 void SctpDataMediaChannel::OnPacketReceived( 795 void SctpTransport::OnPacketRead(rtc::PacketTransportInterface* transport,
689 rtc::CopyOnWriteBuffer* packet, const rtc::PacketTime& packet_time) { 796 const char* data,
690 RTC_DCHECK(rtc::Thread::Current() == worker_thread_); 797 size_t len,
691 LOG(LS_VERBOSE) << debug_name_ << "->OnPacketReceived(...): " 798 const rtc::PacketTime& packet_time,
692 << " length=" << packet->size() << ", sending: " << sending_; 799 int flags) {
800 RTC_DCHECK(network_thread_->IsCurrent());
801 RTC_DCHECK_EQ(transport_channel_, transport);
802 TRACE_EVENT0("webrtc", "SctpTransport::OnPacketRead");
803
804 // TODO(pthatcher): Do this in a more robust way by checking for
805 // SCTP or DTLS.
806 if (IsRtpPacket(data, len)) {
807 return;
808 }
809
810 LOG(LS_VERBOSE) << debug_name_ << "->OnPacketRead(...): "
811 << " length=" << len << ", started: " << started_;
693 // Only give receiving packets to usrsctp after if connected. This enables two 812 // Only give receiving packets to usrsctp after if connected. This enables two
694 // peers to each make a connect call, but for them not to receive an INIT 813 // peers to each make a connect call, but for them not to receive an INIT
695 // packet before they have called connect; least the last receiver of the INIT 814 // packet before they have called connect; least the last receiver of the INIT
696 // packet will have called connect, and a connection will be established. 815 // packet will have called connect, and a connection will be established.
697 if (sending_) { 816 if (sock_) {
698 // Pass received packet to SCTP stack. Once processed by usrsctp, the data 817 // Pass received packet to SCTP stack. Once processed by usrsctp, the data
699 // will be will be given to the global OnSctpInboundData, and then, 818 // will be will be given to the global OnSctpInboundData, and then,
700 // marshalled by a Post and handled with OnMessage. 819 // marshalled by the AsyncInvoker.
701 VerboseLogPacket(packet->cdata(), packet->size(), SCTP_DUMP_INBOUND); 820 VerboseLogPacket(data, len, SCTP_DUMP_INBOUND);
702 usrsctp_conninput(this, packet->cdata(), packet->size(), 0); 821 usrsctp_conninput(this, data, len, 0);
703 } else { 822 } else {
704 // TODO(ldixon): Consider caching the packet for very slightly better 823 // TODO(ldixon): Consider caching the packet for very slightly better
705 // reliability. 824 // reliability.
706 } 825 }
707 } 826 }
708 827
709 void SctpDataMediaChannel::OnInboundPacketFromSctpToChannel( 828 void SctpTransport::OnSendThresholdCallback() {
710 SctpInboundPacket* packet) { 829 RTC_DCHECK(rtc::Thread::Current() == network_thread_);
830 SetReadyToSendData();
831 }
832
833 sockaddr_conn SctpTransport::GetSctpSockAddr(int port) {
834 sockaddr_conn sconn = {0};
835 sconn.sconn_family = AF_CONN;
836 #ifdef HAVE_SCONN_LEN
837 sconn.sconn_len = sizeof(sockaddr_conn);
838 #endif
839 // Note: conversion from int to uint16_t happens here.
840 sconn.sconn_port = rtc::HostToNetwork16(port);
841 sconn.sconn_addr = this;
842 return sconn;
843 }
844
845 void SctpTransport::OnPacketFromSctpToNetwork(
846 const rtc::CopyOnWriteBuffer& buffer) {
847 if (buffer.size() > (kSctpMtu)) {
848 LOG(LS_ERROR) << debug_name_ << "->OnPacketFromSctpToNetwork(...): "
849 << "SCTP seems to have made a packet that is bigger "
850 << "than its official MTU: " << buffer.size() << " vs max of "
851 << kSctpMtu;
852 }
853 TRACE_EVENT0("webrtc", "SctpTransport::OnPacketFromSctpToNetwork");
854
855 // Don't create noise by trying to send a packet when the DTLS channel isn't
856 // even writable.
857 if (!transport_channel_->writable()) {
858 return;
859 }
860
861 // Bon voyage.
862 transport_channel_->SendPacket(buffer.data<char>(), buffer.size(),
863 rtc::PacketOptions(), PF_NORMAL);
864 }
865
866 void SctpTransport::OnInboundPacketFromSctpToChannel(
867 const rtc::CopyOnWriteBuffer& buffer,
868 ReceiveDataParams params,
869 int flags) {
711 LOG(LS_VERBOSE) << debug_name_ << "->OnInboundPacketFromSctpToChannel(...): " 870 LOG(LS_VERBOSE) << debug_name_ << "->OnInboundPacketFromSctpToChannel(...): "
712 << "Received SCTP data:" 871 << "Received SCTP data:"
713 << " ssrc=" << packet->params.ssrc 872 << " ssrc=" << params.ssrc
714 << " notification: " << (packet->flags & MSG_NOTIFICATION) 873 << " notification: " << (flags & MSG_NOTIFICATION)
715 << " length=" << packet->buffer.size(); 874 << " length=" << buffer.size();
716 // Sending a packet with data == NULL (no data) is SCTPs "close the 875 // Sending a packet with data == NULL (no data) is SCTPs "close the
717 // connection" message. This sets sock_ = NULL; 876 // connection" message. This sets sock_ = NULL;
718 if (!packet->buffer.size() || !packet->buffer.data()) { 877 if (!buffer.size() || !buffer.data()) {
719 LOG(LS_INFO) << debug_name_ << "->OnInboundPacketFromSctpToChannel(...): " 878 LOG(LS_INFO) << debug_name_ << "->OnInboundPacketFromSctpToChannel(...): "
720 "No data, closing."; 879 "No data, closing.";
721 return; 880 return;
722 } 881 }
723 if (packet->flags & MSG_NOTIFICATION) { 882 if (flags & MSG_NOTIFICATION) {
724 OnNotificationFromSctp(packet->buffer); 883 OnNotificationFromSctp(buffer);
725 } else { 884 } else {
726 OnDataFromSctpToChannel(packet->params, packet->buffer); 885 OnDataFromSctpToChannel(params, buffer);
727 } 886 }
728 } 887 }
729 888
730 void SctpDataMediaChannel::OnDataFromSctpToChannel( 889 void SctpTransport::OnDataFromSctpToChannel(
731 const ReceiveDataParams& params, const rtc::CopyOnWriteBuffer& buffer) { 890 const ReceiveDataParams& params,
732 if (receiving_) { 891 const rtc::CopyOnWriteBuffer& buffer) {
733 LOG(LS_VERBOSE) << debug_name_ << "->OnDataFromSctpToChannel(...): " 892 LOG(LS_VERBOSE) << debug_name_ << "->OnDataFromSctpToChannel(...): "
734 << "Posting with length: " << buffer.size() 893 << "Posting with length: " << buffer.size() << " on stream "
735 << " on stream " << params.ssrc; 894 << params.ssrc;
736 // Reports all received messages to upper layers, no matter whether the sid 895 // Reports all received messages to upper layers, no matter whether the sid
737 // is known. 896 // is known.
738 SignalDataReceived(params, buffer.data<char>(), buffer.size()); 897 SignalDataReceived(params, buffer);
739 } else {
740 LOG(LS_WARNING) << debug_name_ << "->OnDataFromSctpToChannel(...): "
741 << "Not receiving packet with sid=" << params.ssrc
742 << " len=" << buffer.size() << " before SetReceive(true).";
743 }
744 } 898 }
745 899
746 bool SctpDataMediaChannel::AddStream(const StreamParams& stream) { 900 void SctpTransport::OnNotificationFromSctp(
747 if (!stream.has_ssrcs()) {
748 return false;
749 }
750
751 const uint32_t ssrc = stream.first_ssrc();
752 if (ssrc > kMaxSctpSid) {
753 LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
754 << "Not adding data stream '" << stream.id
755 << "' with sid=" << ssrc << " because sid is too high.";
756 return false;
757 } else if (open_streams_.find(ssrc) != open_streams_.end()) {
758 LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
759 << "Not adding data stream '" << stream.id
760 << "' with sid=" << ssrc
761 << " because stream is already open.";
762 return false;
763 } else if (queued_reset_streams_.find(ssrc) != queued_reset_streams_.end()
764 || sent_reset_streams_.find(ssrc) != sent_reset_streams_.end()) {
765 LOG(LS_WARNING) << debug_name_ << "->Add(Send|Recv)Stream(...): "
766 << "Not adding data stream '" << stream.id
767 << "' with sid=" << ssrc
768 << " because stream is still closing.";
769 return false;
770 }
771
772 open_streams_.insert(ssrc);
773 return true;
774 }
775
776 bool SctpDataMediaChannel::ResetStream(uint32_t ssrc) {
777 // We typically get this called twice for the same stream, once each for
778 // Send and Recv.
779 StreamSet::iterator found = open_streams_.find(ssrc);
780
781 if (found == open_streams_.end()) {
782 LOG(LS_VERBOSE) << debug_name_ << "->ResetStream(" << ssrc << "): "
783 << "stream not found.";
784 return false;
785 } else {
786 LOG(LS_VERBOSE) << debug_name_ << "->ResetStream(" << ssrc << "): "
787 << "Removing and queuing RE-CONFIG chunk.";
788 open_streams_.erase(found);
789 }
790
791 // SCTP won't let you have more than one stream reset pending at a time, but
792 // you can close multiple streams in a single reset. So, we keep an internal
793 // queue of streams-to-reset, and send them as one reset message in
794 // SendQueuedStreamResets().
795 queued_reset_streams_.insert(ssrc);
796
797 // Signal our stream-reset logic that it should try to send now, if it can.
798 SendQueuedStreamResets();
799
800 // The stream will actually get removed when we get the acknowledgment.
801 return true;
802 }
803
804 void SctpDataMediaChannel::OnNotificationFromSctp(
805 const rtc::CopyOnWriteBuffer& buffer) { 901 const rtc::CopyOnWriteBuffer& buffer) {
806 const sctp_notification& notification = 902 const sctp_notification& notification =
807 reinterpret_cast<const sctp_notification&>(*buffer.data()); 903 reinterpret_cast<const sctp_notification&>(*buffer.data());
808 ASSERT(notification.sn_header.sn_length == buffer.size()); 904 RTC_DCHECK(notification.sn_header.sn_length == buffer.size());
809 905
810 // TODO(ldixon): handle notifications appropriately. 906 // TODO(ldixon): handle notifications appropriately.
811 switch (notification.sn_header.sn_type) { 907 switch (notification.sn_header.sn_type) {
812 case SCTP_ASSOC_CHANGE: 908 case SCTP_ASSOC_CHANGE:
813 LOG(LS_VERBOSE) << "SCTP_ASSOC_CHANGE"; 909 LOG(LS_VERBOSE) << "SCTP_ASSOC_CHANGE";
814 OnNotificationAssocChange(notification.sn_assoc_change); 910 OnNotificationAssocChange(notification.sn_assoc_change);
815 break; 911 break;
816 case SCTP_REMOTE_ERROR: 912 case SCTP_REMOTE_ERROR:
817 LOG(LS_INFO) << "SCTP_REMOTE_ERROR"; 913 LOG(LS_INFO) << "SCTP_REMOTE_ERROR";
818 break; 914 break;
819 case SCTP_SHUTDOWN_EVENT: 915 case SCTP_SHUTDOWN_EVENT:
820 LOG(LS_INFO) << "SCTP_SHUTDOWN_EVENT"; 916 LOG(LS_INFO) << "SCTP_SHUTDOWN_EVENT";
821 break; 917 break;
822 case SCTP_ADAPTATION_INDICATION: 918 case SCTP_ADAPTATION_INDICATION:
823 LOG(LS_INFO) << "SCTP_ADAPTATION_INDICATION"; 919 LOG(LS_INFO) << "SCTP_ADAPTATION_INDICATION";
824 break; 920 break;
825 case SCTP_PARTIAL_DELIVERY_EVENT: 921 case SCTP_PARTIAL_DELIVERY_EVENT:
826 LOG(LS_INFO) << "SCTP_PARTIAL_DELIVERY_EVENT"; 922 LOG(LS_INFO) << "SCTP_PARTIAL_DELIVERY_EVENT";
827 break; 923 break;
828 case SCTP_AUTHENTICATION_EVENT: 924 case SCTP_AUTHENTICATION_EVENT:
829 LOG(LS_INFO) << "SCTP_AUTHENTICATION_EVENT"; 925 LOG(LS_INFO) << "SCTP_AUTHENTICATION_EVENT";
830 break; 926 break;
831 case SCTP_SENDER_DRY_EVENT: 927 case SCTP_SENDER_DRY_EVENT:
832 LOG(LS_VERBOSE) << "SCTP_SENDER_DRY_EVENT"; 928 LOG(LS_VERBOSE) << "SCTP_SENDER_DRY_EVENT";
833 SignalReadyToSend(true); 929 SetReadyToSendData();
834 break; 930 break;
835 // TODO(ldixon): Unblock after congestion. 931 // TODO(ldixon): Unblock after congestion.
836 case SCTP_NOTIFICATIONS_STOPPED_EVENT: 932 case SCTP_NOTIFICATIONS_STOPPED_EVENT:
837 LOG(LS_INFO) << "SCTP_NOTIFICATIONS_STOPPED_EVENT"; 933 LOG(LS_INFO) << "SCTP_NOTIFICATIONS_STOPPED_EVENT";
838 break; 934 break;
839 case SCTP_SEND_FAILED_EVENT: 935 case SCTP_SEND_FAILED_EVENT:
840 LOG(LS_INFO) << "SCTP_SEND_FAILED_EVENT"; 936 LOG(LS_INFO) << "SCTP_SEND_FAILED_EVENT";
841 break; 937 break;
842 case SCTP_STREAM_RESET_EVENT: 938 case SCTP_STREAM_RESET_EVENT:
843 OnStreamResetEvent(&notification.sn_strreset_event); 939 OnStreamResetEvent(&notification.sn_strreset_event);
844 break; 940 break;
845 case SCTP_ASSOC_RESET_EVENT: 941 case SCTP_ASSOC_RESET_EVENT:
846 LOG(LS_INFO) << "SCTP_ASSOC_RESET_EVENT"; 942 LOG(LS_INFO) << "SCTP_ASSOC_RESET_EVENT";
847 break; 943 break;
848 case SCTP_STREAM_CHANGE_EVENT: 944 case SCTP_STREAM_CHANGE_EVENT:
849 LOG(LS_INFO) << "SCTP_STREAM_CHANGE_EVENT"; 945 LOG(LS_INFO) << "SCTP_STREAM_CHANGE_EVENT";
850 // An acknowledgment we get after our stream resets have gone through, 946 // An acknowledgment we get after our stream resets have gone through,
851 // if they've failed. We log the message, but don't react -- we don't 947 // if they've failed. We log the message, but don't react -- we don't
852 // keep around the last-transmitted set of SSIDs we wanted to close for 948 // keep around the last-transmitted set of SSIDs we wanted to close for
853 // error recovery. It doesn't seem likely to occur, and if so, likely 949 // error recovery. It doesn't seem likely to occur, and if so, likely
854 // harmless within the lifetime of a single SCTP association. 950 // harmless within the lifetime of a single SCTP association.
855 break; 951 break;
856 default: 952 default:
857 LOG(LS_WARNING) << "Unknown SCTP event: " 953 LOG(LS_WARNING) << "Unknown SCTP event: "
858 << notification.sn_header.sn_type; 954 << notification.sn_header.sn_type;
859 break; 955 break;
860 } 956 }
861 } 957 }
862 958
863 void SctpDataMediaChannel::OnNotificationAssocChange( 959 void SctpTransport::OnNotificationAssocChange(const sctp_assoc_change& change) {
864 const sctp_assoc_change& change) {
865 switch (change.sac_state) { 960 switch (change.sac_state) {
866 case SCTP_COMM_UP: 961 case SCTP_COMM_UP:
867 LOG(LS_VERBOSE) << "Association change SCTP_COMM_UP"; 962 LOG(LS_VERBOSE) << "Association change SCTP_COMM_UP";
868 break; 963 break;
869 case SCTP_COMM_LOST: 964 case SCTP_COMM_LOST:
870 LOG(LS_INFO) << "Association change SCTP_COMM_LOST"; 965 LOG(LS_INFO) << "Association change SCTP_COMM_LOST";
871 break; 966 break;
872 case SCTP_RESTART: 967 case SCTP_RESTART:
873 LOG(LS_INFO) << "Association change SCTP_RESTART"; 968 LOG(LS_INFO) << "Association change SCTP_RESTART";
874 break; 969 break;
875 case SCTP_SHUTDOWN_COMP: 970 case SCTP_SHUTDOWN_COMP:
876 LOG(LS_INFO) << "Association change SCTP_SHUTDOWN_COMP"; 971 LOG(LS_INFO) << "Association change SCTP_SHUTDOWN_COMP";
877 break; 972 break;
878 case SCTP_CANT_STR_ASSOC: 973 case SCTP_CANT_STR_ASSOC:
879 LOG(LS_INFO) << "Association change SCTP_CANT_STR_ASSOC"; 974 LOG(LS_INFO) << "Association change SCTP_CANT_STR_ASSOC";
880 break; 975 break;
881 default: 976 default:
882 LOG(LS_INFO) << "Association change UNKNOWN"; 977 LOG(LS_INFO) << "Association change UNKNOWN";
883 break; 978 break;
884 } 979 }
885 } 980 }
886 981
887 void SctpDataMediaChannel::OnStreamResetEvent( 982 void SctpTransport::OnStreamResetEvent(
888 const struct sctp_stream_reset_event* evt) { 983 const struct sctp_stream_reset_event* evt) {
889 // A stream reset always involves two RE-CONFIG chunks for us -- we always 984 // A stream reset always involves two RE-CONFIG chunks for us -- we always
890 // simultaneously reset a sid's sequence number in both directions. The 985 // simultaneously reset a sid's sequence number in both directions. The
891 // requesting side transmits a RE-CONFIG chunk and waits for the peer to send 986 // requesting side transmits a RE-CONFIG chunk and waits for the peer to send
892 // one back. Both sides get this SCTP_STREAM_RESET_EVENT when they receive 987 // one back. Both sides get this SCTP_STREAM_RESET_EVENT when they receive
893 // RE-CONFIGs. 988 // RE-CONFIGs.
894 const int num_ssrcs = (evt->strreset_length - sizeof(*evt)) / 989 const int num_ssrcs = (evt->strreset_length - sizeof(*evt)) /
895 sizeof(evt->strreset_stream_list[0]); 990 sizeof(evt->strreset_stream_list[0]);
896 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_ 991 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
897 << "): Flags = 0x" 992 << "): Flags = 0x" << std::hex << evt->strreset_flags << " ("
898 << std::hex << evt->strreset_flags << " ("
899 << ListFlags(evt->strreset_flags) << ")"; 993 << ListFlags(evt->strreset_flags) << ")";
900 LOG(LS_VERBOSE) << "Assoc = " << evt->strreset_assoc_id << ", Streams = [" 994 LOG(LS_VERBOSE) << "Assoc = " << evt->strreset_assoc_id << ", Streams = ["
901 << ListArray(evt->strreset_stream_list, num_ssrcs) 995 << ListArray(evt->strreset_stream_list, num_ssrcs)
902 << "], Open: [" 996 << "], Open: [" << ListStreams(open_streams_) << "], Q'd: ["
903 << ListStreams(open_streams_) << "], Q'd: ["
904 << ListStreams(queued_reset_streams_) << "], Sent: [" 997 << ListStreams(queued_reset_streams_) << "], Sent: ["
905 << ListStreams(sent_reset_streams_) << "]"; 998 << ListStreams(sent_reset_streams_) << "]";
906 999
907 // If both sides try to reset some streams at the same time (even if they're 1000 // If both sides try to reset some streams at the same time (even if they're
908 // disjoint sets), we can get reset failures. 1001 // disjoint sets), we can get reset failures.
909 if (evt->strreset_flags & SCTP_STREAM_RESET_FAILED) { 1002 if (evt->strreset_flags & SCTP_STREAM_RESET_FAILED) {
910 // OK, just try again. The stream IDs sent over when the RESET_FAILED flag 1003 // OK, just try again. The stream IDs sent over when the RESET_FAILED flag
911 // is set seem to be garbage values. Ignore them. 1004 // is set seem to be garbage values. Ignore them.
912 queued_reset_streams_.insert( 1005 queued_reset_streams_.insert(sent_reset_streams_.begin(),
913 sent_reset_streams_.begin(), 1006 sent_reset_streams_.end());
914 sent_reset_streams_.end());
915 sent_reset_streams_.clear(); 1007 sent_reset_streams_.clear();
916 1008
917 } else if (evt->strreset_flags & SCTP_STREAM_RESET_INCOMING_SSN) { 1009 } else if (evt->strreset_flags & SCTP_STREAM_RESET_INCOMING_SSN) {
918 // Each side gets an event for each direction of a stream. That is, 1010 // Each side gets an event for each direction of a stream. That is,
919 // closing sid k will make each side receive INCOMING and OUTGOING reset 1011 // closing sid k will make each side receive INCOMING and OUTGOING reset
920 // events for k. As per RFC6525, Section 5, paragraph 2, each side will 1012 // events for k. As per RFC6525, Section 5, paragraph 2, each side will
921 // get an INCOMING event first. 1013 // get an INCOMING event first.
922 for (int i = 0; i < num_ssrcs; i++) { 1014 for (int i = 0; i < num_ssrcs; i++) {
923 const int stream_id = evt->strreset_stream_list[i]; 1015 const int stream_id = evt->strreset_stream_list[i];
924 1016
925 // See if this stream ID was closed by our peer or ourselves. 1017 // See if this stream ID was closed by our peer or ourselves.
926 StreamSet::iterator it = sent_reset_streams_.find(stream_id); 1018 StreamSet::iterator it = sent_reset_streams_.find(stream_id);
927 1019
928 // The reset was requested locally. 1020 // The reset was requested locally.
929 if (it != sent_reset_streams_.end()) { 1021 if (it != sent_reset_streams_.end()) {
930 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_ 1022 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
931 << "): local sid " << stream_id << " acknowledged."; 1023 << "): local sid " << stream_id << " acknowledged.";
932 sent_reset_streams_.erase(it); 1024 sent_reset_streams_.erase(it);
933 1025
934 } else if ((it = open_streams_.find(stream_id)) 1026 } else if ((it = open_streams_.find(stream_id)) != open_streams_.end()) {
935 != open_streams_.end()) {
936 // The peer requested the reset. 1027 // The peer requested the reset.
937 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_ 1028 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
938 << "): closing sid " << stream_id; 1029 << "): closing sid " << stream_id;
939 open_streams_.erase(it); 1030 open_streams_.erase(it);
940 SignalStreamClosedRemotely(stream_id); 1031 SignalStreamClosedRemotely(stream_id);
941 1032
942 } else if ((it = queued_reset_streams_.find(stream_id)) 1033 } else if ((it = queued_reset_streams_.find(stream_id)) !=
943 != queued_reset_streams_.end()) { 1034 queued_reset_streams_.end()) {
944 // The peer requested the reset, but there was a local reset 1035 // The peer requested the reset, but there was a local reset
945 // queued. 1036 // queued.
946 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_ 1037 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
947 << "): double-sided close for sid " << stream_id; 1038 << "): double-sided close for sid " << stream_id;
948 // Both sides want the stream closed, and the peer got to send the 1039 // Both sides want the stream closed, and the peer got to send the
949 // RE-CONFIG first. Treat it like the local Remove(Send|Recv)Stream 1040 // RE-CONFIG first. Treat it like the local Remove(Send|Recv)Stream
950 // finished quickly. 1041 // finished quickly.
951 queued_reset_streams_.erase(it); 1042 queued_reset_streams_.erase(it);
952 1043
953 } else { 1044 } else {
954 // This stream is unknown. Sometimes this can be from an 1045 // This stream is unknown. Sometimes this can be from an
955 // RESET_FAILED-related retransmit. 1046 // RESET_FAILED-related retransmit.
956 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_ 1047 LOG(LS_VERBOSE) << "SCTP_STREAM_RESET_EVENT(" << debug_name_
957 << "): Unknown sid " << stream_id; 1048 << "): Unknown sid " << stream_id;
958 } 1049 }
959 } 1050 }
960 } 1051 }
961 1052
962 // Always try to send the queued RESET because this call indicates that the 1053 // Always try to send the queued RESET because this call indicates that the
963 // last local RESET or remote RESET has made some progress. 1054 // last local RESET or remote RESET has made some progress.
964 SendQueuedStreamResets(); 1055 SendQueuedStreamResets();
965 } 1056 }
966 1057
967 // Puts the specified |param| from the codec identified by |id| into |dest|
968 // and returns true. Or returns false if it wasn't there, leaving |dest|
969 // untouched.
970 static bool GetCodecIntParameter(const std::vector<DataCodec>& codecs,
971 int id, const std::string& name,
972 const std::string& param, int* dest) {
973 std::string value;
974 DataCodec match_pattern(id, name);
975 for (size_t i = 0; i < codecs.size(); ++i) {
976 if (codecs[i].Matches(match_pattern)) {
977 if (codecs[i].GetParam(param, &value)) {
978 *dest = rtc::FromString<int>(value);
979 return true;
980 }
981 }
982 }
983 return false;
984 }
985
986 bool SctpDataMediaChannel::SetSendCodecs(const std::vector<DataCodec>& codecs) {
987 return GetCodecIntParameter(
988 codecs, kGoogleSctpDataCodecPlType, kGoogleSctpDataCodecName,
989 kCodecParamPort, &remote_port_);
990 }
991
992 bool SctpDataMediaChannel::SetRecvCodecs(const std::vector<DataCodec>& codecs) {
993 return GetCodecIntParameter(
994 codecs, kGoogleSctpDataCodecPlType, kGoogleSctpDataCodecName,
995 kCodecParamPort, &local_port_);
996 }
997
998 void SctpDataMediaChannel::OnPacketFromSctpToNetwork(
999 rtc::CopyOnWriteBuffer* buffer) {
1000 if (buffer->size() > (kSctpMtu)) {
1001 LOG(LS_ERROR) << debug_name_ << "->OnPacketFromSctpToNetwork(...): "
1002 << "SCTP seems to have made a packet that is bigger "
1003 << "than its official MTU: " << buffer->size()
1004 << " vs max of " << kSctpMtu;
1005 }
1006 MediaChannel::SendPacket(buffer, rtc::PacketOptions());
1007 }
1008
1009 bool SctpDataMediaChannel::SendQueuedStreamResets() {
1010 if (!sent_reset_streams_.empty() || queued_reset_streams_.empty()) {
1011 return true;
1012 }
1013
1014 LOG(LS_VERBOSE) << "SendQueuedStreamResets[" << debug_name_ << "]: Sending ["
1015 << ListStreams(queued_reset_streams_) << "], Open: ["
1016 << ListStreams(open_streams_) << "], Sent: ["
1017 << ListStreams(sent_reset_streams_) << "]";
1018
1019 const size_t num_streams = queued_reset_streams_.size();
1020 const size_t num_bytes =
1021 sizeof(struct sctp_reset_streams) + (num_streams * sizeof(uint16_t));
1022
1023 std::vector<uint8_t> reset_stream_buf(num_bytes, 0);
1024 struct sctp_reset_streams* resetp = reinterpret_cast<sctp_reset_streams*>(
1025 &reset_stream_buf[0]);
1026 resetp->srs_assoc_id = SCTP_ALL_ASSOC;
1027 resetp->srs_flags = SCTP_STREAM_RESET_INCOMING | SCTP_STREAM_RESET_OUTGOING;
1028 resetp->srs_number_streams = rtc::checked_cast<uint16_t>(num_streams);
1029 int result_idx = 0;
1030 for (StreamSet::iterator it = queued_reset_streams_.begin();
1031 it != queued_reset_streams_.end(); ++it) {
1032 resetp->srs_stream_list[result_idx++] = *it;
1033 }
1034
1035 int ret = usrsctp_setsockopt(
1036 sock_, IPPROTO_SCTP, SCTP_RESET_STREAMS, resetp,
1037 rtc::checked_cast<socklen_t>(reset_stream_buf.size()));
1038 if (ret < 0) {
1039 LOG_ERRNO(LS_ERROR) << debug_name_ << "Failed to send a stream reset for "
1040 << num_streams << " streams";
1041 return false;
1042 }
1043
1044 // sent_reset_streams_ is empty, and all the queued_reset_streams_ go into
1045 // it now.
1046 queued_reset_streams_.swap(sent_reset_streams_);
1047 return true;
1048 }
1049
1050 void SctpDataMediaChannel::OnMessage(rtc::Message* msg) {
1051 switch (msg->message_id) {
1052 case MSG_SCTPINBOUNDPACKET: {
1053 std::unique_ptr<InboundPacketMessage> pdata(
1054 static_cast<InboundPacketMessage*>(msg->pdata));
1055 OnInboundPacketFromSctpToChannel(pdata->data().get());
1056 break;
1057 }
1058 case MSG_SCTPOUTBOUNDPACKET: {
1059 std::unique_ptr<OutboundPacketMessage> pdata(
1060 static_cast<OutboundPacketMessage*>(msg->pdata));
1061 OnPacketFromSctpToNetwork(pdata->data().get());
1062 break;
1063 }
1064 }
1065 }
1066 } // namespace cricket 1058 } // namespace cricket
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698