defaultcircularbuffer.cpp

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

//TRACE("$GDecoding was successful$n\n");
            // Move the read position past the end of the decoded object
            readPosition += decodeResult.GetBytesConsumed();

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

            // Record the position within the circular buffer after the
            // decoded object
            SetAfterPosition(*object, readPosition);

            // Determine whether any more objects are available for the reader
            do
            {
                // If the read position is now coincident with the write position
                // there must be no more objects available for the reader
                if (readPosition == writePosition_)
                {
                    readPositions_[readerNumber].empty_ = true;
                    break;
                }

                // Check whether there is a jump at the new read position
                id = *reinterpret_cast<CircularBufferResidentObject::Id*>(buffer_->GetBuffer() + readPosition.GetOffset());
                if (id >= MAX_SPECIAL_IDS)
                {
                    // No jump, so there must be more objects available for the
                    // reader, as the write position is in advance of the read
                    // position
                    break;
                }

                // Move past the jump ID
                readPosition += sizeof(CircularBufferResidentObject::Id);

                // Follow the jump to its target
                switch(id)
                {
                    case MOVE_TO_START_ID:
                        // Move read position to the start of the buffer
                        // Bear in mind that we may have wrapped if the
                        // MOVE_TO_START_ID was the last thing in the
                        // buffer.
                        if (readPosition.GetOffset() != 0)
                        {
                            readPosition += readPosition.GetMargin();
                        }
                        break;
                    case JUMP_ZERO_ID:
                        // Already positioned after JUMP_ID symbol, so nothing
                        // more to do
                        break;
                    case JUMP_ID:
                        {
                            // Read the length of the unused section of buffer
                            unsigned long jumpLength =
                                *reinterpret_cast<CircularBufferResidentObject::Id*>(
                                    buffer_->GetBuffer() +
                                    readPosition.GetOffset());

                            // Jump over the unused section of the buffer
                            readPosition += sizeof(unsigned long);
                            readPosition += jumpLength;
                        }
                        break;
                    default:
                        TRACE("$RUnknown special ID found\n");
                        return std::pair<bool, CircularBufferResidentObject*>(true, 0);
                }
            } while (true);

            // Record the pool that the decoded object came from
            SetPool(*object, readPool);

            // If a read limit is in effect
            if (readLimitInfo_[readerNumber].first)
            {
                // Increment the count of objects read since the limit was set
                ++limitCount_[readerNumber];
            }

            // Increment the count of the number of objects currently acquired
            // for read for this reader
            ++outstandingReadAcquisitions_[readerNumber];

#ifdef INUSE_TRACKING
            // Ensure the in-use container is large enough
            if ((readerInUse_[readerNumber]->size() + 1) == readerInUse_[readerNumber]->max_size())
            {
                readerInUse_[readerNumber]->expand(readerInUse_[readerNumber]->max_size() * 2);
            }

            // Remember the successful acquistion, by storing the prototype
            // object pointer
            readerInUse_[readerNumber]->push_back(object);
#endif

//TRACE("$GAR leaving: object=0x%lx, readerNumber = %u$n\n", reinterpret_cast<unsigned long>(object), readerNumber);
            return std::pair<bool, CircularBufferResidentObject*>(false, object);
        }
        // }}}

        bool DefaultCircularBuffer::AcquireForWrite(
            CircularBufferResidentObject& object,
            CircularBufferResidentObject* previousObject,
            bool blockIfInsufficientSpace)
        // {{{
        {
            syslib::Lock lock(mutex_);
//TRACE("$YAW entered: object=0x%lx, previousObject = 0x%lx$n\n", reinterpret_cast<unsigned long>(&object), reinterpret_cast<unsigned long>(previousObject));

            bool status = true;

            // If a previous object is provided, use the position after its end
            // within the buffer as the position into which to attempt to encode
            // the object
            CircularBufferPosition writePosition =
                previousObject ? GetAfterPosition(*previousObject): writePosition_;

//TRACE("$YAW writePosition = [%lu:%lu:%lu]$n\n", writePosition.GetLoopCount(), writePosition.GetOffset(), writePosition.GetMargin());
            // Try to associate the object with a range of the circular buffer
            CircularBufferSpaceInfo writeSpaceInfo = GetWriteSpaceInfo(writePosition);
//TRACE("$YAW writeSpaceInfo = [0x%lx:%lu][0x%lx:%lu]$n\n", writeSpaceInfo.GetFirstAvailableSection().GetStartAddress(), writeSpaceInfo.GetFirstAvailableSection().GetLength(), writeSpaceInfo.GetSecondAvailableSection().GetStartAddress(), writeSpaceInfo.GetSecondAvailableSection().GetLength());
            CircularBufferResidentObject::EncodeResult associationResult = object.AssociateWithBuffer(writeSpaceInfo);
//TRACE("$YAW associationResult = %d$n\n", (int)associationResult.GetSuccessful());
//TRACE("$W[%lu:%lu]->%s$n\n", writeSpaceInfo.GetFirstAvailableSection().GetLength(), writeSpaceInfo.GetSecondAvailableSection().GetLength(), associationResult.GetSuccessful() ? "OK" : "Too small");
            while (!associationResult.GetSuccessful())
            {
                // No, so is blocking to await an object allowed?
                if (!blockIfInsufficientSpace || doNotBlock_)
                {
                    // No
                    status = false;
                    break;
                }

                // Should block, so add this thread to the list of waiters for
                // the space to become available for writing
//TRACE("$YBlocking...$n\n");
                ++writeWaitCount_;
                lock.Give();
                writeSemaphore_.take();
                lock.Take();
//TRACE("$YAwoken$n\n");

                // If attempting to acquire for write after a previously acquired
                // object, then are guarenteed that no zeroising of the write
                // and read pointers can have occured
                if (!previousObject)
                {
                    // Zeroising of write and read pointers could have occured,
                    // so re-obtain the write pointer position
                    writePosition = writePosition_;
                }

                // Re-query for the space available for writing
                writeSpaceInfo = GetWriteSpaceInfo(writePosition);
                associationResult = object.AssociateWithBuffer(writeSpaceInfo);
//TRACE("$YAW Awoken from wait, associationResult=%d$n\n", (int)associationResult.GetSuccessful());
            }

            if (status)
            {
                // Association was successful. Did the object decide not to use
                // the space remaining upto the end of the buffer?
                if (associationResult.GetSkippedFirstSection())
                {
                    // Write move-to-start ID in case the object is discarded at
                    // the commit stage, so that it can be determined whether
                    // the associated buffer space included the margin or not
                    *reinterpret_cast<CircularBufferResidentObject::Id*>(
                        buffer_->GetBuffer() + writePosition.GetOffset()) = MOVE_TO_START_ID;
                    writePosition += writePosition.GetMargin();
//TRACE("$MAW writePosition = [%lu:%lu:%lu]$n\n", writePosition.GetLoopCount(), writePosition.GetOffset(), writePosition.GetMargin());
                }
                else
                {
                    // Write an invalid ID in case the object is discarded at
                    // the commit stage, so that it can be determined whether
                    // the associated buffer space included the margin or not.
                    *reinterpret_cast<CircularBufferResidentObject::Id*>(
                        buffer_->GetBuffer() + writePosition.GetOffset()) = MAX_SPECIAL_IDS;
                }

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

                // Ensure the write position finishes on a quad boundary
                unsigned remainder = writePosition.GetOffset() & 0x3UL;
                if (remainder)
                {
                    writePosition += (4 - remainder);
                }
//TRACE("$CAW EndwritePosition = [%lu:%lu:%lu]$n\n", writePosition.GetLoopCount(), writePosition.GetOffset(), writePosition.GetMargin());

                // Have no interest in the pool from which objects for encode
                // originate
                SetPool(object, 0);

                // Record the position within the circular buffer after the
                // associated object
                SetAfterPosition(object, writePosition);

                // Record the position after the most recently acquired object
                mostRecentAcquireAfterPosition_ = writePosition;

                // Determine the reader with the oldest position
                CircularBufferPosition* oldestReadPosition = &readPositions_[0].position_;
                for (unsigned i=1; i < numberOfReaders_; ++i)
                {
                    if (readPositions_[i].position_ < *oldestReadPosition)
                    {
                        oldestReadPosition = &readPositions_[i].position_;
                    }
                }

                // If have now used up all space available for writing
                if (writePosition.GetOffset() == oldestReadPosition->GetOffset())
                {
//TRACE("YAW Setting writeEmpty_ true\n");
                    writeEmpty_ = true;
                }
            }

            if (status)
            {
                // Successfully acquired an object for write, so increment the
                // count of such objects
                ++outstandingWriteAcquisitions_;

#ifdef INUSE_TRACKING
                // Ensure the in-use container is large enough
                if ((writeInUse_.size() + 1) == writeInUse_.max_size())
                {
                    writeInUse_.expand(writeInUse_.max_size() * 2);
                }

                // Remember the successful acquistion, by storing the prototype
                // object pointer
                writeInUse_.push_back(&object);
#endif
            }

//TRACE("$YAW leaving: object=0x%lx, status = %d$n\n", reinterpret_cast<unsigned long>(&object), (int)status);
            return status;
        }
        // }}}

        void DefaultCircularBuffer::CommonReleaseFromRead(bool wasBelowWriteThreshold)
        // {{{
        {
            // Determine the space available to the writer after the reader has
            // released buffers
            bool isBelowWriteThreshold = !IsAboveWriteThreshold();

            // If have crossed the available for write threshold, notify
            // any listeners
            if ( GetWriteThreshold() || (wasBelowWriteThreshold && !isBelowWriteThreshold))
            {
//TRACE("$WNotifying write threshold listeners$n\n");
                Notifier<WriteThresholdEventListener>::NotifyListeners();
            }

            // Determine whether there is any data available for read, i.e.
            // whether the circular buffer is empty
            bool anyReadBuffersAvailable = false;
            bool outstandingReadAcquisitions = false;
            for (unsigned readerNumber=0; readerNumber < numberOfReaders_; ++readerNumber)
            {
                if (!readPositions_[readerNumber].empty_)
                {
                    anyReadBuffersAvailable = true;
                    break;
                }
                if (outstandingReadAcquisitions_[readerNumber])
                {
                    outstandingReadAcquisitions = true;
                }
            }

            // If there is no data in the buffer and there are no buffers
            // currently acquired for writing or reading
            if (!anyReadBuffersAvailable && !outstandingWriteAcquisitions_ && !outstandingReadAcquisitions)
            {
//TRACE("$R*******************ZEROISING******************$n\n");
                // Reposition the write and all read pointers to the start of
                // the buffer, to maximise the space available in a single chunk
                writePosition_.Zeroise(buffer_->GetRequestedSize() -1);
                for (unsigned i=0; i < numberOfReaders_; i++)
                {
                    readPositions_[i].position_.Zeroise(buffer_->GetRequestedSize() - 1);
                }
            }

            // Has the buffer just changed from having some data available for