defaultcircularbuffer.cpp

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

            return (circularBufferSpaceInfo.GetFirstAvailableSection().GetLength() >= circularBufferSpaceInfo.GetSecondAvailableSection().GetLength()) ?
                circularBufferSpaceInfo.GetFirstAvailableSection().GetLength() : circularBufferSpaceInfo.GetSecondAvailableSection().GetLength();
        }
        // }}}

        CircularBufferSpaceInfo DefaultCircularBuffer::GetWriteSpaceInfo(
            const CircularBufferPosition& writePosition) const
        // {{{
        {
            syslib::Lock lock(mutex_);

            CircularBufferPosition* oldestReadPosition = &readPositions_[0].position_;
            for (unsigned i=1; i < numberOfReaders_; ++i)
            {
                if (readPositions_[i].position_ < *oldestReadPosition)
                {
                    oldestReadPosition = &readPositions_[i].position_;
                }
            }

//TRACE("$WGWSI writeEmpty_ = %d$n\n", writeEmpty_);
//TRACE("$WGWSI writePosition = [%lu:%lu:%lu]$n\n", writePosition.GetLoopCount(), writePosition.GetOffset(), writePosition.GetMargin());
//TRACE("$WGWSI oldestReadPosition = [%lu:%lu:%lu]$n\n", oldestReadPosition->GetLoopCount(), oldestReadPosition->GetOffset(), oldestReadPosition->GetMargin());
            // If writer points to the same location as the oldest reader, it
            // means either that the buffer has no write space available for
            // writing, or that the entire buffer is available for writing. Dis-
            // tinguish the cases with the writeEmpty flag
            if ((writePosition.GetOffset() == oldestReadPosition->GetOffset()) && writeEmpty_)
            {
//TRACE("$WGWSI Case 1$n\n");
                return CircularBufferSpaceInfo(
                    BufferSectionInfo(0, 0),
                    BufferSectionInfo(0, 0));
            }

            if (oldestReadPosition->GetOffset() > writePosition.GetOffset())
            {
//TRACE("$WGWSI Case 2$n\n");
                return CircularBufferSpaceInfo(
                    BufferSectionInfo(
                        reinterpret_cast<unsigned long>(buffer_->GetBuffer()) +
                            writePosition.GetOffset(),
                        oldestReadPosition->GetOffset() - writePosition.GetOffset()),
                    BufferSectionInfo(0, 0));
            }
            else
            {
//TRACE("$WGWSI Case 3, buffer_ = 0x%lx, writePosition.GetOffset() = %lu, writePosition.GetMargin() = %lu$n\n", reinterpret_cast<unsigned long>(buffer_->GetBuffer()), writePosition.GetOffset(), writePosition.GetMargin());
//TRACE("$WGWSI 0x%lx,%lu$n\n", reinterpret_cast<unsigned long>(buffer_->GetBuffer()) + writePosition.GetOffset(), writePosition.GetMargin());
                return CircularBufferSpaceInfo(
                    BufferSectionInfo(
                        reinterpret_cast<unsigned long>(buffer_->GetBuffer()) +
                            writePosition.GetOffset(),
                        writePosition.GetMargin()),
                    BufferSectionInfo(
                        reinterpret_cast<unsigned long>(buffer_->GetBuffer()),
                        oldestReadPosition->GetOffset()));
            }
        }
        // }}}

        bool DefaultCircularBuffer::IsComplete() const
        // {{{
        {
            return buffer_->GetBuffer();
        }
        // }}}

        bool DefaultCircularBuffer::IsAboveReadThreshold(unsigned readerNumber) const
        // {{{
        {
            syslib::Lock lock(mutex_);

            CircularBufferSpaceInfo readSpaceInfo;
            bool aboveReadThreshold = false;
            
            if (!readPositions_[readerNumber].empty_)
            {
                readSpaceInfo       = GetReadSpaceInfo(readPositions_[readerNumber].position_);
                aboveReadThreshold  = mpReadThresholds_[readerNumber].IsAboveThreshold(readSpaceInfo);
            }

//TRACE("$MR[%lu+%lu] %lu:%lu ->%s(%u)$n\n", readSpaceInfo.GetFirstAvailableSection().GetLength(), readSpaceInfo.GetSecondAvailableSection().GetLength(), GetReadThreshold(readerNumber), buffer_->GetSize(), aboveReadThreshold ? "above" : "below", readerNumber);
//TRACE("$MR%s:%lu$n\n", aboveReadThreshold ? "above" : "below", totalLength);
            return aboveReadThreshold;
        }
        // }}}

        bool DefaultCircularBuffer::IsAboveWriteThreshold() const
        // {{{
        {
            syslib::Lock lock(mutex_);

            CircularBufferSpaceInfo writeSpaceInfo = GetWriteSpaceInfo(writePosition_);
            
            bool aboveWriteThreshold = writeThreshold_.IsAboveThreshold( writeSpaceInfo );
            
//TRACE("$MW[%lu+%lu]->%s(%lu)$n\n", writeSpaceInfo.GetFirstAvailableSection().GetLength(), writeSpaceInfo.GetSecondAvailableSection().GetLength(), aboveWriteThreshold ? "above" : "below", GetWriteThreshold() );
            return aboveWriteThreshold;
        }
        // }}}

        bool DefaultCircularBuffer::IsEmpty(unsigned readerNumber) const
        // {{{
        {
            return readPositions_[readerNumber].empty_;
        }
        // }}}

        void DefaultCircularBuffer::PostAdditionAction( ReadThresholdEventListener* pListener )
        // {{{
        {
            TRACE("Listener has been added To DCB\n");
            /*
            for (unsigned readerNumber=0; readerNumber < numberOfReaders_; ++readerNumber)
            {
                if ( GetReadThreshold(readerNumber) )
                {
                    if ( IsAboveReadThreshold(readerNumber) )
                    {
                        NotifierWithParameter<ReadThresholdEventListener, unsigned>::NotifyListeners(readerNumber);
                        TRACE("$CReader %u above threshold after addition $n\n", readerNumber );
                    }
                }
            }
            */
        }
        // }}}
        
        
        /**
         * @param readerNumber An unsigned specifying the number of the
         *  reader whose read pointer is to be advanced
         * @return An unsigned int - the number of bytes used in the
         *  circular buffer.
         */

        unsigned int DefaultCircularBuffer::BytesUsed(unsigned int readerNumber) const
        // {{{
        {
            unsigned int totalLength;
            if (readPositions_[readerNumber].empty_)
            {
                totalLength = 0;
            }
            else
            {
                CircularBufferSpaceInfo readSpaceInfo = GetReadSpaceInfo(readPositions_[readerNumber].position_);
                totalLength = (readSpaceInfo.GetFirstAvailableSection().GetLength() +
                              readSpaceInfo.GetSecondAvailableSection().GetLength());
            }
            return totalLength;
        }
        // }}}

        std::pair<bool, unsigned> DefaultCircularBuffer::IsOldestReader(unsigned readerNumber) const
        // {{{
        {
            // Determine the number of readers with the oldest position
            bool isOldestReader = false;
            int numberOfOldestReaders = 0;
            CircularBufferPosition* oldestReadPosition = &readPositions_[0].position_;
            for (unsigned i=0; i < numberOfReaders_; ++i)
            {
                if (readPositions_[i].position_ == *oldestReadPosition)
                {
                    if (readerNumber == i)
                    {
                        isOldestReader = true;
                    }
                    ++numberOfOldestReaders;
                }
                else if (readPositions_[i].position_ < *oldestReadPosition)
                {
                    numberOfOldestReaders = 1;
                    isOldestReader = (readerNumber == i);
                    oldestReadPosition = &readPositions_[i].position_;
                }
            }

            return std::make_pair(isOldestReader, numberOfOldestReaders);
        }
        // }}}

        void DefaultCircularBuffer::NotifyReaders()
        // {{{
        {
            // Notify all readers blocked awaiting a buffer that one may now be
            //  available
//TRACE("$WNotifyReaders$n");
            for (unsigned i = 0; i < readWaitCount_; ++i)
            {
//TRACE("$W.$n");
                readSemaphore_.give();
            }
//TRACE("\n");
            readWaitCount_ = 0;
        }
        // }}}

        void DefaultCircularBuffer::NotifyWriters()
        // {{{
        {
//TRACE("$WNotifyWriters$n\n");
            // Notify all writers blocked awaiting a buffer that one may now be
            //  available
            for (unsigned i = 0; i < writeWaitCount_; ++i)
            {
                writeSemaphore_.give();
            }
            writeWaitCount_ = 0;
        }
        // }}}

        bool DefaultCircularBuffer::RegisterResidentClass(
            CircularBufferResidentObject::Id& id,
            ObjectPool& pool)
        // {{{
        {
            syslib::Lock lock(mutex_);

            if (knownObjectTypes_.size() == knownObjectTypes_.max_size())
            {
                knownObjectTypes_.expand(knownObjectTypes_.max_size() * 2);
            }

            knownObjectTypes_.insert(
                std::lower_bound(
                    knownObjectTypes_.begin(),
                    knownObjectTypes_.end(),
                    nextFreeId_,
                    SequenceOfPairsCompare<KNOWN_TYPES_CONTAINTER_TYPE::value_type>()),
                std::make_pair(nextFreeId_, &pool));

            id = nextFreeId_++;

            return true;
        }
        // }}}

        void DefaultCircularBuffer::ReleaseFromRead(
            unsigned readerNumber,
            CircularBufferResidentObject* object,
            bool pending)
        // {{{
        {
#ifdef INUSE_TRACKING
            if (!object)
            {
                TRACE("$RDefaultCircularBuffer::ReleaseFromRead() Object pointer is null\n");
                return;
            }
#endif

            syslib::Lock lock(mutex_);
//TRACE("$BRR entered: object=0x%lx, readerNumber=%u$n\n", reinterpret_cast<unsigned long>(object), readerNumber);

#ifdef INUSE_TRACKING
            // Catch trying to release object when none are in-use
            if (!readerInUse_[readerNumber]->size())
            {
                TRACE("$RDefaultCircularBuffer::ReleaseFromRead() Attempt to release when no buffers in-use$n\n");
                return;
            }

            // Check that the object being release is the object in the current
            // in-use set that was acquired earliest
            if (*readerInUse_[readerNumber]->begin() != object)
            {
                TRACE("$RDefaultCircularBuffer::ReleaseFromRead() 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
            readerInUse_[readerNumber]->erase(readerInUse_[readerNumber]->begin());
#endif

            // There will be one less buffer currently acquired for read by the
            // specified reader
            --outstandingReadAcquisitions_[readerNumber];

            if (pending)
            {
                // Return the object to its pool, but leave the circular buffer
                // resident data untouched
                GetPool(*object)->Free(object);

                // The empty status for the read is set when a buffer is acquired
                // for read, so we must ensure that it is reset here as a buffer
                // is definitely now available for the specified reader
                readPositions_[readerNumber].empty_ = false;
//TRACE("$BRR Pending true, so restting empty flag\n");
            }
            else
            {
                // Determine whether the specified reader is the oldest and how
                // many oldest readers there are