defaultcircularbuffer.cpp

来自「C++封装的视频采集代码」· C++ 代码 · 共 1,498 行 · 第 1/5 页

                std::pair<bool, unsigned> oldestResult = IsOldestReader(readerNumber);

                // Determine the space available to the writer before releasing
                // the object
                bool wasBelowWriteThreshold = !GetWriteThreshold() ? true : !IsAboveWriteThreshold();

                // Advance the read position beyond the last object read from the
                // buffer
                readPositions_[readerNumber].position_ = GetAfterPosition(*object);

                // Return the object to its pool
                GetPool(*object)->Free(object);

                // If the reader is the sole oldest reader
                if (oldestResult.first && (oldestResult.second == 1))
                {
                    // Must be space available for write
//TRACE("$BRR Setting writeEmpty_ false\n");
                    writeEmpty_ = false;
                }

                // One less object available for read
                --numberOfObjects_[readerNumber];
//TRACE("$B<$n");

                // Notify writers that space has been freed in the buffer and inform
                // any listeners of empty or threshold crossing events
                CommonReleaseFromRead(wasBelowWriteThreshold);
            }
//TRACE("$BRR leaving: object=0x%lx, readerNumber = %u$n\n", reinterpret_cast<unsigned long>(object), readerNumber);
        }
        // }}}

        void DefaultCircularBuffer::ReleaseFromWrite(
            const CircularBufferResidentObject& object,
            bool discard)
        // {{{
        {
            using namespace std::rel_ops;

            syslib::Lock lock(mutex_);
//TRACE("$CRW entered: object=0x%lx, discard = %d$n\n", reinterpret_cast<unsigned long>(&object), (int)discard);

#ifdef INUSE_TRACKING
            // Catch trying to release object when none are in-use
            if (!writeInUse_.size())
            {
                TRACE("$RDefaultCircularBuffer::ReleaseFromWrite() Attempt to release when no buffers in-use$n\n");
                return;
            }

            // Check that the object being released is the object in the current
            // in-use set that was acquired earliest
            if (*writeInUse_.begin() != &object)
            {
                TRACE("$RDefaultCircularBuffer::ReleaseFromWrite() Attempt to release buffer 0x%lx which is not the earliest in-use$n\n",
                    reinterpret_cast<unsigned long>(&object));
                return;
            }

            // The earliest in-use is to be released, so remove from record
            // of in-use objects
            writeInUse_.erase(writeInUse_.begin());
#endif

            // If discard is required then if object is the most recently acquired
            // then just don't advance the write pointer, however if the object is
            // not the most recent then write the JUMP_ID and object's in-buffer
            // length then update the write pointer to point to the next object
            // which is currently acquired for write as if discard was not required

            bool isMostRecent = GetAfterPosition(object) == mostRecentAcquireAfterPosition_;
            if (!discard)
            {
//TRACE("$CCommiting object$n\n");
                // Query the object for it's in-buffer commit attributes
                CircularBufferResidentObject::EncodeResult commitResult = object.CommitToBuffer();
                if (!commitResult.GetSuccessful())
                {
                    TRACE("$RDefaultCircularBuffer::ReleaseFromWrite() CommitToBuffer failed$n\n");
//TRACE("$CRW() leaving: nil$n\n");
                    return;
                }
                else
                {
//TRACE("$CAdjusting writePosition_, commitResult=[%d:%lu]$n\n", (int)commitResult.GetSkippedFirstSection(), commitResult.GetBytesConsumed());

                    // Determine the available-for-read threshold status for
                    // each of the readers
                    bool wasBelowThreshold[numberOfReaders_];
                    for (unsigned readerNumber=0; readerNumber < numberOfReaders_; ++readerNumber)
                    {
                        if ( GetReadThreshold(readerNumber) )
                        {
                            wasBelowThreshold[readerNumber] = !IsAboveReadThreshold(readerNumber);
//TRACE("$CReader %u wasBelowThreshold = %u$n\n", readerNumber, wasBelowThreshold[readerNumber]);
                        }
                    }

                    // Commit was successful. Did the object decide not to use
                    // the space remaining upto the end of the buffer?
                    if (commitResult.GetSkippedFirstSection())
                    {
//TRACE("$CSkipped first section$n\n");
                        writePosition_ += writePosition_.GetMargin();
                    }

                    // Move the write position past the end of the associated
                    // object
                    writePosition_ += commitResult.GetBytesConsumed();

                    // Ensure the write position finishes on a quad boundary
                    unsigned remainder = writePosition_.GetOffset() & 0x3UL;
                    if (remainder)
                    {
                        writePosition_ += (4 - remainder);
                    }

                    // Has the object's size been trimmed since it was initially
                    // acquired for write?
//TRACE("$CAfter=[%lu:%lu], writePosition_=[%lu:%lu]$n\n", GetAfterPosition(object).GetLoopCount(), GetAfterPosition(object).GetOffset(), writePosition_.GetLoopCount(), writePosition_.GetOffset());
                    unsigned long unusedSpace = GetDifference(writePosition_, GetAfterPosition(object));
//TRACE("$CunusedSpace = %lu$n\n", unusedSpace);
                    if (unusedSpace)
                    {
//TRACE("$CIs unused space$n\n");
                        if (isMostRecent)
                        {
//TRACE("$CIs most recent$n\n");
                            // Object is the most recent and has not used all
                            // its associated space, so adjust the most recent
                            // acquire position to match the end of the used
                            // space
                            mostRecentAcquireAfterPosition_ = writePosition_;

                            //TRACE("$GSaving %d.$n ", unusedSpace);
                        }
                        else
                        {
                            TRACE("$RWasting %d!$n ", unusedSpace);
//TRACE("$CIs $RNOT$C most recent$n\n");
                            // Object is not the most recently acquired, so must
                            // add skip info. to enable the readers to avoid
                            // the unused portion of the buffer

                            // Usused space will be a multiple of the size of a
                            // quad (which is assumed here to be the same size
                            // as a CircularBufferResidentObject::Id)
                            if (unusedSpace == sizeof(CircularBufferResidentObject::Id))
                            {
//TRACE("$CWritting jump-zero$n\n");
                                // Write the jump-zero ID to the buffer, which
                                // will result in readers just skipping the ID
                                // itself
                                *reinterpret_cast<CircularBufferResidentObject::Id*>(
                                    buffer_->GetBuffer() + writePosition_.GetOffset()) = JUMP_ZERO_ID;
                            }
                            else
                            {
                                // There is room to write the jump ID and a jump
                                // length to the buffer
                                *reinterpret_cast<CircularBufferResidentObject::Id*>(
                                    buffer_->GetBuffer() + writePosition_.GetOffset()) = JUMP_ID;

                                // Write the jump length to the buffer
                                *reinterpret_cast<unsigned long*>(
                                    buffer_->GetBuffer() + writePosition_.GetOffset() + sizeof(CircularBufferResidentObject::Id)) =
                                        unusedSpace -
                                        sizeof(CircularBufferResidentObject::Id) -
                                        sizeof(unsigned long);
//TRACE("$CWritting jump %lu$n\n", unusedSpace - sizeof(CircularBufferResidentObject::Id) - sizeof(unsigned long));
                            }

                            // Update write position to point to start of next
                            // object acquired for write
                            writePosition_ = GetAfterPosition(object);
                        }
                    }

                    // New data has been made available for readers, so clear
                    // all readers' empty flags and increment their available-
                    // for-read counts
                    for (unsigned readerNumber=0; readerNumber < numberOfReaders_; ++readerNumber)
                    {
//TRACE("$CSetting reader %u to not empty$n\n", readerNumber);
                        readPositions_[readerNumber].empty_ = false;
                        ++numberOfObjects_[readerNumber];
//TRACE("$B>$n");
                    }

                    // There must now be some buffers available for read, so
                    // allow checking for the condition of no read becoming
                    // available again
                    lastAnyReadBuffersAvailable_ = true;

                    // For any of the readers which was below the available-for-
                    // read threshold, check whether the threshold has now been
                    // exceeded
                    for (unsigned readerNumber=0; readerNumber < numberOfReaders_; ++readerNumber)
                    {
                        if ( !GetReadThreshold(readerNumber)  ||
                            (wasBelowThreshold[readerNumber] && IsAboveReadThreshold(readerNumber)))
                        {
                            // Threshold has been exceeded, so notify the listener
//TRACE("$CNot discard: Notifying read threshold listener %u$n\n", readerNumber);
                            NotifierWithParameter<ReadThresholdEventListener, unsigned>::NotifyListeners(readerNumber);
                        }
                    }

                    // Decrement count of objects acquired for write
                    --outstandingWriteAcquisitions_;

//TRACE("$CNotifying readers$n\n");
                    // Wake any readers blocked awaiting objects, as have just
                    // made one available
                    NotifyReaders();
                }
            }
            else
            {
//TRACE("$CDiscarding object$n\n");
                // Is the object to be discarded the most recently acquired?
                if (isMostRecent)
                {
//TRACE("$CIs most recent$n\n");
                    // Yes, so adjust the most recent after position back to the
                    // write position, as if the acquire for write did not hap-
                    // pen
                    mostRecentAcquireAfterPosition_ = writePosition_;
                }
                else
                {
//TRACE("$CIs NOT most recent$n\n");
                    // Determine the available-for-read threshold status for
                    // each of the readers
                    bool wasBelowThreshold[numberOfReaders_];
                    for (unsigned readerNumber=0; readerNumber < numberOfReaders_; ++readerNumber)
                    {
                        if ( GetReadThreshold(readerNumber) )
                        {
                            wasBelowThreshold[readerNumber] = !IsAboveReadThreshold(readerNumber);
//TRACE("$CReader %u wasBelowThreshold = %u$n\n", readerNumber, wasBelowThreshold[readerNumber]);
                        }
                    }

                    // No. Discarding an object which is not the most recently
                    // acquired requires inserting jump codes into the buffer,
                    // such that readers know to skip the discarded object
                    if (*reinterpret_cast<CircularBufferResidentObject::Id*>(
                            buffer_->GetBuffer() + writePosition_.GetOffset()) == MOVE_TO_START_ID)
                    {
//TRACE("$CMoving to start$n\n");
                        // Move write position to start
                        writePosition_ += writePosition_.GetMargin();
                    }

                    // Insert jump ID where object would have started had it not
                    // been decided to discard it
                    *reinterpret_cast<CircularBufferResidentObject::Id*>(
                        buffer_->GetBuffer() + writePosition_.GetOffset()) = JUMP_ID;

                    // Determine the object's association length, which includes
                    // rounding-up to next quad boundary
                    unsigned long associationLength = GetDifference(writePosition_, GetAfterPosition(object));

                    // Fill jump length with object's associated size minus the
                    // space occupied by the object ID, which has just been
                    // filled with the jump ID, and the size of the space occu-
                    // pied by the jump length itself
                    *reinterpret_cast<unsigned long*>(buffer_->GetBuffer() + writePosition_.GetOffset() + sizeof(CircularBufferResidentObject::Id)) =
                        associationLength - sizeof(CircularBufferResidentObject::Id) - sizeof(unsigned long);
//TRACE("$CWrote jump and length = %lu$n\n", associationLength - sizeof(CircularBufferResidentObject::Id) - sizeof(unsigned long));

                    // Update the write position to point to the first location
                    // after the just-discarded object
                    writePosition_ = GetAfterPosition(object);

                    // New data has been made available for readers (even though
                    // that data is to be skipped), so clear all readers' empty
                    // flags
                    for (unsigned readerNumber=0; readerNumber < numberOfReaders_; ++readerNumber)
                    {
//TRACE("$CSetting reader %u to not empty$n\n", readerNumber);
                        readPositions_[readerNumber].empty_ = false;
                    }

                    // For any of the readers which was below the available-for-
                    // read threshold, check whether the threshold has now been
                    // exceeded
                    for (unsigned readerNumber=0; readerNumber < numberOfReaders_; ++readerNumber)
                    {
                        if ( !GetReadThreshold(readerNumber) ||
                            (wasBelowThreshold[readerNumber] && IsAboveReadThreshold(readerNumber)))
                        {
                            // Threshold has been exceeded, so notify the listener
//TRACE("$CDiscard not most recent: Notifying read threshold listener %u$n\n", readerNumber);
                            NotifierWithParameter<ReadThresholdEventListener, unsigned>::NotifyListeners(readerNumber);