import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public void allocate(long size, OpOrder.Group opGroup)
{
    assert size >= 0;

    while (true)
    {
        if (parent.tryAllocate(size))
        {
            acquired(size);
            return;
        }
        WaitQueue.Signal signal = opGroup.isBlockingSignal(parent.hasRoom().register(parent.blockedTimerContext()));
        boolean allocated = parent.tryAllocate(size);
        if (allocated || opGroup.isBlocking())
        {
            signal.cancel();
            if (allocated) // if we allocated, take ownership
                acquired(size);
            else // otherwise we're blocking so we're permitted to overshoot our constraints, to just allocate without blocking
                allocated(size);
            return;
        }
        else
            signal.awaitUninterruptibly();
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public long allocate(int size, OpOrder.Group opGroup)
{
    assert size >= 0;
    offHeap().allocate(size, opGroup);
    // satisfy large allocations directly from JVM since they don't cause fragmentation
    // as badly, and fill up our regions quickly
    if (size > MAX_CLONED_SIZE)
        return allocateOversize(size);

    while (true)
    {
        Region region = currentRegion.get();
        long peer;
        if (region != null && (peer = region.allocate(size)) > 0)
            return peer;

        trySwapRegion(region, size);
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public Future<Pair<RowCacheKey, IRowCacheEntry>> deserialize(DataInputPlus in, final ColumnFamilyStore cfs) throws IOException
{
    //Keyspace and CF name are deserialized by AutoSaving cache and used to fetch the CFS provided as a
    //parameter so they aren't deserialized here, even though they are serialized by this serializer
    final ByteBuffer buffer = ByteBufferUtil.readWithLength(in);
    final int rowsToCache = cfs.metadata.params.caching.rowsPerPartitionToCache();
    if (cfs == null  || !cfs.isRowCacheEnabled())
        return null;
    assert(!cfs.isIndex());//Shouldn't have row cache entries for indexes

    return StageManager.getStage(Stage.READ).submit(new Callable<Pair<RowCacheKey, IRowCacheEntry>>()
    {
        public Pair<RowCacheKey, IRowCacheEntry> call() throws Exception
        {
            DecoratedKey key = cfs.decorateKey(buffer);
            int nowInSec = FBUtilities.nowInSeconds();
            try (OpOrder.Group op = cfs.readOrdering.start(); UnfilteredRowIterator iter = SinglePartitionReadCommand.fullPartitionRead(cfs.metadata, nowInSec, key).queryMemtableAndDisk(cfs, op))
            {
                CachedPartition toCache = CachedBTreePartition.create(DataLimits.cqlLimits(rowsToCache).filter(iter, nowInSec), nowInSec);
                return Pair.create(new RowCacheKey(cfs.metadata.ksAndCFName, key), (IRowCacheEntry)toCache);
            }
        }
    });
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
private static int estimateRowOverhead(final int count)
{
    // calculate row overhead
    try (final OpOrder.Group group = new OpOrder().start())
    {
        int rowOverhead;
        MemtableAllocator allocator = MEMORY_POOL.newAllocator();
        ConcurrentNavigableMap<PartitionPosition, Object> partitions = new ConcurrentSkipListMap<>();
        final Object val = new Object();
        for (int i = 0 ; i < count ; i++)
            partitions.put(allocator.clone(new BufferDecoratedKey(new LongToken(i), ByteBufferUtil.EMPTY_BYTE_BUFFER), group), val);
        double avgSize = ObjectSizes.measureDeep(partitions) / (double) count;
        rowOverhead = (int) ((avgSize - Math.floor(avgSize)) < 0.05 ? Math.floor(avgSize) : Math.ceil(avgSize));
        rowOverhead -= ObjectSizes.measureDeep(new LongToken(0));
        rowOverhead += AtomicBTreePartition.EMPTY_SIZE;
        allocator.setDiscarding();
        allocator.setDiscarded();
        return rowOverhead;
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public long allocate(int size, OpOrder.Group opGroup)
{
    assert size >= 0;
    offHeap().allocate(size, opGroup);
    // satisfy large allocations directly from JVM since they don't cause fragmentation
    // as badly, and fill up our regions quickly
    if (size > MAX_CLONED_SIZE)
        return allocateOversize(size, opGroup);

    while (true)
    {
        Region region = currentRegion.get();
        long peer;
        if (region != null && (peer = region.allocate(size)) > 0)
            return peer;

        trySwapRegion(region, size);
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
/**
 * Allocate space in this buffer for the provided mutation, and return the allocated Allocation object.
 * Returns null if there is not enough space in this segment, and a new segment is needed.
 */
Allocation allocate(Mutation mutation, int size)
{
    final OpOrder.Group opGroup = appendOrder.start();
    try
    {
        int position = allocate(size);
        if (position < 0)
        {
            opGroup.close();
            return null;
        }
        markDirty(mutation, position);
        return new Allocation(this, opGroup, position, (ByteBuffer) buffer.duplicate().position(position).limit(position + size));
    }
    catch (Throwable t)
    {
        opGroup.close();
        throw t;
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
private static int estimateRowOverhead(final int count)
{
    // calculate row overhead
    final OpOrder.Group group = new OpOrder().start();
    int rowOverhead;
    MemtableAllocator allocator = MEMORY_POOL.newAllocator();
    ConcurrentNavigableMap<RowPosition, Object> rows = new ConcurrentSkipListMap<>();
    final Object val = new Object();
    for (int i = 0 ; i < count ; i++)
        rows.put(allocator.clone(new BufferDecoratedKey(new LongToken((long) i), ByteBufferUtil.EMPTY_BYTE_BUFFER), group), val);
    double avgSize = ObjectSizes.measureDeep(rows) / (double) count;
    rowOverhead = (int) ((avgSize - Math.floor(avgSize)) < 0.05 ? Math.floor(avgSize) : Math.ceil(avgSize));
    rowOverhead -= ObjectSizes.measureDeep(new LongToken((long) 0));
    rowOverhead += AtomicBTreeColumns.EMPTY_SIZE;
    allocator.setDiscarding();
    allocator.setDiscarded();
    return rowOverhead;
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
/**
 * @param key row to index
 * @param cfs ColumnFamily to index row in
 * @param idxNames columns to index, in comparator order
 */
public static void indexRow(DecoratedKey key, ColumnFamilyStore cfs, Set<String> idxNames)
{
    if (logger.isDebugEnabled())
        logger.debug("Indexing row {} ", cfs.metadata.getKeyValidator().getString(key.getKey()));

    try (OpOrder.Group opGroup = cfs.keyspace.writeOrder.start())
    {
        Set<SecondaryIndex> indexes = cfs.indexManager.getIndexesByNames(idxNames);

        Iterator<ColumnFamily> pager = QueryPagers.pageRowLocally(cfs, key.getKey(), DEFAULT_PAGE_SIZE);
        while (pager.hasNext())
        {
            ColumnFamily cf = pager.next();
            ColumnFamily cf2 = cf.cloneMeShallow();
            for (Cell cell : cf)
            {
                if (cfs.indexManager.indexes(cell.name(), indexes))
                    cf2.addColumn(cell);
            }
            cfs.indexManager.indexRow(key.getKey(), cf2, opGroup);
        }
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
/**
 * get the Memtable that the ordered writeOp should be directed to
 */
public Memtable getMemtableFor(OpOrder.Group opGroup, ReplayPosition replayPosition)
{
    // since any new memtables appended to the list after we fetch it will be for operations started
    // after us, we can safely assume that we will always find the memtable that 'accepts' us;
    // if the barrier for any memtable is set whilst we are reading the list, it must accept us.

    // there may be multiple memtables in the list that would 'accept' us, however we only ever choose
    // the oldest such memtable, as accepts() only prevents us falling behind (i.e. ensures we don't
    // assign operations to a memtable that was retired/queued before we started)
    for (Memtable memtable : view.get().liveMemtables)
    {
        if (memtable.accepts(opGroup, replayPosition))
            return memtable;
    }
    throw new AssertionError(view.get().liveMemtables.toString());
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public void insert(ByteBuffer rowKey, Cell cell, OpOrder.Group opGroup)
{
    DecoratedKey valueKey = getIndexKeyFor(getIndexedValue(rowKey, cell));
    ColumnFamily cfi = ArrayBackedSortedColumns.factory.create(indexCfs.metadata, false, 1);
    CellName name = makeIndexColumnName(rowKey, cell);
    if (cell instanceof ExpiringCell)
    {
        ExpiringCell ec = (ExpiringCell) cell;
        cfi.addColumn(new BufferExpiringCell(name, ByteBufferUtil.EMPTY_BYTE_BUFFER, ec.timestamp(), ec.getTimeToLive(), ec.getLocalDeletionTime()));
    }
    else
    {
        cfi.addColumn(new BufferCell(name, ByteBufferUtil.EMPTY_BYTE_BUFFER, cell.timestamp()));
    }
    if (logger.isDebugEnabled())
        logger.debug("applying index row {} in {}", indexCfs.metadata.getKeyValidator().getString(valueKey.getKey()), cfi);

    indexCfs.apply(valueKey, cfi, SecondaryIndexManager.nullUpdater, opGroup, null);
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public void commit()
{
    if (row == null && partitionDelete == null)
        return;

    try (OpOrder.Group opGroup = Keyspace.writeOrder.start())
    {
        for (Index index : indexes)
        {
            Index.Indexer indexer = index.indexerFor(key, columns, nowInSec, opGroup, Type.CLEANUP);
            if (indexer == null)
                continue;

            indexer.begin();

            if (partitionDelete != null)
                indexer.partitionDelete(partitionDelete);

            if (row != null)
                indexer.removeRow(row);

            indexer.finish();
        }
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public Memtable getMemtableFor(OpOrder.Group opGroup, ReplayPosition replayPosition)
{
    // since any new memtables appended to the list after we fetch it will be for operations started
    // after us, we can safely assume that we will always find the memtable that 'accepts' us;
    // if the barrier for any memtable is set whilst we are reading the list, it must accept us.

    // there may be multiple memtables in the list that would 'accept' us, however we only ever choose
    // the oldest such memtable, as accepts() only prevents us falling behind (i.e. ensures we don't
    // assign operations to a memtable that was retired/queued before we started)
    for (Memtable memtable : view.get().liveMemtables)
    {
        if (memtable.accepts(opGroup, replayPosition))
            return memtable;
    }
    throw new AssertionError(view.get().liveMemtables.toString());
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public void run()
{
    while (true)
    {
        AtomicInteger c;
        try (OpOrder.Group opGroup = order.start())
        {
            if (null == (c = count.get(opGroup)))
            {
                count.putIfAbsent(opGroup, new AtomicInteger());
                c = count.get(opGroup);
            }
            c.incrementAndGet();
            State s = state;
            while (!s.accept(opGroup))
                s = s.replacement;
        }
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
/**
 * Insert/Update the column family for this key.
 * Caller is responsible for acquiring Keyspace.switchLock
 * param @ lock - lock that needs to be used.
 * param @ key - key for update/insert
 * param @ columnFamily - columnFamily changes
 */
public void apply(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup, ReplayPosition replayPosition)

{
    long start = System.nanoTime();
    Memtable mt = data.getMemtableFor(opGroup, replayPosition);
    try
    {
        long timeDelta = mt.put(update, indexer, opGroup);
        DecoratedKey key = update.partitionKey();
        maybeUpdateRowCache(key);
        metric.samplers.get(Sampler.WRITES).addSample(key.getKey(), key.hashCode(), 1);
        metric.writeLatency.addNano(System.nanoTime() - start);
        if(timeDelta < Long.MAX_VALUE)
            metric.colUpdateTimeDeltaHistogram.update(timeDelta);
    }
    catch (RuntimeException e)
    {
        throw new RuntimeException(e.getMessage()
                                                 + " for ks: "
                                                 + keyspace.getName() + ", table: " + name, e);
    }

}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
/**
 * Allocate space in this buffer for the provided mutation, and return the allocated Allocation object.
 * Returns null if there is not enough space in this segment, and a new segment is needed.
 */
@SuppressWarnings("resource") //we pass the op order around
Allocation allocate(Mutation mutation, int size)
{
    final OpOrder.Group opGroup = appendOrder.start();
    try
    {
        int position = allocate(size);
        if (position < 0)
        {
            opGroup.close();
            return null;
        }
        markDirty(mutation, position);
        return new Allocation(this, opGroup, position, (ByteBuffer) buffer.duplicate().position(position).limit(position + size));
    }
    catch (Throwable t)
    {
        opGroup.close();
        throw t;
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public void adjust(long size, OpOrder.Group opGroup)
{
    if (size <= 0)
        released(-size);
    else
        allocate(size, opGroup);
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public ByteBuffer allocate(int size, OpOrder.Group opGroup)
{
    assert size >= 0;
    if (size == 0)
        return ByteBufferUtil.EMPTY_BYTE_BUFFER;

    (allocateOnHeapOnly ? onHeap() : offHeap()).allocate(size, opGroup);
    // satisfy large allocations directly from JVM since they don't cause fragmentation
    // as badly, and fill up our regions quickly
    if (size > MAX_CLONED_SIZE)
    {
        unslabbedSize.addAndGet(size);
        if (allocateOnHeapOnly)
            return ByteBuffer.allocate(size);
        Region region = new Region(ByteBuffer.allocateDirect(size));
        offHeapRegions.add(region);
        return region.allocate(size);
    }

    while (true)
    {
        Region region = getRegion();

        // Try to allocate from this region
        ByteBuffer cloned = region.allocate(size);
        if (cloned != null)
            return cloned;

        // not enough space!
        currentRegion.compareAndSet(region, null);
    }
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
private RowUpdater(AtomicBTreePartition updating, MemtableAllocator allocator, OpOrder.Group writeOp, UpdateTransaction indexer)
{
    this.updating = updating;
    this.allocator = allocator;
    this.writeOp = writeOp;
    this.indexer = indexer;
    this.nowInSec = FBUtilities.nowInSeconds();
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public NativeDecoratedKey(Token token, NativeAllocator allocator, OpOrder.Group writeOp, ByteBuffer key)
{
    super(token);
    assert key != null;
    assert key.order() == ByteOrder.BIG_ENDIAN;

    int size = key.remaining();
    this.peer = allocator.allocate(4 + size, writeOp);
    MemoryUtil.setInt(peer, size);
    MemoryUtil.setBytes(peer + 4, key);
}
 

import org.apache.cassandra.utils.concurrent.OpOrder; //导入依赖的package包/类
public NativeClustering(NativeAllocator allocator, OpOrder.Group writeOp, Clustering clustering)
{
    int count = clustering.size();
    int metadataSize = (count * 2) + 4;
    int dataSize = clustering.dataSize();
    int bitmapSize = ((count + 7) >>> 3);

    assert count < 64 << 10;
    assert dataSize < 64 << 10;

    peer = allocator.allocate(metadataSize + dataSize + bitmapSize, writeOp);
    long bitmapStart = peer + metadataSize;
    MemoryUtil.setShort(peer, (short) count);
    MemoryUtil.setShort(peer + (metadataSize - 2), (short) dataSize); // goes at the end of the other offsets

    MemoryUtil.setByte(bitmapStart, bitmapSize, (byte) 0);
    long dataStart = peer + metadataSize + bitmapSize;
    int dataOffset = 0;
    for (int i = 0 ; i < count ; i++)
    {
        MemoryUtil.setShort(peer + 2 + i * 2, (short) dataOffset);

        ByteBuffer value = clustering.get(i);
        if (value == null)
        {
            long boffset = bitmapStart + (i >>> 3);
            int b = MemoryUtil.getByte(boffset);
            b |= 1 << (i & 7);
            MemoryUtil.setByte(boffset, (byte) b);
            continue;
        }

        assert value.order() == ByteOrder.BIG_ENDIAN;

        int size = value.remaining();
        MemoryUtil.setBytes(dataStart + dataOffset, value);
        dataOffset += size;
    }
}