import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public void doVerb(MessageIn<AbstractRangeCommand> message, int id)
{
    try
    {
        if (StorageService.instance.isBootstrapMode())
        {
            /* Don't service reads! */
            throw new RuntimeException("Cannot service reads while bootstrapping!");
        }
        RangeSliceReply reply = new RangeSliceReply(message.payload.executeLocally());
        Tracing.trace("Enqueuing response to {}", message.from);
        MessagingService.instance().sendReply(reply.createMessage(), id, message.from);
    }
    catch (TombstoneOverwhelmingException e)
    {
        // error already logged.  Drop the request
    }
    catch (Exception ex)
    {
        throw new RuntimeException(ex);
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public static boolean shouldHint(InetAddress ep)
{
    if (DatabaseDescriptor.shouldHintByDC())
    {
        final String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(ep);
        //Disable DC specific hints
        if(!DatabaseDescriptor.hintedHandoffEnabled(dc))
        {
            HintedHandOffManager.instance.metrics.incrPastWindow(ep);
            return false;
        }
    }
    else if (!DatabaseDescriptor.hintedHandoffEnabled())
    {
        HintedHandOffManager.instance.metrics.incrPastWindow(ep);
        return false;
    }

    boolean hintWindowExpired = Gossiper.instance.getEndpointDowntime(ep) > DatabaseDescriptor.getMaxHintWindow();
    if (hintWindowExpired)
    {
        HintedHandOffManager.instance.metrics.incrPastWindow(ep);
        Tracing.trace("Not hinting {} which has been down {}ms", ep, Gossiper.instance.getEndpointDowntime(ep));
    }
    return !hintWindowExpired;
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
private ClockAndCount[] getCurrentValues(List<CounterUpdateCell> counterUpdateCells, ColumnFamilyStore cfs)
{
    ClockAndCount[] currentValues = new ClockAndCount[counterUpdateCells.size()];
    int remaining = counterUpdateCells.size();

    if (CacheService.instance.counterCache.getCapacity() != 0)
    {
        Tracing.trace("Fetching {} counter values from cache", counterUpdateCells.size());
        remaining = getCurrentValuesFromCache(counterUpdateCells, cfs, currentValues);
        if (remaining == 0)
            return currentValues;
    }

    Tracing.trace("Reading {} counter values from the CF", remaining);
    getCurrentValuesFromCFS(counterUpdateCells, cfs, currentValues);

    return currentValues;
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
/**
 * Older version (< 1.0) will not send this message at all, hence we don't
 * need to check the version of the data.
 */
private static void forwardToLocalNodes(Mutation mutation, MessagingService.Verb verb, byte[] forwardBytes, InetAddress from) throws IOException
{
    try (DataInputStream in = new DataInputStream(new FastByteArrayInputStream(forwardBytes)))
    {
        int size = in.readInt();

        // tell the recipients who to send their ack to
        MessageOut<Mutation> message = new MessageOut<>(verb, mutation, Mutation.serializer).withParameter(Mutation.FORWARD_FROM, from.getAddress());
        // Send a message to each of the addresses on our Forward List
        for (int i = 0; i < size; i++)
        {
            InetAddress address = CompactEndpointSerializationHelper.deserialize(in);
            int id = in.readInt();
            Tracing.trace("Enqueuing forwarded write to {}", address);
            MessagingService.instance().sendOneWay(message, id, address);
        }
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
/**
 * Older version (< 1.0) will not send this message at all, hence we don't
 * need to check the version of the data.
 */
private void forwardToLocalNodes(Mutation mutation, MessagingService.Verb verb, byte[] forwardBytes, InetAddress from) throws IOException
{
    DataInputStream in = new DataInputStream(new FastByteArrayInputStream(forwardBytes));
    int size = in.readInt();

    // tell the recipients who to send their ack to
    MessageOut<Mutation> message = new MessageOut<>(verb, mutation, Mutation.serializer).withParameter(Mutation.FORWARD_FROM, from.getAddress());
    // Send a message to each of the addresses on our Forward List
    for (int i = 0; i < size; i++)
    {
        InetAddress address = CompactEndpointSerializationHelper.deserialize(in);
        int id = in.readInt();
        Tracing.trace("Enqueuing forwarded write to {}", address);
        MessagingService.instance().sendOneWay(message, id, address);
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public void doVerb(MessageIn<Truncation> message, int id)
{
    Truncation t = message.payload;
    Tracing.trace("Applying truncation of {}.{}", t.keyspace, t.columnFamily);
    try
    {
        ColumnFamilyStore cfs = Keyspace.open(t.keyspace).getColumnFamilyStore(t.columnFamily);
        cfs.truncateBlocking();
    }
    catch (Exception e)
    {
        logger.error("Error in truncation", e);
        respondError(t, message);

        if (FSError.findNested(e) != null)
            throw FSError.findNested(e);
    }
    Tracing.trace("Enqueuing response to truncate operation to {}", message.from);

    TruncateResponse response = new TruncateResponse(t.keyspace, t.columnFamily, true);
    logger.trace("{} applied.  Enqueuing response to {}@{} ", new Object[]{ t, id, message.from });
    MessagingService.instance().sendReply(response.createMessage(), id, message.from);
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public static PrepareResponse prepare(Commit toPrepare)
{
    synchronized (lockFor(toPrepare.key))
    {
        PaxosState state = SystemKeyspace.loadPaxosState(toPrepare.key, toPrepare.update.metadata());
        if (toPrepare.isAfter(state.promised))
        {
            Tracing.trace("Promising ballot {}", toPrepare.ballot);
            SystemKeyspace.savePaxosPromise(toPrepare);
            return new PrepareResponse(true, state.accepted, state.mostRecentCommit);
        }
        else
        {
            Tracing.trace("Promise rejected; {} is not sufficiently newer than {}", toPrepare, state.promised);
            // return the currently promised ballot (not the last accepted one) so the coordinator can make sure it uses newer ballot next time (#5667)
            return new PrepareResponse(false, state.promised, state.mostRecentCommit);
        }
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public static Boolean propose(Commit proposal)
{
    synchronized (lockFor(proposal.key))
    {
        PaxosState state = SystemKeyspace.loadPaxosState(proposal.key, proposal.update.metadata());
        if (proposal.hasBallot(state.promised.ballot) || proposal.isAfter(state.promised))
        {
            Tracing.trace("Accepting proposal {}", proposal);
            SystemKeyspace.savePaxosProposal(proposal);
            return true;
        }
        else
        {
            Tracing.trace("Rejecting proposal for {} because inProgress is now {}", proposal, state.promised);
            return false;
        }
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public void doVerb(MessageIn<ReadCommand> message, int id)
{
    if (StorageService.instance.isBootstrapMode())
    {
        throw new RuntimeException("Cannot service reads while bootstrapping!");
    }

    ReadCommand command = message.payload;
    ReadResponse response;
    try (ReadOrderGroup opGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(opGroup))
    {
        response = command.createResponse(iterator);
    }

    MessageOut<ReadResponse> reply = new MessageOut<>(MessagingService.Verb.REQUEST_RESPONSE, response, serializer());

    Tracing.trace("Enqueuing response to {}", message.from);
    MessagingService.instance().sendReply(reply, id, message.from);
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public void close()
{
    try
    {
        FBUtilities.waitOnFutures(repairResults, DatabaseDescriptor.getWriteRpcTimeout());
    }
    catch (TimeoutException ex)
    {
        // We got all responses, but timed out while repairing
        int blockFor = consistency.blockFor(keyspace);
        if (Tracing.isTracing())
            Tracing.trace("Timed out while read-repairing after receiving all {} data and digest responses", blockFor);
        else
            logger.debug("Timeout while read-repairing after receiving all {} data and digest responses", blockFor);

        throw new ReadTimeoutException(consistency, blockFor-1, blockFor, true);
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
AbstractReadExecutor(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistencyLevel, List<InetAddress> targetReplicas)
{
    this.command = command;
    this.targetReplicas = targetReplicas;
    this.handler = new ReadCallback(new DigestResolver(keyspace, command, consistencyLevel, targetReplicas.size()), consistencyLevel, command, targetReplicas);
    this.traceState = Tracing.instance.get();

    // Set the digest version (if we request some digests). This is the smallest version amongst all our target replicas since new nodes
    // knows how to produce older digest but the reverse is not true.
    // TODO: we need this when talking with pre-3.0 nodes. So if we preserve the digest format moving forward, we can get rid of this once
    // we stop being compatible with pre-3.0 nodes.
    int digestVersion = MessagingService.current_version;
    for (InetAddress replica : targetReplicas)
        digestVersion = Math.min(digestVersion, MessagingService.instance().getVersion(replica));
    command.setDigestVersion(digestVersion);
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
private PartitionUpdate processModifications(PartitionUpdate changes)
{
    ColumnFamilyStore cfs = Keyspace.open(getKeyspaceName()).getColumnFamilyStore(changes.metadata().cfId);

    List<PartitionUpdate.CounterMark> marks = changes.collectCounterMarks();

    if (CacheService.instance.counterCache.getCapacity() != 0)
    {
        Tracing.trace("Fetching {} counter values from cache", marks.size());
        updateWithCurrentValuesFromCache(marks, cfs);
        if (marks.isEmpty())
            return changes;
    }

    Tracing.trace("Reading {} counter values from the CF", marks.size());
    updateWithCurrentValuesFromCFS(marks, cfs);

    // What's remain is new counters
    for (PartitionUpdate.CounterMark mark : marks)
        updateWithCurrentValue(mark, ClockAndCount.BLANK, cfs);

    return changes;
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public static ParsedStatement.Prepared getStatement(String queryStr, ClientState clientState)
throws RequestValidationException
{
    Tracing.trace("Parsing {}", queryStr);
    ParsedStatement statement = parseStatement(queryStr);

    // Set keyspace for statement that require login
    if (statement instanceof CFStatement)
        ((CFStatement)statement).prepareKeyspace(clientState);

    Tracing.trace("Preparing statement");
    return statement.prepare();
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public void doVerb(MessageIn<Commit> message, int id)
{
    PaxosState.commit(message.payload);

    WriteResponse response = new WriteResponse();
    Tracing.trace("Enqueuing acknowledge to {}", message.from);
    MessagingService.instance().sendReply(response.createMessage(), id, message.from);
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public static PrepareResponse prepare(Commit toPrepare)
{
    long start = System.nanoTime();
    try
    {
        Lock lock = LOCKS.get(toPrepare.key);
        lock.lock();
        try
        {
            PaxosState state = SystemKeyspace.loadPaxosState(toPrepare.key, toPrepare.update.metadata());
            if (toPrepare.isAfter(state.promised))
            {
                Tracing.trace("Promising ballot {}", toPrepare.ballot);
                SystemKeyspace.savePaxosPromise(toPrepare);
                return new PrepareResponse(true, state.accepted, state.mostRecentCommit);
            }
            else
            {
                Tracing.trace("Promise rejected; {} is not sufficiently newer than {}", toPrepare, state.promised);
                // return the currently promised ballot (not the last accepted one) so the coordinator can make sure it uses newer ballot next time (#5667)
                return new PrepareResponse(false, state.promised, state.mostRecentCommit);
            }
        }
        finally
        {
            lock.unlock();
        }
    }
    finally
    {
        Keyspace.open(toPrepare.update.metadata().ksName).getColumnFamilyStore(toPrepare.update.metadata().cfId).metric.casPrepare.addNano(System.nanoTime() - start);
    }

}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public static Boolean propose(Commit proposal)
{
    long start = System.nanoTime();
    try
    {
        Lock lock = LOCKS.get(proposal.key);
        lock.lock();
        try
        {
            PaxosState state = SystemKeyspace.loadPaxosState(proposal.key, proposal.update.metadata());
            if (proposal.hasBallot(state.promised.ballot) || proposal.isAfter(state.promised))
            {
                Tracing.trace("Accepting proposal {}", proposal);
                SystemKeyspace.savePaxosProposal(proposal);
                return true;
            }
            else
            {
                Tracing.trace("Rejecting proposal for {} because inProgress is now {}", proposal, state.promised);
                return false;
            }
        }
        finally
        {
            lock.unlock();
        }
    }
    finally
    {
        Keyspace.open(proposal.update.metadata().ksName).getColumnFamilyStore(proposal.update.metadata().cfId).metric.casPropose.addNano(System.nanoTime() - start);
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public static void commit(Commit proposal)
{
    long start = System.nanoTime();
    try
    {
        // There is no guarantee we will see commits in the right order, because messages
        // can get delayed, so a proposal can be older than our current most recent ballot/commit.
        // Committing it is however always safe due to column timestamps, so always do it. However,
        // if our current in-progress ballot is strictly greater than the proposal one, we shouldn't
        // erase the in-progress update.
        // The table may have been truncated since the proposal was initiated. In that case, we
        // don't want to perform the mutation and potentially resurrect truncated data
        if (UUIDGen.unixTimestamp(proposal.ballot) >= SystemKeyspace.getTruncatedAt(proposal.update.metadata().cfId))
        {
            Tracing.trace("Committing proposal {}", proposal);
            Mutation mutation = proposal.makeMutation();
            Keyspace.open(mutation.getKeyspaceName()).apply(mutation, true);
        }
        else
        {
            Tracing.trace("Not committing proposal {} as ballot timestamp predates last truncation time", proposal);
        }
        // We don't need to lock, we're just blindly updating
        SystemKeyspace.savePaxosCommit(proposal);
    }
    finally
    {
        Keyspace.open(proposal.update.metadata().ksName).getColumnFamilyStore(proposal.update.metadata().cfId).metric.casCommit.addNano(System.nanoTime() - start);
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
public void createTracingSession()
{
    if (this.preparedTracingSession == null)
    {
        Tracing.instance.newSession();
    }
    else
    {
        UUID session = this.preparedTracingSession;
        this.preparedTracingSession = null;
        Tracing.instance.newSession(session);
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
/**
 * Handle counter mutation on the coordinator host.
 *
 * A counter mutation needs to first be applied to a replica (that we'll call the leader for the mutation) before being
 * replicated to the other endpoint. To achieve so, there is two case:
 *   1) the coordinator host is a replica: we proceed to applying the update locally and replicate throug
 *   applyCounterMutationOnCoordinator
 *   2) the coordinator is not a replica: we forward the (counter)mutation to a chosen replica (that will proceed through
 *   applyCounterMutationOnLeader upon receive) and wait for its acknowledgment.
 *
 * Implementation note: We check if we can fulfill the CL on the coordinator host even if he is not a replica to allow
 * quicker response and because the WriteResponseHandlers don't make it easy to send back an error. We also always gather
 * the write latencies at the coordinator node to make gathering point similar to the case of standard writes.
 */
public static AbstractWriteResponseHandler mutateCounter(CounterMutation cm, String localDataCenter) throws UnavailableException, OverloadedException
{
    InetAddress endpoint = findSuitableEndpoint(cm.getKeyspaceName(), cm.key(), localDataCenter, cm.consistency());

    if (endpoint.equals(FBUtilities.getBroadcastAddress()))
    {
        return applyCounterMutationOnCoordinator(cm, localDataCenter);
    }
    else
    {
        // Exit now if we can't fulfill the CL here instead of forwarding to the leader replica
        String keyspaceName = cm.getKeyspaceName();
        AbstractReplicationStrategy rs = Keyspace.open(keyspaceName).getReplicationStrategy();
        Token tk = StorageService.getPartitioner().getToken(cm.key());
        List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(keyspaceName, tk);
        Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, keyspaceName);

        rs.getWriteResponseHandler(naturalEndpoints, pendingEndpoints, cm.consistency(), null, WriteType.COUNTER).assureSufficientLiveNodes();

        // Forward the actual update to the chosen leader replica
        AbstractWriteResponseHandler responseHandler = new WriteResponseHandler(endpoint, WriteType.COUNTER);

        Tracing.trace("Enqueuing counter update to {}", endpoint);
        MessagingService.instance().sendRR(cm.makeMutationMessage(), endpoint, responseHandler, false);
        return responseHandler;
    }
}
 

import org.apache.cassandra.tracing.Tracing; //导入依赖的package包/类
/**
 * Performs the truncate operatoin, which effectively deletes all data from
 * the column family cfname
 * @param keyspace
 * @param cfname
 * @throws UnavailableException If some of the hosts in the ring are down.
 * @throws TimeoutException
 * @throws IOException
 */
public static void truncateBlocking(String keyspace, String cfname) throws UnavailableException, TimeoutException, IOException
{
    logger.debug("Starting a blocking truncate operation on keyspace {}, CF {}", keyspace, cfname);
    if (isAnyStorageHostDown())
    {
        logger.info("Cannot perform truncate, some hosts are down");
        // Since the truncate operation is so aggressive and is typically only
        // invoked by an admin, for simplicity we require that all nodes are up
        // to perform the operation.
        int liveMembers = Gossiper.instance.getLiveMembers().size();
        throw new UnavailableException(ConsistencyLevel.ALL, liveMembers + Gossiper.instance.getUnreachableMembers().size(), liveMembers);
    }

    Set<InetAddress> allEndpoints = Gossiper.instance.getLiveTokenOwners();
    
    int blockFor = allEndpoints.size();
    final TruncateResponseHandler responseHandler = new TruncateResponseHandler(blockFor);

    // Send out the truncate calls and track the responses with the callbacks.
    Tracing.trace("Enqueuing truncate messages to hosts {}", allEndpoints);
    final Truncation truncation = new Truncation(keyspace, cfname);
    MessageOut<Truncation> message = truncation.createMessage();
    for (InetAddress endpoint : allEndpoints)
        MessagingService.instance().sendRR(message, endpoint, responseHandler);

    // Wait for all
    try
    {
        responseHandler.get();
    }
    catch (TimeoutException e)
    {
        Tracing.trace("Timed out");
        throw e;
    }
}