private void advanceAllocatingFrom ( CommitLogSegment old ) { while ( true ) { CommitLogSegment next ; synchronized ( this ) { // do this in a critical section so we can atomically remove from availableSegments and add to allocatingFrom/activeSegments // see https://issues.apache.org/jira/browse/CASSANDRA-6557?focusedCommentId=13874432&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-13874432 if ( allocatingFrom != old ) return ; next = availableSegments . poll ( ) ; if ( next != null ) { allocatingFrom = next ; activeSegments . add ( next ) ; } } if ( next != null ) { if ( old != null ) { // Now we can run the user defined command just after switching to the new commit log. // (Do this here instead of in the recycle call so we can get a head start on the archive.) CommitLog . instance . archiver . maybeArchive ( old ) ; // ensure we don't continue to use the old file; not strictly necessary, but cleaner to enforce it old . discardUnusedTail ( ) ; } // request that the CL be synced out-of-band, as we've finished a segment CommitLog . instance . requestExtraSync ( ) ; return ; } // no more segments, so register to receive a signal when not empty WaitQueue . Signal signal = hasAvailableSegments . register ( CommitLog . instance . metrics . waitingOnSegmentAllocation . time ( ) ) ; // trigger the management thread; this must occur after registering // the signal to ensure we are woken by any new segment creation wakeManager ( ) ; // check if the queue has already been added to before waiting on the signal, to catch modifications // that happened prior to registering the signal; *then* check to see if we've been beaten to making the change if ( ! availableSegments . isEmpty ( ) || allocatingFrom != old ) { signal . cancel ( ) ; // if we've been beaten, just stop immediately if ( allocatingFrom != old ) return ; // otherwise try again, as there should be an available segment continue ; } // can only reach here if the queue hasn't been inserted into // before we registered the signal, as we only remove items from the queue // after updating allocatingFrom. Can safely block until we are signalled // by the allocator that new segments have been published signal . awaitUninterruptibly ( ) ; } }
Fetches a new segment from the queue creating a new one if necessary and activates it