src/java/org/apache/cassandra/db/memtable/AbstractMemtableWithCommitlog.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.cassandra.db.memtable;

import java.util.concurrent.atomic.AtomicReference;

import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.concurrent.OpOrder;

/**
 * Memtable that uses a commit log for persistence. Provides methods of tracking the commit log positions covered by
 * it and safely switching between memtables.
 */
public abstract class AbstractMemtableWithCommitlog extends AbstractMemtable
{
    // The approximate lower bound by this memtable; must be <= commitLogLowerBound once our predecessor
    // has been finalised, and this is enforced in the ColumnFamilyStore.setCommitLogUpperBound
    private final CommitLogPosition approximateCommitLogLowerBound = CommitLog.instance.getCurrentPosition();
    // the precise lower bound of CommitLogPosition owned by this memtable; equal to its predecessor's commitLogUpperBound
    private final AtomicReference<CommitLogPosition> commitLogLowerBound;
    // the write barrier for directing writes to this memtable or the next during a switch
    private volatile OpOrder.Barrier writeBarrier;
    // the precise upper bound of CommitLogPosition owned by this memtable
    private volatile AtomicReference<CommitLogPosition> commitLogUpperBound;

    public AbstractMemtableWithCommitlog(TableMetadataRef metadataRef, AtomicReference<CommitLogPosition> commitLogLowerBound)
    {
        super(metadataRef);
        this.commitLogLowerBound = commitLogLowerBound;
    }

    public CommitLogPosition getApproximateCommitLogLowerBound()
    {
        return approximateCommitLogLowerBound;
    }

    public void switchOut(OpOrder.Barrier writeBarrier, AtomicReference<CommitLogPosition> commitLogUpperBound)
    {
        // This can prepare the memtable data for deletion; it will still be used while the flush is proceeding.
        // A setDiscarded call will follow.
        assert this.writeBarrier == null;
        this.commitLogUpperBound = commitLogUpperBound;
        this.writeBarrier = writeBarrier;
    }

    public void discard()
    {
        assert writeBarrier != null : "Memtable must be switched out before being discarded.";
    }

    // decide if this memtable should take the write, or if it should go to the next memtable
    @Override
    public boolean accepts(OpOrder.Group opGroup, CommitLogPosition commitLogPosition)
    {
        // if the barrier hasn't been set yet, then this memtable is still the newest and is taking ALL writes.
        OpOrder.Barrier barrier = this.writeBarrier;
        if (barrier == null)
            return true;
        // Note that if this races with taking the barrier the opGroup and commit log position we were given must
        // necessarily be before the barrier and any LastCommitLogPosition is set, thus this function will return true
        // and no update to commitLogUpperBound is necessary.

        // If the barrier has been set and issued, but is in the past, we are definitely destined for a future memtable.
        // Because we issue the barrier after taking LastCommitLogPosition and mutations take their position after
        // taking the opGroup, this condition also ensures the given commit log position is greater than the chosen
        // upper bound.
        if (!barrier.isAfter(opGroup))
            return false;
        // We are in the segment of time between the barrier is constructed (and the memtable is switched out)
        // and the barrier is issued.
        // if we aren't durable we are directed only by the barrier
        if (commitLogPosition == null)
            return true;
        while (true)
        {
            // If the CL boundary has been set, the mutation can be accepted depending on whether it falls before it.
            // However, if it has not been set, the old sstable must still accept writes but we must also ensure that
            // their positions are accounted for in the boundary (as there may be a delay between taking the log
            // position for the boundary and setting it where a mutation sneaks in).
            // Thus, if the boundary hasn't been finalised yet, we simply update it to the max of its current value and
            // ours; this permits us to coordinate a safe boundary, as the boundary choice is made atomically wrt our
            // max() maintenance, so an operation cannot sneak into the past.
            CommitLogPosition currentLast = commitLogUpperBound.get();
            if (currentLast instanceof LastCommitLogPosition)
                return currentLast.compareTo(commitLogPosition) >= 0;
            if (currentLast != null && currentLast.compareTo(commitLogPosition) >= 0)
                return true;
            if (commitLogUpperBound.compareAndSet(currentLast, commitLogPosition))
                return true;
        }
    }

    public CommitLogPosition getCommitLogLowerBound()
    {
        return commitLogLowerBound.get();
    }

    public LastCommitLogPosition getFinalCommitLogUpperBound()
    {
        assert commitLogUpperBound != null : "Commit log upper bound should be set before flushing";
        assert commitLogUpperBound.get() instanceof LastCommitLogPosition : "Commit log upper bound has not been sealed yet? " + commitLogUpperBound.get();
        return (LastCommitLogPosition) commitLogUpperBound.get();
    }

    public boolean mayContainDataBefore(CommitLogPosition position)
    {
        return approximateCommitLogLowerBound.compareTo(position) < 0;
    }
}