defaultcircularbuffer.cpp

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

            // read to having no data available for read?
            if (lastAnyReadBuffersAvailable_ && !anyReadBuffersAvailable)
            {
                // Yes, so notify any read empty listeners
                lastAnyReadBuffersAvailable_ = false;
                Notifier<EmptyEventListener>::NotifyListeners();
            }

            // Notify all writers blocked awaiting a buffer that one may now
            // be available
            NotifyWriters();
        }
        // }}}

        void DefaultCircularBuffer::DoNotify(EmptyEventListener* listener)
        // {{{
        {
            //TRACE("$YDCB::DoNotify - EmptyEvent\n");
            listener->EmptyEventOccured();
        }
        // }}}

        void DefaultCircularBuffer::DoNotify(
            ReadThresholdEventListener* listener,
            const unsigned& readerNumber)
        // {{{
        {
            //TRACE("$YDCB::DoNotify - ReadThreshold for %u\n", readerNumber);
            listener->ReadThresholdEventOccured(readerNumber);
        }
        // }}}

        void DefaultCircularBuffer::DoNotify(WriteThresholdEventListener* listener)
        // {{{
        {
            //TRACE("$YDCB::DoNotify - WriteThreshold\n");
            listener->WriteThresholdEventOccured();
        }
        // }}}

        ObjectPool* DefaultCircularBuffer::FindPoolForType(CircularBufferResidentObject::Id id)
        // {{{
        {
            std::pair<KNOWN_TYPES_CONTAINTER_TYPE::iterator, KNOWN_TYPES_CONTAINTER_TYPE::iterator> searchResult =
                std::equal_range(
                    knownObjectTypes_.begin(),
                    knownObjectTypes_.end(),
                    id,
                    SequenceOfPairsCompare<KNOWN_TYPES_CONTAINTER_TYPE::value_type>());

            return (searchResult.first == searchResult.second) ? 0 : searchResult.first->second;
        }
        // }}}

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

#ifdef INUSE_TRACKING
            bool anyIsUse = false;
            for (unsigned readerNumber = 0; readerNumber < numberOfReaders_; ++readerNumber)
            {
                if (readerInUse_[readerNumber]->size())
                {
                    TRACE("$RDefaultCircularBuffer::FlushAvailableForRead() Attempt to flush while %u objects in-use for reader %u$n\n",
                        readerInUse_[readerNumber]->size(), readerNumber);
                    anyIsUse = true;
                }
            }
            if (anyIsUse)
            {
                return false;
            }
#endif

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

            bool readPointerAdvanced = false;
            for (unsigned readerNumber = 0; readerNumber < numberOfReaders_; ++readerNumber)
            {
                if (outstandingReadAcquisitions_[readerNumber])
                {
                    TRACE(
                        "$RAttempt to flush when reader %u has %u outstanding %s for read$n\n",
                        readerNumber,
                        outstandingReadAcquisitions_[readerNumber],
                        (outstandingReadAcquisitions_[readerNumber] > 1) ? "acquisitions" : "acquisition");

                    ASSERT(false);
                }
                else if (!readPositions_[readerNumber].empty_)
                {
                    // Flush buffers available to the specified reader, by
                    // advancing the read pointer to the write pointer's position
                    readPositions_[readerNumber].position_ = writePosition_;
                    readPositions_[readerNumber].empty_ = true;
                    readPointerAdvanced = true;

                    // No objects available for read
                    numberOfObjects_[readerNumber] = 0;

                    // There were objects left to be read and there was a limit set
                    // so the reader may have been blocked on the limit, so unblock
                    // it as there are no longer any objects left
                    if (readLimitInfo_[readerNumber].first)
                    {
                        SetReadLimit(readerNumber, 0);
                    }
                }
            }

            if (readPointerAdvanced)
            {
                // Notify writers that space has been freed in the buffer and inform
                // any listeners of empty or threshold crossing events
                CommonReleaseFromRead(wasBelowWriteThreshold);
            }

            // Whole buffer must now be available for writing
//TRACE("DefaultCircularBuffer::FlushAvailableForRead() Setting writeEmpty_ false\n");
            writeEmpty_ = false;

            return readPointerAdvanced;
        }
        // }}}

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

#ifdef INUSE_TRACKING
            if (readerInUse_[readerNumber]->size())
            {
                TRACE("$RDefaultCircularBuffer::FlushAvailableForRead(unsigned) Attempt to flush while %u objects in-use for reader %u$n\n",
                    readerInUse_[readerNumber]->size(), readerNumber);
                return false;
            }
#endif

//TRACE("DefaultCircularBuffer::FlushAvailableForRead(unsigned) Entered for reader %u\n", readerNumber);
            // Determine whether the specified reader is the oldest and how
            // many oldest readers there are
            std::pair<bool, unsigned> oldestResult = IsOldestReader(readerNumber);

            bool readPointerAdvanced = false;
            if (outstandingReadAcquisitions_[readerNumber])
            {
                TRACE(
                    "$RAttempt to flush when reader %u has %u outstanding %s for read$n\n",
                    readerNumber,
                    outstandingReadAcquisitions_[readerNumber],
                    (outstandingReadAcquisitions_[readerNumber] > 1) ? "acquisitions" : "acquisition");

                ASSERT(false);
            }
            else if (!readPositions_[readerNumber].empty_)
            {
                // Determine the space available to the writer before flushing
                // the readers
                bool wasBelowWriteThreshold = !GetWriteThreshold() ? true : !IsAboveWriteThreshold();

                // Flush buffers available to the specified reader, by
                // advancing the read pointer to the write pointer's position
                readPositions_[readerNumber].position_ = writePosition_;
                readPositions_[readerNumber].empty_ = true;
                readPointerAdvanced = true;

                // No objects available for read
                numberOfObjects_[readerNumber] = 0;

                // Notify writers that space has been freed in the buffer and
                // inform any listeners of empty or threshold crossing events
                CommonReleaseFromRead(wasBelowWriteThreshold);

                // There were objects left to be read and there was a limit set
                // so the reader may have been blocked on the limit, so unblock
                // it as there are no longer any objects left
                if (readLimitInfo_[readerNumber].first)
                {
//TRACE("DefaultCircularBuffer::FlushAvailableForRead(unsigned) Zeroing read limit for reader %u\n", readerNumber);
                    SetReadLimit(readerNumber, 0);
                }
            }

            // If the sole oldest reader was flushed
            if (readPointerAdvanced && oldestResult.first && (oldestResult.second == 1))
            {
                // Must be space available for write
//TRACE("DefaultCircularBuffer::FlushAvailableForRead(unsigned) Setting writeEmpty_ false\n");
                writeEmpty_ = false;
            }

            return readPointerAdvanced;
        }
        // }}}

        void DefaultCircularBuffer::ForceUnblockAll()
        // {{{
        {
            syslib::Lock lock(mutex_);
//TRACE("$RForceUnblockAll()$n\n");
            // Don't allow readers/or writer to block in future
            doNotBlock_ = true;

            // Deal with blocking readers
            SetReaderBlockState(false);

            // Unblock any writers awaiting buffers
            NotifyWriters();
        }
        // }}}

        unsigned long DefaultCircularBuffer::GetDifference(
            const CircularBufferPosition& earlierPosition,
            const CircularBufferPosition& laterPosition)
        // {{{
        {
//TRACE("$BearlierPosition=[%lu:%lu:%lu], laterPosition=[%lu:%lu:%lu]$n\n", earlierPosition.GetLoopCount(), earlierPosition.GetOffset(), earlierPosition.GetMargin(), laterPosition.GetLoopCount(), laterPosition.GetOffset(), laterPosition.GetMargin());
            unsigned long unusedSpace = 0;
            if (earlierPosition < laterPosition)
            {
//TRACE("$B1$n\n");
                if (earlierPosition.GetOffset() < laterPosition.GetOffset())
                {
//TRACE("$B2$n\n");
                    unusedSpace = laterPosition.GetOffset() - earlierPosition.GetOffset();
                }
                else
                {
//TRACE("$B3$n\n");
                    unusedSpace = earlierPosition.GetMargin() + laterPosition.GetOffset();
                }
            }
//TRACE("$BunusedSpace=%lu$n\n", unusedSpace);
            return unusedSpace;
        }
        // }}}

        CircularBufferSpaceInfo DefaultCircularBuffer::GetReadSpaceInfo(
            const CircularBufferPosition& readPosition) const
        // {{{
        {
            syslib::Lock lock(mutex_);

            // Will assume here that if the write and read positions are the
            // same then all the buffer is available for read. The test for
            // whether in fact no data is available for read will have been made
            // prior to calling this method

            //TRACE("$Y%d %d -$n", writePosition_.GetOffset(), readPosition.GetOffset());

            if (writePosition_.GetOffset() > readPosition.GetOffset())
            {
                return CircularBufferSpaceInfo(
                    BufferSectionInfo(
                        reinterpret_cast<unsigned long>(buffer_->GetBuffer()) +
                            readPosition.GetOffset(),
                        writePosition_.GetOffset() - readPosition.GetOffset()),
                    BufferSectionInfo(0, 0));
            }
            else
            {
                return CircularBufferSpaceInfo(
                    BufferSectionInfo(
                        reinterpret_cast<unsigned long>(buffer_->GetBuffer()) +
                            readPosition.GetOffset(),
                        readPosition.GetMargin()),
                    BufferSectionInfo(
                        reinterpret_cast<unsigned long>(buffer_->GetBuffer()),
                        writePosition_.GetOffset()));
            }
        }
        // }}}

        unsigned long DefaultCircularBuffer::GetReadThreshold(unsigned readerNumber) const
        // {{{
        {
            syslib::Lock lock(mutex_);
            return mpReadThresholds_[readerNumber].GetThreshold();
        }
        // }}}

        unsigned long DefaultCircularBuffer::GetWriteThreshold() const
        // {{{
        {
            syslib::Lock lock(mutex_);
            return writeThreshold_.GetThreshold();
        }
        // }}}

        unsigned long DefaultCircularBuffer::GetSpaceAvailableForWrite() const
        // {{{
        {
            syslib::Lock lock(mutex_);

            CircularBufferSpaceInfo circularBufferSpaceInfo = GetWriteSpaceInfo(writePosition_);