enum {
// Video PES packets contain exactly one (aligned) access unit.
kFlagAlignedVideoData = 1,
+
+ // Timestamps are in ALooper::GetNowUs() units.
+ kFlagIsRealTimeData = 2,
};
virtual uint32_t flags() const { return 0; }
};
mSource->start();
+ uint32_t flags = 0;
+
+ if (mSource->isRealTime()) {
+ flags |= Renderer::FLAG_REAL_TIME;
+ }
+
mRenderer = new Renderer(
mAudioSink,
- new AMessage(kWhatRendererNotify, id()));
+ new AMessage(kWhatRendererNotify, id()),
+ flags);
looper()->registerHandler(mRenderer);
int mode = request.readInt32();
return mPlayer->setVideoScalingMode(mode);
}
+
default:
{
return INVALID_OPERATION;
NuPlayer::Renderer::Renderer(
const sp<MediaPlayerBase::AudioSink> &sink,
- const sp<AMessage> ¬ify)
+ const sp<AMessage> ¬ify,
+ uint32_t flags)
: mAudioSink(sink),
mNotify(notify),
+ mFlags(flags),
mNumFramesWritten(0),
mDrainAudioQueuePending(false),
mDrainVideoQueuePending(false),
if (entry.mBuffer == NULL) {
// EOS doesn't carry a timestamp.
delayUs = 0;
+ } else if (mFlags & FLAG_REAL_TIME) {
+ int64_t mediaTimeUs;
+ CHECK(entry.mBuffer->meta()->findInt64("timeUs", &mediaTimeUs));
+
+ delayUs = mediaTimeUs - ALooper::GetNowUs();
} else {
int64_t mediaTimeUs;
CHECK(entry.mBuffer->meta()->findInt64("timeUs", &mediaTimeUs));
return;
}
- int64_t mediaTimeUs;
- CHECK(entry->mBuffer->meta()->findInt64("timeUs", &mediaTimeUs));
+ int64_t realTimeUs;
+ if (mFlags & FLAG_REAL_TIME) {
+ CHECK(entry->mBuffer->meta()->findInt64("timeUs", &realTimeUs));
+ } else {
+ int64_t mediaTimeUs;
+ CHECK(entry->mBuffer->meta()->findInt64("timeUs", &mediaTimeUs));
+
+ realTimeUs = mediaTimeUs - mAnchorTimeMediaUs + mAnchorTimeRealUs;
+ }
- int64_t realTimeUs = mediaTimeUs - mAnchorTimeMediaUs + mAnchorTimeRealUs;
mVideoLateByUs = ALooper::GetNowUs() - realTimeUs;
-
bool tooLate = (mVideoLateByUs > 40000);
if (tooLate) {
struct ABuffer;
struct NuPlayer::Renderer : public AHandler {
+ enum Flags {
+ FLAG_REAL_TIME = 1,
+ };
Renderer(const sp<MediaPlayerBase::AudioSink> &sink,
- const sp<AMessage> ¬ify);
+ const sp<AMessage> ¬ify,
+ uint32_t flags = 0);
void queueBuffer(
bool audio,
sp<MediaPlayerBase::AudioSink> mAudioSink;
sp<AMessage> mNotify;
+ uint32_t mFlags;
List<QueueEntry> mAudioQueue;
List<QueueEntry> mVideoQueue;
uint32_t mNumFramesWritten;
return INVALID_OPERATION;
}
+ virtual bool isRealTime() const {
+ return false;
+ }
+
protected:
virtual ~Source() {}
return err;
}
+bool NuPlayer::StreamingSource::isRealTime() const {
+ return mSource->flags() & IStreamSource::kFlagIsRealTimeData;
+}
+
} // namespace android
virtual status_t dequeueAccessUnit(bool audio, sp<ABuffer> *accessUnit);
+ virtual bool isRealTime() const;
+
protected:
virtual ~StreamingSource();
timeUs += mParser->mAbsoluteTimeAnchorUs;
}
+ if (mParser->mTimeOffsetValid) {
+ timeUs += mParser->mTimeOffsetUs;
+ }
+
return timeUs;
}
ATSParser::ATSParser(uint32_t flags)
: mFlags(flags),
mAbsoluteTimeAnchorUs(-1ll),
+ mTimeOffsetValid(false),
+ mTimeOffsetUs(0ll),
mNumTSPacketsParsed(0),
mNumPCRs(0) {
mPSISections.add(0 /* PID */, new PSISection);
CHECK(mPrograms.empty());
mAbsoluteTimeAnchorUs = timeUs;
return;
+ } else if (type == DISCONTINUITY_TIME_OFFSET) {
+ int64_t offset;
+ CHECK(extra->findInt64("offset", &offset));
+
+ mTimeOffsetValid = true;
+ mTimeOffsetUs = offset;
+ return;
}
for (size_t i = 0; i < mPrograms.size(); ++i) {
DISCONTINUITY_AUDIO_FORMAT = 2,
DISCONTINUITY_VIDEO_FORMAT = 4,
DISCONTINUITY_ABSOLUTE_TIME = 8,
+ DISCONTINUITY_TIME_OFFSET = 16,
DISCONTINUITY_SEEK = DISCONTINUITY_TIME,
int64_t mAbsoluteTimeAnchorUs;
+ bool mTimeOffsetValid;
+ int64_t mTimeOffsetUs;
+
size_t mNumTSPacketsParsed;
void parseProgramAssociationTable(ABitReader *br);
#include <net/if.h>
#include <netdb.h>
#include <netinet/in.h>
+#include <sys/ioctl.h>
#include <sys/socket.h>
#include <media/stagefright/foundation/ABuffer.h>
mSawSendFailure = true;
}
+#if 0
+ int numBytesQueued;
+ int res = ioctl(mSocket, SIOCOUTQ, &numBytesQueued);
+ if (res == 0 && numBytesQueued > 102400) {
+ ALOGI("numBytesQueued = %d", numBytesQueued);
+ }
+#endif
+
return err;
}
sink/TunnelRenderer.cpp \
sink/WifiDisplaySink.cpp \
SNTPClient.cpp \
+ TimeSyncer.cpp \
source/Converter.cpp \
source/MediaPuller.cpp \
source/PlaybackSession.cpp \
notifyInitDone(mInitStatus);
}
- mTSParser = new ATSParser(ATSParser::ALIGNED_VIDEO_DATA);
+ mTSParser = new ATSParser(
+ ATSParser::ALIGNED_VIDEO_DATA
+ | ATSParser::TS_TIMESTAMPS_ARE_ABSOLUTE);
+
mFormatKnownMask = 0;
break;
}
notify->post();
}
+status_t MediaReceiver::notifyLateness(size_t trackIndex, int64_t latenessUs) {
+ if (trackIndex >= mTrackInfos.size()) {
+ return -ERANGE;
+ }
+
+ TrackInfo *info = &mTrackInfos.editItemAt(trackIndex);
+ return info->mReceiver->notifyLateness(latenessUs);
+}
+
} // namespace android
};
status_t initAsync(Mode mode);
+ status_t notifyLateness(size_t trackIndex, int64_t latenessUs);
+
protected:
virtual void onMessageReceived(const sp<AMessage> &msg);
virtual ~MediaReceiver();
--- /dev/null
+/*
+ * 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_NEBUG 0
+#define LOG_TAG "TimeSyncer"
+#include <utils/Log.h>
+
+#include "TimeSyncer.h"
+
+#include "ANetworkSession.h"
+
+#include <media/stagefright/foundation/ABuffer.h>
+#include <media/stagefright/foundation/ADebug.h>
+#include <media/stagefright/foundation/AHandler.h>
+#include <media/stagefright/foundation/ALooper.h>
+#include <media/stagefright/foundation/AMessage.h>
+#include <media/stagefright/Utils.h>
+
+namespace android {
+
+TimeSyncer::TimeSyncer(
+ const sp<ANetworkSession> &netSession, const sp<AMessage> ¬ify)
+ : mNetSession(netSession),
+ mNotify(notify),
+ mIsServer(false),
+ mConnected(false),
+ mUDPSession(0),
+ mSeqNo(0),
+ mTotalTimeUs(0.0),
+ mPendingT1(0ll),
+ mTimeoutGeneration(0) {
+}
+
+TimeSyncer::~TimeSyncer() {
+}
+
+void TimeSyncer::startServer(unsigned localPort) {
+ sp<AMessage> msg = new AMessage(kWhatStartServer, id());
+ msg->setInt32("localPort", localPort);
+ msg->post();
+}
+
+void TimeSyncer::startClient(const char *remoteHost, unsigned remotePort) {
+ sp<AMessage> msg = new AMessage(kWhatStartClient, id());
+ msg->setString("remoteHost", remoteHost);
+ msg->setInt32("remotePort", remotePort);
+ msg->post();
+}
+
+void TimeSyncer::onMessageReceived(const sp<AMessage> &msg) {
+ switch (msg->what()) {
+ case kWhatStartClient:
+ {
+ AString remoteHost;
+ CHECK(msg->findString("remoteHost", &remoteHost));
+
+ int32_t remotePort;
+ CHECK(msg->findInt32("remotePort", &remotePort));
+
+ sp<AMessage> notify = new AMessage(kWhatUDPNotify, id());
+
+ CHECK_EQ((status_t)OK,
+ mNetSession->createUDPSession(
+ 0 /* localPort */,
+ remoteHost.c_str(),
+ remotePort,
+ notify,
+ &mUDPSession));
+
+ postSendPacket();
+ break;
+ }
+
+ case kWhatStartServer:
+ {
+ mIsServer = true;
+
+ int32_t localPort;
+ CHECK(msg->findInt32("localPort", &localPort));
+
+ sp<AMessage> notify = new AMessage(kWhatUDPNotify, id());
+
+ CHECK_EQ((status_t)OK,
+ mNetSession->createUDPSession(
+ localPort, notify, &mUDPSession));
+
+ break;
+ }
+
+ case kWhatSendPacket:
+ {
+ TimeInfo ti;
+ memset(&ti, 0, sizeof(ti));
+
+ ti.mT1 = ALooper::GetNowUs();
+
+ CHECK_EQ((status_t)OK,
+ mNetSession->sendRequest(
+ mUDPSession, &ti, sizeof(ti)));
+
+ mPendingT1 = ti.mT1;
+ postTimeout();
+ break;
+ }
+
+ case kWhatTimedOut:
+ {
+ int32_t generation;
+ CHECK(msg->findInt32("generation", &generation));
+
+ if (generation != mTimeoutGeneration) {
+ break;
+ }
+
+ ALOGI("timed out, sending another request");
+ postSendPacket();
+ break;
+ }
+
+ case kWhatUDPNotify:
+ {
+ 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));
+
+ AString detail;
+ CHECK(msg->findString("detail", &detail));
+
+ ALOGE("An error occurred in session %d (%d, '%s/%s').",
+ sessionID,
+ err,
+ detail.c_str(),
+ strerror(-err));
+
+ mNetSession->destroySession(sessionID);
+
+ cancelTimeout();
+
+ notifyError(err);
+ break;
+ }
+
+ case ANetworkSession::kWhatDatagram:
+ {
+ int32_t sessionID;
+ CHECK(msg->findInt32("sessionID", &sessionID));
+
+ sp<ABuffer> packet;
+ CHECK(msg->findBuffer("data", &packet));
+
+ int64_t arrivalTimeUs;
+ CHECK(packet->meta()->findInt64(
+ "arrivalTimeUs", &arrivalTimeUs));
+
+ CHECK_EQ(packet->size(), sizeof(TimeInfo));
+
+ TimeInfo *ti = (TimeInfo *)packet->data();
+
+ if (mIsServer) {
+ if (!mConnected) {
+ AString fromAddr;
+ CHECK(msg->findString("fromAddr", &fromAddr));
+
+ int32_t fromPort;
+ CHECK(msg->findInt32("fromPort", &fromPort));
+
+ CHECK_EQ((status_t)OK,
+ mNetSession->connectUDPSession(
+ mUDPSession, fromAddr.c_str(), fromPort));
+
+ mConnected = true;
+ }
+
+ ti->mT2 = arrivalTimeUs;
+ ti->mT3 = ALooper::GetNowUs();
+
+ CHECK_EQ((status_t)OK,
+ mNetSession->sendRequest(
+ mUDPSession, ti, sizeof(*ti)));
+ } else {
+ if (ti->mT1 != mPendingT1) {
+ break;
+ }
+
+ cancelTimeout();
+ mPendingT1 = 0;
+
+ ti->mT4 = arrivalTimeUs;
+
+ // One way delay for a packet to travel from client
+ // to server or back (assumed to be the same either way).
+ int64_t delay =
+ (ti->mT2 - ti->mT1 + ti->mT4 - ti->mT3) / 2;
+
+ // Offset between the client clock (T1, T4) and the
+ // server clock (T2, T3) timestamps.
+ int64_t offset =
+ (ti->mT2 - ti->mT1 - ti->mT4 + ti->mT3) / 2;
+
+ mHistory.push_back(*ti);
+
+ ALOGV("delay = %lld us,\toffset %lld us",
+ delay,
+ offset);
+
+ if (mHistory.size() < kNumPacketsPerBatch) {
+ postSendPacket(1000000ll / 30);
+ } else {
+ notifyOffset();
+
+ mHistory.clear();
+ postSendPacket(kBatchDelayUs);
+ }
+ }
+ break;
+ }
+
+ default:
+ TRESPASS();
+ }
+
+ break;
+ }
+
+ default:
+ TRESPASS();
+ }
+}
+
+void TimeSyncer::postSendPacket(int64_t delayUs) {
+ (new AMessage(kWhatSendPacket, id()))->post(delayUs);
+}
+
+void TimeSyncer::postTimeout() {
+ sp<AMessage> msg = new AMessage(kWhatTimedOut, id());
+ msg->setInt32("generation", mTimeoutGeneration);
+ msg->post(kTimeoutDelayUs);
+}
+
+void TimeSyncer::cancelTimeout() {
+ ++mTimeoutGeneration;
+}
+
+void TimeSyncer::notifyError(status_t err) {
+ if (mNotify == NULL) {
+ looper()->stop();
+ return;
+ }
+
+ sp<AMessage> notify = mNotify->dup();
+ notify->setInt32("what", kWhatError);
+ notify->setInt32("err", err);
+ notify->post();
+}
+
+// static
+int TimeSyncer::CompareRountripTime(const TimeInfo *ti1, const TimeInfo *ti2) {
+ int64_t rt1 = ti1->mT4 - ti1->mT1;
+ int64_t rt2 = ti2->mT4 - ti2->mT1;
+
+ if (rt1 < rt2) {
+ return -1;
+ } else if (rt1 > rt2) {
+ return 1;
+ }
+
+ return 0;
+}
+
+void TimeSyncer::notifyOffset() {
+ mHistory.sort(CompareRountripTime);
+
+ int64_t sum = 0ll;
+ size_t count = 0;
+
+ // Only consider the third of the information associated with the best
+ // (smallest) roundtrip times.
+ for (size_t i = 0; i < mHistory.size() / 3; ++i) {
+ const TimeInfo *ti = &mHistory[i];
+
+#if 0
+ // One way delay for a packet to travel from client
+ // to server or back (assumed to be the same either way).
+ int64_t delay =
+ (ti->mT2 - ti->mT1 + ti->mT4 - ti->mT3) / 2;
+#endif
+
+ // Offset between the client clock (T1, T4) and the
+ // server clock (T2, T3) timestamps.
+ int64_t offset =
+ (ti->mT2 - ti->mT1 - ti->mT4 + ti->mT3) / 2;
+
+ ALOGV("(%d) RT: %lld us, offset: %lld us",
+ i, ti->mT4 - ti->mT1, offset);
+
+ sum += offset;
+ ++count;
+ }
+
+ if (mNotify == NULL) {
+ ALOGI("avg. offset is %lld", sum / count);
+ return;
+ }
+
+ sp<AMessage> notify = mNotify->dup();
+ notify->setInt32("what", kWhatTimeOffset);
+ notify->setInt64("offset", sum / count);
+ notify->post();
+}
+
+} // namespace android
--- /dev/null
+/*
+ * 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.
+ */
+
+#ifndef TIME_SYNCER_H_
+
+#define TIME_SYNCER_H_
+
+#include <media/stagefright/foundation/AHandler.h>
+
+namespace android {
+
+struct ANetworkSession;
+
+/*
+ TimeSyncer allows us to synchronize time between a client and a server.
+ The client sends a UDP packet containing its send-time to the server,
+ the server sends that packet back to the client amended with information
+ about when it was received as well as the time the reply was sent back.
+ Finally the client receives the reply and has now enough information to
+ compute the clock offset between client and server assuming that packet
+ exchange is symmetric, i.e. time for a packet client->server and
+ server->client is roughly equal.
+ This exchange is repeated a number of times and the average offset computed
+ over the 30% of packets that had the lowest roundtrip times.
+ The offset is determined every 10 secs to account for slight differences in
+ clock frequency.
+*/
+struct TimeSyncer : public AHandler {
+ enum {
+ kWhatError,
+ kWhatTimeOffset,
+ };
+ TimeSyncer(
+ const sp<ANetworkSession> &netSession,
+ const sp<AMessage> ¬ify);
+
+ void startServer(unsigned localPort);
+ void startClient(const char *remoteHost, unsigned remotePort);
+
+protected:
+ virtual ~TimeSyncer();
+
+ virtual void onMessageReceived(const sp<AMessage> &msg);
+
+private:
+ enum {
+ kWhatStartServer,
+ kWhatStartClient,
+ kWhatUDPNotify,
+ kWhatSendPacket,
+ kWhatTimedOut,
+ };
+
+ struct TimeInfo {
+ int64_t mT1; // client timestamp at send
+ int64_t mT2; // server timestamp at receive
+ int64_t mT3; // server timestamp at send
+ int64_t mT4; // client timestamp at receive
+ };
+
+ enum {
+ kNumPacketsPerBatch = 30,
+ };
+ static const int64_t kTimeoutDelayUs = 500000ll;
+ static const int64_t kBatchDelayUs = 10000000ll; // every 10 secs
+
+ sp<ANetworkSession> mNetSession;
+ sp<AMessage> mNotify;
+
+ bool mIsServer;
+ bool mConnected;
+ int32_t mUDPSession;
+ uint32_t mSeqNo;
+ double mTotalTimeUs;
+
+ Vector<TimeInfo> mHistory;
+
+ int64_t mPendingT1;
+ int32_t mTimeoutGeneration;
+
+ void postSendPacket(int64_t delayUs = 0ll);
+
+ void postTimeout();
+ void cancelTimeout();
+
+ void notifyError(status_t err);
+ void notifyOffset();
+
+ static int CompareRountripTime(const TimeInfo *ti1, const TimeInfo *ti2);
+
+ DISALLOW_EVIL_CONSTRUCTORS(TimeSyncer);
+};
+
+} // namespace android
+
+#endif // TIME_SYNCER_H_
}
status_t RTPReceiver::TSAssembler::processPacket(const sp<ABuffer> &packet) {
+ int32_t rtpTime;
+ CHECK(packet->meta()->findInt32("rtp-time", &rtpTime));
+
+ packet->meta()->setInt64("timeUs", (rtpTime * 100ll) / 9);
+
postAccessUnit(packet, mSawDiscontinuity);
if (mSawDiscontinuity) {
mNumDeclaredLostPrior = mNumDeclaredLost;
- ALOGI("lost %lld packets (%.2f %%), declared %d lost\n",
- lostInterval,
- 100.0f * lostInterval / expectedInterval,
- declaredLostInterval);
+ if (declaredLostInterval > 0) {
+ ALOGI("lost %lld packets (%.2f %%), declared %d lost\n",
+ lostInterval,
+ 100.0f * lostInterval / expectedInterval,
+ declaredLostInterval);
+ }
}
mNextReportTimeUs = nowUs + kReportIntervalUs;
return OK;
}
+status_t RTPReceiver::notifyLateness(int64_t latenessUs) {
+ sp<ABuffer> buf = new ABuffer(20);
+
+ uint8_t *ptr = buf->data();
+ ptr[0] = 0x80 | 0;
+ ptr[1] = 204; // APP
+ ptr[2] = 0;
+
+ CHECK((buf->size() % 4) == 0u);
+ ptr[3] = (buf->size() / 4) - 1;
+
+ ptr[4] = kSourceID >> 24; // SSRC
+ ptr[5] = (kSourceID >> 16) & 0xff;
+ ptr[6] = (kSourceID >> 8) & 0xff;
+ ptr[7] = kSourceID & 0xff;
+ ptr[8] = 'l';
+ ptr[9] = 'a';
+ ptr[10] = 't';
+ ptr[11] = 'e';
+
+ ptr[12] = latenessUs >> 56;
+ ptr[13] = (latenessUs >> 48) & 0xff;
+ ptr[14] = (latenessUs >> 40) & 0xff;
+ ptr[15] = (latenessUs >> 32) & 0xff;
+ ptr[16] = (latenessUs >> 24) & 0xff;
+ ptr[17] = (latenessUs >> 16) & 0xff;
+ ptr[18] = (latenessUs >> 8) & 0xff;
+ ptr[19] = latenessUs & 0xff;
+
+ mNetSession->sendRequest(mRTCPSessionID, buf->data(), buf->size());
+
+ return OK;
+}
+
void RTPReceiver::onMessageReceived(const sp<AMessage> &msg) {
switch (msg->what()) {
case kWhatRTPNotify:
int32_t remoteRTPPort,
int32_t remoteRTCPPort);
+ status_t notifyLateness(int64_t latenessUs);
+
protected:
virtual ~RTPReceiver();
virtual void onMessageReceived(const sp<AMessage> &msg);
case 202: // SDES
case 203:
+ break;
+
case 204: // APP
break;
: mSurfaceTex(bufferProducer),
mVideoDecoderNotificationPending(false),
mRenderPending(false),
- mFirstRenderTimeUs(-1ll),
- mFirstRenderRealUs(-1ll) {
+ mTimeOffsetUs(0ll),
+ mLatencySum(0ll),
+ mLatencyCount(0),
+ mNumFramesLate(0),
+ mNumFrames(0) {
}
DirectRenderer::~DirectRenderer() {
}
}
+void DirectRenderer::setTimeOffset(int64_t offset) {
+ mTimeOffsetUs = offset;
+}
+
+int64_t DirectRenderer::getAvgLatenessUs() {
+ if (mLatencyCount == 0) {
+ return 0ll;
+ }
+
+ int64_t avgLatencyUs = mLatencySum / mLatencyCount;
+
+ mLatencySum = 0ll;
+ mLatencyCount = 0;
+
+ if (mNumFrames > 0) {
+ ALOGI("%d / %d frames late", mNumFramesLate, mNumFrames);
+ mNumFramesLate = 0;
+ mNumFrames = 0;
+ }
+
+ return avgLatencyUs;
+}
+
void DirectRenderer::onMessageReceived(const sp<AMessage> &msg) {
switch (msg->what()) {
case kWhatVideoDecoderNotify:
}
void DirectRenderer::queueOutputBuffer(size_t index, int64_t timeUs) {
-#if 0
+#if 1
OutputInfo info;
info.mIndex = index;
- info.mTimeUs = timeUs;
+ info.mTimeUs = timeUs + mTimeOffsetUs;
mOutputBuffers.push_back(info);
scheduleRenderIfNecessary();
#else
+ mLatencySum += ALooper::GetNowUs() - (timeUs + mTimeOffsetUs);
+ ++mLatencyCount;
+
status_t err = mVideoDecoder->renderOutputBufferAndRelease(index);
CHECK_EQ(err, (status_t)OK);
#endif
int64_t timeUs = (*mOutputBuffers.begin()).mTimeUs;
int64_t nowUs = ALooper::GetNowUs();
- if (mFirstRenderTimeUs < 0ll) {
- mFirstRenderTimeUs = timeUs;
- mFirstRenderRealUs = nowUs;
- }
-
- int64_t whenUs = timeUs - mFirstRenderTimeUs + mFirstRenderRealUs;
- int64_t delayUs = whenUs - nowUs;
+ int64_t delayUs = timeUs - nowUs;
(new AMessage(kWhatRender, id()))->post(delayUs);
}
break;
}
+ if (info.mTimeUs + 15000ll < nowUs) {
+ ++mNumFramesLate;
+ }
+ ++mNumFrames;
+
+ mLatencySum += nowUs - info.mTimeUs;
+ ++mLatencyCount;
+
status_t err = mVideoDecoder->renderOutputBufferAndRelease(info.mIndex);
CHECK_EQ(err, (status_t)OK);
void setFormat(size_t trackIndex, const sp<AMessage> &format);
void queueAccessUnit(size_t trackIndex, const sp<ABuffer> &accessUnit);
+ void setTimeOffset(int64_t offset);
+
+ int64_t getAvgLatenessUs();
+
protected:
virtual void onMessageReceived(const sp<AMessage> &msg);
virtual ~DirectRenderer();
List<OutputInfo> mOutputBuffers;
bool mRenderPending;
- int64_t mFirstRenderTimeUs;
- int64_t mFirstRenderRealUs;
+
+ int64_t mTimeOffsetUs;
+
+ int64_t mLatencySum;
+ size_t mLatencyCount;
+
+ int32_t mNumFramesLate;
+ int32_t mNumFrames;
void onVideoDecoderNotify();
void onRender();
#include <gui/SurfaceComposerClient.h>
#include <media/IMediaPlayerService.h>
#include <media/IStreamSource.h>
+#include <media/mediaplayer.h>
#include <media/stagefright/foundation/ABuffer.h>
#include <media/stagefright/foundation/ADebug.h>
#include <media/stagefright/foundation/AMessage.h>
void doSomeWork();
+ void setTimeOffset(int64_t offset);
+
protected:
virtual ~StreamSource();
size_t mNumDeqeued;
+ int64_t mTimeOffsetUs;
+ bool mTimeOffsetChanged;
+
DISALLOW_EVIL_CONSTRUCTORS(StreamSource);
};
TunnelRenderer::StreamSource::StreamSource(TunnelRenderer *owner)
: mOwner(owner),
- mNumDeqeued(0) {
+ mNumDeqeued(0),
+ mTimeOffsetUs(0ll),
+ mTimeOffsetChanged(false) {
}
TunnelRenderer::StreamSource::~StreamSource() {
}
uint32_t TunnelRenderer::StreamSource::flags() const {
- return kFlagAlignedVideoData;
+ return kFlagAlignedVideoData | kFlagIsRealTimeData;
}
void TunnelRenderer::StreamSource::doSomeWork() {
++mNumDeqeued;
- if (mNumDeqeued == 1) {
- ALOGI("fixing real time now.");
-
+ if (mTimeOffsetChanged) {
sp<AMessage> extra = new AMessage;
extra->setInt32(
IStreamListener::kKeyDiscontinuityMask,
- ATSParser::DISCONTINUITY_ABSOLUTE_TIME);
+ ATSParser::DISCONTINUITY_TIME_OFFSET);
- extra->setInt64("timeUs", ALooper::GetNowUs());
+ extra->setInt64("offset", mTimeOffsetUs);
mListener->issueCommand(
IStreamListener::DISCONTINUITY,
false /* synchronous */,
extra);
+
+ mTimeOffsetChanged = false;
}
ALOGV("dequeue TS packet of size %d", srcBuffer->size());
}
}
+void TunnelRenderer::StreamSource::setTimeOffset(int64_t offset) {
+ Mutex::Autolock autoLock(mLock);
+
+ if (offset != mTimeOffsetUs) {
+ mTimeOffsetUs = offset;
+ mTimeOffsetChanged = true;
+ }
+}
+
////////////////////////////////////////////////////////////////////////////////
TunnelRenderer::TunnelRenderer(
const sp<IGraphicBufferProducer> &bufferProducer)
: mSurfaceTex(bufferProducer),
mStartup(true) {
+ mStreamSource = new StreamSource(this);
}
TunnelRenderer::~TunnelRenderer() {
destroyPlayer();
}
+void TunnelRenderer::setTimeOffset(int64_t offset) {
+ mStreamSource->setTimeOffset(offset);
+}
+
void TunnelRenderer::onMessageReceived(const sp<AMessage> &msg) {
switch (msg->what()) {
default:
sp<IMediaPlayerService> service = interface_cast<IMediaPlayerService>(binder);
CHECK(service.get() != NULL);
- mStreamSource = new StreamSource(this);
-
mPlayerClient = new PlayerClient;
mPlayer = service->create(mPlayerClient, 0);
void TunnelRenderer::destroyPlayer() {
mStreamSource.clear();
+ mPlayer->setVideoSurfaceTexture(NULL);
+
mPlayer->stop();
mPlayer.clear();
void queueBuffer(const sp<ABuffer> &buffer);
sp<ABuffer> dequeueBuffer();
+ void setTimeOffset(int64_t offset);
+
+ int64_t getAvgLatenessUs() {
+ return 0ll;
+ }
+
protected:
virtual void onMessageReceived(const sp<AMessage> &msg);
virtual ~TunnelRenderer();
#include "DirectRenderer.h"
#include "MediaReceiver.h"
#include "ParsedMessage.h"
+#include "TimeSyncer.h"
#include "TunnelRenderer.h"
+#include <cutils/properties.h>
#include <media/stagefright/foundation/ABuffer.h>
#include <media/stagefright/foundation/ADebug.h>
#include <media/stagefright/foundation/AMessage.h>
#include <media/stagefright/MediaErrors.h>
-#include <cutils/properties.h>
-
namespace android {
WifiDisplaySink::WifiDisplaySink(
+ uint32_t flags,
const sp<ANetworkSession> &netSession,
const sp<IGraphicBufferProducer> &bufferProducer,
const sp<AMessage> ¬ify)
: mState(UNDEFINED),
+ mFlags(flags),
mNetSession(netSession),
mSurfaceTex(bufferProducer),
mNotify(notify),
mUsingTCPInterleaving(false),
mSessionID(0),
mNextCSeq(1),
- mIDRFrameRequestPending(false) {
+ mIDRFrameRequestPending(false),
+ mTimeOffsetUs(0ll),
+ mTimeOffsetValid(false),
+ mTargetLatencyUs(-1ll),
+ mSetupDeferred(false) {
// We support any and all resolutions, but prefer 720p30
mSinkSupportedVideoFormats.setNativeResolution(
VideoFormats::RESOLUTION_CEA, 5); // 1280 x 720 p30
{
ALOGI("We're now connected.");
mState = CONNECTED;
+
+ if (mFlags & FLAG_SPECIAL_MODE) {
+ sp<AMessage> notify = new AMessage(
+ kWhatTimeSyncerNotify, id());
+
+ mTimeSyncer = new TimeSyncer(mNetSession, notify);
+ looper()->registerHandler(mTimeSyncer);
+
+ mTimeSyncer->startClient(mRTSPHost.c_str(), 8123);
+ }
break;
}
break;
}
+ case kWhatTimeSyncerNotify:
+ {
+ int32_t what;
+ CHECK(msg->findInt32("what", &what));
+
+ if (what == TimeSyncer::kWhatTimeOffset) {
+ CHECK(msg->findInt64("offset", &mTimeOffsetUs));
+ mTimeOffsetValid = true;
+
+ if (mSetupDeferred) {
+ CHECK_EQ((status_t)OK,
+ sendSetup(
+ mSessionID,
+ "rtsp://x.x.x.x:x/wfd1.0/streamid=0"));
+
+ mSetupDeferred = false;
+ }
+ }
+ break;
+ }
+
+ case kWhatReportLateness:
+ {
+ int64_t latenessUs = mRenderer->getAvgLatenessUs();
+
+ ALOGI("avg. lateness = %lld ms",
+ (latenessUs + mTargetLatencyUs) / 1000ll);
+
+ mMediaReceiver->notifyLateness(
+ 0 /* trackIndex */, latenessUs);
+
+ msg->post(kReportLatenessEveryUs);
+ break;
+ }
+
default:
TRESPASS();
}
looper()->registerHandler(mRenderer);
}
+ CHECK(mTimeOffsetValid);
+
+ int64_t latencyUs = 300000ll; // 300ms by default
+
+ char val[PROPERTY_VALUE_MAX];
+ if (property_get("media.wfd-sink.latency", val, NULL)) {
+ char *end;
+ int64_t x = strtoll(val, &end, 10);
+
+ if (end > val && *end == '\0' && x >= 0ll) {
+ latencyUs = x;
+ }
+ }
+
+ if (latencyUs != mTargetLatencyUs) {
+ mTargetLatencyUs = latencyUs;
+
+ ALOGI("Assuming %lld ms of latency.", latencyUs / 1000ll);
+ }
+
+ // We are the timesync _client_,
+ // client time = server time - time offset.
+ mRenderer->setTimeOffset(-mTimeOffsetUs + mTargetLatencyUs);
+
sp<ABuffer> accessUnit;
CHECK(msg->findBuffer("accessUnit", &accessUnit));
-#if USE_TUNNEL_RENDERER
- mRenderer->queueBuffer(accessUnit);
-#else
size_t trackIndex;
CHECK(msg->findSize("trackIndex", &trackIndex));
+#if USE_TUNNEL_RENDERER
+ mRenderer->queueBuffer(accessUnit);
+#else
sp<AMessage> format;
if (msg->findMessage("format", &format)) {
mRenderer->setFormat(trackIndex, format);
mState = PLAYING;
+ (new AMessage(kWhatReportLateness, id()))->post(kReportLatenessEveryUs);
+
return OK;
}
mUsingTCPTransport = true;
mUsingTCPInterleaving = true;
}
+ } else if (mFlags & FLAG_SPECIAL_MODE) {
+ mUsingTCPTransport = true;
}
body = "wfd_video_formats: ";
const char *content = data->getContent();
if (strstr(content, "wfd_trigger_method: SETUP\r\n") != NULL) {
- status_t err =
- sendSetup(
- sessionID,
- "rtsp://x.x.x.x:x/wfd1.0/streamid=0");
+ if ((mFlags & FLAG_SPECIAL_MODE) && !mTimeOffsetValid) {
+ mSetupDeferred = true;
+ } else {
+ status_t err =
+ sendSetup(
+ sessionID,
+ "rtsp://x.x.x.x:x/wfd1.0/streamid=0");
- CHECK_EQ(err, (status_t)OK);
+ CHECK_EQ(err, (status_t)OK);
+ }
}
AString response = "RTSP/1.0 200 OK\r\n";
struct DirectRenderer;
struct MediaReceiver;
struct ParsedMessage;
+struct TimeSyncer;
struct TunnelRenderer;
#define USE_TUNNEL_RENDERER 0
kWhatDisconnected,
};
+ enum Flags {
+ FLAG_SPECIAL_MODE = 1,
+ };
+
// If no notification message is specified (notify == NULL)
// the sink will stop its looper() once the session ends,
// otherwise it will post an appropriate notification but leave
// the looper() running.
WifiDisplaySink(
+ uint32_t flags,
const sp<ANetworkSession> &netSession,
const sp<IGraphicBufferProducer> &bufferProducer = NULL,
const sp<AMessage> ¬ify = NULL);
kWhatRTSPNotify,
kWhatStop,
kWhatMediaReceiverNotify,
+ kWhatTimeSyncerNotify,
+ kWhatReportLateness,
};
struct ResponseID {
typedef status_t (WifiDisplaySink::*HandleRTSPResponseFunc)(
int32_t sessionID, const sp<ParsedMessage> &msg);
+ static const int64_t kReportLatenessEveryUs = 1000000ll;
+
State mState;
+ uint32_t mFlags;
VideoFormats mSinkSupportedVideoFormats;
sp<ANetworkSession> mNetSession;
sp<IGraphicBufferProducer> mSurfaceTex;
sp<AMessage> mNotify;
+ sp<TimeSyncer> mTimeSyncer;
bool mUsingTCPTransport;
bool mUsingTCPInterleaving;
AString mRTSPHost;
bool mIDRFrameRequestPending;
+ int64_t mTimeOffsetUs;
+ bool mTimeOffsetValid;
+
+ int64_t mTargetLatencyUs;
+
+ bool mSetupDeferred;
+
status_t sendM2(int32_t sessionID);
status_t sendSetup(int32_t sessionID, const char *uri);
status_t sendPlay(int32_t sessionID, const char *uri);
#include "Parameters.h"
#include "ParsedMessage.h"
#include "rtp/RTPSender.h"
+#include "TimeSyncer.h"
#include <binder/IServiceManager.h>
#include <gui/IGraphicBufferProducer.h>
}
if (err == OK) {
+ sp<AMessage> notify = new AMessage(kWhatTimeSyncerNotify, id());
+ mTimeSyncer = new TimeSyncer(mNetSession, notify);
+ looper()->registerHandler(mTimeSyncer);
+
+ mTimeSyncer->startServer(8123);
+
mState = AWAITING_CLIENT_CONNECTION;
}
break;
}
+ case kWhatTimeSyncerNotify:
+ {
+ break;
+ }
+
default:
TRESPASS();
}
struct IHDCP;
struct IRemoteDisplayClient;
struct ParsedMessage;
+struct TimeSyncer;
// Represents the RTSP server acting as a wifi display source.
// Manages incoming connections, sets up Playback sessions as necessary.
kWhatHDCPNotify,
kWhatFinishStop2,
kWhatTeardownTriggerTimedOut,
+ kWhatTimeSyncerNotify,
};
struct ResponseID {
VideoFormats mSupportedSourceVideoFormats;
sp<ANetworkSession> mNetSession;
sp<IRemoteDisplayClient> mClient;
+ sp<TimeSyncer> mTimeSyncer;
struct in_addr mInterfaceAddr;
int32_t mSessionID;
#include <utils/Log.h>
#include "ANetworkSession.h"
+#include "TimeSyncer.h"
#include <binder/ProcessState.h>
-#include <media/stagefright/foundation/ABuffer.h>
-#include <media/stagefright/foundation/ADebug.h>
-#include <media/stagefright/foundation/AHandler.h>
-#include <media/stagefright/foundation/ALooper.h>
#include <media/stagefright/foundation/AMessage.h>
-#include <media/stagefright/Utils.h>
namespace android {
-struct TestHandler : public AHandler {
- TestHandler(const sp<ANetworkSession> &netSession);
-
- void startServer(unsigned localPort);
- void startClient(const char *remoteHost, unsigned remotePort);
-
-protected:
- virtual ~TestHandler();
-
- virtual void onMessageReceived(const sp<AMessage> &msg);
-
-private:
- enum {
- kWhatStartServer,
- kWhatStartClient,
- kWhatUDPNotify,
- kWhatSendPacket,
- kWhatTimedOut,
- };
-
- struct TimeInfo {
- int64_t mT1; // client timestamp at send
- int64_t mT2; // server timestamp at receive
- int64_t mT3; // server timestamp at send
- int64_t mT4; // client timestamp at receive
- };
-
- static const int64_t kTimeoutDelayUs = 1000000ll;
-
- sp<ANetworkSession> mNetSession;
-
- bool mIsServer;
- bool mConnected;
- int32_t mUDPSession;
- uint32_t mSeqNo;
- double mTotalTimeUs;
- int32_t mCount;
- int64_t mSumOffsets;
-
- int64_t mPendingT1;
- int32_t mTimeoutGeneration;
-
- void postSendPacket(int64_t delayUs = 0ll);
-
- void postTimeout();
- void cancelTimeout();
-
- DISALLOW_EVIL_CONSTRUCTORS(TestHandler);
-};
-
-TestHandler::TestHandler(const sp<ANetworkSession> &netSession)
- : mNetSession(netSession),
- mIsServer(false),
- mConnected(false),
- mUDPSession(0),
- mSeqNo(0),
- mTotalTimeUs(0.0),
- mCount(0),
- mSumOffsets(0ll),
- mPendingT1(0ll),
- mTimeoutGeneration(0) {
-}
-
-TestHandler::~TestHandler() {
-}
-
-void TestHandler::startServer(unsigned localPort) {
- sp<AMessage> msg = new AMessage(kWhatStartServer, id());
- msg->setInt32("localPort", localPort);
- msg->post();
-}
-
-void TestHandler::startClient(const char *remoteHost, unsigned remotePort) {
- sp<AMessage> msg = new AMessage(kWhatStartClient, id());
- msg->setString("remoteHost", remoteHost);
- msg->setInt32("remotePort", remotePort);
- msg->post();
-}
-
-void TestHandler::onMessageReceived(const sp<AMessage> &msg) {
- switch (msg->what()) {
- case kWhatStartClient:
- {
- AString remoteHost;
- CHECK(msg->findString("remoteHost", &remoteHost));
-
- int32_t remotePort;
- CHECK(msg->findInt32("remotePort", &remotePort));
-
- sp<AMessage> notify = new AMessage(kWhatUDPNotify, id());
-
- CHECK_EQ((status_t)OK,
- mNetSession->createUDPSession(
- 0 /* localPort */,
- remoteHost.c_str(),
- remotePort,
- notify,
- &mUDPSession));
-
- postSendPacket();
- break;
- }
-
- case kWhatStartServer:
- {
- mIsServer = true;
-
- int32_t localPort;
- CHECK(msg->findInt32("localPort", &localPort));
-
- sp<AMessage> notify = new AMessage(kWhatUDPNotify, id());
-
- CHECK_EQ((status_t)OK,
- mNetSession->createUDPSession(
- localPort, notify, &mUDPSession));
-
- break;
- }
-
- case kWhatSendPacket:
- {
- TimeInfo ti;
- memset(&ti, 0, sizeof(ti));
-
- ti.mT1 = ALooper::GetNowUs();
-
- CHECK_EQ((status_t)OK,
- mNetSession->sendRequest(
- mUDPSession, &ti, sizeof(ti)));
-
- mPendingT1 = ti.mT1;
- postTimeout();
- break;
- }
-
- case kWhatTimedOut:
- {
- int32_t generation;
- CHECK(msg->findInt32("generation", &generation));
-
- if (generation != mTimeoutGeneration) {
- break;
- }
-
- ALOGI("timed out, sending another request");
- postSendPacket();
- break;
- }
-
- case kWhatUDPNotify:
- {
- 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));
-
- AString detail;
- CHECK(msg->findString("detail", &detail));
-
- ALOGE("An error occurred in session %d (%d, '%s/%s').",
- sessionID,
- err,
- detail.c_str(),
- strerror(-err));
-
- mNetSession->destroySession(sessionID);
-
- cancelTimeout();
- looper()->stop();
- break;
- }
-
- case ANetworkSession::kWhatDatagram:
- {
- int32_t sessionID;
- CHECK(msg->findInt32("sessionID", &sessionID));
-
- sp<ABuffer> packet;
- CHECK(msg->findBuffer("data", &packet));
-
- int64_t arrivalTimeUs;
- CHECK(packet->meta()->findInt64(
- "arrivalTimeUs", &arrivalTimeUs));
-
- CHECK_EQ(packet->size(), sizeof(TimeInfo));
-
- TimeInfo *ti = (TimeInfo *)packet->data();
-
- if (mIsServer) {
- if (!mConnected) {
- AString fromAddr;
- CHECK(msg->findString("fromAddr", &fromAddr));
-
- int32_t fromPort;
- CHECK(msg->findInt32("fromPort", &fromPort));
-
- CHECK_EQ((status_t)OK,
- mNetSession->connectUDPSession(
- mUDPSession, fromAddr.c_str(), fromPort));
-
- mConnected = true;
- }
-
- ti->mT2 = arrivalTimeUs;
- ti->mT3 = ALooper::GetNowUs();
-
- CHECK_EQ((status_t)OK,
- mNetSession->sendRequest(
- mUDPSession, ti, sizeof(*ti)));
- } else {
- if (ti->mT1 != mPendingT1) {
- break;
- }
-
- cancelTimeout();
- mPendingT1 = 0;
-
- ti->mT4 = arrivalTimeUs;
-
- // One way delay for a packet to travel from client
- // to server or back (assumed to be the same either way).
- int64_t delay =
- (ti->mT2 - ti->mT1 + ti->mT4 - ti->mT3) / 2;
-
- // Offset between the client clock (T1, T4) and the
- // server clock (T2, T3) timestamps.
- int64_t offset =
- (ti->mT2 - ti->mT1 - ti->mT4 + ti->mT3) / 2;
-
- mSumOffsets += offset;
- ++mCount;
-
- printf("delay = %lld us,\toffset %lld us\n",
- delay,
- offset);
- fflush(stdout);
-
- postSendPacket(1000000ll / 30);
- }
- break;
- }
-
- default:
- TRESPASS();
- }
-
- break;
- }
-
- default:
- TRESPASS();
- }
-}
-
-void TestHandler::postSendPacket(int64_t delayUs) {
- (new AMessage(kWhatSendPacket, id()))->post(delayUs);
-}
-
-void TestHandler::postTimeout() {
- sp<AMessage> msg = new AMessage(kWhatTimedOut, id());
- msg->setInt32("generation", mTimeoutGeneration);
- msg->post(kTimeoutDelayUs);
-}
-
-void TestHandler::cancelTimeout() {
- ++mTimeoutGeneration;
-}
-
} // namespace android
static void usage(const char *me) {
sp<ALooper> looper = new ALooper;
- sp<TestHandler> handler = new TestHandler(netSession);
+ sp<TimeSyncer> handler = new TimeSyncer(netSession, NULL /* notify */);
looper->registerHandler(handler);
if (localPort >= 0) {
sp<ALooper> looper = new ALooper;
sp<WifiDisplaySink> sink = new WifiDisplaySink(
- session, surface->getIGraphicBufferProducer());
+ 0 /* flags */,
+ session,
+ surface->getIGraphicBufferProducer());
+
looper->registerHandler(sink);
if (connectToPort >= 0) {
OSDN Git Service
