From a556c4822fc205db0d27834ba5b637c351d73ffa Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Tue, 5 Mar 2013 10:56:27 -0800 Subject: Squashed commit of the following: commit e5919b1f57ea61fa1d380dfdb4e3e832ce73d79d Author: Andreas Huber Date: Wed Feb 27 16:38:48 2013 -0800 Configure TCP datagram sockets to be TCP_NODELAY. Change-Id: Ia724a81e6e27dccd00ac84603e712d69ca77a0cd commit 1b52b393183db8a6dc000a7c31baac544ccfc50c Author: Andreas Huber Date: Wed Feb 27 14:26:01 2013 -0800 Send IDR frame requests on packet loss. Change-Id: I53b7fb85cbd6923491113b93ec3e2175726d654a commit 68d76b4b3a0181b30abc57cd2915273210530a6d Author: Andreas Huber Date: Tue Feb 26 15:12:34 2013 -0800 Revive TunnelRenderer Change-Id: I8c5a9d982793b1c5b841c828227b354f1dab618c commit 3df28a8e9d8bcdc1430016bb088d097eca653b56 Author: Andreas Huber Date: Tue Feb 26 13:53:14 2013 -0800 Disable suspension of video updates. Change-Id: I7e3a16b8d7dd7a55d9f962a2236388931f664106 commit 2ec7a79de019a26ec415016c1478afd762f069cd Author: Andreas Huber Date: Tue Feb 26 08:54:40 2013 -0800 Adds an SNTP client to wfd. Change-Id: Icd7d6104e951e1443e4c1b81ccf6b3731d79d3ec commit c81c3bb5725bb4079a4d7fb02151ad0bb540632f Author: Andreas Huber Date: Mon Feb 25 10:00:58 2013 -0800 Squashed commit of the following: commit b83a4ec96659ef6f6b7c2090fdd866abe3ab78ba Author: Andreas Huber Date: Mon Feb 25 09:28:11 2013 -0800 Some reorganization of the rtp code, renamed StreamHub -> MediaSender Change-Id: I8cf67444960e60426bf74880af1acce41e8b2fef commit 7769cbd739f2a67c58e0c6a7b1a21a12210c7c4d Author: Andreas Huber Date: Fri Feb 22 16:12:18 2013 -0800 Choose a smaller MTU to avoid fragmented IPv4 packets, fix AVC assembler. Change-Id: I274b3cc1483c4e9f4d146dbf9f3d9f7557ef7ef9 commit 1f687ee80a88b56d614c2cf408ff729114ff86a0 Author: Andreas Huber Date: Fri Feb 22 11:38:31 2013 -0800 better reporting. Change-Id: I67f0bb51f106ea77f5cc75938b053c8e8e8f688e commit 7950c1cd59213eb5f281fcde44a772ecffae473d Author: Andreas Huber Date: Fri Feb 22 09:07:41 2013 -0800 stuff Change-Id: Ib99416366d3eec6e6ad69b4d791a8a9408410f3b commit 33c09045b0f86fcaa4619cbd679b47a074f71231 Author: Andreas Huber Date: Thu Feb 21 15:54:01 2013 -0800 Render frames according to their timestamps. Change-Id: I8143a95cffe775799d6a4bb093558bd7abb1f063 commit d8b6daae2160bf1c016d7c6251256b46bb89db42 Author: Andreas Huber Date: Thu Feb 21 15:01:27 2013 -0800 Better packet-lost logic. Change-Id: I611eee5a42bd089638cf45b0e16f628ff2a955ab commit 782c6b15717e2d062d96665a089d06c0577733d0 Author: Andreas Huber Date: Wed Feb 20 15:06:47 2013 -0800 Add a dedicated looper for the MediaReceiver Change-Id: I3b79cad367fb69c9a160a8d009af8c5f5142b98e commit 4c7b8b10861674b773270103bcabd1a99486a691 Author: Andreas Huber Date: Wed Feb 20 14:30:28 2013 -0800 Tweaks to RTPSender and RTPReceiver Change-Id: Ib535552f289a26cfead6df8c63e4c63d3987d4e9 commit 39226b28177a816cda5c67b321745d396b18277d Author: Andreas Huber Date: Tue Feb 19 08:48:25 2013 -0800 Playing around with non muxed delivery Change-Id: I845375f6938d04bc30502840c2ceb7688dc9b237 commit c16d21de75d8ecdbcd9abce14934afe484970061 Author: Andreas Huber Date: Wed Feb 13 14:43:35 2013 -0800 A more solid base for RTP communication. Change-Id: I52033eeb0feba0ff029d61553a821c82f2fa1c3f Change-Id: I57e3bcfc1c59a012b15aaaa42ed81f09c34c26bb Change-Id: I4b09db4a44d0eeded7a1658f6dc6c97d4b8be720 --- .../libstagefright/wifi-display/rtp/RTPSender.cpp | 701 +++++++++++++++++++++ 1 file changed, 701 insertions(+) create mode 100644 media/libstagefright/wifi-display/rtp/RTPSender.cpp (limited to 'media/libstagefright/wifi-display/rtp/RTPSender.cpp') diff --git a/media/libstagefright/wifi-display/rtp/RTPSender.cpp b/media/libstagefright/wifi-display/rtp/RTPSender.cpp new file mode 100644 index 0000000..85c5933 --- /dev/null +++ b/media/libstagefright/wifi-display/rtp/RTPSender.cpp @@ -0,0 +1,701 @@ +/* + * Copyright 2013, The Android Open Source Project + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +//#define LOG_NDEBUG 0 +#define LOG_TAG "RTPSender" +#include + +#include "RTPSender.h" + +#include "ANetworkSession.h" + +#include +#include +#include +#include +#include +#include + +#include "include/avc_utils.h" + +namespace android { + +RTPSender::RTPSender( + const sp &netSession, + const sp ¬ify) + : mNetSession(netSession), + mNotify(notify), + mMode(TRANSPORT_UNDEFINED), + mRTPSessionID(0), + mRTCPSessionID(0), + mRTPConnected(false), + mRTCPConnected(false), + mLastNTPTime(0), + mLastRTPTime(0), + mNumRTPSent(0), + mNumRTPOctetsSent(0), + mNumSRsSent(0), + mRTPSeqNo(0), + mHistorySize(0) { +} + +RTPSender::~RTPSender() { + if (mRTCPSessionID != 0) { + mNetSession->destroySession(mRTCPSessionID); + mRTCPSessionID = 0; + } + + if (mRTPSessionID != 0) { + mNetSession->destroySession(mRTPSessionID); + mRTPSessionID = 0; + } +} + +// static +int32_t RTPBase::PickRandomRTPPort() { + // Pick an even integer in range [1024, 65534) + + static const size_t kRange = (65534 - 1024) / 2; + + return (int32_t)(((float)(kRange + 1) * rand()) / RAND_MAX) * 2 + 1024; +} + +status_t RTPSender::initAsync( + TransportMode mode, + const char *remoteHost, + int32_t remoteRTPPort, + int32_t remoteRTCPPort, + int32_t *outLocalRTPPort) { + if (mMode != TRANSPORT_UNDEFINED || mode == TRANSPORT_UNDEFINED) { + return INVALID_OPERATION; + } + + CHECK_NE(mMode, TRANSPORT_TCP_INTERLEAVED); + + if (mode == TRANSPORT_TCP && remoteRTCPPort >= 0) { + return INVALID_OPERATION; + } + + sp rtpNotify = new AMessage(kWhatRTPNotify, id()); + + sp rtcpNotify; + if (remoteRTCPPort >= 0) { + rtcpNotify = new AMessage(kWhatRTCPNotify, id()); + } + + CHECK_EQ(mRTPSessionID, 0); + CHECK_EQ(mRTCPSessionID, 0); + + int32_t localRTPPort; + + for (;;) { + localRTPPort = PickRandomRTPPort(); + + status_t err; + if (mode == TRANSPORT_UDP) { + err = mNetSession->createUDPSession( + localRTPPort, + remoteHost, + remoteRTPPort, + rtpNotify, + &mRTPSessionID); + } else { + CHECK_EQ(mode, TRANSPORT_TCP); + err = mNetSession->createTCPDatagramSession( + localRTPPort, + remoteHost, + remoteRTPPort, + rtpNotify, + &mRTPSessionID); + } + + if (err != OK) { + continue; + } + + if (remoteRTCPPort < 0) { + break; + } + + if (mode == TRANSPORT_UDP) { + err = mNetSession->createUDPSession( + localRTPPort + 1, + remoteHost, + remoteRTCPPort, + rtcpNotify, + &mRTCPSessionID); + } else { + CHECK_EQ(mode, TRANSPORT_TCP); + err = mNetSession->createTCPDatagramSession( + localRTPPort + 1, + remoteHost, + remoteRTCPPort, + rtcpNotify, + &mRTCPSessionID); + } + + if (err == OK) { + break; + } + + mNetSession->destroySession(mRTPSessionID); + mRTPSessionID = 0; + } + + if (mode == TRANSPORT_UDP) { + mRTPConnected = true; + mRTCPConnected = true; + } + + mMode = mode; + *outLocalRTPPort = localRTPPort; + + if (mMode == TRANSPORT_UDP) { + notifyInitDone(OK); + } + + return OK; +} + +status_t RTPSender::queueBuffer( + const sp &buffer, uint8_t packetType, PacketizationMode mode) { + status_t err; + + switch (mode) { + case PACKETIZATION_TRANSPORT_STREAM: + err = queueTSPackets(buffer, packetType); + break; + + case PACKETIZATION_H264: + err = queueAVCBuffer(buffer, packetType); + break; + + default: + TRESPASS(); + } + + return err; +} + +status_t RTPSender::queueTSPackets( + const sp &tsPackets, uint8_t packetType) { + CHECK_EQ(0, tsPackets->size() % 188); + + const size_t numTSPackets = tsPackets->size() / 188; + + size_t srcOffset = 0; + while (srcOffset < tsPackets->size()) { + sp udpPacket = + new ABuffer(12 + kMaxNumTSPacketsPerRTPPacket * 188); + + udpPacket->setInt32Data(mRTPSeqNo); + + uint8_t *rtp = udpPacket->data(); + rtp[0] = 0x80; + rtp[1] = packetType; + + rtp[2] = (mRTPSeqNo >> 8) & 0xff; + rtp[3] = mRTPSeqNo & 0xff; + ++mRTPSeqNo; + + int64_t nowUs = ALooper::GetNowUs(); + uint32_t rtpTime = (nowUs * 9) / 100ll; + + rtp[4] = rtpTime >> 24; + rtp[5] = (rtpTime >> 16) & 0xff; + rtp[6] = (rtpTime >> 8) & 0xff; + rtp[7] = rtpTime & 0xff; + + rtp[8] = kSourceID >> 24; + rtp[9] = (kSourceID >> 16) & 0xff; + rtp[10] = (kSourceID >> 8) & 0xff; + rtp[11] = kSourceID & 0xff; + + size_t numTSPackets = (tsPackets->size() - srcOffset) / 188; + if (numTSPackets > kMaxNumTSPacketsPerRTPPacket) { + numTSPackets = kMaxNumTSPacketsPerRTPPacket; + } + + memcpy(&rtp[12], tsPackets->data() + srcOffset, numTSPackets * 188); + + udpPacket->setRange(0, 12 + numTSPackets * 188); + status_t err = sendRTPPacket(udpPacket, true /* storeInHistory */); + + if (err != OK) { + return err; + } + + srcOffset += numTSPackets * 188; + } + + return OK; +} + +status_t RTPSender::queueAVCBuffer( + const sp &accessUnit, uint8_t packetType) { + int64_t timeUs; + CHECK(accessUnit->meta()->findInt64("timeUs", &timeUs)); + + uint32_t rtpTime = (timeUs * 9 / 100ll); + + List > packets; + + sp out = new ABuffer(kMaxUDPPacketSize); + size_t outBytesUsed = 12; // Placeholder for RTP header. + + const uint8_t *data = accessUnit->data(); + size_t size = accessUnit->size(); + const uint8_t *nalStart; + size_t nalSize; + while (getNextNALUnit( + &data, &size, &nalStart, &nalSize, + true /* startCodeFollows */) == OK) { + size_t bytesNeeded = nalSize + 2; + if (outBytesUsed == 12) { + ++bytesNeeded; + } + + if (outBytesUsed + bytesNeeded > out->capacity()) { + bool emitSingleNALPacket = false; + + if (outBytesUsed == 12 + && outBytesUsed + nalSize <= out->capacity()) { + // We haven't emitted anything into the current packet yet and + // this NAL unit fits into a single-NAL-unit-packet while + // it wouldn't have fit as part of a STAP-A packet. + + memcpy(out->data() + outBytesUsed, nalStart, nalSize); + outBytesUsed += nalSize; + + emitSingleNALPacket = true; + } + + if (outBytesUsed > 12) { + out->setRange(0, outBytesUsed); + packets.push_back(out); + out = new ABuffer(kMaxUDPPacketSize); + outBytesUsed = 12; // Placeholder for RTP header + } + + if (emitSingleNALPacket) { + continue; + } + } + + if (outBytesUsed + bytesNeeded <= out->capacity()) { + uint8_t *dst = out->data() + outBytesUsed; + + if (outBytesUsed == 12) { + *dst++ = 24; // STAP-A header + } + + *dst++ = (nalSize >> 8) & 0xff; + *dst++ = nalSize & 0xff; + memcpy(dst, nalStart, nalSize); + + outBytesUsed += bytesNeeded; + continue; + } + + // This single NAL unit does not fit into a single RTP packet, + // we need to emit an FU-A. + + CHECK_EQ(outBytesUsed, 12u); + + uint8_t nalType = nalStart[0] & 0x1f; + uint8_t nri = (nalStart[0] >> 5) & 3; + + size_t srcOffset = 1; + while (srcOffset < nalSize) { + size_t copy = out->capacity() - outBytesUsed - 2; + if (copy > nalSize - srcOffset) { + copy = nalSize - srcOffset; + } + + uint8_t *dst = out->data() + outBytesUsed; + dst[0] = (nri << 5) | 28; + + dst[1] = nalType; + + if (srcOffset == 1) { + dst[1] |= 0x80; + } + + if (srcOffset + copy == nalSize) { + dst[1] |= 0x40; + } + + memcpy(&dst[2], nalStart + srcOffset, copy); + srcOffset += copy; + + out->setRange(0, outBytesUsed + copy + 2); + + packets.push_back(out); + out = new ABuffer(kMaxUDPPacketSize); + outBytesUsed = 12; // Placeholder for RTP header + } + } + + if (outBytesUsed > 12) { + out->setRange(0, outBytesUsed); + packets.push_back(out); + } + + while (!packets.empty()) { + sp out = *packets.begin(); + packets.erase(packets.begin()); + + out->setInt32Data(mRTPSeqNo); + + bool last = packets.empty(); + + uint8_t *dst = out->data(); + + dst[0] = 0x80; + + dst[1] = packetType; + if (last) { + dst[1] |= 1 << 7; // M-bit + } + + dst[2] = (mRTPSeqNo >> 8) & 0xff; + dst[3] = mRTPSeqNo & 0xff; + ++mRTPSeqNo; + + dst[4] = rtpTime >> 24; + dst[5] = (rtpTime >> 16) & 0xff; + dst[6] = (rtpTime >> 8) & 0xff; + dst[7] = rtpTime & 0xff; + dst[8] = kSourceID >> 24; + dst[9] = (kSourceID >> 16) & 0xff; + dst[10] = (kSourceID >> 8) & 0xff; + dst[11] = kSourceID & 0xff; + + status_t err = sendRTPPacket(out, true /* storeInHistory */); + + if (err != OK) { + return err; + } + } + + return OK; +} + +status_t RTPSender::sendRTPPacket( + const sp &buffer, bool storeInHistory) { + CHECK(mRTPConnected); + + status_t err = mNetSession->sendRequest( + mRTPSessionID, buffer->data(), buffer->size()); + + if (err != OK) { + return err; + } + + mLastNTPTime = GetNowNTP(); + mLastRTPTime = U32_AT(buffer->data() + 4); + + ++mNumRTPSent; + mNumRTPOctetsSent += buffer->size() - 12; + + if (storeInHistory) { + if (mHistorySize == kMaxHistorySize) { + mHistory.erase(mHistory.begin()); + } else { + ++mHistorySize; + } + mHistory.push_back(buffer); + } + + return OK; +} + +// static +uint64_t RTPSender::GetNowNTP() { + struct timeval tv; + gettimeofday(&tv, NULL /* timezone */); + + uint64_t nowUs = tv.tv_sec * 1000000ll + tv.tv_usec; + + nowUs += ((70ll * 365 + 17) * 24) * 60 * 60 * 1000000ll; + + uint64_t hi = nowUs / 1000000ll; + uint64_t lo = ((1ll << 32) * (nowUs % 1000000ll)) / 1000000ll; + + return (hi << 32) | lo; +} + +void RTPSender::onMessageReceived(const sp &msg) { + switch (msg->what()) { + case kWhatRTPNotify: + case kWhatRTCPNotify: + onNetNotify(msg->what() == kWhatRTPNotify, msg); + break; + + default: + TRESPASS(); + } +} + +void RTPSender::onNetNotify(bool isRTP, const sp &msg) { + int32_t reason; + CHECK(msg->findInt32("reason", &reason)); + + switch (reason) { + case ANetworkSession::kWhatError: + { + int32_t sessionID; + CHECK(msg->findInt32("sessionID", &sessionID)); + + int32_t err; + CHECK(msg->findInt32("err", &err)); + + int32_t errorOccuredDuringSend; + CHECK(msg->findInt32("send", &errorOccuredDuringSend)); + + AString detail; + CHECK(msg->findString("detail", &detail)); + + ALOGE("An error occurred during %s in session %d " + "(%d, '%s' (%s)).", + errorOccuredDuringSend ? "send" : "receive", + sessionID, + err, + detail.c_str(), + strerror(-err)); + + mNetSession->destroySession(sessionID); + + if (sessionID == mRTPSessionID) { + mRTPSessionID = 0; + } else if (sessionID == mRTCPSessionID) { + mRTCPSessionID = 0; + } + + if (mMode == TRANSPORT_TCP) { + if (!mRTPConnected + || (mRTCPSessionID > 0 && !mRTCPConnected)) { + notifyInitDone(err); + break; + } + } + + notifyError(err); + break; + } + + case ANetworkSession::kWhatDatagram: + { + sp data; + CHECK(msg->findBuffer("data", &data)); + + if (isRTP) { + ALOGW("Huh? Received data on RTP connection..."); + } else { + onRTCPData(data); + } + break; + } + + case ANetworkSession::kWhatConnected: + { + CHECK_EQ(mMode, TRANSPORT_TCP); + + int32_t sessionID; + CHECK(msg->findInt32("sessionID", &sessionID)); + + if (isRTP) { + CHECK_EQ(sessionID, mRTPSessionID); + mRTPConnected = true; + } else { + CHECK_EQ(sessionID, mRTCPSessionID); + mRTCPConnected = true; + } + + if (mRTPConnected && (mRTCPSessionID == 0 || mRTCPConnected)) { + notifyInitDone(OK); + } + break; + } + } +} + +status_t RTPSender::onRTCPData(const sp &buffer) { + const uint8_t *data = buffer->data(); + size_t size = buffer->size(); + + while (size > 0) { + if (size < 8) { + // Too short to be a valid RTCP header + return ERROR_MALFORMED; + } + + if ((data[0] >> 6) != 2) { + // Unsupported version. + return ERROR_UNSUPPORTED; + } + + if (data[0] & 0x20) { + // Padding present. + + size_t paddingLength = data[size - 1]; + + if (paddingLength + 12 > size) { + // If we removed this much padding we'd end up with something + // that's too short to be a valid RTP header. + return ERROR_MALFORMED; + } + + size -= paddingLength; + } + + size_t headerLength = 4 * (data[2] << 8 | data[3]) + 4; + + if (size < headerLength) { + // Only received a partial packet? + return ERROR_MALFORMED; + } + + switch (data[1]) { + case 200: + case 201: // RR + parseReceiverReport(data, headerLength); + break; + + case 202: // SDES + case 203: + case 204: // APP + break; + + case 205: // TSFB (transport layer specific feedback) + parseTSFB(data, headerLength); + break; + + case 206: // PSFB (payload specific feedback) + // hexdump(data, headerLength); + break; + + default: + { + ALOGW("Unknown RTCP packet type %u of size %d", + (unsigned)data[1], headerLength); + break; + } + } + + data += headerLength; + size -= headerLength; + } + + return OK; +} + +status_t RTPSender::parseReceiverReport(const uint8_t *data, size_t size) { + // hexdump(data, size); + + float fractionLost = data[12] / 256.0f; + + ALOGI("lost %.2f %% of packets during report interval.", + 100.0f * fractionLost); + + return OK; +} + +status_t RTPSender::parseTSFB(const uint8_t *data, size_t size) { + if ((data[0] & 0x1f) != 1) { + return ERROR_UNSUPPORTED; // We only support NACK for now. + } + + uint32_t srcId = U32_AT(&data[8]); + if (srcId != kSourceID) { + return ERROR_MALFORMED; + } + + for (size_t i = 12; i < size; i += 4) { + uint16_t seqNo = U16_AT(&data[i]); + uint16_t blp = U16_AT(&data[i + 2]); + + List >::iterator it = mHistory.begin(); + bool foundSeqNo = false; + while (it != mHistory.end()) { + const sp &buffer = *it; + + uint16_t bufferSeqNo = buffer->int32Data() & 0xffff; + + bool retransmit = false; + if (bufferSeqNo == seqNo) { + retransmit = true; + } else if (blp != 0) { + for (size_t i = 0; i < 16; ++i) { + if ((blp & (1 << i)) + && (bufferSeqNo == ((seqNo + i + 1) & 0xffff))) { + blp &= ~(1 << i); + retransmit = true; + } + } + } + + if (retransmit) { + ALOGV("retransmitting seqNo %d", bufferSeqNo); + + CHECK_EQ((status_t)OK, + sendRTPPacket(buffer, false /* storeInHistory */)); + + if (bufferSeqNo == seqNo) { + foundSeqNo = true; + } + + if (foundSeqNo && blp == 0) { + break; + } + } + + ++it; + } + + if (!foundSeqNo || blp != 0) { + ALOGI("Some sequence numbers were no longer available for " + "retransmission (seqNo = %d, foundSeqNo = %d, blp = 0x%04x)", + seqNo, foundSeqNo, blp); + + if (!mHistory.empty()) { + int32_t earliest = (*mHistory.begin())->int32Data() & 0xffff; + int32_t latest = (*--mHistory.end())->int32Data() & 0xffff; + + ALOGI("have seq numbers from %d - %d", earliest, latest); + } + } + } + + return OK; +} + +void RTPSender::notifyInitDone(status_t err) { + sp notify = mNotify->dup(); + notify->setInt32("what", kWhatInitDone); + notify->setInt32("err", err); + notify->post(); +} + +void RTPSender::notifyError(status_t err) { + sp notify = mNotify->dup(); + notify->setInt32("what", kWhatError); + notify->setInt32("err", err); + notify->post(); +} + +} // namespace android + -- cgit v1.1