diff options
Diffstat (limited to 'netwerk/sctp/datachannel/DataChannel.cpp')
| -rw-r--r-- | netwerk/sctp/datachannel/DataChannel.cpp | 3548 |
1 files changed, 3548 insertions, 0 deletions
diff --git a/netwerk/sctp/datachannel/DataChannel.cpp b/netwerk/sctp/datachannel/DataChannel.cpp new file mode 100644 index 0000000000..8af9a558b8 --- /dev/null +++ b/netwerk/sctp/datachannel/DataChannel.cpp @@ -0,0 +1,3548 @@ +/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */ +/* vim: set ts=2 et sw=2 tw=80: */ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this file, + * You can obtain one at http://mozilla.org/MPL/2.0/. */ + +#include <algorithm> +#include <stdio.h> +#include <stdlib.h> +#if !defined(__Userspace_os_Windows) +# include <arpa/inet.h> +#endif +// usrsctp.h expects to have errno definitions prior to its inclusion. +#include <errno.h> + +#define SCTP_DEBUG 1 +#define SCTP_STDINT_INCLUDE <stdint.h> + +#ifdef _MSC_VER +// Disable "warning C4200: nonstandard extension used : zero-sized array in +// struct/union" +// ...which the third-party file usrsctp.h runs afoul of. +# pragma warning(push) +# pragma warning(disable : 4200) +#endif + +#include "usrsctp.h" + +#ifdef _MSC_VER +# pragma warning(pop) +#endif + +#include "nsServiceManagerUtils.h" +#include "nsIInputStream.h" +#include "nsIPrefBranch.h" +#include "nsIPrefService.h" +#include "mozilla/Services.h" +#include "mozilla/Sprintf.h" +#include "nsProxyRelease.h" +#include "nsThread.h" +#include "nsThreadUtils.h" +#include "nsNetUtil.h" +#include "nsNetCID.h" +#include "mozilla/RandomNum.h" +#include "mozilla/StaticMutex.h" +#include "mozilla/UniquePtrExtensions.h" +#include "mozilla/Unused.h" +#include "mozilla/dom/RTCDataChannelBinding.h" +#include "mozilla/dom/RTCStatsReportBinding.h" +#include "mozilla/media/MediaUtils.h" +#ifdef MOZ_PEERCONNECTION +# include "transport/runnable_utils.h" +# include "jsapi/MediaTransportHandler.h" +# include "mediapacket.h" +#endif + +#include "DataChannel.h" +#include "DataChannelProtocol.h" + +// Let us turn on and off important assertions in non-debug builds +#ifdef DEBUG +# define ASSERT_WEBRTC(x) MOZ_ASSERT((x)) +#elif defined(MOZ_WEBRTC_ASSERT_ALWAYS) +# define ASSERT_WEBRTC(x) \ + do { \ + if (!(x)) { \ + MOZ_CRASH(); \ + } \ + } while (0) +#endif + +namespace mozilla { + +LazyLogModule gDataChannelLog("DataChannel"); +static LazyLogModule gSCTPLog("SCTP"); + +#define SCTP_LOG(args) \ + MOZ_LOG(mozilla::gSCTPLog, mozilla::LogLevel::Debug, args) + +static void debug_printf(const char* format, ...) { + va_list ap; + char buffer[1024]; + + if (MOZ_LOG_TEST(gSCTPLog, LogLevel::Debug)) { + va_start(ap, format); +#ifdef _WIN32 + if (vsnprintf_s(buffer, sizeof(buffer), _TRUNCATE, format, ap) > 0) { +#else + if (VsprintfLiteral(buffer, format, ap) > 0) { +#endif + SCTP_LOG(("%s", buffer)); + } + va_end(ap); + } +} + +static constexpr const char* ToString(DataChannelState state) { + switch (state) { + case DataChannelState::Connecting: + return "CONNECTING"; + case DataChannelState::Open: + return "OPEN"; + case DataChannelState::Closing: + return "CLOSING"; + case DataChannelState::Closed: + return "CLOSED"; + } + return ""; +}; + +static constexpr const char* ToString(DataChannelConnectionState state) { + switch (state) { + case DataChannelConnectionState::Connecting: + return "CONNECTING"; + case DataChannelConnectionState::Open: + return "OPEN"; + case DataChannelConnectionState::Closed: + return "CLOSED"; + } + return ""; +}; + +static constexpr const char* ToString( + DataChannelOnMessageAvailable::EventType type) { + switch (type) { + case DataChannelOnMessageAvailable::EventType::OnConnection: + return "ON_CONNECTION"; + case DataChannelOnMessageAvailable::EventType::OnDisconnected: + return "ON_DISCONNECTED"; + case DataChannelOnMessageAvailable::EventType::OnChannelCreated: + return "ON_CHANNEL_CREATED"; + case DataChannelOnMessageAvailable::EventType::OnDataString: + return "ON_DATA_STRING"; + case DataChannelOnMessageAvailable::EventType::OnDataBinary: + return "ON_DATA_BINARY"; + } + return ""; +}; + +static constexpr const char* ToString(DataChannelConnection::PendingType type) { + switch (type) { + case DataChannelConnection::PendingType::None: + return "NONE"; + case DataChannelConnection::PendingType::Dcep: + return "DCEP"; + case DataChannelConnection::PendingType::Data: + return "DATA"; + } + return ""; +}; + +static constexpr const char* ToString(DataChannelReliabilityPolicy type) { + switch (type) { + case DataChannelReliabilityPolicy::Reliable: + return "RELIABLE"; + case DataChannelReliabilityPolicy::LimitedRetransmissions: + return "LIMITED_RETRANSMISSIONS"; + case DataChannelReliabilityPolicy::LimitedLifetime: + return "LIMITED_LIFETIME"; + } + return ""; +}; + +static constexpr uint16_t ToUsrsctpValue(DataChannelReliabilityPolicy type) { + switch (type) { + case DataChannelReliabilityPolicy::Reliable: + return SCTP_PR_SCTP_NONE; + case DataChannelReliabilityPolicy::LimitedRetransmissions: + return SCTP_PR_SCTP_RTX; + case DataChannelReliabilityPolicy::LimitedLifetime: + return SCTP_PR_SCTP_TTL; + } + return SCTP_PR_SCTP_NONE; +}; + +class DataChannelRegistry { + public: + NS_INLINE_DECL_THREADSAFE_REFCOUNTING(DataChannelRegistry) + + static uintptr_t Register(DataChannelConnection* aConnection) { + StaticMutexAutoLock lock(sInstanceMutex); + uintptr_t result = EnsureInstance()->RegisterImpl(aConnection); + DC_DEBUG( + ("Registering connection %p as ulp %p", aConnection, (void*)result)); + return result; + } + + static void Deregister(uintptr_t aId) { + RefPtr<DataChannelRegistry> maybeTrash; + + { + StaticMutexAutoLock lock(sInstanceMutex); + DC_DEBUG(("Deregistering connection ulp = %p", (void*)aId)); + if (NS_WARN_IF(!Instance())) { + return; + } + Instance()->DeregisterImpl(aId); + if (Instance()->Empty()) { + // Unset singleton inside mutex lock, but don't call Shutdown until we + // unlock, since that involves calling into libusrsctp, which invites + // deadlock. + maybeTrash = Instance().forget(); + } + } + } + + static RefPtr<DataChannelConnection> Lookup(uintptr_t aId) { + StaticMutexAutoLock lock(sInstanceMutex); + if (NS_WARN_IF(!Instance())) { + return nullptr; + } + return Instance()->LookupImpl(aId); + } + + private: + // This is a singleton class, so don't let just anyone create one of these + DataChannelRegistry() { + ASSERT_WEBRTC(NS_IsMainThread()); + mShutdownBlocker = media::ShutdownBlockingTicket::Create( + u"DataChannelRegistry::mShutdownBlocker"_ns, + NS_LITERAL_STRING_FROM_CSTRING(__FILE__), __LINE__); + InitUsrSctp(); + } + + static RefPtr<DataChannelRegistry>& Instance() { + static RefPtr<DataChannelRegistry> sRegistry; + return sRegistry; + } + + static RefPtr<DataChannelRegistry>& EnsureInstance() { + ASSERT_WEBRTC(NS_IsMainThread()); + if (!Instance()) { + Instance() = new DataChannelRegistry(); + } + return Instance(); + } + + uintptr_t RegisterImpl(DataChannelConnection* aConnection) { + ASSERT_WEBRTC(NS_IsMainThread()); + mConnections.emplace(mNextId, aConnection); + return mNextId++; + } + + void DeregisterImpl(uintptr_t aId) { + ASSERT_WEBRTC(NS_IsMainThread()); + mConnections.erase(aId); + } + + bool Empty() const { return mConnections.empty(); } + + RefPtr<DataChannelConnection> LookupImpl(uintptr_t aId) { + auto it = mConnections.find(aId); + if (NS_WARN_IF(it == mConnections.end())) { + DC_DEBUG(("Can't find connection ulp %p", (void*)aId)); + return nullptr; + } + return it->second; + } + + virtual ~DataChannelRegistry() { + ASSERT_WEBRTC(NS_IsMainThread()); + + if (NS_WARN_IF(!mConnections.empty())) { + MOZ_ASSERT(false); + mConnections.clear(); + } + + DeinitUsrSctp(); + } + +#ifdef SCTP_DTLS_SUPPORTED + static int SctpDtlsOutput(void* addr, void* buffer, size_t length, + uint8_t tos, uint8_t set_df) { + uintptr_t id = reinterpret_cast<uintptr_t>(addr); + RefPtr<DataChannelConnection> connection = DataChannelRegistry::Lookup(id); + if (NS_WARN_IF(!connection) || connection->InShutdown()) { + return 0; + } + return connection->SctpDtlsOutput(addr, buffer, length, tos, set_df); + } +#endif + + void InitUsrSctp() { +#ifndef MOZ_PEERCONNECTION + MOZ_CRASH("Trying to use SCTP/DTLS without dom/media/webrtc/transport"); +#endif + + DC_DEBUG(("Calling usrsctp_init %p", this)); + + usrsctp_init(0, DataChannelRegistry::SctpDtlsOutput, debug_printf); + + // Set logging to SCTP:LogLevel::Debug to get SCTP debugs + if (MOZ_LOG_TEST(gSCTPLog, LogLevel::Debug)) { + usrsctp_sysctl_set_sctp_debug_on(SCTP_DEBUG_ALL); + } + + // Do not send ABORTs in response to INITs (1). + // Do not send ABORTs for received Out of the Blue packets (2). + usrsctp_sysctl_set_sctp_blackhole(2); + + // Disable the Explicit Congestion Notification extension (currently not + // supported by the Firefox code) + usrsctp_sysctl_set_sctp_ecn_enable(0); + + // Enable interleaving messages for different streams (incoming) + // See: https://tools.ietf.org/html/rfc6458#section-8.1.20 + usrsctp_sysctl_set_sctp_default_frag_interleave(2); + + // Disabling authentication and dynamic address reconfiguration as neither + // of them are used for data channel and only result in additional code + // paths being used. + usrsctp_sysctl_set_sctp_asconf_enable(0); + usrsctp_sysctl_set_sctp_auth_enable(0); + } + + void DeinitUsrSctp() { + DC_DEBUG(("Calling usrsctp_finish %p", this)); + usrsctp_finish(); + } + + uintptr_t mNextId = 1; + std::map<uintptr_t, RefPtr<DataChannelConnection>> mConnections; + UniquePtr<media::ShutdownBlockingTicket> mShutdownBlocker; + static StaticMutex sInstanceMutex MOZ_UNANNOTATED; +}; + +StaticMutex DataChannelRegistry::sInstanceMutex; + +OutgoingMsg::OutgoingMsg(struct sctp_sendv_spa& info, const uint8_t* data, + size_t length) + : mLength(length), mData(data) { + mInfo = &info; + mPos = 0; +} + +void OutgoingMsg::Advance(size_t offset) { + mPos += offset; + if (mPos > mLength) { + mPos = mLength; + } +} + +BufferedOutgoingMsg::BufferedOutgoingMsg(OutgoingMsg& msg) { + size_t length = msg.GetLeft(); + auto* tmp = new uint8_t[length]; // infallible malloc! + memcpy(tmp, msg.GetData(), length); + mLength = length; + mData = tmp; + mInfo = new sctp_sendv_spa; + *mInfo = msg.GetInfo(); + mPos = 0; +} + +BufferedOutgoingMsg::~BufferedOutgoingMsg() { + delete mInfo; + delete[] mData; +} + +static int receive_cb(struct socket* sock, union sctp_sockstore addr, + void* data, size_t datalen, struct sctp_rcvinfo rcv, + int flags, void* ulp_info) { + DC_DEBUG(("In receive_cb, ulp_info=%p", ulp_info)); + uintptr_t id = reinterpret_cast<uintptr_t>(ulp_info); + RefPtr<DataChannelConnection> connection = DataChannelRegistry::Lookup(id); + if (!connection) { + // Unfortunately, we can get callbacks after calling + // usrsctp_close(socket), so we need to simply ignore them if we've + // already killed the DataChannelConnection object + DC_DEBUG(( + "Ignoring receive callback for terminated Connection ulp=%p, %zu bytes", + ulp_info, datalen)); + return 0; + } + return connection->ReceiveCallback(sock, data, datalen, rcv, flags); +} + +static RefPtr<DataChannelConnection> GetConnectionFromSocket( + 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]); + uintptr_t id = reinterpret_cast<uintptr_t>(sconn->sconn_addr); + RefPtr<DataChannelConnection> connection = DataChannelRegistry::Lookup(id); + usrsctp_freeladdrs(addrs); + + return connection; +} + +// Called when the buffer empties to the threshold value. This is called +// from SctpDtlsInput() through the sctp stack. SctpDtlsInput() calls +// usrsctp_conninput() under lock +int DataChannelConnection::OnThresholdEvent(struct socket* sock, + uint32_t sb_free, void* ulp_info) { + RefPtr<DataChannelConnection> connection = GetConnectionFromSocket(sock); + connection->mLock.AssertCurrentThreadOwns(); + if (connection) { + connection->SendDeferredMessages(); + } else { + DC_ERROR(("Can't find connection for socket %p", sock)); + } + return 0; +} + +DataChannelConnection::~DataChannelConnection() { + DC_DEBUG(("Deleting DataChannelConnection %p", (void*)this)); + // This may die on the MainThread, or on the STS thread, or on an + // sctp thread if we were in a callback when the DOM side shut things down. + ASSERT_WEBRTC(mState == DataChannelConnectionState::Closed); + MOZ_ASSERT(!mMasterSocket); + MOZ_ASSERT(mPending.GetSize() == 0); + + if (!IsSTSThread()) { + // We may be on MainThread *or* on an sctp thread (being called from + // receive_cb() or SctpDtlsOutput()) + if (mInternalIOThread) { + // Avoid spinning the event thread from here (which if we're mainthread + // is in the event loop already) + nsCOMPtr<nsIRunnable> r = WrapRunnable( + nsCOMPtr<nsIThread>(mInternalIOThread), &nsIThread::AsyncShutdown); + Dispatch(r.forget()); + } + } else { + // on STS, safe to call shutdown + if (mInternalIOThread) { + mInternalIOThread->Shutdown(); + } + } +} + +void DataChannelConnection::Destroy() { + // Though it's probably ok to do this and close the sockets; + // if we really want it to do true clean shutdowns it can + // create a dependant Internal object that would remain around + // until the network shut down the association or timed out. + DC_DEBUG(("Destroying DataChannelConnection %p", (void*)this)); + ASSERT_WEBRTC(NS_IsMainThread()); + CloseAll(); + + MutexAutoLock lock(mLock); + // If we had a pending reset, we aren't waiting for it - clear the list so + // we can deregister this DataChannelConnection without leaking. + ClearResets(); + + MOZ_ASSERT(mSTS); + ASSERT_WEBRTC(NS_IsMainThread()); + mListener = nullptr; + // Finish Destroy on STS thread to avoid bug 876167 - once that's fixed, + // the usrsctp_close() calls can move back here (and just proxy the + // disconnect_all()) + RUN_ON_THREAD(mSTS, + WrapRunnable(RefPtr<DataChannelConnection>(this), + &DataChannelConnection::DestroyOnSTS, mSocket, + mMasterSocket), + NS_DISPATCH_NORMAL); + + // These will be released on STS + mSocket = nullptr; + mMasterSocket = nullptr; // also a flag that we've Destroyed this connection + + // We can't get any more *new* callbacks from the SCTP library + + // All existing callbacks have refs to DataChannelConnection - however, + // we need to handle their destroying the object off mainthread/STS + + // nsDOMDataChannel objects have refs to DataChannels that have refs to us +} + +void DataChannelConnection::DestroyOnSTS(struct socket* aMasterSocket, + struct socket* aSocket) { + if (aSocket && aSocket != aMasterSocket) usrsctp_close(aSocket); + if (aMasterSocket) usrsctp_close(aMasterSocket); + + usrsctp_deregister_address(reinterpret_cast<void*>(mId)); + DC_DEBUG( + ("Deregistered %p from the SCTP stack.", reinterpret_cast<void*>(mId))); +#ifdef MOZ_DIAGNOSTIC_ASSERT_ENABLED + mShutdown = true; + DC_DEBUG(("Shutting down connection %p, id %p", this, (void*)mId)); +#endif + + disconnect_all(); + mTransportHandler = nullptr; + GetMainThreadSerialEventTarget()->Dispatch(NS_NewRunnableFunction( + "DataChannelConnection::Destroy", + [id = mId]() { DataChannelRegistry::Deregister(id); })); +} + +Maybe<RefPtr<DataChannelConnection>> DataChannelConnection::Create( + DataChannelConnection::DataConnectionListener* aListener, + nsISerialEventTarget* aTarget, MediaTransportHandler* aHandler, + const uint16_t aLocalPort, const uint16_t aNumStreams, + const Maybe<uint64_t>& aMaxMessageSize) { + ASSERT_WEBRTC(NS_IsMainThread()); + + RefPtr<DataChannelConnection> connection = new DataChannelConnection( + aListener, aTarget, aHandler); // Walks into a bar + return connection->Init(aLocalPort, aNumStreams, aMaxMessageSize) + ? Some(connection) + : Nothing(); +} + +DataChannelConnection::DataChannelConnection( + DataChannelConnection::DataConnectionListener* aListener, + nsISerialEventTarget* aTarget, MediaTransportHandler* aHandler) + : NeckoTargetHolder(aTarget), + mLock("netwerk::sctp::DataChannelConnection"), + mListener(aListener), + mTransportHandler(aHandler) { + DC_VERBOSE(("Constructor DataChannelConnection=%p, listener=%p", this, + mListener.get())); +#ifdef MOZ_DIAGNOSTIC_ASSERT_ENABLED + mShutdown = false; +#endif +} + +bool DataChannelConnection::Init(const uint16_t aLocalPort, + const uint16_t aNumStreams, + const Maybe<uint64_t>& aMaxMessageSize) { + ASSERT_WEBRTC(NS_IsMainThread()); + + struct sctp_initmsg initmsg = {}; + struct sctp_assoc_value av = {}; + struct sctp_event event = {}; + socklen_t len; + + uint16_t event_types[] = { + SCTP_ASSOC_CHANGE, SCTP_PEER_ADDR_CHANGE, + SCTP_REMOTE_ERROR, SCTP_SHUTDOWN_EVENT, + SCTP_ADAPTATION_INDICATION, SCTP_PARTIAL_DELIVERY_EVENT, + SCTP_SEND_FAILED_EVENT, SCTP_STREAM_RESET_EVENT, + SCTP_STREAM_CHANGE_EVENT}; + { + // MutexAutoLock lock(mLock); Not needed since we're on mainthread always + mLocalPort = aLocalPort; + SetMaxMessageSize(aMaxMessageSize.isSome(), aMaxMessageSize.valueOr(0)); + } + + mId = DataChannelRegistry::Register(this); + + // XXX FIX! make this a global we get once + // Find the STS thread + nsresult rv; + mSTS = do_GetService(NS_SOCKETTRANSPORTSERVICE_CONTRACTID, &rv); + MOZ_ASSERT(NS_SUCCEEDED(rv)); + + // Open sctp with a callback + if ((mMasterSocket = + usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP, receive_cb, + &DataChannelConnection::OnThresholdEvent, + usrsctp_sysctl_get_sctp_sendspace() / 2, + reinterpret_cast<void*>(mId))) == nullptr) { + return false; + } + + int buf_size = 1024 * 1024; + + if (usrsctp_setsockopt(mMasterSocket, SOL_SOCKET, SO_RCVBUF, + (const void*)&buf_size, sizeof(buf_size)) < 0) { + DC_ERROR(("Couldn't change receive buffer size on SCTP socket")); + goto error_cleanup; + } + if (usrsctp_setsockopt(mMasterSocket, SOL_SOCKET, SO_SNDBUF, + (const void*)&buf_size, sizeof(buf_size)) < 0) { + DC_ERROR(("Couldn't change send buffer size on SCTP socket")); + goto error_cleanup; + } + + // Make non-blocking for bind/connect. SCTP over UDP defaults to non-blocking + // in associations for normal IO + if (usrsctp_set_non_blocking(mMasterSocket, 1) < 0) { + DC_ERROR(("Couldn't set non_blocking on SCTP socket")); + // We can't handle connect() safely if it will block, not that this will + // even happen. + goto error_cleanup; + } + + // Make sure when we close the socket, make sure it doesn't call us back + // again! This would cause it try to use an invalid DataChannelConnection + // pointer + struct linger l; + l.l_onoff = 1; + l.l_linger = 0; + if (usrsctp_setsockopt(mMasterSocket, SOL_SOCKET, SO_LINGER, (const void*)&l, + (socklen_t)sizeof(struct linger)) < 0) { + DC_ERROR(("Couldn't set SO_LINGER on SCTP socket")); + // unsafe to allow it to continue if this fails + goto error_cleanup; + } + + // XXX Consider disabling this when we add proper SDP negotiation. + // We may want to leave enabled for supporting 'cloning' of SDP offers, which + // implies re-use of the same pseudo-port number, or forcing a renegotiation. + { + const int option_value = 1; + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_REUSE_PORT, + (const void*)&option_value, + (socklen_t)sizeof(option_value)) < 0) { + DC_WARN(("Couldn't set SCTP_REUSE_PORT on SCTP socket")); + } + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_NODELAY, + (const void*)&option_value, + (socklen_t)sizeof(option_value)) < 0) { + DC_WARN(("Couldn't set SCTP_NODELAY on SCTP socket")); + } + } + + // Set explicit EOR + { + const int option_value = 1; + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_EXPLICIT_EOR, + (const void*)&option_value, + (socklen_t)sizeof(option_value)) < 0) { + DC_ERROR(("*** failed to enable explicit EOR mode %d", errno)); + goto error_cleanup; + } + } + + // Enable ndata + // TODO: Bug 1381145, enable this once ndata has been deployed +#if 0 + av.assoc_id = SCTP_FUTURE_ASSOC; + av.assoc_value = 1; + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_INTERLEAVING_SUPPORTED, &av, + (socklen_t)sizeof(struct sctp_assoc_value)) < 0) { + DC_ERROR(("*** failed enable ndata errno %d", errno)); + goto error_cleanup; + } +#endif + + av.assoc_id = SCTP_ALL_ASSOC; + av.assoc_value = SCTP_ENABLE_RESET_STREAM_REQ | SCTP_ENABLE_CHANGE_ASSOC_REQ; + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_ENABLE_STREAM_RESET, + &av, (socklen_t)sizeof(struct sctp_assoc_value)) < 0) { + DC_ERROR(("*** failed enable stream reset errno %d", errno)); + goto error_cleanup; + } + + /* Enable the events of interest. */ + event.se_assoc_id = SCTP_ALL_ASSOC; + event.se_on = 1; + for (unsigned short event_type : event_types) { + event.se_type = event_type; + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_EVENT, &event, + sizeof(event)) < 0) { + DC_ERROR(("*** failed setsockopt SCTP_EVENT errno %d", errno)); + goto error_cleanup; + } + } + + len = sizeof(initmsg); + if (usrsctp_getsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_INITMSG, &initmsg, + &len) < 0) { + DC_ERROR(("*** failed getsockopt SCTP_INITMSG")); + goto error_cleanup; + } + DC_DEBUG(("Setting number of SCTP streams to %u, was %u/%u", aNumStreams, + initmsg.sinit_num_ostreams, initmsg.sinit_max_instreams)); + initmsg.sinit_num_ostreams = aNumStreams; + initmsg.sinit_max_instreams = MAX_NUM_STREAMS; + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_INITMSG, &initmsg, + (socklen_t)sizeof(initmsg)) < 0) { + DC_ERROR(("*** failed setsockopt SCTP_INITMSG, errno %d", errno)); + goto error_cleanup; + } + + mSocket = nullptr; + mSTS->Dispatch( + NS_NewRunnableFunction("DataChannelConnection::Init", [id = mId]() { + usrsctp_register_address(reinterpret_cast<void*>(id)); + DC_DEBUG(("Registered %p within the SCTP stack.", + reinterpret_cast<void*>(id))); + })); + + return true; + +error_cleanup: + usrsctp_close(mMasterSocket); + mMasterSocket = nullptr; + return false; +} + +// Only called on MainThread, mMaxMessageSize is read on other threads +void DataChannelConnection::SetMaxMessageSize(bool aMaxMessageSizeSet, + uint64_t aMaxMessageSize) { + ASSERT_WEBRTC(NS_IsMainThread()); + MutexAutoLock lock(mLock); + + if (mMaxMessageSizeSet && !aMaxMessageSizeSet) { + // Don't overwrite already set MMS with default values + return; + } + + mMaxMessageSizeSet = aMaxMessageSizeSet; + mMaxMessageSize = aMaxMessageSize; + + nsresult rv; + nsCOMPtr<nsIPrefService> prefs = + do_GetService("@mozilla.org/preferences-service;1", &rv); + if (!NS_WARN_IF(NS_FAILED(rv))) { + nsCOMPtr<nsIPrefBranch> branch = do_QueryInterface(prefs); + + if (branch) { + int32_t temp; + if (!NS_FAILED(branch->GetIntPref( + "media.peerconnection.sctp.force_maximum_message_size", &temp))) { + if (temp >= 0) { + mMaxMessageSize = (uint64_t)temp; + } + } + } + } + + // Fix remote MMS. This code exists, so future implementations of + // RTCSctpTransport.maxMessageSize can simply provide that value from + // GetMaxMessageSize. + + // TODO: Bug 1382779, once resolved, can be increased to + // min(Uint8ArrayMaxSize, UINT32_MAX) + // TODO: Bug 1381146, once resolved, can be increased to whatever we support + // then (hopefully + // SIZE_MAX) + if (mMaxMessageSize == 0 || + mMaxMessageSize > WEBRTC_DATACHANNEL_MAX_MESSAGE_SIZE_REMOTE) { + mMaxMessageSize = WEBRTC_DATACHANNEL_MAX_MESSAGE_SIZE_REMOTE; + } + + DC_DEBUG(("Maximum message size (outgoing data): %" PRIu64 + " (set=%s, enforced=%s)", + mMaxMessageSize, mMaxMessageSizeSet ? "yes" : "no", + aMaxMessageSize != mMaxMessageSize ? "yes" : "no")); +} + +uint64_t DataChannelConnection::GetMaxMessageSize() { + MutexAutoLock lock(mLock); + return mMaxMessageSize; +} + +void DataChannelConnection::AppendStatsToReport( + const UniquePtr<dom::RTCStatsCollection>& aReport, + const DOMHighResTimeStamp aTimestamp) const { + ASSERT_WEBRTC(NS_IsMainThread()); + nsString temp; + for (const RefPtr<DataChannel>& chan : mChannels.GetAll()) { + // If channel is empty, ignore + if (!chan) { + continue; + } + mozilla::dom::RTCDataChannelStats stats; + nsString id = u"dc"_ns; + id.AppendInt(chan->GetStream()); + stats.mId.Construct(id); + chan->GetLabel(temp); + stats.mTimestamp.Construct(aTimestamp); + stats.mType.Construct(mozilla::dom::RTCStatsType::Data_channel); + stats.mLabel.Construct(temp); + chan->GetProtocol(temp); + stats.mProtocol.Construct(temp); + stats.mDataChannelIdentifier.Construct(chan->GetStream()); + { + using State = mozilla::dom::RTCDataChannelState; + State state; + switch (chan->GetReadyState()) { + case DataChannelState::Connecting: + state = State::Connecting; + break; + case DataChannelState::Open: + state = State::Open; + break; + case DataChannelState::Closing: + state = State::Closing; + break; + case DataChannelState::Closed: + state = State::Closed; + break; + }; + stats.mState.Construct(state); + } + auto counters = chan->GetTrafficCounters(); + stats.mMessagesSent.Construct(counters.mMessagesSent); + stats.mBytesSent.Construct(counters.mBytesSent); + stats.mMessagesReceived.Construct(counters.mMessagesReceived); + stats.mBytesReceived.Construct(counters.mBytesReceived); + if (!aReport->mDataChannelStats.AppendElement(stats, fallible)) { + mozalloc_handle_oom(0); + } + } +} + +#ifdef MOZ_PEERCONNECTION + +bool DataChannelConnection::ConnectToTransport(const std::string& aTransportId, + const bool aClient, + const uint16_t aLocalPort, + const uint16_t aRemotePort) { + MutexAutoLock lock(mLock); + + MOZ_ASSERT(mMasterSocket, + "SCTP wasn't initialized before ConnectToTransport!"); + static const auto paramString = + [](const std::string& tId, const Maybe<bool>& client, + const uint16_t localPort, const uint16_t remotePort) -> std::string { + std::ostringstream stream; + stream << "Transport ID: '" << tId << "', Role: '" + << (client ? (client.value() ? "client" : "server") : "") + << "', Local Port: '" << localPort << "', Remote Port: '" + << remotePort << "'"; + return stream.str(); + }; + + const auto params = + paramString(aTransportId, Some(aClient), aLocalPort, aRemotePort); + DC_DEBUG(("ConnectToTransport connecting DTLS transport with parameters: %s", + params.c_str())); + + DataChannelConnectionState state = GetState(); + if (state == DataChannelConnectionState::Open) { + if (aTransportId == mTransportId && mAllocateEven.isSome() && + mAllocateEven.value() == aClient && mLocalPort == aLocalPort && + mRemotePort == aRemotePort) { + DC_WARN( + ("Skipping attempt to connect to an already OPEN transport with " + "identical parameters.")); + return true; + } + DC_WARN( + ("Attempting to connect to an already OPEN transport, because " + "different parameters were provided.")); + DC_WARN(("Original transport parameters: %s", + paramString(mTransportId, mAllocateEven, mLocalPort, aRemotePort) + .c_str())); + DC_WARN(("New transport parameters: %s", params.c_str())); + } + if (NS_WARN_IF(aTransportId.empty())) { + return false; + } + + mLocalPort = aLocalPort; + mRemotePort = aRemotePort; + SetState(DataChannelConnectionState::Connecting); + mAllocateEven = Some(aClient); + + // Could be faster. Probably doesn't matter. + while (auto channel = mChannels.Get(INVALID_STREAM)) { + mChannels.Remove(channel); + channel->mStream = FindFreeStream(); + if (channel->mStream != INVALID_STREAM) { + mChannels.Insert(channel); + } + } + RUN_ON_THREAD(mSTS, + WrapRunnable(RefPtr<DataChannelConnection>(this), + &DataChannelConnection::SetSignals, aTransportId), + NS_DISPATCH_NORMAL); + return true; +} + +void DataChannelConnection::SetSignals(const std::string& aTransportId) { + ASSERT_WEBRTC(IsSTSThread()); + { + MutexAutoLock lock(mLock); + mTransportId = aTransportId; + } + if (!mConnectedToTransportHandler) { + mTransportHandler->SignalPacketReceived.connect( + this, &DataChannelConnection::SctpDtlsInput); + mTransportHandler->SignalStateChange.connect( + this, &DataChannelConnection::TransportStateChange); + mConnectedToTransportHandler = true; + } + // SignalStateChange() doesn't call you with the initial state + if (mTransportHandler->GetState(mTransportId, false) == + TransportLayer::TS_OPEN) { + DC_DEBUG(("Setting transport signals, dtls already open")); + CompleteConnect(); + } else { + DC_DEBUG(("Setting transport signals, dtls not open yet")); + } +} + +void DataChannelConnection::TransportStateChange( + const std::string& aTransportId, TransportLayer::State aState) { + ASSERT_WEBRTC(IsSTSThread()); + if (aTransportId == mTransportId) { + if (aState == TransportLayer::TS_OPEN) { + DC_DEBUG(("Transport is open!")); + CompleteConnect(); + } else if (aState == TransportLayer::TS_CLOSED || + aState == TransportLayer::TS_NONE || + aState == TransportLayer::TS_ERROR) { + DC_DEBUG(("Transport is closed!")); + Stop(); + } + } +} + +void DataChannelConnection::CompleteConnect() { + MutexAutoLock lock(mLock); + + DC_DEBUG(("dtls open")); + ASSERT_WEBRTC(IsSTSThread()); + if (!mMasterSocket) { + return; + } + + struct sockaddr_conn addr = {}; + addr.sconn_family = AF_CONN; +# if defined(__Userspace_os_Darwin) + addr.sconn_len = sizeof(addr); +# endif + addr.sconn_port = htons(mLocalPort); + addr.sconn_addr = reinterpret_cast<void*>(mId); + + DC_DEBUG(("Calling usrsctp_bind")); + int r = usrsctp_bind(mMasterSocket, reinterpret_cast<struct sockaddr*>(&addr), + sizeof(addr)); + if (r < 0) { + DC_ERROR(("usrsctp_bind failed: %d", r)); + } else { + // This is the remote addr + addr.sconn_port = htons(mRemotePort); + DC_DEBUG(("Calling usrsctp_connect")); + r = usrsctp_connect( + mMasterSocket, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)); + if (r >= 0 || errno == EINPROGRESS) { + struct sctp_paddrparams paddrparams = {}; + socklen_t opt_len; + + memcpy(&paddrparams.spp_address, &addr, sizeof(struct sockaddr_conn)); + opt_len = (socklen_t)sizeof(struct sctp_paddrparams); + r = usrsctp_getsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_PEER_ADDR_PARAMS, + &paddrparams, &opt_len); + if (r < 0) { + DC_ERROR(("usrsctp_getsockopt failed: %d", r)); + } else { + // This field is misnamed. |spp_pathmtu| represents the maximum + // _payload_ size in libusrsctp. So: + // 1280 (a reasonable IPV6 MTU according to RFC 8831) + // -12 (sctp header) + // -24 (GCM sipher) + // -13 (DTLS record header) + // -8 (UDP header) + // -4 (TURN ChannelData) + // -40 (IPV6 header) + // = 1179 + // We could further restrict this, because RFC 8831 suggests a starting + // IPV4 path MTU of 1200, which would lead to a value of 1115. + // I suspect that in practice the path MTU for IPV4 is substantially + // larger than 1200. + paddrparams.spp_pathmtu = 1179; + paddrparams.spp_flags &= ~SPP_PMTUD_ENABLE; + paddrparams.spp_flags |= SPP_PMTUD_DISABLE; + opt_len = (socklen_t)sizeof(struct sctp_paddrparams); + r = usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, + SCTP_PEER_ADDR_PARAMS, &paddrparams, opt_len); + if (r < 0) { + DC_ERROR(("usrsctp_getsockopt failed: %d", r)); + } else { + DC_ERROR(("usrsctp: PMTUD disabled, MTU set to %u", + paddrparams.spp_pathmtu)); + } + } + } + if (r < 0) { + if (errno == EINPROGRESS) { + // non-blocking + return; + } + DC_ERROR(("usrsctp_connect failed: %d", errno)); + SetState(DataChannelConnectionState::Closed); + } else { + // We fire ON_CONNECTION via SCTP_COMM_UP when we get that + return; + } + } + // Note: currently this doesn't actually notify the application + Dispatch(do_AddRef(new DataChannelOnMessageAvailable( + DataChannelOnMessageAvailable::EventType::OnConnection, this))); +} + +// Process any pending Opens +void DataChannelConnection::ProcessQueuedOpens() { + // The nsDeque holds channels with an AddRef applied. Another reference + // (may) be held by the DOMDataChannel, unless it's been GC'd. No other + // references should exist. + + // Can't copy nsDeque's. Move into temp array since any that fail will + // go back to mPending + nsRefPtrDeque<DataChannel> temp; + RefPtr<DataChannel> temp_channel; + while (nullptr != (temp_channel = mPending.PopFront())) { + temp.Push(temp_channel.forget()); + } + + RefPtr<DataChannel> channel; + + while (nullptr != (channel = temp.PopFront())) { + if (channel->mHasFinishedOpen) { + DC_DEBUG(("Processing queued open for %p (%u)", channel.get(), + channel->mStream)); + channel->mHasFinishedOpen = false; + // OpenFinish returns a reference itself, so we need to take it can + // Release it + channel = OpenFinish(channel.forget()); // may reset the flag and re-push + } else { + NS_ASSERTION(false, + "How did a DataChannel get queued without the " + "mHasFinishedOpen flag?"); + } + } +} + +void DataChannelConnection::SctpDtlsInput(const std::string& aTransportId, + const MediaPacket& packet) { + MutexAutoLock lock(mLock); + if ((packet.type() != MediaPacket::SCTP) || (mTransportId != aTransportId)) { + return; + } + + if (MOZ_LOG_TEST(gSCTPLog, LogLevel::Debug)) { + char* buf; + + if ((buf = usrsctp_dumppacket((void*)packet.data(), packet.len(), + SCTP_DUMP_INBOUND)) != nullptr) { + SCTP_LOG(("%s", buf)); + usrsctp_freedumpbuffer(buf); + } + } + // Pass the data to SCTP + usrsctp_conninput(reinterpret_cast<void*>(mId), packet.data(), packet.len(), + 0); +} + +void DataChannelConnection::SendPacket(std::unique_ptr<MediaPacket>&& packet) { + mSTS->Dispatch(NS_NewRunnableFunction( + "DataChannelConnection::SendPacket", + [this, self = RefPtr<DataChannelConnection>(this), + packet = std::move(packet)]() mutable { + // DC_DEBUG(("%p: SCTP/DTLS sent %ld bytes", this, len)); + if (!mTransportId.empty() && mTransportHandler) { + mTransportHandler->SendPacket(mTransportId, std::move(*packet)); + } + })); +} + +int DataChannelConnection::SctpDtlsOutput(void* addr, void* buffer, + size_t length, uint8_t tos, + uint8_t set_df) { + if (MOZ_LOG_TEST(gSCTPLog, LogLevel::Debug)) { + char* buf; + + if ((buf = usrsctp_dumppacket(buffer, length, SCTP_DUMP_OUTBOUND)) != + nullptr) { + SCTP_LOG(("%s", buf)); + usrsctp_freedumpbuffer(buf); + } + } + + // We're async proxying even if on the STSThread because this is called + // with internal SCTP locks held in some cases (such as in usrsctp_connect()). + // SCTP has an option for Apple, on IP connections only, to release at least + // one of the locks before calling a packet output routine; with changes to + // the underlying SCTP stack this might remove the need to use an async proxy. + std::unique_ptr<MediaPacket> packet(new MediaPacket); + packet->SetType(MediaPacket::SCTP); + packet->Copy(static_cast<const uint8_t*>(buffer), length); + + if (NS_IsMainThread() && mDeferSend) { + mDeferredSend.emplace_back(std::move(packet)); + return 0; + } + + SendPacket(std::move(packet)); + return 0; // cheat! Packets can always be dropped later anyways +} +#endif + +#ifdef ALLOW_DIRECT_SCTP_LISTEN_CONNECT +// listen for incoming associations +// Blocks! - Don't call this from main thread! + +bool DataChannelConnection::Listen(unsigned short port) { + struct sockaddr_in addr = {}; + socklen_t addr_len; + + NS_WARNING_ASSERTION(!NS_IsMainThread(), + "Blocks, do not call from main thread!!!"); + + /* Acting as the 'server' */ +# ifdef HAVE_SIN_LEN + addr.sin_len = sizeof(struct sockaddr_in); +# endif + addr.sin_family = AF_INET; + addr.sin_port = htons(port); + addr.sin_addr.s_addr = htonl(INADDR_ANY); + DC_DEBUG(("Waiting for connections on port %u", ntohs(addr.sin_port))); + { + MutexAutoLock lock(mLock); + SetState(DataChannelConnectionState::Connecting); + } + if (usrsctp_bind(mMasterSocket, reinterpret_cast<struct sockaddr*>(&addr), + sizeof(struct sockaddr_in)) < 0) { + DC_ERROR(("***Failed userspace_bind")); + return false; + } + if (usrsctp_listen(mMasterSocket, 1) < 0) { + DC_ERROR(("***Failed userspace_listen")); + return false; + } + + DC_DEBUG(("Accepting connection")); + addr_len = 0; + if ((mSocket = usrsctp_accept(mMasterSocket, nullptr, &addr_len)) == + nullptr) { + DC_ERROR(("***Failed accept")); + return false; + } + + { + MutexAutoLock lock(mLock); + SetState(DataChannelConnectionState::Open); + } + + struct linger l; + l.l_onoff = 1; + l.l_linger = 0; + if (usrsctp_setsockopt(mSocket, SOL_SOCKET, SO_LINGER, (const void*)&l, + (socklen_t)sizeof(struct linger)) < 0) { + DC_WARN(("Couldn't set SO_LINGER on SCTP socket")); + } + + // Notify Connection open + // XXX We need to make sure connection sticks around until the message is + // delivered + DC_DEBUG(("%s: sending ON_CONNECTION for %p", __FUNCTION__, this)); + Dispatch(do_AddRef(new DataChannelOnMessageAvailable( + DataChannelOnMessageAvailable::EventType::OnConnection, this, + (DataChannel*)nullptr))); + return true; +} + +// Blocks! - Don't call this from main thread! +bool DataChannelConnection::Connect(const char* addr, unsigned short port) { + struct sockaddr_in addr4 = {}; + struct sockaddr_in6 addr6 = {}; + + NS_WARNING_ASSERTION(!NS_IsMainThread(), + "Blocks, do not call from main thread!!!"); + + /* Acting as the connector */ + DC_DEBUG(("Connecting to %s, port %u", addr, port)); +# ifdef HAVE_SIN_LEN + addr4.sin_len = sizeof(struct sockaddr_in); +# endif +# ifdef HAVE_SIN6_LEN + addr6.sin6_len = sizeof(struct sockaddr_in6); +# endif + addr4.sin_family = AF_INET; + addr6.sin6_family = AF_INET6; + addr4.sin_port = htons(port); + addr6.sin6_port = htons(port); + { + MutexAutoLock lock(mLock); + SetState(DataChannelConnectionState::Connecting); + } +# if !defined(__Userspace_os_Windows) + if (inet_pton(AF_INET6, addr, &addr6.sin6_addr) == 1) { + if (usrsctp_connect(mMasterSocket, + reinterpret_cast<struct sockaddr*>(&addr6), + sizeof(struct sockaddr_in6)) < 0) { + DC_ERROR(("*** Failed userspace_connect")); + return false; + } + } else if (inet_pton(AF_INET, addr, &addr4.sin_addr) == 1) { + if (usrsctp_connect(mMasterSocket, + reinterpret_cast<struct sockaddr*>(&addr4), + sizeof(struct sockaddr_in)) < 0) { + DC_ERROR(("*** Failed userspace_connect")); + return false; + } + } else { + DC_ERROR(("*** Illegal destination address.")); + } +# else + { + struct sockaddr_storage ss; + int sslen = sizeof(ss); + + if (!WSAStringToAddressA(const_cast<char*>(addr), AF_INET6, nullptr, + (struct sockaddr*)&ss, &sslen)) { + addr6.sin6_addr = + (reinterpret_cast<struct sockaddr_in6*>(&ss))->sin6_addr; + if (usrsctp_connect(mMasterSocket, + reinterpret_cast<struct sockaddr*>(&addr6), + sizeof(struct sockaddr_in6)) < 0) { + DC_ERROR(("*** Failed userspace_connect")); + return false; + } + } else if (!WSAStringToAddressA(const_cast<char*>(addr), AF_INET, nullptr, + (struct sockaddr*)&ss, &sslen)) { + addr4.sin_addr = (reinterpret_cast<struct sockaddr_in*>(&ss))->sin_addr; + if (usrsctp_connect(mMasterSocket, + reinterpret_cast<struct sockaddr*>(&addr4), + sizeof(struct sockaddr_in)) < 0) { + DC_ERROR(("*** Failed userspace_connect")); + return false; + } + } else { + DC_ERROR(("*** Illegal destination address.")); + } + } +# endif + + mSocket = mMasterSocket; + + DC_DEBUG(("connect() succeeded! Entering connected mode")); + { + MutexAutoLock lock(mLock); + SetState(DataChannelConnectionState::Open); + } + // Notify Connection open + // XXX We need to make sure connection sticks around until the message is + // delivered + DC_DEBUG(("%s: sending ON_CONNECTION for %p", __FUNCTION__, this)); + Dispatch(do_AddRef(new DataChannelOnMessageAvailable( + DataChannelOnMessageAvailable::EventType::OnConnection, this, + (DataChannel*)nullptr))); + return true; +} +#endif + +DataChannel* DataChannelConnection::FindChannelByStream(uint16_t stream) { + return mChannels.Get(stream).get(); +} + +uint16_t DataChannelConnection::FindFreeStream() const { + ASSERT_WEBRTC(NS_IsMainThread()); + uint16_t i, limit; + + limit = MAX_NUM_STREAMS; + + MOZ_ASSERT(mAllocateEven.isSome()); + for (i = (*mAllocateEven ? 0 : 1); i < limit; i += 2) { + if (mChannels.Get(i)) { + continue; + } + + // Verify it's not still in the process of closing + size_t j; + for (j = 0; j < mStreamsResetting.Length(); ++j) { + if (mStreamsResetting[j] == i) { + break; + } + } + + if (j == mStreamsResetting.Length()) { + return i; + } + } + return INVALID_STREAM; +} + +uint32_t DataChannelConnection::UpdateCurrentStreamIndex() { + RefPtr<DataChannel> channel = mChannels.GetNextChannel(mCurrentStream); + if (!channel) { + mCurrentStream = 0; + } else { + mCurrentStream = channel->mStream; + } + return mCurrentStream; +} + +uint32_t DataChannelConnection::GetCurrentStreamIndex() { + if (!mChannels.Get(mCurrentStream)) { + // The stream muse have been removed, reset + DC_DEBUG(("Reset mCurrentChannel")); + mCurrentStream = 0; + } + return mCurrentStream; +} + +bool DataChannelConnection::RequestMoreStreams(int32_t aNeeded) { + struct sctp_status status = {}; + struct sctp_add_streams sas = {}; + uint32_t outStreamsNeeded; + socklen_t len; + + if (aNeeded + mNegotiatedIdLimit > MAX_NUM_STREAMS) { + aNeeded = MAX_NUM_STREAMS - mNegotiatedIdLimit; + } + if (aNeeded <= 0) { + return false; + } + + len = (socklen_t)sizeof(struct sctp_status); + if (usrsctp_getsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_STATUS, &status, + &len) < 0) { + DC_ERROR(("***failed: getsockopt SCTP_STATUS")); + return false; + } + outStreamsNeeded = aNeeded; // number to add + + // Note: if multiple channel opens happen when we don't have enough space, + // we'll call RequestMoreStreams() multiple times + sas.sas_instrms = 0; + sas.sas_outstrms = (uint16_t)outStreamsNeeded; /* XXX error handling */ + // Doesn't block, we get an event when it succeeds or fails + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_ADD_STREAMS, &sas, + (socklen_t)sizeof(struct sctp_add_streams)) < 0) { + if (errno == EALREADY) { + DC_DEBUG(("Already have %u output streams", outStreamsNeeded)); + return true; + } + + DC_ERROR(("***failed: setsockopt ADD errno=%d", errno)); + return false; + } + DC_DEBUG(("Requested %u more streams", outStreamsNeeded)); + // We add to mNegotiatedIdLimit when we get a SCTP_STREAM_CHANGE_EVENT and the + // values are larger than mNegotiatedIdLimit + return true; +} + +// Returns a POSIX error code. +int DataChannelConnection::SendControlMessage(const uint8_t* data, uint32_t len, + uint16_t stream) { + struct sctp_sendv_spa info = {}; + + // General flags + info.sendv_flags = SCTP_SEND_SNDINFO_VALID; + + // Set stream identifier, protocol identifier and flags + info.sendv_sndinfo.snd_sid = stream; + info.sendv_sndinfo.snd_flags = SCTP_EOR; + info.sendv_sndinfo.snd_ppid = htonl(DATA_CHANNEL_PPID_CONTROL); + + // Create message instance and send + // Note: Main-thread IO, but doesn't block +#if (UINT32_MAX > SIZE_MAX) + if (len > SIZE_MAX) { + return EMSGSIZE; + } +#endif + OutgoingMsg msg(info, data, (size_t)len); + bool buffered; + int error = SendMsgInternalOrBuffer(mBufferedControl, msg, buffered, nullptr); + + // Set pending type (if buffered) + if (!error && buffered && mPendingType == PendingType::None) { + mPendingType = PendingType::Dcep; + } + return error; +} + +// Returns a POSIX error code. +int DataChannelConnection::SendOpenAckMessage(uint16_t stream) { + struct rtcweb_datachannel_ack ack = {}; + ack.msg_type = DATA_CHANNEL_ACK; + + return SendControlMessage((const uint8_t*)&ack, sizeof(ack), stream); +} + +// Returns a POSIX error code. +int DataChannelConnection::SendOpenRequestMessage( + const nsACString& label, const nsACString& protocol, uint16_t stream, + bool unordered, DataChannelReliabilityPolicy prPolicy, uint32_t prValue) { + const size_t label_len = label.Length(); // not including nul + const size_t proto_len = protocol.Length(); // not including nul + // careful - request struct include one char for the label + const size_t req_size = sizeof(struct rtcweb_datachannel_open_request) - 1 + + label_len + proto_len; + UniqueFreePtr<struct rtcweb_datachannel_open_request> req( + (struct rtcweb_datachannel_open_request*)moz_xmalloc(req_size)); + + memset(req.get(), 0, req_size); + req->msg_type = DATA_CHANNEL_OPEN_REQUEST; + switch (prPolicy) { + case DataChannelReliabilityPolicy::Reliable: + req->channel_type = DATA_CHANNEL_RELIABLE; + break; + case DataChannelReliabilityPolicy::LimitedLifetime: + req->channel_type = DATA_CHANNEL_PARTIAL_RELIABLE_TIMED; + break; + case DataChannelReliabilityPolicy::LimitedRetransmissions: + req->channel_type = DATA_CHANNEL_PARTIAL_RELIABLE_REXMIT; + break; + default: + return EINVAL; + } + if (unordered) { + // Per the current types, all differ by 0x80 between ordered and unordered + req->channel_type |= + 0x80; // NOTE: be careful if new types are added in the future + } + + req->reliability_param = htonl(prValue); + req->priority = htons(0); /* XXX: add support */ + req->label_length = htons(label_len); + req->protocol_length = htons(proto_len); + memcpy(&req->label[0], PromiseFlatCString(label).get(), label_len); + memcpy(&req->label[label_len], PromiseFlatCString(protocol).get(), proto_len); + + // TODO: req_size is an int... that looks hairy + int error = SendControlMessage((const uint8_t*)req.get(), req_size, stream); + return error; +} + +// XXX This should use a separate thread (outbound queue) which should +// select() to know when to *try* to send data to the socket again. +// Alternatively, it can use a timeout, but that's guaranteed to be wrong +// (just not sure in what direction). We could re-implement NSPR's +// PR_POLL_WRITE/etc handling... with a lot of work. + +// Better yet, use the SCTP stack's notifications on buffer state to avoid +// filling the SCTP's buffers. + +// returns if we're still blocked (true) +bool DataChannelConnection::SendDeferredMessages() { + RefPtr<DataChannel> channel; // we may null out the refs to this + + // This may block while something is modifying channels, but should not block + // for IO + ASSERT_WEBRTC(!NS_IsMainThread()); + mLock.AssertCurrentThreadOwns(); + + DC_DEBUG(("SendDeferredMessages called, pending type: %s", + ToString(mPendingType))); + if (mPendingType == PendingType::None) { + return false; + } + + // Send pending control messages + // Note: If ndata is not active, check if DCEP messages are currently + // outstanding. These need to + // be sent first before other streams can be used for sending. + if (!mBufferedControl.IsEmpty() && + (mSendInterleaved || mPendingType == PendingType::Dcep)) { + if (SendBufferedMessages(mBufferedControl, nullptr)) { + return true; + } + + // Note: There may or may not be pending data messages + mPendingType = PendingType::Data; + } + + bool blocked = false; + uint32_t i = GetCurrentStreamIndex(); + uint32_t end = i; + do { + channel = mChannels.Get(i); + // Should already be cleared if closing/closed + if (!channel || channel->mBufferedData.IsEmpty()) { + i = UpdateCurrentStreamIndex(); + continue; + } + + // Send buffered data messages + // Warning: This will fail in case ndata is inactive and a previously + // deallocated data channel has not been closed properly. If you + // ever see that no messages can be sent on any channel, this is + // likely the cause (an explicit EOR message partially sent whose + // remaining chunks are still being waited for). + size_t written = 0; + mDeferSend = true; + blocked = SendBufferedMessages(channel->mBufferedData, &written); + mDeferSend = false; + if (written) { + channel->DecrementBufferedAmount(written); + } + + for (auto&& packet : mDeferredSend) { + MOZ_ASSERT(written); + SendPacket(std::move(packet)); + } + mDeferredSend.clear(); + + // Update current stream index + // Note: If ndata is not active, the outstanding data messages on this + // stream need to be sent first before other streams can be used for + // sending. + if (mSendInterleaved || !blocked) { + i = UpdateCurrentStreamIndex(); + } + } while (!blocked && i != end); + + if (!blocked) { + mPendingType = + mBufferedControl.IsEmpty() ? PendingType::None : PendingType::Dcep; + } + return blocked; +} + +// Called with mLock locked! +// buffer MUST have at least one item! +// returns if we're still blocked (true) +bool DataChannelConnection::SendBufferedMessages( + nsTArray<UniquePtr<BufferedOutgoingMsg>>& buffer, size_t* aWritten) { + do { + // Re-send message + int error = SendMsgInternal(*buffer[0], aWritten); + switch (error) { + case 0: + buffer.RemoveElementAt(0); + break; + case EAGAIN: +#if (EAGAIN != EWOULDBLOCK) + case EWOULDBLOCK: +#endif + return true; + default: + buffer.RemoveElementAt(0); + DC_ERROR(("error on sending: %d", error)); + break; + } + } while (!buffer.IsEmpty()); + + return false; +} + +// Caller must ensure that length <= SIZE_MAX +void DataChannelConnection::HandleOpenRequestMessage( + const struct rtcweb_datachannel_open_request* req, uint32_t length, + uint16_t stream) { + RefPtr<DataChannel> channel; + uint32_t prValue; + DataChannelReliabilityPolicy prPolicy; + + ASSERT_WEBRTC(!NS_IsMainThread()); + mLock.AssertCurrentThreadOwns(); + + const size_t requiredLength = (sizeof(*req) - 1) + ntohs(req->label_length) + + ntohs(req->protocol_length); + if (((size_t)length) != requiredLength) { + if (((size_t)length) < requiredLength) { + DC_ERROR( + ("%s: insufficient length: %u, should be %zu. Unable to continue.", + __FUNCTION__, length, requiredLength)); + return; + } + DC_WARN(("%s: Inconsistent length: %u, should be %zu", __FUNCTION__, length, + requiredLength)); + } + + DC_DEBUG(("%s: length %u, sizeof(*req) = %zu", __FUNCTION__, length, + sizeof(*req))); + + switch (req->channel_type) { + case DATA_CHANNEL_RELIABLE: + case DATA_CHANNEL_RELIABLE_UNORDERED: + prPolicy = DataChannelReliabilityPolicy::Reliable; + break; + case DATA_CHANNEL_PARTIAL_RELIABLE_REXMIT: + case DATA_CHANNEL_PARTIAL_RELIABLE_REXMIT_UNORDERED: + prPolicy = DataChannelReliabilityPolicy::LimitedRetransmissions; + break; + case DATA_CHANNEL_PARTIAL_RELIABLE_TIMED: + case DATA_CHANNEL_PARTIAL_RELIABLE_TIMED_UNORDERED: + prPolicy = DataChannelReliabilityPolicy::LimitedLifetime; + break; + default: + DC_ERROR(("Unknown channel type %d", req->channel_type)); + /* XXX error handling */ + return; + } + prValue = ntohl(req->reliability_param); + bool ordered = !(req->channel_type & 0x80); + + if ((channel = FindChannelByStream(stream))) { + if (!channel->mNegotiated) { + DC_ERROR( + ("HandleOpenRequestMessage: channel for pre-existing stream " + "%u that was not externally negotiated. JS is lying to us, or " + "there's an id collision.", + stream)); + /* XXX: some error handling */ + } else { + DC_DEBUG(("Open for externally negotiated channel %u", stream)); + // XXX should also check protocol, maybe label + if (prPolicy != channel->mPrPolicy || prValue != channel->mPrValue || + ordered != channel->mOrdered) { + DC_WARN( + ("external negotiation mismatch with OpenRequest:" + "channel %u, policy %s/%s, value %u/%u, ordered %d/%d", + stream, ToString(prPolicy), ToString(channel->mPrPolicy), prValue, + channel->mPrValue, static_cast<int>(ordered), + static_cast<int>(channel->mOrdered))); + } + } + return; + } + if (stream >= mNegotiatedIdLimit) { + DC_ERROR(("%s: stream %u out of bounds (%zu)", __FUNCTION__, stream, + mNegotiatedIdLimit)); + return; + } + + nsCString label( + nsDependentCSubstring(&req->label[0], ntohs(req->label_length))); + nsCString protocol(nsDependentCSubstring( + &req->label[ntohs(req->label_length)], ntohs(req->protocol_length))); + + channel = + new DataChannel(this, stream, DataChannelState::Open, label, protocol, + prPolicy, prValue, ordered, false, nullptr, nullptr); + mChannels.Insert(channel); + + DC_DEBUG(("%s: sending ON_CHANNEL_CREATED for %s/%s: %u", __FUNCTION__, + channel->mLabel.get(), channel->mProtocol.get(), stream)); + Dispatch(do_AddRef(new DataChannelOnMessageAvailable( + DataChannelOnMessageAvailable::EventType::OnChannelCreated, this, + channel))); + + DC_DEBUG(("%s: deferring sending ON_CHANNEL_OPEN for %p", __FUNCTION__, + channel.get())); + channel->AnnounceOpen(); + + // Note that any message can be buffered; SendOpenAckMessage may + // error later than this check. + const auto error = SendOpenAckMessage(channel->mStream); + if (error) { + DC_ERROR(("SendOpenRequest failed, error = %d", error)); + Dispatch(NS_NewRunnableFunction( + "DataChannelConnection::HandleOpenRequestMessage", + [channel, connection = RefPtr<DataChannelConnection>(this)]() { + // Close the channel on failure + connection->Close(channel); + })); + return; + } + DeliverQueuedData(channel->mStream); +} + +// NOTE: the updated spec from the IETF says we should set in-order until we +// receive an ACK. That would make this code moot. Keep it for now for +// backwards compatibility. +void DataChannelConnection::DeliverQueuedData(uint16_t stream) { + mLock.AssertCurrentThreadOwns(); + + mQueuedData.RemoveElementsBy([stream, this](const auto& dataItem) { + mLock.AssertCurrentThreadOwns(); + const bool match = dataItem->mStream == stream; + if (match) { + DC_DEBUG(("Delivering queued data for stream %u, length %u", stream, + dataItem->mLength)); + // Deliver the queued data + HandleDataMessage(dataItem->mData, dataItem->mLength, dataItem->mPpid, + dataItem->mStream, dataItem->mFlags); + } + return match; + }); +} + +// Caller must ensure that length <= SIZE_MAX +void DataChannelConnection::HandleOpenAckMessage( + const struct rtcweb_datachannel_ack* ack, uint32_t length, + uint16_t stream) { + DataChannel* channel; + + mLock.AssertCurrentThreadOwns(); + + channel = FindChannelByStream(stream); + if (NS_WARN_IF(!channel)) { + return; + } + + DC_DEBUG(("OpenAck received for stream %u, waiting=%d", stream, + channel->mWaitingForAck ? 1 : 0)); + + channel->mWaitingForAck = false; +} + +// Caller must ensure that length <= SIZE_MAX +void DataChannelConnection::HandleUnknownMessage(uint32_t ppid, uint32_t length, + uint16_t stream) { + /* XXX: Send an error message? */ + DC_ERROR(("unknown DataChannel message received: %u, len %u on stream %d", + ppid, length, stream)); + // XXX Log to JS error console if possible +} + +uint8_t DataChannelConnection::BufferMessage(nsACString& recvBuffer, + const void* data, uint32_t length, + uint32_t ppid, int flags) { + const char* buffer = (const char*)data; + uint8_t bufferFlags = 0; + + if ((flags & MSG_EOR) && ppid != DATA_CHANNEL_PPID_BINARY_PARTIAL && + ppid != DATA_CHANNEL_PPID_DOMSTRING_PARTIAL) { + bufferFlags |= DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_COMPLETE; + + // Return directly if nothing has been buffered + if (recvBuffer.IsEmpty()) { + return bufferFlags; + } + } + + // Ensure it doesn't blow up our buffer + // TODO: Change 'WEBRTC_DATACHANNEL_MAX_MESSAGE_SIZE_LOCAL' to whatever the + // new buffer is capable of holding. + if (((uint64_t)recvBuffer.Length()) + ((uint64_t)length) > + WEBRTC_DATACHANNEL_MAX_MESSAGE_SIZE_LOCAL) { + bufferFlags |= DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_TOO_LARGE; + return bufferFlags; + } + + // Copy & add to receive buffer + recvBuffer.Append(buffer, length); + bufferFlags |= DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_BUFFERED; + return bufferFlags; +} + +void DataChannelConnection::HandleDataMessage(const void* data, size_t length, + uint32_t ppid, uint16_t stream, + int flags) { + DataChannel* channel; + const char* buffer = (const char*)data; + + mLock.AssertCurrentThreadOwns(); + channel = FindChannelByStream(stream); + + // Note: Until we support SIZE_MAX sized messages, we need this check +#if (SIZE_MAX > UINT32_MAX) + if (length > UINT32_MAX) { + DC_ERROR(("DataChannel: Cannot handle message of size %zu (max=%" PRIu32 + ")", + length, UINT32_MAX)); + CloseLocked(channel); + return; + } +#endif + uint32_t data_length = (uint32_t)length; + + // XXX A closed channel may trip this... check + // NOTE: the updated spec from the IETF says we should set in-order until we + // receive an ACK. That would make this code moot. Keep it for now for + // backwards compatibility. + if (!channel) { + // In the updated 0-RTT open case, the sender can send data immediately + // after Open, and doesn't set the in-order bit (since we don't have a + // response or ack). Also, with external negotiation, data can come in + // before we're told about the external negotiation. We need to buffer + // data until either a) Open comes in, if the ordering get messed up, + // or b) the app tells us this channel was externally negotiated. When + // these occur, we deliver the data. + + // Since this is rare and non-performance, keep a single list of queued + // data messages to deliver once the channel opens. + DC_DEBUG(("Queuing data for stream %u, length %u", stream, data_length)); + // Copies data + mQueuedData.AppendElement( + new QueuedDataMessage(stream, ppid, flags, data, data_length)); + return; + } + + // RFC8832: "MUST be sent ordered, ... After the DATA_CHANNEL_ACK **or any + // other message** has been received on the data channel". + // If the channel was opened on this side, and a message is received, this + // indicates that the peer has already received the DATA_CHANNEL_ACK, as the + // channel is ordered initially. + channel->mWaitingForAck = false; + + bool is_binary = true; + uint8_t bufferFlags; + DataChannelOnMessageAvailable::EventType type; + const char* info = ""; + + if (ppid == DATA_CHANNEL_PPID_DOMSTRING_PARTIAL || + ppid == DATA_CHANNEL_PPID_DOMSTRING || + ppid == DATA_CHANNEL_PPID_DOMSTRING_EMPTY) { + is_binary = false; + } + if (is_binary != channel->mIsRecvBinary && !channel->mRecvBuffer.IsEmpty()) { + NS_WARNING("DataChannel message aborted by fragment type change!"); + // TODO: Maybe closing would be better as this is a hard to detect protocol + // violation? + channel->mRecvBuffer.Truncate(0); + } + channel->mIsRecvBinary = is_binary; + + // Remaining chunks of previously truncated message (due to the buffer being + // full)? + if (channel->mClosingTooLarge) { + DC_ERROR( + ("DataChannel: Ignoring partial message of length %u, buffer full and " + "closing", + data_length)); + // Only unblock if unordered + if (!channel->mOrdered && (flags & MSG_EOR)) { + channel->mClosingTooLarge = false; + } + } + + // Buffer message until complete + bufferFlags = + BufferMessage(channel->mRecvBuffer, buffer, data_length, ppid, flags); + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_TOO_LARGE) { + DC_ERROR( + ("DataChannel: Buffered message would become too large to handle, " + "closing channel")); + channel->mRecvBuffer.Truncate(0); + channel->mClosingTooLarge = true; + CloseLocked(channel); + return; + } + if (!(bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_COMPLETE)) { + DC_DEBUG( + ("DataChannel: Partial %s message of length %u (total %zu) on channel " + "id %u", + is_binary ? "binary" : "string", data_length, + channel->mRecvBuffer.Length(), channel->mStream)); + return; // Not ready to notify application + } + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_BUFFERED) { + data_length = channel->mRecvBuffer.Length(); + } + + // Complain about large messages (only complain - we can handle it) + if (data_length > WEBRTC_DATACHANNEL_MAX_MESSAGE_SIZE_LOCAL) { + DC_WARN( + ("DataChannel: Received message of length %u is > announced maximum " + "message size (%u)", + data_length, WEBRTC_DATACHANNEL_MAX_MESSAGE_SIZE_LOCAL)); + } + + bool is_empty = false; + + switch (ppid) { + case DATA_CHANNEL_PPID_DOMSTRING: + DC_DEBUG( + ("DataChannel: Received string message of length %u on channel %u", + data_length, channel->mStream)); + type = DataChannelOnMessageAvailable::EventType::OnDataString; + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_BUFFERED) { + info = " (string fragmented)"; + } + // else send using recvData normally + + // WebSockets checks IsUTF8() here; we can try to deliver it + break; + + case DATA_CHANNEL_PPID_DOMSTRING_EMPTY: + DC_DEBUG( + ("DataChannel: Received empty string message of length %u on channel " + "%u", + data_length, channel->mStream)); + type = DataChannelOnMessageAvailable::EventType::OnDataString; + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_BUFFERED) { + info = " (string fragmented)"; + } + is_empty = true; + break; + + case DATA_CHANNEL_PPID_BINARY: + DC_DEBUG( + ("DataChannel: Received binary message of length %u on channel id %u", + data_length, channel->mStream)); + type = DataChannelOnMessageAvailable::EventType::OnDataBinary; + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_BUFFERED) { + info = " (binary fragmented)"; + } + + // else send using recvData normally + break; + + case DATA_CHANNEL_PPID_BINARY_EMPTY: + DC_DEBUG( + ("DataChannel: Received empty binary message of length %u on channel " + "id %u", + data_length, channel->mStream)); + type = DataChannelOnMessageAvailable::EventType::OnDataBinary; + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_BUFFERED) { + info = " (binary fragmented)"; + } + is_empty = true; + break; + + default: + NS_ERROR("Unknown data PPID"); + DC_ERROR(("Unknown data PPID %" PRIu32, ppid)); + return; + } + + channel->WithTrafficCounters( + [&data_length](DataChannel::TrafficCounters& counters) { + counters.mMessagesReceived++; + counters.mBytesReceived += data_length; + }); + + // Notify onmessage + DC_DEBUG( + ("%s: sending %s%s for %p", __FUNCTION__, ToString(type), info, channel)); + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_BUFFERED) { + channel->SendOrQueue(new DataChannelOnMessageAvailable( + type, this, channel, channel->mRecvBuffer)); + channel->mRecvBuffer.Truncate(0); + } else { + nsAutoCString recvData(is_empty ? "" : buffer, + is_empty ? 0 : data_length); // allocates >64 + channel->SendOrQueue( + new DataChannelOnMessageAvailable(type, this, channel, recvData)); + } +} + +void DataChannelConnection::HandleDCEPMessage(const void* buffer, size_t length, + uint32_t ppid, uint16_t stream, + int flags) { + const struct rtcweb_datachannel_open_request* req; + const struct rtcweb_datachannel_ack* ack; + + // Note: Until we support SIZE_MAX sized messages, we need this check +#if (SIZE_MAX > UINT32_MAX) + if (length > UINT32_MAX) { + DC_ERROR(("DataChannel: Cannot handle message of size %zu (max=%u)", length, + UINT32_MAX)); + Stop(); + return; + } +#endif + uint32_t data_length = (uint32_t)length; + + mLock.AssertCurrentThreadOwns(); + + // Buffer message until complete + const uint8_t bufferFlags = + BufferMessage(mRecvBuffer, buffer, data_length, ppid, flags); + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_TOO_LARGE) { + DC_ERROR( + ("DataChannel: Buffered message would become too large to handle, " + "closing connection")); + mRecvBuffer.Truncate(0); + Stop(); + return; + } + if (!(bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_COMPLETE)) { + DC_DEBUG(("Buffered partial DCEP message of length %u", data_length)); + return; + } + if (bufferFlags & DATA_CHANNEL_BUFFER_MESSAGE_FLAGS_BUFFERED) { + buffer = reinterpret_cast<const void*>(mRecvBuffer.BeginReading()); + data_length = mRecvBuffer.Length(); + } + + req = static_cast<const struct rtcweb_datachannel_open_request*>(buffer); + DC_DEBUG(("Handling DCEP message of length %u", data_length)); + + // Ensure minimum message size (ack is the smallest DCEP message) + if ((size_t)data_length < sizeof(*ack)) { + DC_WARN(("Ignored invalid DCEP message (too short)")); + return; + } + + switch (req->msg_type) { + case DATA_CHANNEL_OPEN_REQUEST: + // structure includes a possibly-unused char label[1] (in a packed + // structure) + if (NS_WARN_IF((size_t)data_length < sizeof(*req) - 1)) { + return; + } + + HandleOpenRequestMessage(req, data_length, stream); + break; + case DATA_CHANNEL_ACK: + // >= sizeof(*ack) checked above + + ack = static_cast<const struct rtcweb_datachannel_ack*>(buffer); + HandleOpenAckMessage(ack, data_length, stream); + break; + default: + HandleUnknownMessage(ppid, data_length, stream); + break; + } + + // Reset buffer + mRecvBuffer.Truncate(0); +} + +// Called with mLock locked! +void DataChannelConnection::HandleMessage(const void* buffer, size_t length, + uint32_t ppid, uint16_t stream, + int flags) { + mLock.AssertCurrentThreadOwns(); + + switch (ppid) { + case DATA_CHANNEL_PPID_CONTROL: + HandleDCEPMessage(buffer, length, ppid, stream, flags); + break; + case DATA_CHANNEL_PPID_DOMSTRING_PARTIAL: + case DATA_CHANNEL_PPID_DOMSTRING: + case DATA_CHANNEL_PPID_DOMSTRING_EMPTY: + case DATA_CHANNEL_PPID_BINARY_PARTIAL: + case DATA_CHANNEL_PPID_BINARY: + case DATA_CHANNEL_PPID_BINARY_EMPTY: + HandleDataMessage(buffer, length, ppid, stream, flags); + break; + default: + DC_ERROR(( + "Unhandled message of length %zu PPID %u on stream %u received (%s).", + length, ppid, stream, (flags & MSG_EOR) ? "complete" : "partial")); + break; + } +} + +void DataChannelConnection::HandleAssociationChangeEvent( + const struct sctp_assoc_change* sac) { + mLock.AssertCurrentThreadOwns(); + + uint32_t i, n; + DataChannelConnectionState state = GetState(); + switch (sac->sac_state) { + case SCTP_COMM_UP: + DC_DEBUG(("Association change: SCTP_COMM_UP")); + if (state == DataChannelConnectionState::Connecting) { + mSocket = mMasterSocket; + SetState(DataChannelConnectionState::Open); + + DC_DEBUG(("Negotiated number of incoming streams: %" PRIu16, + sac->sac_inbound_streams)); + DC_DEBUG(("Negotiated number of outgoing streams: %" PRIu16, + sac->sac_outbound_streams)); + mNegotiatedIdLimit = + std::max(mNegotiatedIdLimit, + static_cast<size_t>(std::max(sac->sac_outbound_streams, + sac->sac_inbound_streams))); + + Dispatch(do_AddRef(new DataChannelOnMessageAvailable( + DataChannelOnMessageAvailable::EventType::OnConnection, this))); + DC_DEBUG(("DTLS connect() succeeded! Entering connected mode")); + + // Open any streams pending... + ProcessQueuedOpens(); + + } else if (state == DataChannelConnectionState::Open) { + DC_DEBUG(("DataConnection Already OPEN")); + } else { + DC_ERROR(("Unexpected state: %s", ToString(state))); + } + break; + case SCTP_COMM_LOST: + DC_DEBUG(("Association change: SCTP_COMM_LOST")); + // This association is toast, so also close all the channels -- from + // mainthread! + Stop(); + break; + case SCTP_RESTART: + DC_DEBUG(("Association change: SCTP_RESTART")); + break; + case SCTP_SHUTDOWN_COMP: + DC_DEBUG(("Association change: SCTP_SHUTDOWN_COMP")); + Stop(); + break; + case SCTP_CANT_STR_ASSOC: + DC_DEBUG(("Association change: SCTP_CANT_STR_ASSOC")); + break; + default: + DC_DEBUG(("Association change: UNKNOWN")); + break; + } + DC_DEBUG(("Association change: streams (in/out) = (%u/%u)", + sac->sac_inbound_streams, sac->sac_outbound_streams)); + + if (NS_WARN_IF(!sac)) { + return; + } + + n = sac->sac_length - sizeof(*sac); + if ((sac->sac_state == SCTP_COMM_UP) || (sac->sac_state == SCTP_RESTART)) { + if (n > 0) { + for (i = 0; i < n; ++i) { + switch (sac->sac_info[i]) { + case SCTP_ASSOC_SUPPORTS_PR: + DC_DEBUG(("Supports: PR")); + break; + case SCTP_ASSOC_SUPPORTS_AUTH: + DC_DEBUG(("Supports: AUTH")); + break; + case SCTP_ASSOC_SUPPORTS_ASCONF: + DC_DEBUG(("Supports: ASCONF")); + break; + case SCTP_ASSOC_SUPPORTS_MULTIBUF: + DC_DEBUG(("Supports: MULTIBUF")); + break; + case SCTP_ASSOC_SUPPORTS_RE_CONFIG: + DC_DEBUG(("Supports: RE-CONFIG")); + break; +#if defined(SCTP_ASSOC_SUPPORTS_INTERLEAVING) + case SCTP_ASSOC_SUPPORTS_INTERLEAVING: + DC_DEBUG(("Supports: NDATA")); + // TODO: This should probably be set earlier above in 'case + // SCTP_COMM_UP' but we also need this for 'SCTP_RESTART'. + mSendInterleaved = true; + break; +#endif + default: + DC_ERROR(("Supports: UNKNOWN(0x%02x)", sac->sac_info[i])); + break; + } + } + } + } else if (((sac->sac_state == SCTP_COMM_LOST) || + (sac->sac_state == SCTP_CANT_STR_ASSOC)) && + (n > 0)) { + DC_DEBUG(("Association: ABORT =")); + for (i = 0; i < n; ++i) { + DC_DEBUG((" 0x%02x", sac->sac_info[i])); + } + } + if ((sac->sac_state == SCTP_CANT_STR_ASSOC) || + (sac->sac_state == SCTP_SHUTDOWN_COMP) || + (sac->sac_state == SCTP_COMM_LOST)) { + return; + } +} + +void DataChannelConnection::HandlePeerAddressChangeEvent( + const struct sctp_paddr_change* spc) { + const char* addr = ""; +#if !defined(__Userspace_os_Windows) + char addr_buf[INET6_ADDRSTRLEN]; + struct sockaddr_in* sin; + struct sockaddr_in6* sin6; +#endif + + switch (spc->spc_aaddr.ss_family) { + case AF_INET: +#if !defined(__Userspace_os_Windows) + sin = (struct sockaddr_in*)&spc->spc_aaddr; + addr = inet_ntop(AF_INET, &sin->sin_addr, addr_buf, INET6_ADDRSTRLEN); +#endif + break; + case AF_INET6: +#if !defined(__Userspace_os_Windows) + sin6 = (struct sockaddr_in6*)&spc->spc_aaddr; + addr = inet_ntop(AF_INET6, &sin6->sin6_addr, addr_buf, INET6_ADDRSTRLEN); +#endif + break; + case AF_CONN: + addr = "DTLS connection"; + break; + default: + break; + } + DC_DEBUG(("Peer address %s is now ", addr)); + switch (spc->spc_state) { + case SCTP_ADDR_AVAILABLE: + DC_DEBUG(("SCTP_ADDR_AVAILABLE")); + break; + case SCTP_ADDR_UNREACHABLE: + DC_DEBUG(("SCTP_ADDR_UNREACHABLE")); + break; + case SCTP_ADDR_REMOVED: + DC_DEBUG(("SCTP_ADDR_REMOVED")); + break; + case SCTP_ADDR_ADDED: + DC_DEBUG(("SCTP_ADDR_ADDED")); + break; + case SCTP_ADDR_MADE_PRIM: + DC_DEBUG(("SCTP_ADDR_MADE_PRIM")); + break; + case SCTP_ADDR_CONFIRMED: + DC_DEBUG(("SCTP_ADDR_CONFIRMED")); + break; + default: + DC_ERROR(("UNKNOWN SCP STATE")); + break; + } + if (spc->spc_error) { + DC_ERROR((" (error = 0x%08x).\n", spc->spc_error)); + } +} + +void DataChannelConnection::HandleRemoteErrorEvent( + const struct sctp_remote_error* sre) { + size_t i, n; + + n = sre->sre_length - sizeof(struct sctp_remote_error); + DC_WARN(("Remote Error (error = 0x%04x): ", sre->sre_error)); + for (i = 0; i < n; ++i) { + DC_WARN((" 0x%02x", sre->sre_data[i])); + } +} + +void DataChannelConnection::HandleShutdownEvent( + const struct sctp_shutdown_event* sse) { + DC_DEBUG(("Shutdown event.")); + /* XXX: notify all channels. */ + // Attempts to actually send anything will fail +} + +void DataChannelConnection::HandleAdaptationIndication( + const struct sctp_adaptation_event* sai) { + DC_DEBUG(("Adaptation indication: %x.", sai->sai_adaptation_ind)); +} + +void DataChannelConnection::HandlePartialDeliveryEvent( + const struct sctp_pdapi_event* spde) { + // Note: Be aware that stream and sequence number being u32 instead of u16 is + // a bug in the SCTP API. This may change in the future. + + DC_DEBUG(("Partial delivery event: ")); + switch (spde->pdapi_indication) { + case SCTP_PARTIAL_DELIVERY_ABORTED: + DC_DEBUG(("delivery aborted ")); + break; + default: + DC_ERROR(("??? ")); + break; + } + DC_DEBUG(("(flags = %x), stream = %" PRIu32 ", sn = %" PRIu32, + spde->pdapi_flags, spde->pdapi_stream, spde->pdapi_seq)); + + // Validate stream ID + if (spde->pdapi_stream >= UINT16_MAX) { + DC_ERROR(("Invalid stream id in partial delivery event: %" PRIu32 "\n", + spde->pdapi_stream)); + return; + } + + // Find channel and reset buffer + DataChannel* channel = FindChannelByStream((uint16_t)spde->pdapi_stream); + if (channel) { + DC_WARN(("Abort partially delivered message of %zu bytes\n", + channel->mRecvBuffer.Length())); + channel->mRecvBuffer.Truncate(0); + } +} + +void DataChannelConnection::HandleSendFailedEvent( + const struct sctp_send_failed_event* ssfe) { + size_t i, n; + + if (ssfe->ssfe_flags & SCTP_DATA_UNSENT) { + DC_DEBUG(("Unsent ")); + } + if (ssfe->ssfe_flags & SCTP_DATA_SENT) { + DC_DEBUG(("Sent ")); + } + if (ssfe->ssfe_flags & ~(SCTP_DATA_SENT | SCTP_DATA_UNSENT)) { + DC_DEBUG(("(flags = %x) ", ssfe->ssfe_flags)); + } +#ifdef XP_WIN +# define PRIPPID "lu" +#else +# define PRIPPID "u" +#endif + DC_DEBUG(("message with PPID = %" PRIPPID + ", SID = %d, flags: 0x%04x due to error = 0x%08x", + ntohl(ssfe->ssfe_info.snd_ppid), ssfe->ssfe_info.snd_sid, + ssfe->ssfe_info.snd_flags, ssfe->ssfe_error)); +#undef PRIPPID + n = ssfe->ssfe_length - sizeof(struct sctp_send_failed_event); + for (i = 0; i < n; ++i) { + DC_DEBUG((" 0x%02x", ssfe->ssfe_data[i])); + } +} + +void DataChannelConnection::ClearResets() { + // Clear all pending resets + if (!mStreamsResetting.IsEmpty()) { + DC_DEBUG(("Clearing resets for %zu streams", mStreamsResetting.Length())); + } + + for (uint32_t i = 0; i < mStreamsResetting.Length(); ++i) { + RefPtr<DataChannel> channel; + channel = FindChannelByStream(mStreamsResetting[i]); + if (channel) { + DC_DEBUG(("Forgetting channel %u (%p) with pending reset", + channel->mStream, channel.get())); + // TODO: Do we _really_ want to remove this? Are we allowed to reuse the + // id? + mChannels.Remove(channel); + } + } + mStreamsResetting.Clear(); +} + +void DataChannelConnection::ResetOutgoingStream(uint16_t stream) { + uint32_t i; + + mLock.AssertCurrentThreadOwns(); + DC_DEBUG( + ("Connection %p: Resetting outgoing stream %u", (void*)this, stream)); + // Rarely has more than a couple items and only for a short time + for (i = 0; i < mStreamsResetting.Length(); ++i) { + if (mStreamsResetting[i] == stream) { + return; + } + } + mStreamsResetting.AppendElement(stream); +} + +void DataChannelConnection::SendOutgoingStreamReset() { + struct sctp_reset_streams* srs; + uint32_t i; + size_t len; + + DC_DEBUG(("Connection %p: Sending outgoing stream reset for %zu streams", + (void*)this, mStreamsResetting.Length())); + mLock.AssertCurrentThreadOwns(); + if (mStreamsResetting.IsEmpty()) { + DC_DEBUG(("No streams to reset")); + return; + } + len = sizeof(sctp_assoc_t) + + (2 + mStreamsResetting.Length()) * sizeof(uint16_t); + srs = static_cast<struct sctp_reset_streams*>( + moz_xmalloc(len)); // infallible malloc + memset(srs, 0, len); + srs->srs_flags = SCTP_STREAM_RESET_OUTGOING; + srs->srs_number_streams = mStreamsResetting.Length(); + for (i = 0; i < mStreamsResetting.Length(); ++i) { + srs->srs_stream_list[i] = mStreamsResetting[i]; + } + if (usrsctp_setsockopt(mMasterSocket, IPPROTO_SCTP, SCTP_RESET_STREAMS, srs, + (socklen_t)len) < 0) { + DC_ERROR(("***failed: setsockopt RESET, errno %d", errno)); + // if errno == EALREADY, this is normal - we can't send another reset + // with one pending. + // When we get an incoming reset (which may be a response to our + // outstanding one), see if we have any pending outgoing resets and + // send them + } else { + mStreamsResetting.Clear(); + } + free(srs); +} + +void DataChannelConnection::HandleStreamResetEvent( + const struct sctp_stream_reset_event* strrst) { + uint32_t n, i; + RefPtr<DataChannel> channel; // since we may null out the ref to the channel + + if (!(strrst->strreset_flags & SCTP_STREAM_RESET_DENIED) && + !(strrst->strreset_flags & SCTP_STREAM_RESET_FAILED)) { + n = (strrst->strreset_length - sizeof(struct sctp_stream_reset_event)) / + sizeof(uint16_t); + for (i = 0; i < n; ++i) { + if (strrst->strreset_flags & SCTP_STREAM_RESET_INCOMING_SSN) { + channel = FindChannelByStream(strrst->strreset_stream_list[i]); + if (channel) { + // The other side closed the channel + // We could be in three states: + // 1. Normal state (input and output streams (OPEN) + // Notify application, send a RESET in response on our + // outbound channel. Go to CLOSED + // 2. We sent our own reset (CLOSING); either they crossed on the + // wire, or this is a response to our Reset. + // Go to CLOSED + // 3. We've sent a open but haven't gotten a response yet (CONNECTING) + // I believe this is impossible, as we don't have an input stream + // yet. + + DC_DEBUG(("Incoming: Channel %u closed", channel->mStream)); + if (mChannels.Remove(channel)) { + // Mark the stream for reset (the reset is sent below) + ResetOutgoingStream(channel->mStream); + } + + DC_DEBUG(("Disconnected DataChannel %p from connection %p", + (void*)channel.get(), (void*)channel->mConnection.get())); + channel->StreamClosedLocked(); + } else { + DC_WARN(("Can't find incoming channel %d", i)); + } + } + } + } + + // Process any pending resets now: + if (!mStreamsResetting.IsEmpty()) { + DC_DEBUG(("Sending %zu pending resets", mStreamsResetting.Length())); + SendOutgoingStreamReset(); + } +} + +void DataChannelConnection::HandleStreamChangeEvent( + const struct sctp_stream_change_event* strchg) { + ASSERT_WEBRTC(!NS_IsMainThread()); + if (strchg->strchange_flags == SCTP_STREAM_CHANGE_DENIED) { + DC_ERROR(("*** Failed increasing number of streams from %zu (%u/%u)", + mNegotiatedIdLimit, strchg->strchange_instrms, + strchg->strchange_outstrms)); + // XXX FIX! notify pending opens of failure + return; + } + if (strchg->strchange_instrms > mNegotiatedIdLimit) { + DC_DEBUG(("Other side increased streams from %zu to %u", mNegotiatedIdLimit, + strchg->strchange_instrms)); + } + uint16_t old_limit = mNegotiatedIdLimit; + uint16_t new_limit = + std::max(strchg->strchange_outstrms, strchg->strchange_instrms); + if (new_limit > mNegotiatedIdLimit) { + DC_DEBUG(("Increasing number of streams from %u to %u - adding %u (in: %u)", + old_limit, new_limit, new_limit - old_limit, + strchg->strchange_instrms)); + // make sure both are the same length + mNegotiatedIdLimit = new_limit; + DC_DEBUG(("New length = %zu (was %d)", mNegotiatedIdLimit, old_limit)); + // Re-process any channels waiting for streams. + // Linear search, but we don't increase channels often and + // the array would only get long in case of an app error normally + + // Make sure we request enough streams if there's a big jump in streams + // Could make a more complex API for OpenXxxFinish() and avoid this loop + auto channels = mChannels.GetAll(); + size_t num_needed = + channels.Length() ? (channels.LastElement()->mStream + 1) : 0; + MOZ_ASSERT(num_needed != INVALID_STREAM); + if (num_needed > new_limit) { + int32_t more_needed = num_needed - ((int32_t)mNegotiatedIdLimit) + 16; + DC_DEBUG(("Not enough new streams, asking for %d more", more_needed)); + // TODO: parameter is an int32_t but we pass size_t + RequestMoreStreams(more_needed); + } else if (strchg->strchange_outstrms < strchg->strchange_instrms) { + DC_DEBUG(("Requesting %d output streams to match partner", + strchg->strchange_instrms - strchg->strchange_outstrms)); + RequestMoreStreams(strchg->strchange_instrms - + strchg->strchange_outstrms); + } + + ProcessQueuedOpens(); + } + // else probably not a change in # of streams + + if ((strchg->strchange_flags & SCTP_STREAM_CHANGE_DENIED) || + (strchg->strchange_flags & SCTP_STREAM_CHANGE_FAILED)) { + // Other side denied our request. Need to AnnounceClosed some stuff. + for (auto& channel : mChannels.GetAll()) { + if (channel->mStream >= mNegotiatedIdLimit) { + /* XXX: Signal to the other end. */ + channel->AnnounceClosed(); + // maybe fire onError (bug 843625) + } + } + } +} + +// Called with mLock locked! +void DataChannelConnection::HandleNotification( + const union sctp_notification* notif, size_t n) { + mLock.AssertCurrentThreadOwns(); + if (notif->sn_header.sn_length != (uint32_t)n) { + return; + } + switch (notif->sn_header.sn_type) { + case SCTP_ASSOC_CHANGE: + HandleAssociationChangeEvent(&(notif->sn_assoc_change)); + break; + case SCTP_PEER_ADDR_CHANGE: + HandlePeerAddressChangeEvent(&(notif->sn_paddr_change)); + break; + case SCTP_REMOTE_ERROR: + HandleRemoteErrorEvent(&(notif->sn_remote_error)); + break; + case SCTP_SHUTDOWN_EVENT: + HandleShutdownEvent(&(notif->sn_shutdown_event)); + break; + case SCTP_ADAPTATION_INDICATION: + HandleAdaptationIndication(&(notif->sn_adaptation_event)); + break; + case SCTP_AUTHENTICATION_EVENT: + DC_DEBUG(("SCTP_AUTHENTICATION_EVENT")); + break; + case SCTP_SENDER_DRY_EVENT: + // DC_DEBUG(("SCTP_SENDER_DRY_EVENT")); + break; + case SCTP_NOTIFICATIONS_STOPPED_EVENT: + DC_DEBUG(("SCTP_NOTIFICATIONS_STOPPED_EVENT")); + break; + case SCTP_PARTIAL_DELIVERY_EVENT: + HandlePartialDeliveryEvent(&(notif->sn_pdapi_event)); + break; + case SCTP_SEND_FAILED_EVENT: + HandleSendFailedEvent(&(notif->sn_send_failed_event)); + break; + case SCTP_STREAM_RESET_EVENT: + HandleStreamResetEvent(&(notif->sn_strreset_event)); + break; + case SCTP_ASSOC_RESET_EVENT: + DC_DEBUG(("SCTP_ASSOC_RESET_EVENT")); + break; + case SCTP_STREAM_CHANGE_EVENT: + HandleStreamChangeEvent(&(notif->sn_strchange_event)); + break; + default: + DC_ERROR(("unknown SCTP event: %u", (uint32_t)notif->sn_header.sn_type)); + break; + } +} + +int DataChannelConnection::ReceiveCallback( + struct socket* sock, void* data, size_t datalen, struct sctp_rcvinfo rcv, + int flags) MOZ_NO_THREAD_SAFETY_ANALYSIS { + ASSERT_WEBRTC(!NS_IsMainThread()); + DC_DEBUG(("In ReceiveCallback")); + + // libusrsctp just went reentrant on us. Put a stop to this. + mSTS->Dispatch(NS_NewRunnableFunction( + "DataChannelConnection::ReceiveCallback", + [data, datalen, rcv, flags, this, + self = RefPtr<DataChannelConnection>(this)]() mutable { + if (!data) { + DC_DEBUG(("ReceiveCallback: SCTP has finished shutting down")); + } else { + mLock.Lock(); + if (flags & MSG_NOTIFICATION) { + HandleNotification(static_cast<union sctp_notification*>(data), + datalen); + } else { + HandleMessage(data, datalen, ntohl(rcv.rcv_ppid), rcv.rcv_sid, + flags); + } + mLock.Unlock(); + // sctp allocates 'data' with malloc(), and expects the receiver to + // free it (presumably with free). + // XXX future optimization: try to deliver messages without an + // internal alloc/copy, and if so delay the free until later. + free(data); + } + })); + + // usrsctp defines the callback as returning an int, but doesn't use it + return 1; +} + +already_AddRefed<DataChannel> DataChannelConnection::Open( + const nsACString& label, const nsACString& protocol, + DataChannelReliabilityPolicy prPolicy, bool inOrder, uint32_t prValue, + DataChannelListener* aListener, nsISupports* aContext, + bool aExternalNegotiated, uint16_t aStream) { + ASSERT_WEBRTC(NS_IsMainThread()); + MutexAutoLock lock(mLock); // OpenFinish assumes this + if (!aExternalNegotiated) { + if (mAllocateEven.isSome()) { + aStream = FindFreeStream(); + if (aStream == INVALID_STREAM) { + return nullptr; + } + } else { + // We do not yet know whether we are client or server, and an id has not + // been chosen for us. We will need to choose later. + aStream = INVALID_STREAM; + } + } + + DC_DEBUG( + ("DC Open: label %s/%s, type %s, inorder %d, prValue %u, listener %p, " + "context %p, external: %s, stream %u", + PromiseFlatCString(label).get(), PromiseFlatCString(protocol).get(), + ToString(prPolicy), inOrder, prValue, aListener, aContext, + aExternalNegotiated ? "true" : "false", aStream)); + + if ((prPolicy == DataChannelReliabilityPolicy::Reliable) && (prValue != 0)) { + return nullptr; + } + + if (aStream != INVALID_STREAM && mChannels.Get(aStream)) { + DC_ERROR(("external negotiation of already-open channel %u", aStream)); + return nullptr; + } + + RefPtr<DataChannel> channel(new DataChannel( + this, aStream, DataChannelState::Connecting, label, protocol, prPolicy, + prValue, inOrder, aExternalNegotiated, aListener, aContext)); + mChannels.Insert(channel); + + return OpenFinish(channel.forget()); +} + +// Separate routine so we can also call it to finish up from pending opens +already_AddRefed<DataChannel> DataChannelConnection::OpenFinish( + already_AddRefed<DataChannel>&& aChannel) { + RefPtr<DataChannel> channel(aChannel); // takes the reference passed in + // Normally 1 reference if called from ::Open(), or 2 if called from + // ProcessQueuedOpens() unless the DOMDataChannel was gc'd + const uint16_t stream = channel->mStream; + + mLock.AssertCurrentThreadOwns(); + + // Cases we care about: + // Pre-negotiated: + // Not Open: + // Doesn't fit: + // -> change initial ask or renegotiate after open + // -> queue open + // Open: + // Doesn't fit: + // -> RequestMoreStreams && queue + // Does fit: + // -> open + // Not negotiated: + // Not Open: + // -> queue open + // Open: + // -> Try to get a stream + // Doesn't fit: + // -> RequestMoreStreams && queue + // Does fit: + // -> open + // So the Open cases are basically the same + // Not Open cases are simply queue for non-negotiated, and + // either change the initial ask or possibly renegotiate after open. + DataChannelConnectionState state = GetState(); + if (state != DataChannelConnectionState::Open || + stream >= mNegotiatedIdLimit) { + if (state == DataChannelConnectionState::Open) { + MOZ_ASSERT(stream != INVALID_STREAM); + // RequestMoreStreams() limits to MAX_NUM_STREAMS -- allocate extra + // streams to avoid going back immediately for more if the ask to N, N+1, + // etc + int32_t more_needed = stream - ((int32_t)mNegotiatedIdLimit) + 16; + if (!RequestMoreStreams(more_needed)) { + // Something bad happened... we're done + goto request_error_cleanup; + } + } + DC_DEBUG(("Queuing channel %p (%u) to finish open", channel.get(), stream)); + // Also serves to mark we told the app + channel->mHasFinishedOpen = true; + mPending.Push(channel); + return channel.forget(); + } + + MOZ_ASSERT(stream != INVALID_STREAM); + MOZ_ASSERT(stream < mNegotiatedIdLimit); + +#ifdef TEST_QUEUED_DATA + // It's painful to write a test for this... + channel->AnnounceOpen(); + SendDataMsgInternalOrBuffer(channel, "Help me!", 8, + DATA_CHANNEL_PPID_DOMSTRING); +#endif + + if (!channel->mNegotiated) { + if (!channel->mOrdered) { + // Don't send unordered until this gets cleared. + channel->mWaitingForAck = true; + } + + int error = SendOpenRequestMessage(channel->mLabel, channel->mProtocol, + stream, !channel->mOrdered, + channel->mPrPolicy, channel->mPrValue); + if (error) { + DC_ERROR(("SendOpenRequest failed, error = %d", error)); + if (channel->mHasFinishedOpen) { + // We already returned the channel to the app. + NS_ERROR("Failed to send open request"); + channel->AnnounceClosed(); + } + // If we haven't returned the channel yet, it will get destroyed when we + // exit this function. + mChannels.Remove(channel); + // we'll be destroying the channel + return nullptr; + /* NOTREACHED */ + } + } + + // Either externally negotiated or we sent Open + // FIX? Move into DOMDataChannel? I don't think we can send it yet here + channel->AnnounceOpen(); + + return channel.forget(); + +request_error_cleanup: + if (channel->mHasFinishedOpen) { + // We already returned the channel to the app. + NS_ERROR("Failed to request more streams"); + channel->AnnounceClosed(); + return channel.forget(); + } + // we'll be destroying the channel, but it never really got set up + // Alternative would be to RUN_ON_THREAD(channel.forget(),::Destroy,...) and + // Dispatch it to ourselves + return nullptr; +} + +// Requires mLock to be locked! +// Returns a POSIX error code directly instead of setting errno. +int DataChannelConnection::SendMsgInternal(OutgoingMsg& msg, size_t* aWritten) { + struct sctp_sndinfo& info = msg.GetInfo().sendv_sndinfo; + int error; + + // EOR set? + bool eor_set = (info.snd_flags & SCTP_EOR) != 0; + + // Send until buffer is empty + size_t left = msg.GetLeft(); + do { + size_t length; + + // Carefully chunk the buffer + if (left > DATA_CHANNEL_MAX_BINARY_FRAGMENT) { + length = DATA_CHANNEL_MAX_BINARY_FRAGMENT; + + // Unset EOR flag + info.snd_flags &= ~SCTP_EOR; + } else { + length = left; + + // Set EOR flag + if (eor_set) { + info.snd_flags |= SCTP_EOR; + } + } + + // Send (or try at least) + // SCTP will return EMSGSIZE if the message is bigger than the buffer + // size (or EAGAIN if there isn't space). However, we can avoid EMSGSIZE + // by carefully crafting small enough message chunks. + ssize_t written = usrsctp_sendv( + mSocket, msg.GetData(), length, nullptr, 0, (void*)&msg.GetInfo(), + (socklen_t)sizeof(struct sctp_sendv_spa), SCTP_SENDV_SPA, 0); + + if (written < 0) { + error = errno; + goto out; + } + + if (aWritten) { + *aWritten += written; + } + DC_DEBUG(("Sent buffer (written=%zu, len=%zu, left=%zu)", (size_t)written, + length, left - (size_t)written)); + + // TODO: Remove once resolved + // (https://github.com/sctplab/usrsctp/issues/132) + if (written == 0) { + DC_ERROR(("@tuexen: usrsctp_sendv returned 0")); + error = EAGAIN; + goto out; + } + + // If not all bytes have been written, this obviously means that usrsctp's + // buffer is full and we need to try again later. + if ((size_t)written < length) { + msg.Advance((size_t)written); + error = EAGAIN; + goto out; + } + + // Update buffer position + msg.Advance((size_t)written); + + // Get amount of bytes left in the buffer + left = msg.GetLeft(); + } while (left > 0); + + // Done + error = 0; + +out: + // Reset EOR flag + if (eor_set) { + info.snd_flags |= SCTP_EOR; + } + + return error; +} + +// Requires mLock to be locked! +// Returns a POSIX error code directly instead of setting errno. +// IMPORTANT: Ensure that the buffer passed is guarded by mLock! +int DataChannelConnection::SendMsgInternalOrBuffer( + nsTArray<UniquePtr<BufferedOutgoingMsg>>& buffer, OutgoingMsg& msg, + bool& buffered, size_t* aWritten) { + NS_WARNING_ASSERTION(msg.GetLength() > 0, "Length is 0?!"); + + int error = 0; + bool need_buffering = false; + + // Note: Main-thread IO, but doesn't block! + // XXX FIX! to deal with heavy overruns of JS trying to pass data in + // (more than the buffersize) queue data onto another thread to do the + // actual sends. See netwerk/protocol/websocket/WebSocketChannel.cpp + + // Avoid a race between buffer-full-failure (where we have to add the + // packet to the buffered-data queue) and the buffer-now-only-half-full + // callback, which happens on a different thread. Otherwise we might + // fail here, then before we add it to the queue get the half-full + // callback, find nothing to do, then on this thread add it to the + // queue - which would sit there. Also, if we later send more data, it + // would arrive ahead of the buffered message, but if the buffer ever + // got to 1/2 full, the message would get sent - but at a semi-random + // time, after other data it was supposed to be in front of. + + // Must lock before empty check for similar reasons! + mLock.AssertCurrentThreadOwns(); + if (buffer.IsEmpty() && + (mSendInterleaved || mPendingType == PendingType::None)) { + error = SendMsgInternal(msg, aWritten); + switch (error) { + case 0: + break; + case EAGAIN: +#if (EAGAIN != EWOULDBLOCK) + case EWOULDBLOCK: +#endif + need_buffering = true; + break; + default: + DC_ERROR(("error %d on sending", error)); + break; + } + } else { + need_buffering = true; + } + + if (need_buffering) { + // queue data for resend! And queue any further data for the stream until + // it is... + auto* bufferedMsg = new BufferedOutgoingMsg(msg); // infallible malloc + buffer.AppendElement(bufferedMsg); // owned by mBufferedData array + DC_DEBUG(("Queued %zu buffers (left=%zu, total=%zu)", buffer.Length(), + msg.GetLeft(), msg.GetLength())); + buffered = true; + return 0; + } + + buffered = false; + return error; +} + +// Caller must ensure that length <= UINT32_MAX +// Returns a POSIX error code. +int DataChannelConnection::SendDataMsgInternalOrBuffer(DataChannel& channel, + const uint8_t* data, + size_t len, + uint32_t ppid) { + if (NS_WARN_IF(channel.GetReadyState() != DataChannelState::Open)) { + return EINVAL; // TODO: Find a better error code + } + + struct sctp_sendv_spa info = {}; + + // General flags + info.sendv_flags = SCTP_SEND_SNDINFO_VALID; + + // Set stream identifier, protocol identifier and flags + info.sendv_sndinfo.snd_sid = channel.mStream; + info.sendv_sndinfo.snd_flags = SCTP_EOR; + info.sendv_sndinfo.snd_ppid = htonl(ppid); + + // Unordered? + // To avoid problems where an in-order OPEN is lost and an + // out-of-order data message "beats" it, require data to be in-order + // until we get an ACK. + if (!channel.mOrdered && !channel.mWaitingForAck) { + info.sendv_sndinfo.snd_flags |= SCTP_UNORDERED; + } + + // Partial reliability policy + if (channel.mPrPolicy != DataChannelReliabilityPolicy::Reliable) { + info.sendv_prinfo.pr_policy = ToUsrsctpValue(channel.mPrPolicy); + info.sendv_prinfo.pr_value = channel.mPrValue; + info.sendv_flags |= SCTP_SEND_PRINFO_VALID; + } + + // Create message instance and send + OutgoingMsg msg(info, data, len); + bool buffered; + size_t written = 0; + mDeferSend = true; + int error = + SendMsgInternalOrBuffer(channel.mBufferedData, msg, buffered, &written); + mDeferSend = false; + if (written && ppid != DATA_CHANNEL_PPID_DOMSTRING_EMPTY && + ppid != DATA_CHANNEL_PPID_BINARY_EMPTY) { + channel.DecrementBufferedAmount(written); + } + + for (auto&& packet : mDeferredSend) { + MOZ_ASSERT(written); + SendPacket(std::move(packet)); + } + mDeferredSend.clear(); + + // Set pending type and stream index (if buffered) + if (!error && buffered && mPendingType == PendingType::None) { + mPendingType = PendingType::Data; + mCurrentStream = channel.mStream; + } + return error; +} + +// Caller must ensure that length <= UINT32_MAX +// Returns a POSIX error code. +int DataChannelConnection::SendDataMsg(DataChannel& channel, + const uint8_t* data, size_t len, + uint32_t ppidPartial, + uint32_t ppidFinal) { + // We *really* don't want to do this from main thread! - and + // SendDataMsgInternalOrBuffer avoids blocking. + mLock.AssertCurrentThreadOwns(); + + if (mMaxMessageSize != 0 && len > mMaxMessageSize) { + DC_ERROR(("Message rejected, too large (%zu > %" PRIu64 ")", len, + mMaxMessageSize)); + return EMSGSIZE; + } + + // This will use EOR-based fragmentation if the message is too large (> 64 + // KiB) + return SendDataMsgInternalOrBuffer(channel, data, len, ppidFinal); +} + +class ReadBlobRunnable : public Runnable { + public: + ReadBlobRunnable(DataChannelConnection* aConnection, uint16_t aStream, + nsIInputStream* aBlob) + : Runnable("ReadBlobRunnable"), + mConnection(aConnection), + mStream(aStream), + mBlob(aBlob) {} + + NS_IMETHOD Run() override { + // ReadBlob() is responsible to releasing the reference + DataChannelConnection* self = mConnection; + self->ReadBlob(mConnection.forget(), mStream, mBlob); + return NS_OK; + } + + private: + // Make sure the Connection doesn't die while there are jobs outstanding. + // Let it die (if released by PeerConnectionImpl while we're running) + // when we send our runnable back to MainThread. Then ~DataChannelConnection + // can send the IOThread to MainThread to die in a runnable, avoiding + // unsafe event loop recursion. Evil. + RefPtr<DataChannelConnection> mConnection; + uint16_t mStream; + // Use RefCount for preventing the object is deleted when SendBlob returns. + RefPtr<nsIInputStream> mBlob; +}; + +// Returns a POSIX error code. +int DataChannelConnection::SendBlob(uint16_t stream, nsIInputStream* aBlob) { + MutexAutoLock lock(mLock); + RefPtr<DataChannel> channel = mChannels.Get(stream); + if (NS_WARN_IF(!channel)) { + return EINVAL; // TODO: Find a better error code + } + + // Spawn a thread to send the data + if (!mInternalIOThread) { + nsresult rv = + NS_NewNamedThread("DataChannel IO", getter_AddRefs(mInternalIOThread)); + if (NS_FAILED(rv)) { + return EINVAL; // TODO: Find a better error code + } + } + + mInternalIOThread->Dispatch( + do_AddRef(new ReadBlobRunnable(this, stream, aBlob)), NS_DISPATCH_NORMAL); + return 0; +} + +class DataChannelBlobSendRunnable : public Runnable { + public: + DataChannelBlobSendRunnable( + already_AddRefed<DataChannelConnection>& aConnection, uint16_t aStream) + : Runnable("DataChannelBlobSendRunnable"), + mConnection(aConnection), + mStream(aStream) {} + + ~DataChannelBlobSendRunnable() override { + if (!NS_IsMainThread() && mConnection) { + MOZ_ASSERT(false); + // explicitly leak the connection if destroyed off mainthread + Unused << mConnection.forget().take(); + } + } + + NS_IMETHOD Run() override { + ASSERT_WEBRTC(NS_IsMainThread()); + + mConnection->SendBinaryMsg(mStream, mData); + mConnection = nullptr; + return NS_OK; + } + + // explicitly public so we can avoid allocating twice and copying + nsCString mData; + + private: + // Note: we can be destroyed off the target thread, so be careful not to let + // this get Released()ed on the temp thread! + RefPtr<DataChannelConnection> mConnection; + uint16_t mStream; +}; + +void DataChannelConnection::SetState(DataChannelConnectionState aState) { + mLock.AssertCurrentThreadOwns(); + + DC_DEBUG( + ("DataChannelConnection labeled %s (%p) switching connection state %s -> " + "%s", + mTransportId.c_str(), this, ToString(mState), ToString(aState))); + + mState = aState; +} + +void DataChannelConnection::ReadBlob( + already_AddRefed<DataChannelConnection> aThis, uint16_t aStream, + nsIInputStream* aBlob) { + // NOTE: 'aThis' has been forgotten by the caller to avoid releasing + // it off mainthread; if PeerConnectionImpl has released then we want + // ~DataChannelConnection() to run on MainThread + + // XXX to do this safely, we must enqueue these atomically onto the + // output socket. We need a sender thread(s?) to enqueue data into the + // socket and to avoid main-thread IO that might block. Even on a + // background thread, we may not want to block on one stream's data. + // I.e. run non-blocking and service multiple channels. + + // Must not let Dispatching it cause the DataChannelConnection to get + // released on the wrong thread. Using + // WrapRunnable(RefPtr<DataChannelConnection>(aThis),... will occasionally + // cause aThis to get released on this thread. Also, an explicit Runnable + // lets us avoid copying the blob data an extra time. + RefPtr<DataChannelBlobSendRunnable> runnable = + new DataChannelBlobSendRunnable(aThis, aStream); + // avoid copying the blob data by passing the mData from the runnable + if (NS_FAILED(NS_ReadInputStreamToString(aBlob, runnable->mData, -1))) { + // Bug 966602: Doesn't return an error to the caller via onerror. + // We must release DataChannelConnection on MainThread to avoid issues (bug + // 876167) aThis is now owned by the runnable; release it there + NS_ReleaseOnMainThread("DataChannelBlobSendRunnable", runnable.forget()); + return; + } + aBlob->Close(); + Dispatch(runnable.forget()); +} + +// Returns a POSIX error code. +int DataChannelConnection::SendDataMsgCommon(uint16_t stream, + const nsACString& aMsg, + bool isBinary) { + ASSERT_WEBRTC(NS_IsMainThread()); + // We really could allow this from other threads, so long as we deal with + // asynchronosity issues with channels closing, in particular access to + // mChannels, and issues with the association closing (access to mSocket). + + const uint8_t* data = (const uint8_t*)aMsg.BeginReading(); + uint32_t len = aMsg.Length(); +#if (UINT32_MAX > SIZE_MAX) + if (len > SIZE_MAX) { + return EMSGSIZE; + } +#endif + + DC_DEBUG(("Sending %sto stream %u: %u bytes", isBinary ? "binary " : "", + stream, len)); + // XXX if we want more efficiency, translate flags once at open time + RefPtr<DataChannel> channelPtr = mChannels.Get(stream); + if (NS_WARN_IF(!channelPtr)) { + return EINVAL; // TODO: Find a better error code + } + bool is_empty = len == 0; + const uint8_t byte = 0; + if (is_empty) { + data = &byte; + len = 1; + } + auto& channel = *channelPtr; + int err = 0; + MutexAutoLock lock(mLock); + if (isBinary) { + err = SendDataMsg( + channel, data, len, DATA_CHANNEL_PPID_BINARY_PARTIAL, + is_empty ? DATA_CHANNEL_PPID_BINARY_EMPTY : DATA_CHANNEL_PPID_BINARY); + } else { + err = SendDataMsg(channel, data, len, DATA_CHANNEL_PPID_DOMSTRING_PARTIAL, + is_empty ? DATA_CHANNEL_PPID_DOMSTRING_EMPTY + : DATA_CHANNEL_PPID_DOMSTRING); + } + if (!err) { + channel.WithTrafficCounters([&len](DataChannel::TrafficCounters& counters) { + counters.mMessagesSent++; + counters.mBytesSent += len; + }); + } + + return err; +} + +void DataChannelConnection::Stop() { + // Note: This will call 'CloseAll' from the main thread + Dispatch(do_AddRef(new DataChannelOnMessageAvailable( + DataChannelOnMessageAvailable::EventType::OnDisconnected, this))); +} + +void DataChannelConnection::Close(DataChannel* aChannel) { + MutexAutoLock lock(mLock); + CloseLocked(aChannel); +} + +// So we can call Close() with the lock already held +// Called from someone who holds a ref via ::Close(), or from ~DataChannel +void DataChannelConnection::CloseLocked(DataChannel* aChannel) { + MOZ_ASSERT(aChannel); + RefPtr<DataChannel> channel(aChannel); // make sure it doesn't go away on us + + mLock.AssertCurrentThreadOwns(); + DC_DEBUG(("Connection %p/Channel %p: Closing stream %u", + channel->mConnection.get(), channel.get(), channel->mStream)); + + aChannel->mBufferedData.Clear(); + if (GetState() == DataChannelConnectionState::Closed) { + // If we're CLOSING, we might leave this in place until we can send a + // reset. + mChannels.Remove(channel); + } + + // This is supposed to only be accessed from Main thread, but this has + // been accessed here from the STS thread for a long time now. + // See Bug 1586475 + DataChannelState channelState = aChannel->GetReadyState(); + // re-test since it may have closed before the lock was grabbed + if (channelState == DataChannelState::Closed || + channelState == DataChannelState::Closing) { + DC_DEBUG(("Channel already closing/closed (%s)", ToString(channelState))); + return; + } + + if (channel->mStream != INVALID_STREAM) { + ResetOutgoingStream(channel->mStream); + if (GetState() != DataChannelConnectionState::Closed) { + // Individual channel is being closed, send reset now. + SendOutgoingStreamReset(); + } + } + aChannel->SetReadyState(DataChannelState::Closing); + if (GetState() == DataChannelConnectionState::Closed) { + // we're not going to hang around waiting + channel->StreamClosedLocked(); + } + // At this point when we leave here, the object is a zombie held alive only by + // the DOM object +} + +void DataChannelConnection::CloseAll() { + DC_DEBUG(("Closing all channels (connection %p)", (void*)this)); + + // Make sure no more channels will be opened + MutexAutoLock lock(mLock); + SetState(DataChannelConnectionState::Closed); + + // Close current channels + // If there are runnables, they hold a strong ref and keep the channel + // and/or connection alive (even if in a CLOSED state) + for (auto& channel : mChannels.GetAll()) { + MutexAutoUnlock lock(mLock); + channel->Close(); + } + + // Clean up any pending opens for channels + RefPtr<DataChannel> channel; + while (nullptr != (channel = mPending.PopFront())) { + DC_DEBUG(("closing pending channel %p, stream %u", channel.get(), + channel->mStream)); + MutexAutoUnlock lock(mLock); + channel->Close(); // also releases the ref on each iteration + } + // It's more efficient to let the Resets queue in shutdown and then + // SendOutgoingStreamReset() here. + SendOutgoingStreamReset(); +} + +bool DataChannelConnection::Channels::IdComparator::Equals( + const RefPtr<DataChannel>& aChannel, uint16_t aId) const { + return aChannel->mStream == aId; +} + +bool DataChannelConnection::Channels::IdComparator::LessThan( + const RefPtr<DataChannel>& aChannel, uint16_t aId) const { + return aChannel->mStream < aId; +} + +bool DataChannelConnection::Channels::IdComparator::Equals( + const RefPtr<DataChannel>& a1, const RefPtr<DataChannel>& a2) const { + return Equals(a1, a2->mStream); +} + +bool DataChannelConnection::Channels::IdComparator::LessThan( + const RefPtr<DataChannel>& a1, const RefPtr<DataChannel>& a2) const { + return LessThan(a1, a2->mStream); +} + +void DataChannelConnection::Channels::Insert( + const RefPtr<DataChannel>& aChannel) { + DC_DEBUG(("Inserting channel %u : %p", aChannel->mStream, aChannel.get())); + MutexAutoLock lock(mMutex); + if (aChannel->mStream != INVALID_STREAM) { + MOZ_ASSERT(!mChannels.ContainsSorted(aChannel, IdComparator())); + } + + MOZ_ASSERT(!mChannels.Contains(aChannel)); + + mChannels.InsertElementSorted(aChannel, IdComparator()); +} + +bool DataChannelConnection::Channels::Remove( + const RefPtr<DataChannel>& aChannel) { + DC_DEBUG(("Removing channel %u : %p", aChannel->mStream, aChannel.get())); + MutexAutoLock lock(mMutex); + if (aChannel->mStream == INVALID_STREAM) { + return mChannels.RemoveElement(aChannel); + } + + return mChannels.RemoveElementSorted(aChannel, IdComparator()); +} + +RefPtr<DataChannel> DataChannelConnection::Channels::Get(uint16_t aId) const { + MutexAutoLock lock(mMutex); + auto index = mChannels.BinaryIndexOf(aId, IdComparator()); + if (index == ChannelArray::NoIndex) { + return nullptr; + } + return mChannels[index]; +} + +RefPtr<DataChannel> DataChannelConnection::Channels::GetNextChannel( + uint16_t aCurrentId) const { + MutexAutoLock lock(mMutex); + if (mChannels.IsEmpty()) { + return nullptr; + } + + auto index = mChannels.IndexOfFirstElementGt(aCurrentId, IdComparator()); + if (index == mChannels.Length()) { + index = 0; + } + return mChannels[index]; +} + +DataChannel::~DataChannel() { + // NS_ASSERTION since this is more "I think I caught all the cases that + // can cause this" than a true kill-the-program assertion. If this is + // wrong, nothing bad happens. A worst it's a leak. + NS_ASSERTION(mReadyState == DataChannelState::Closed || + mReadyState == DataChannelState::Closing, + "unexpected state in ~DataChannel"); +} + +void DataChannel::Close() { + if (mConnection) { + // ensure we don't get deleted + RefPtr<DataChannelConnection> connection(mConnection); + connection->Close(this); + } +} + +// Used when disconnecting from the DataChannelConnection +void DataChannel::StreamClosedLocked() { + MOZ_ASSERT(mConnection); + if (!mConnection) { + return; + } + mConnection->mLock.AssertCurrentThreadOwns(); + + DC_DEBUG(("Destroying Data channel %u", mStream)); + MOZ_ASSERT_IF(mStream != INVALID_STREAM, + !mConnection->FindChannelByStream(mStream)); + AnnounceClosed(); + // We leave mConnection live until the DOM releases us, to avoid races +} + +void DataChannel::ReleaseConnection() { + ASSERT_WEBRTC(NS_IsMainThread()); + mConnection = nullptr; +} + +void DataChannel::SetListener(DataChannelListener* aListener, + nsISupports* aContext) { + ASSERT_WEBRTC(NS_IsMainThread()); + mContext = aContext; + mListener = aListener; +} + +void DataChannel::SendErrnoToErrorResult(int error, size_t aMessageSize, + ErrorResult& aRv) { + switch (error) { + case 0: + break; + case EMSGSIZE: { + nsPrintfCString err("Message size (%zu) exceeds maxMessageSize", + aMessageSize); + aRv.ThrowTypeError(err); + break; + } + default: + aRv.Throw(NS_ERROR_DOM_OPERATION_ERR); + break; + } +} + +void DataChannel::IncrementBufferedAmount(uint32_t aSize, ErrorResult& aRv) { + ASSERT_WEBRTC(NS_IsMainThread()); + if (mBufferedAmount > UINT32_MAX - aSize) { + aRv.Throw(NS_ERROR_FILE_TOO_BIG); + return; + } + + mBufferedAmount += aSize; +} + +void DataChannel::DecrementBufferedAmount(uint32_t aSize) { + mMainThreadEventTarget->Dispatch(NS_NewRunnableFunction( + "DataChannel::DecrementBufferedAmount", + [this, self = RefPtr<DataChannel>(this), aSize] { + MOZ_ASSERT(aSize <= mBufferedAmount); + bool wasLow = mBufferedAmount <= mBufferedThreshold; + mBufferedAmount -= aSize; + if (!wasLow && mBufferedAmount <= mBufferedThreshold) { + DC_DEBUG(("%s: sending BUFFER_LOW_THRESHOLD for %s/%s: %u", + __FUNCTION__, mLabel.get(), mProtocol.get(), mStream)); + mListener->OnBufferLow(mContext); + } + if (mBufferedAmount == 0) { + DC_DEBUG(("%s: sending NO_LONGER_BUFFERED for %s/%s: %u", + __FUNCTION__, mLabel.get(), mProtocol.get(), mStream)); + mListener->NotBuffered(mContext); + } + })); +} + +void DataChannel::AnnounceOpen() { + mMainThreadEventTarget->Dispatch(NS_NewRunnableFunction( + "DataChannel::AnnounceOpen", [this, self = RefPtr<DataChannel>(this)] { + DataChannelState state = GetReadyState(); + // Special-case; spec says to put brand-new remote-created DataChannel + // in "open", but queue the firing of the "open" event. + if (state != DataChannelState::Closing && + state != DataChannelState::Closed) { + if (!mEverOpened && mConnection && mConnection->mListener) { + mEverOpened = true; + mConnection->mListener->NotifyDataChannelOpen(this); + } + SetReadyState(DataChannelState::Open); + DC_DEBUG(("%s: sending ON_CHANNEL_OPEN for %s/%s: %u", __FUNCTION__, + mLabel.get(), mProtocol.get(), mStream)); + if (mListener) { + mListener->OnChannelConnected(mContext); + } + } + })); +} + +void DataChannel::AnnounceClosed() { + mMainThreadEventTarget->Dispatch(NS_NewRunnableFunction( + "DataChannel::AnnounceClosed", [this, self = RefPtr<DataChannel>(this)] { + if (GetReadyState() == DataChannelState::Closed) { + return; + } + if (mEverOpened && mConnection && mConnection->mListener) { + mConnection->mListener->NotifyDataChannelClosed(this); + } + SetReadyState(DataChannelState::Closed); + mBufferedData.Clear(); + if (mListener) { + DC_DEBUG(("%s: sending ON_CHANNEL_CLOSED for %s/%s: %u", __FUNCTION__, + mLabel.get(), mProtocol.get(), mStream)); + mListener->OnChannelClosed(mContext); + } + })); +} + +// Set ready state +void DataChannel::SetReadyState(const DataChannelState aState) { + MOZ_ASSERT(NS_IsMainThread()); + + DC_DEBUG( + ("DataChannelConnection labeled %s(%p) (stream %d) changing ready state " + "%s -> %s", + mLabel.get(), this, mStream, ToString(mReadyState), ToString(aState))); + + mReadyState = aState; +} + +void DataChannel::SendMsg(const nsACString& aMsg, ErrorResult& aRv) { + if (!EnsureValidStream(aRv)) { + return; + } + + SendErrnoToErrorResult(mConnection->SendMsg(mStream, aMsg), aMsg.Length(), + aRv); + if (!aRv.Failed()) { + IncrementBufferedAmount(aMsg.Length(), aRv); + } +} + +void DataChannel::SendBinaryMsg(const nsACString& aMsg, ErrorResult& aRv) { + if (!EnsureValidStream(aRv)) { + return; + } + + SendErrnoToErrorResult(mConnection->SendBinaryMsg(mStream, aMsg), + aMsg.Length(), aRv); + if (!aRv.Failed()) { + IncrementBufferedAmount(aMsg.Length(), aRv); + } +} + +void DataChannel::SendBinaryBlob(dom::Blob& aBlob, ErrorResult& aRv) { + if (!EnsureValidStream(aRv)) { + return; + } + + uint64_t msgLength = aBlob.GetSize(aRv); + if (aRv.Failed()) { + return; + } + + if (msgLength > UINT32_MAX) { + aRv.Throw(NS_ERROR_FILE_TOO_BIG); + return; + } + + // We convert to an nsIInputStream here, because Blob is not threadsafe, and + // we don't convert it earlier because we need to know how large this is so we + // can update bufferedAmount. + nsCOMPtr<nsIInputStream> msgStream; + aBlob.CreateInputStream(getter_AddRefs(msgStream), aRv); + if (NS_WARN_IF(aRv.Failed())) { + return; + } + + SendErrnoToErrorResult(mConnection->SendBlob(mStream, msgStream), msgLength, + aRv); + if (!aRv.Failed()) { + IncrementBufferedAmount(msgLength, aRv); + } +} + +dom::Nullable<uint16_t> DataChannel::GetMaxPacketLifeTime() const { + if (mPrPolicy == DataChannelReliabilityPolicy::LimitedLifetime) { + return dom::Nullable<uint16_t>(mPrValue); + } + return dom::Nullable<uint16_t>(); +} + +dom::Nullable<uint16_t> DataChannel::GetMaxRetransmits() const { + if (mPrPolicy == DataChannelReliabilityPolicy::LimitedRetransmissions) { + return dom::Nullable<uint16_t>(mPrValue); + } + return dom::Nullable<uint16_t>(); +} + +uint32_t DataChannel::GetBufferedAmountLowThreshold() const { + return mBufferedThreshold; +} + +// Never fire immediately, as it's defined to fire on transitions, not state +void DataChannel::SetBufferedAmountLowThreshold(uint32_t aThreshold) { + mBufferedThreshold = aThreshold; +} + +// Called with mLock locked! +void DataChannel::SendOrQueue(DataChannelOnMessageAvailable* aMessage) { + nsCOMPtr<nsIRunnable> runnable = aMessage; + mMainThreadEventTarget->Dispatch(runnable.forget()); +} + +DataChannel::TrafficCounters DataChannel::GetTrafficCounters() const { + MutexAutoLock lock(mStatsLock); + return mTrafficCounters; +} + +bool DataChannel::EnsureValidStream(ErrorResult& aRv) { + MOZ_ASSERT(mConnection); + if (mConnection && mStream != INVALID_STREAM) { + return true; + } + aRv.Throw(NS_ERROR_DOM_INVALID_STATE_ERR); + return false; +} + +void DataChannel::WithTrafficCounters( + const std::function<void(TrafficCounters&)>& aFn) { + MutexAutoLock lock(mStatsLock); + aFn(mTrafficCounters); +} + +nsresult DataChannelOnMessageAvailable::Run() { + MOZ_ASSERT(NS_IsMainThread()); + + // Note: calling the listeners can indirectly cause the listeners to be + // made available for GC (by removing event listeners), especially for + // OnChannelClosed(). We hold a ref to the Channel and the listener + // while calling this. + switch (mType) { + case EventType::OnDataString: + case EventType::OnDataBinary: + if (!mChannel->mListener) { + DC_ERROR(("DataChannelOnMessageAvailable (%s) with null Listener!", + ToString(mType))); + return NS_OK; + } + + if (mChannel->GetReadyState() == DataChannelState::Closed || + mChannel->GetReadyState() == DataChannelState::Closing) { + // Closed by JS, probably + return NS_OK; + } + + if (mType == EventType::OnDataString) { + mChannel->mListener->OnMessageAvailable(mChannel->mContext, mData); + } else { + mChannel->mListener->OnBinaryMessageAvailable(mChannel->mContext, + mData); + } + break; + case EventType::OnDisconnected: + // If we've disconnected, make sure we close all the streams - from + // mainthread! + if (mConnection->mListener) { + mConnection->mListener->NotifySctpClosed(); + } + mConnection->CloseAll(); + break; + case EventType::OnChannelCreated: + if (!mConnection->mListener) { + DC_ERROR(("DataChannelOnMessageAvailable (%s) with null Listener!", + ToString(mType))); + return NS_OK; + } + + // important to give it an already_AddRefed pointer! + mConnection->mListener->NotifyDataChannel(mChannel.forget()); + break; + case EventType::OnConnection: + if (mConnection->mListener) { + mConnection->mListener->NotifySctpConnected(); + } + break; + } + return NS_OK; +} + +} // namespace mozilla |