path: root/media/libaah_rtp/aah_rx_player_core.cpp

/*
 * Copyright (C) 2011 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_TAG "LibAAH_RTP"
//#define LOG_NDEBUG 0
#include <utils/Log.h>

#include <fcntl.h>
#include <poll.h>
#include <sys/socket.h>
#include <time.h>
#include <utils/misc.h>

#include <media/stagefright/Utils.h>

#include "aah_rx_player.h"
#include "aah_tx_packet.h"

namespace android {

const uint32_t AAH_RXPlayer::kRetransRequestMagic =
    FOURCC('T','r','e','q');
const uint32_t AAH_RXPlayer::kRetransNAKMagic =
    FOURCC('T','n','a','k');
const uint32_t AAH_RXPlayer::kFastStartRequestMagic =
    FOURCC('T','f','s','t');
const uint32_t AAH_RXPlayer::kGapRerequestTimeoutUSec = 75000;
const uint32_t AAH_RXPlayer::kFastStartTimeoutUSec = 800000;
const uint32_t AAH_RXPlayer::kRTPActivityTimeoutUSec = 10000000;

static inline int16_t fetchInt16(uint8_t* data) {
    return static_cast<int16_t>(U16_AT(data));
}

static inline int32_t fetchInt32(uint8_t* data) {
    return static_cast<int32_t>(U32_AT(data));
}

static inline int64_t fetchInt64(uint8_t* data) {
    return static_cast<int64_t>(U64_AT(data));
}

uint64_t AAH_RXPlayer::monotonicUSecNow() {
    struct timespec now;
    int res = clock_gettime(CLOCK_MONOTONIC, &now);
    CHECK(res >= 0);

    uint64_t ret = static_cast<uint64_t>(now.tv_sec) * 1000000;
    ret += now.tv_nsec / 1000;

    return ret;
}

status_t AAH_RXPlayer::startWorkThread() {
    status_t res;
    stopWorkThread();
    res = thread_wrapper_->run("TRX_Player", PRIORITY_AUDIO);

    if (res != OK) {
        ALOGE("Failed to start work thread (res = %d)", res);
    }

    return res;
}

void AAH_RXPlayer::stopWorkThread() {
    thread_wrapper_->requestExit();  // set the exit pending flag
    wakeup_work_thread_evt_.setEvent();

    status_t res;
    res = thread_wrapper_->requestExitAndWait(); // block until thread exit.
    if (res != OK) {
        ALOGE("Failed to stop work thread (res = %d)", res);
    }

    wakeup_work_thread_evt_.clearPendingEvents();
}

void AAH_RXPlayer::cleanupSocket() {
    if (sock_fd_ >= 0) {
        if (multicast_joined_) {
            int res;
            struct ip_mreq mreq;
            mreq.imr_multiaddr = listen_addr_.sin_addr;
            mreq.imr_interface.s_addr = htonl(INADDR_ANY);
            res = setsockopt(sock_fd_,
                             IPPROTO_IP,
                             IP_DROP_MEMBERSHIP,
                             &mreq, sizeof(mreq));
            if (res < 0) {
                ALOGW("Failed to leave multicast group. (%d, %d)", res, errno);
            }
            multicast_joined_ = false;
        }

        close(sock_fd_);
        sock_fd_ = -1;
    }

    resetPipeline();
}

void AAH_RXPlayer::resetPipeline() {
    ring_buffer_.reset();

    // Explicitly shudown all of the active substreams, then call clear out the
    // collection.  Failure to clear out a substream can result in its decoder
    // holding a reference to itself and therefor not going away when the
    // collection is cleared.
    for (size_t i = 0; i < substreams_.size(); ++i)
        substreams_.valueAt(i)->shutdown();

    substreams_.clear();

    current_gap_status_ = kGS_NoGap;
}

bool AAH_RXPlayer::setupSocket() {
    long flags;
    int  res, buf_size;
    socklen_t opt_size;

    cleanupSocket();
    CHECK(sock_fd_ < 0);

    // Make the socket
    sock_fd_ = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
    if (sock_fd_ < 0) {
        ALOGE("Failed to create listen socket (errno %d)", errno);
        goto bailout;
    }

    // Set non-blocking operation
    flags = fcntl(sock_fd_, F_GETFL);
    res   = fcntl(sock_fd_, F_SETFL, flags | O_NONBLOCK);
    if (res < 0) {
        ALOGE("Failed to set socket (%d) to non-blocking mode (errno %d)",
              sock_fd_, errno);
        goto bailout;
    }

    // Bind to our port
    struct sockaddr_in bind_addr;
    memset(&bind_addr, 0, sizeof(bind_addr));
    bind_addr.sin_family = AF_INET;
    bind_addr.sin_addr.s_addr = INADDR_ANY;
    bind_addr.sin_port = listen_addr_.sin_port;
    res = bind(sock_fd_,
               reinterpret_cast<const sockaddr*>(&bind_addr),
               sizeof(bind_addr));
    if (res < 0) {
        uint32_t a = ntohl(bind_addr.sin_addr.s_addr);
        uint16_t p = ntohs(bind_addr.sin_port);
        ALOGE("Failed to bind socket (%d) to %d.%d.%d.%d:%hd. (errno %d)",
              sock_fd_,
              (a >> 24) & 0xFF,
              (a >> 16) & 0xFF,
              (a >>  8) & 0xFF,
              (a      ) & 0xFF,
              p,
              errno);

        goto bailout;
    }

    buf_size = 1 << 16;   // 64k
    res = setsockopt(sock_fd_,
                     SOL_SOCKET, SO_RCVBUF,
                     &buf_size, sizeof(buf_size));
    if (res < 0) {
        ALOGW("Failed to increase socket buffer size to %d.  (errno %d)",
              buf_size, errno);
    }

    buf_size = 0;
    opt_size = sizeof(buf_size);
    res = getsockopt(sock_fd_,
                     SOL_SOCKET, SO_RCVBUF,
                     &buf_size, &opt_size);
    if (res < 0) {
        ALOGW("Failed to fetch socket buffer size.  (errno %d)", errno);
    } else {
        ALOGI("RX socket buffer size is now %d bytes",  buf_size);
    }

    if (listen_addr_.sin_addr.s_addr) {
        // Join the multicast group and we should be good to go.
        struct ip_mreq mreq;
        mreq.imr_multiaddr = listen_addr_.sin_addr;
        mreq.imr_interface.s_addr = htonl(INADDR_ANY);
        res = setsockopt(sock_fd_,
                         IPPROTO_IP,
                         IP_ADD_MEMBERSHIP,
                         &mreq, sizeof(mreq));
        if (res < 0) {
            ALOGE("Failed to join multicast group. (errno %d)", errno);
            goto bailout;
        }
        multicast_joined_ = true;
    }

    return true;

bailout:
    cleanupSocket();
    return false;
}

bool AAH_RXPlayer::threadLoop() {
    struct pollfd poll_fds[2];
    bool process_more_right_now = false;

    if (!setupSocket()) {
        sendEvent(MEDIA_ERROR);
        goto bailout;
    }

    while (!thread_wrapper_->exitPending()) {
        // Step 1: Wait until there is something to do.
        int gap_timeout = computeNextGapRetransmitTimeout();
        int ring_timeout = ring_buffer_.computeInactivityTimeout();
        int timeout = -1;

        if (!ring_timeout) {
            ALOGW("RTP inactivity timeout reached, resetting pipeline.");
            resetPipeline();
            timeout = gap_timeout;
        } else {
            if (gap_timeout < 0) {
                timeout = ring_timeout;
            } else if (ring_timeout < 0) {
                timeout = gap_timeout;
            } else {
                timeout = (gap_timeout < ring_timeout) ? gap_timeout
                                                       : ring_timeout;
            }
        }

        if ((0 != timeout) && (!process_more_right_now)) {
            // Set up the events to wait on.  Start with the wakeup pipe.
            memset(&poll_fds, 0, sizeof(poll_fds));
            poll_fds[0].fd     = wakeup_work_thread_evt_.getWakeupHandle();
            poll_fds[0].events = POLLIN;

            // Add the RX socket.
            poll_fds[1].fd     = sock_fd_;
            poll_fds[1].events = POLLIN;

            // Wait for something interesing to happen.
            int poll_res = poll(poll_fds, NELEM(poll_fds), timeout);
            if (poll_res < 0) {
                ALOGE("Fatal error (%d,%d) while waiting on events",
                      poll_res, errno);
                sendEvent(MEDIA_ERROR);
                goto bailout;
            }
        }

        if (thread_wrapper_->exitPending()) {
            break;
        }

        wakeup_work_thread_evt_.clearPendingEvents();
        process_more_right_now = false;

        // Step 2: Do we have data waiting in the socket?  If so, drain the
        // socket moving valid RTP information into the ring buffer to be
        // processed.
        if (poll_fds[1].revents) {
            struct sockaddr_in from;
            socklen_t from_len;

            ssize_t res = 0;
            while (!thread_wrapper_->exitPending()) {
                // Check the size of any pending packet.
                res = recv(sock_fd_, NULL, 0, MSG_PEEK | MSG_TRUNC);

                // Error?
                if (res < 0) {
                    // If the error is anything other than would block,
                    // something has gone very wrong.
                    if ((errno != EAGAIN) && (errno != EWOULDBLOCK)) {
                        ALOGE("Fatal socket error during recvfrom (%d, %d)",
                              (int)res, errno);
                        goto bailout;
                    }

                    // Socket is out of data, just break out of processing and
                    // wait for more.
                    break;
                }

                // Allocate a payload.
                PacketBuffer* pb = PacketBuffer::allocate(res);
                if (NULL == pb) {
                    ALOGE("Fatal error, failed to allocate packet buffer of"
                          " length %u", static_cast<uint32_t>(res));
                    goto bailout;
                }

                // Fetch the data.
                from_len = sizeof(from);
                res = recvfrom(sock_fd_, pb->data_, pb->length_, 0,
                               reinterpret_cast<struct sockaddr*>(&from),
                               &from_len);
                if (res != pb->length_) {
                    ALOGE("Fatal error, fetched packet length (%d) does not"
                          " match peeked packet length (%u).  This should never"
                          " happen.  (errno = %d)",
                          static_cast<int>(res),
                          static_cast<uint32_t>(pb->length_),
                          errno);
                }

                bool drop_packet = false;
                if (transmitter_known_) {
                    if (from.sin_addr.s_addr !=
                        transmitter_addr_.sin_addr.s_addr) {
                        uint32_t a = ntohl(from.sin_addr.s_addr);
                        uint16_t p = ntohs(from.sin_port);
                        ALOGV("Dropping packet from unknown transmitter"
                              " %u.%u.%u.%u:%hu",
                              ((a >> 24) & 0xFF),
                              ((a >> 16) & 0xFF),
                              ((a >>  8) & 0xFF),
                              ( a        & 0xFF),
                              p);

                        drop_packet = true;
                    } else {
                        transmitter_addr_.sin_port = from.sin_port;
                    }
                } else {
                    memcpy(&transmitter_addr_, &from, sizeof(from));
                    transmitter_known_ = true;
                }

                if (!drop_packet) {
                    bool serious_error = !processRX(pb);

                    if (serious_error) {
                        // Something went "seriously wrong".  Currently, the
                        // only trigger for this should be a ring buffer
                        // overflow.  The current failsafe behavior for when
                        // something goes seriously wrong is to just reset the
                        // pipeline.  The system should behave as if this
                        // AAH_RXPlayer was just set up for the first time.
                        ALOGE("Something just went seriously wrong with the"
                              " pipeline.  Resetting.");
                        resetPipeline();
                    }
                } else {
                    PacketBuffer::destroy(pb);
                }
            }
        }

        // Step 3: Process any data we mave have accumulated in the ring buffer
        // so far.
        if (!thread_wrapper_->exitPending()) {
            processRingBuffer();
        }

        // Step 4: At this point in time, the ring buffer should either be
        // empty, or stalled in front of a gap caused by some dropped packets.
        // Check on the current gap situation and deal with it in an appropriate
        // fashion.  If processGaps returns true, it means that it has given up
        // on a gap and that we should try to process some more data
        // immediately.
        if (!thread_wrapper_->exitPending()) {
            process_more_right_now = processGaps();
        }

        // Step 5: Check for fatal errors.  If any of our substreams has
        // encountered a fatal, unrecoverable, error, then propagate the error
        // up to user level and shut down.
        for (size_t i = 0; i < substreams_.size(); ++i) {
            status_t status;
            CHECK(substreams_.valueAt(i) != NULL);

            status = substreams_.valueAt(i)->getStatus();
            if (OK != status) {
                ALOGE("Substream index %d has encountered an unrecoverable"
                      " error (%d).  Signalling application level and shutting"
                      " down.", i, status);
                sendEvent(MEDIA_ERROR);
                goto bailout;
            }
        }
    }

bailout:
    cleanupSocket();
    return false;
}

bool AAH_RXPlayer::processRX(PacketBuffer* pb) {
    CHECK(NULL != pb);

    uint8_t* data = pb->data_;
    ssize_t  amt  = pb->length_;
    uint32_t nak_magic;
    uint16_t seq_no;
    uint32_t epoch;

    // Every packet either starts with an RTP header which is at least 12 bytes
    // long or is a retry NAK which is 14 bytes long.  If there are fewer than
    // 12 bytes here, this cannot be a proper RTP packet.
    if (amt < 12) {
        ALOGV("Dropping packet, too short to contain RTP header (%u bytes)",
              static_cast<uint32_t>(amt));
        goto drop_packet;
    }

    // Check to see if this is the special case of a NAK packet.
    nak_magic = ntohl(*(reinterpret_cast<uint32_t*>(data)));
    if (nak_magic == kRetransNAKMagic) {
        // Looks like a NAK packet; make sure its long enough.

        if (amt < static_cast<ssize_t>(sizeof(RetransRequest))) {
            ALOGV("Dropping packet, too short to contain NAK payload"
                  " (%u bytes)", static_cast<uint32_t>(amt));
            goto drop_packet;
        }

        SeqNoGap gap;
        RetransRequest* rtr = reinterpret_cast<RetransRequest*>(data);
        gap.start_seq_ = ntohs(rtr->start_seq_);
        gap.end_seq_   = ntohs(rtr->end_seq_);

        ALOGV("Process NAK for gap at [%hu, %hu]",
                gap.start_seq_, gap.end_seq_);
        ring_buffer_.processNAK(&gap);

        return true;
    }

    // According to the TRTP spec, version should be 2, padding should be 0,
    // extension should be 0 and CSRCCnt should be 0.  If any of these tests
    // fail, we chuck the packet.
    if (data[0] != 0x80) {
        ALOGV("Dropping packet, bad V/P/X/CSRCCnt field (0x%02x)",
              data[0]);
        goto drop_packet;
    }

    // Check the payload type.  For TRTP, it should always be 100.
    if ((data[1] & 0x7F) != 100) {
        ALOGV("Dropping packet, bad payload type. (%u)",
              data[1] & 0x7F);
        goto drop_packet;
    }

    // Check whether the transmitter has begun a new epoch.
    epoch = (U32_AT(data + 8) >> 10) & 0x3FFFFF;
    if (current_epoch_known_) {
        if (epoch != current_epoch_) {
            ALOGV("%s: new epoch %u", __PRETTY_FUNCTION__, epoch);
            current_epoch_ = epoch;
            resetPipeline();
        }
    } else {
        current_epoch_ = epoch;
        current_epoch_known_ = true;
    }

    // Extract the sequence number and hand the packet off to the ring buffer
    // for dropped packet detection and later processing.
    seq_no = U16_AT(data + 2);
    return ring_buffer_.pushBuffer(pb, seq_no);

drop_packet:
    PacketBuffer::destroy(pb);
    return true;
}

void AAH_RXPlayer::processRingBuffer() {
    PacketBuffer* pb;
    bool is_discon;
    sp<Substream> substream;
    LinearTransform trans;
    bool foundTrans = false;

    while (NULL != (pb = ring_buffer_.fetchBuffer(&is_discon))) {
        if (is_discon) {
            // Abort all partially assembled payloads.
            for (size_t i = 0; i < substreams_.size(); ++i) {
                CHECK(substreams_.valueAt(i) != NULL);
                substreams_.valueAt(i)->cleanupBufferInProgress();
            }
        }

        uint8_t* data = pb->data_;
        ssize_t  amt  = pb->length_;

        // Should not have any non-RTP packets in the ring buffer.  RTP packets
        // must be at least 12 bytes long.
        CHECK(amt >= 12);

        // Extract the marker bit and the SSRC field.
        bool     marker = (data[1] & 0x80) != 0;
        uint32_t ssrc   = U32_AT(data + 8);

        // Is this the start of a new TRTP payload?  If so, the marker bit
        // should be set and there are some things we should be checking for.
        if (marker) {
            // TRTP headers need to have at least a byte for version, a byte for
            // payload type and flags, and 4 bytes for length.
            if (amt < 18) {
                ALOGV("Dropping packet, too short to contain TRTP header"
                      " (%u bytes)", static_cast<uint32_t>(amt));
                goto process_next_packet;
            }

            // Check the TRTP version and extract the payload type/flags.
            uint8_t trtp_version =  data[12];
            uint8_t payload_type = (data[13] >> 4) & 0xF;
            uint8_t trtp_flags   =  data[13]       & 0xF;

            if (1 != trtp_version) {
                ALOGV("Dropping packet, bad trtp version %hhu", trtp_version);
                goto process_next_packet;
            }

            // Is there a timestamp transformation present on this packet?  If
            // so, extract it and pass it to the appropriate substreams.
            if (trtp_flags & 0x02) {
                ssize_t offset = 18 + ((trtp_flags & 0x01) ? 4 : 0);
                if (amt < (offset + 24)) {
                    ALOGV("Dropping packet, too short to contain TRTP Timestamp"
                          " Transformation (%u bytes)",
                          static_cast<uint32_t>(amt));
                    goto process_next_packet;
                }

                trans.a_zero = fetchInt64(data + offset);
                trans.b_zero = fetchInt64(data + offset + 16);
                trans.a_to_b_numer = static_cast<int32_t>(
                        fetchInt32 (data + offset + 8));
                trans.a_to_b_denom = U32_AT(data + offset + 12);
                foundTrans = true;

                uint32_t program_id = (ssrc >> 5) & 0x1F;
                for (size_t i = 0; i < substreams_.size(); ++i) {
                    sp<Substream> iter = substreams_.valueAt(i);
                    CHECK(iter != NULL);

                    if (iter->getProgramID() == program_id) {
                        iter->processTSTransform(trans);
                    }
                }
            }

            // Is this a command packet?  If so, its not necessarily associate
            // with one particular substream.  Just give it to the command
            // packet handler and then move on.
            if (4 == payload_type) {
                processCommandPacket(pb);
                goto process_next_packet;
            }
        }

        // If we got to here, then we are a normal packet.  Find (or allocate)
        // the substream we belong to and send the packet off to be processed.
        substream = substreams_.valueFor(ssrc);
        if (substream == NULL) {
            substream = new Substream(ssrc, omx_);
            if (substream == NULL) {
                ALOGE("Failed to allocate substream for SSRC 0x%08x", ssrc);
                goto process_next_packet;
            }
            substreams_.add(ssrc, substream);

            if (foundTrans) {
                substream->processTSTransform(trans);
            }
        }

        CHECK(substream != NULL);

        if (marker) {
            // Start of a new TRTP payload for this substream.  Extract the
            // lower 32 bits of the timestamp and hand the buffer to the
            // substream for processing.
            uint32_t ts_lower = U32_AT(data + 4);
            substream->processPayloadStart(data + 12, amt - 12, ts_lower);
        } else {
            // Continuation of an existing TRTP payload.  Just hand it off to
            // the substream for processing.
            substream->processPayloadCont(data + 12, amt - 12);
        }

process_next_packet:
        PacketBuffer::destroy(pb);
    }  // end of main processing while loop.
}

void AAH_RXPlayer::processCommandPacket(PacketBuffer* pb) {
    CHECK(NULL != pb);

    uint8_t* data = pb->data_;
    ssize_t  amt  = pb->length_;

    // verify that this packet meets the minimum length of a command packet
    if (amt < 20) {
        return;
    }

    uint8_t trtp_version =  data[12];
    uint8_t trtp_flags   =  data[13]       & 0xF;

    if (1 != trtp_version) {
        ALOGV("Dropping packet, bad trtp version %hhu", trtp_version);
        return;
    }

    // calculate the start of the command payload
    ssize_t offset = 18;
    if (trtp_flags & 0x01) {
        // timestamp is present (4 bytes)
        offset += 4;
    }
    if (trtp_flags & 0x02) {
        // transform is present (24 bytes)
        offset += 24;
    }

    // the packet must contain 2 bytes of command payload beyond the TRTP header
    if (amt < offset + 2) {
        return;
    }

    uint16_t command_id = U16_AT(data + offset);

    switch (command_id) {
        case TRTPControlPacket::kCommandNop:
            break;

        case TRTPControlPacket::kCommandEOS:
        case TRTPControlPacket::kCommandFlush: {
            uint16_t program_id = (U32_AT(data + 8) >> 5) & 0x1F;
            ALOGI("*** %s flushing program_id=%d",
                  __PRETTY_FUNCTION__, program_id);

            Vector<uint32_t> substreams_to_remove;
            for (size_t i = 0; i < substreams_.size(); ++i) {
                sp<Substream> iter = substreams_.valueAt(i);
                if (iter->getProgramID() == program_id) {
                    iter->shutdown();
                    substreams_to_remove.add(iter->getSSRC());
                }
            }

            for (size_t i = 0; i < substreams_to_remove.size(); ++i) {
                substreams_.removeItem(substreams_to_remove[i]);
            }
        } break;
    }
}

bool AAH_RXPlayer::processGaps() {
    // Deal with the current gap situation.  Specifically...
    //
    // 1) If a new gap has shown up, send a retransmit request to the
    //    transmitter.
    // 2) If a gap we were working on has had a packet in the middle or at
    //    the end filled in, send another retransmit request for the begining
    //    portion of the gap.  TRTP was designed for LANs where packet
    //    re-ordering is very unlikely; so see the middle or end of a gap
    //    filled in before the begining is an almost certain indication that
    //    a retransmission packet was also dropped.
    // 3) If we have been working on a gap for a while and it still has not
    //    been filled in, send another retransmit request.
    // 4) If the are no more gaps in the ring, clear the current_gap_status_
    //    flag to indicate that all is well again.

    // Start by fetching the active gap status.
    SeqNoGap gap;
    bool send_retransmit_request = false;
    bool ret_val = false;
    GapStatus gap_status;
    if (kGS_NoGap != (gap_status = ring_buffer_.fetchCurrentGap(&gap))) {
        // Note: checking for a change in the end sequence number should cover
        // moving on to an entirely new gap for case #1 as well as resending the
        // begining of a gap range for case #2.
        send_retransmit_request = (kGS_NoGap == current_gap_status_) ||
                                  (current_gap_.end_seq_ != gap.end_seq_);

        // If this is the same gap we have been working on, and it has timed
        // out, then check to see if our substreams are about to underflow.  If
        // so, instead of sending another retransmit request, just give up on
        // this gap and move on.
        if (!send_retransmit_request &&
           (kGS_NoGap != current_gap_status_) &&
           (0 == computeNextGapRetransmitTimeout())) {

            // If out current gap is the fast-start gap, don't bother to skip it
            // because substreams look like the are about to underflow.
            if ((kGS_FastStartGap != gap_status) ||
                (current_gap_.end_seq_ != gap.end_seq_)) {
                for (size_t i = 0; i < substreams_.size(); ++i) {
                    if (substreams_.valueAt(i)->isAboutToUnderflow()) {
                        ALOGV("About to underflow, giving up on gap [%hu, %hu]",
                              gap.start_seq_, gap.end_seq_);
                        ring_buffer_.processNAK();
                        current_gap_status_ = kGS_NoGap;
                        return true;
                    }
                }
            }

            // Looks like no one is about to underflow.  Just go ahead and send
            // the request.
            send_retransmit_request = true;
        }
    } else {
        current_gap_status_ = kGS_NoGap;
    }

    if (send_retransmit_request) {
        // If we have been working on a fast start, and it is still not filled
        // in, even after the extended retransmit time out, give up and skip it.
        // The system should fall back into its normal slow-start behavior.
        if ((kGS_FastStartGap == current_gap_status_) &&
            (current_gap_.end_seq_ == gap.end_seq_)) {
            ALOGV("Fast start is taking forever; giving up.");
            ring_buffer_.processNAK();
            current_gap_status_ = kGS_NoGap;
            return true;
        }

        // Send the request.
        RetransRequest req;
        uint32_t magic  = (kGS_FastStartGap == gap_status)
                        ? kFastStartRequestMagic
                        : kRetransRequestMagic;
        req.magic_      = htonl(magic);
        req.mcast_ip_   = listen_addr_.sin_addr.s_addr;
        req.mcast_port_ = listen_addr_.sin_port;
        req.start_seq_  = htons(gap.start_seq_);
        req.end_seq_    = htons(gap.end_seq_);

        {
            uint32_t a = ntohl(transmitter_addr_.sin_addr.s_addr);
            uint16_t p = ntohs(transmitter_addr_.sin_port);
            ALOGV("Sending to transmitter %u.%u.%u.%u:%hu",
                    ((a >> 24) & 0xFF),
                    ((a >> 16) & 0xFF),
                    ((a >>  8) & 0xFF),
                    ( a        & 0xFF),
                    p);
        }

        int res = sendto(sock_fd_, &req, sizeof(req), 0,
                         reinterpret_cast<struct sockaddr*>(&transmitter_addr_),
                         sizeof(transmitter_addr_));
        if (res < 0) {
            ALOGE("Error when sending retransmit request (%d)", errno);
        } else {
            ALOGV("%s request for range [%hu, %hu] sent",
                  (kGS_FastStartGap == gap_status) ? "Fast Start"
                                                   : "Retransmit",
                  gap.start_seq_, gap.end_seq_);
        }

        // Update the current gap info.
        current_gap_ = gap;
        current_gap_status_ = gap_status;
        next_retrans_req_time_ = monotonicUSecNow() +
                               ((kGS_FastStartGap == current_gap_status_)
                                ? kFastStartTimeoutUSec
                                : kGapRerequestTimeoutUSec);
    }

    return false;
}

// Compute when its time to send the next gap retransmission in milliseconds.
// Returns < 0 for an infinite timeout (no gap) and 0 if its time to retransmit
// right now.
int AAH_RXPlayer::computeNextGapRetransmitTimeout() {
    if (kGS_NoGap == current_gap_status_) {
        return -1;
    }

    int64_t timeout_delta = next_retrans_req_time_ - monotonicUSecNow();

    timeout_delta /= 1000;
    if (timeout_delta <= 0) {
        return 0;
    }

    return static_cast<uint32_t>(timeout_delta);
}

}  // namespace android