jitter.cxx

sloedgy open sip stack source code

{
  PTRACE(3, "RTP\tRemoving jitter buffer " << this << ' ' << GetThreadName());

  shuttingDown = TRUE;
  PAssert(WaitForTermination(10000), "Jitter buffer thread did not terminate");

  bufferMutex.Wait();

  // Free up all the memory allocated
  while (oldestFrame != NULL) {
    Entry * frame = oldestFrame;
    oldestFrame = oldestFrame->next;
    delete frame;
  }

  while (freeFrames != NULL) {
    Entry * frame = freeFrames;
    freeFrames = freeFrames->next;
    delete frame;
  }

  delete currentWriteFrame;

  bufferMutex.Signal();

#if PTRACING && !defined(NO_ANALYSER)
  PTRACE(5, "Jitter buffer analysis: size=" << bufferSize
         << " time=" << currentJitterTime << '\n' << *analyser);
  delete analyser;
#endif
}


void RTP_JitterBuffer::SetDelay(unsigned minJitterDelay, unsigned maxJitterDelay)
{
  if (shuttingDown)
    PAssert(WaitForTermination(10000), "Jitter buffer thread did not terminate");

  bufferMutex.Wait();

  minJitterTime = minJitterDelay;
  maxJitterTime = maxJitterDelay;
  currentJitterTime = minJitterDelay;
  targetJitterTime = currentJitterTime;

  PINDEX newBufferSize = maxJitterTime/40+1;
  while (bufferSize < newBufferSize) {
    Entry * frame = new Entry;
    frame->prev = NULL;
    frame->next = freeFrames;
    freeFrames->prev = frame;
    freeFrames = frame;
    bufferSize++;
  }

  if (IsTerminated()) {
    packetsTooLate = 0;
    bufferOverruns = 0;
    consecutiveBufferOverruns = 0;
    consecutiveMarkerBits = 0;
    consecutiveEarlyPacketStartTime = 0;

    oldestFrame = newestFrame = currentWriteFrame = NULL;

    shuttingDown = FALSE;
    preBuffering = TRUE;

    PTRACE(2, "RTP\tJitter buffer restarted:"
              " size=" << bufferSize <<
              " delay=" << minJitterTime << '-' << maxJitterTime << '/' << currentJitterTime <<
              " (" << (currentJitterTime/timeUnits) << "ms)");
    Restart();
  }

  bufferMutex.Signal();
}


void RTP_JitterBuffer::Main()
{
  PTRACE(3, "RTP\tJitter RTP receive thread started: " << this);

  bufferMutex.Wait();

  for (;;) {

    // Get the next free frame available for use for reading from the RTP
    // transport. Place it into a parking spot.
    Entry * currentReadFrame;
    if (freeFrames != NULL) {
      // Take the next free frame and make it the current for reading
      currentReadFrame = freeFrames;
      freeFrames = freeFrames->next;
      if (freeFrames != NULL)
        freeFrames->prev = NULL;
      PTRACE_IF(2, consecutiveBufferOverruns > 1,
                "RTP\tJitter buffer full, threw away "
                << consecutiveBufferOverruns << " oldest frames");
      consecutiveBufferOverruns = 0;
    }
    else {
      // We have a full jitter buffer, need a new frame so take the oldest one
      currentReadFrame = oldestFrame;
      if (oldestFrame != NULL)
	oldestFrame = oldestFrame->next;
      if (oldestFrame != NULL)
        oldestFrame->prev = NULL;
      currentDepth--;
      bufferOverruns++;
      consecutiveBufferOverruns++;
      if (consecutiveBufferOverruns > MAX_BUFFER_OVERRUNS) {
        PTRACE(2, "RTP\tJitter buffer continuously full, throwing away entire buffer.");
        freeFrames = oldestFrame;
        oldestFrame = newestFrame = NULL;
        preBuffering = TRUE;
      }
      else {
        PTRACE_IF(2, consecutiveBufferOverruns == 1,
                  "RTP\tJitter buffer full, throwing away oldest frame ("
                  << currentReadFrame->GetTimestamp() << ')');
      }
    }

    if (currentReadFrame == NULL) {
      bufferMutex.Signal();
      return;
    }

    currentReadFrame->next = NULL;

    bufferMutex.Signal();

    // Keep reading from the RTP transport frames
    if (!session.ReadData(*currentReadFrame)) {
      if (currentReadFrame != NULL)
	delete currentReadFrame;  // Destructor won't delete this one, so do it here.
      shuttingDown = TRUE; // Flag to stop the reading side thread
      PTRACE(3, "RTP\tJitter RTP receive thread ended");
      return;
    }

    currentReadFrame->tick = PTimer::Tick();

    if (consecutiveMarkerBits < maxConsecutiveMarkerBits) {
      if (currentReadFrame != NULL && currentReadFrame->GetMarker()) {
        PTRACE(3, "RTP\tReceived start of talk burst: " << currentReadFrame->GetTimestamp());
        //preBuffering = TRUE;
        consecutiveMarkerBits++;
      }
      else
        consecutiveMarkerBits = 0;
    }
    else {
      if (currentReadFrame != NULL && currentReadFrame->GetMarker())
        currentReadFrame->SetMarker(FALSE);
      if (consecutiveMarkerBits == maxConsecutiveMarkerBits) {
        PTRACE(3, "RTP\tEvery packet has Marker bit, ignoring them from this client!");
      }
    }
    
    
#if PTRACING && !defined(NO_ANALYSER)
    analyser->In(currentReadFrame->GetTimestamp(), currentDepth, preBuffering ? "PreBuf" : "");
#endif

    // Queue the frame for playing by the thread at other end of jitter buffer
    bufferMutex.Wait();

    // Have been reading a frame, put it into the queue now, at correct position
    if (newestFrame == NULL || oldestFrame == NULL)
      oldestFrame = newestFrame = currentReadFrame; // Was empty
    else {
      DWORD time = currentReadFrame->GetTimestamp();

      if (time > newestFrame->GetTimestamp()) {
        // Is newer than newst, put at that end of queue
        currentReadFrame->prev = newestFrame;
        newestFrame->next = currentReadFrame;
        newestFrame = currentReadFrame;
      }
      else if (time <= oldestFrame->GetTimestamp()) {
        // Is older than the oldest, put at that end of queue
        currentReadFrame->next = oldestFrame;
        oldestFrame->prev = currentReadFrame;
        oldestFrame = currentReadFrame;
      }
      else {
        // Somewhere in between, locate its position
        Entry * frame = newestFrame->prev;
        while (time < frame->GetTimestamp())
          frame = frame->prev;

        currentReadFrame->prev = frame;
        currentReadFrame->next = frame->next;
        frame->next->prev = currentReadFrame;
        frame->next = currentReadFrame;
      }
    }

    currentDepth++;
  }
}


BOOL RTP_JitterBuffer::ReadData(DWORD timestamp, RTP_DataFrame & frame)
{
  if (shuttingDown)
    return FALSE;

  /*Free the frame just written to codec, putting it back into
    the free list and clearing the parking spot for it.
   */
  bufferMutex.Wait();
  if (currentWriteFrame != NULL) {

    // Move frame from current to free list
    currentWriteFrame->next = freeFrames;
    if (freeFrames != NULL)
      freeFrames->prev = currentWriteFrame;
    freeFrames = currentWriteFrame;

    currentWriteFrame = NULL;
  }
  bufferMutex.Signal();


  // Default response is an empty frame, ie silence
  frame.SetPayloadSize(0);

  PWaitAndSignal mutex(bufferMutex);

  /*Get the next frame to write to the codec. Takes it from the oldest
    position in the queue, if it is time to do so, and parks it in the
    special member so can unlock the mutex while the writer thread has its
    way with the buffer.
   */
  if (oldestFrame == NULL) {
    /*No data to play! We ran the buffer down to empty, restart buffer by
      setting flag that will fill it again before returning any data.
     */
    preBuffering = TRUE;
    currentJitterTime = targetJitterTime;
    

#if PTRACING && !defined(NO_ANALYSER)
    analyser->Out(0, currentDepth, "Empty");
#endif
    return TRUE;
  }

 
  DWORD oldestTimestamp = oldestFrame->GetTimestamp();
  DWORD newestTimestamp = newestFrame->GetTimestamp();

  /* If there is an opportunity (due to silence in the buffer) to implement a desired 
  reduction in the size of the jitter buffer, effect it */

  if (targetJitterTime < currentJitterTime &&
      (newestTimestamp - oldestTimestamp) < currentJitterTime) {
    currentJitterTime = ( targetJitterTime > (newestTimestamp - oldestTimestamp)) ?
                          targetJitterTime : (newestTimestamp - oldestTimestamp);

    PTRACE(3, "RTP\tJitter buffer size decreased to "
           << currentJitterTime << " (" << (currentJitterTime/timeUnits) << "ms)");
  }

  /* See if time for this packet, if our oldest frame is older than the
     required age, then use it. If it is not time yet, make sure that the
     writer thread isn't falling behind (not enough MIPS). If the time
     between the oldest and the newest entries in the jitter buffer is
     greater than the size specified for the buffer, then return the oldest
     entry regardless, making the writer thread catch up.
  */

  if (preBuffering) {
    // Reset jitter baseline (should be handled by GetMarker() condition, but just in case...)
    lastWriteTimestamp = 0;
    lastWriteTick = 0;

    /*
    // Check for requesting something that already exceeds the maximum time,
    // or have filled the jitter buffer, not filling if this is so
    if ((timestamp - oldestTimestamp) < currentJitterTime &&
        (newestTimestamp - oldestTimestamp) < currentJitterTime/2) {
    */

    // If oldest frame has not been in the buffer long enough, don't return anything yet
    if ((PTimer::Tick() - oldestFrame->tick).GetInterval() * timeUnits
         < currentJitterTime / 2) {
#if PTRACING && !defined(NO_ANALYSER)
      analyser->Out(oldestTimestamp, currentDepth, "PreBuf");
#endif
      return TRUE;
    }

    preBuffering = FALSE;
  }


  //Handle short silence bursts in the middle of the buffer
  // - if we think we're getting marker bit information, use that
  BOOL shortSilence = FALSE;
  if (consecutiveMarkerBits < maxConsecutiveMarkerBits) {
      if (oldestFrame->GetMarker() &&
          (PTimer::Tick() - oldestFrame->tick).GetInterval()* timeUnits < currentJitterTime / 2)
        shortSilence = TRUE;
  }
  else if (timestamp < oldestTimestamp && timestamp > (newestTimestamp - currentJitterTime))
    shortSilence = TRUE;
  
  if (shortSilence) {
    // It is not yet time for something in the buffer
#if PTRACING && !defined(NO_ANALYSER)
    analyser->Out(oldestTimestamp, currentDepth, "Wait");
#endif