import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata)
{
    String req = "SELECT * FROM system.%s WHERE row_key = ? AND cf_id = ?";
    UntypedResultSet results = executeInternal(String.format(req, PAXOS_CF), key, metadata.cfId);
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = row.has("in_progress_ballot")
                    ? new Commit(key, row.getUUID("in_progress_ballot"), ArrayBackedSortedColumns.factory.create(metadata))
                    : Commit.emptyCommit(key, metadata);
    // either we have both a recently accepted ballot and update or we have neither
    Commit accepted = row.has("proposal")
                    ? new Commit(key, row.getUUID("proposal_ballot"), ColumnFamily.fromBytes(row.getBytes("proposal")))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(key, row.getUUID("most_recent_commit_at"), ColumnFamily.fromBytes(row.getBytes("most_recent_commit")))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata)
{
    String req = "SELECT * FROM system.%s WHERE row_key = 0x%s AND cf_id = %s";
    UntypedResultSet results = processInternal(String.format(req, PAXOS_CF, ByteBufferUtil.bytesToHex(key), metadata.cfId));
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = new Commit(key, row.getUUID("in_progress_ballot"), EmptyColumns.factory.create(metadata));
    // either we have both a recently accepted ballot and update or we have neither
    Commit accepted = row.has("proposal")
                    ? new Commit(key, row.getUUID("proposal_ballot"), ColumnFamily.fromBytes(row.getBytes("proposal")))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(key, row.getUUID("most_recent_commit_at"), ColumnFamily.fromBytes(row.getBytes("most_recent_commit")))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
static RowIterator casInternal(CQL3CasRequest request, QueryState state)
{
    UUID ballot = UUIDGen.getTimeUUIDFromMicros(state.getTimestamp());

    SinglePartitionReadCommand readCommand = request.readCommand(FBUtilities.nowInSeconds());
    FilteredPartition current;
    try (ReadOrderGroup orderGroup = readCommand.startOrderGroup(); PartitionIterator iter = readCommand.executeInternal(orderGroup))
    {
        current = FilteredPartition.create(PartitionIterators.getOnlyElement(iter, readCommand));
    }

    if (!request.appliesTo(current))
        return current.rowIterator();

    PartitionUpdate updates = request.makeUpdates(current);
    updates = TriggerExecutor.instance.execute(updates);

    Commit proposal = Commit.newProposal(ballot, updates);
    proposal.makeMutation().apply();
    return null;
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
private static boolean proposePaxos(Commit proposal, List<InetAddress> endpoints, int requiredParticipants, boolean timeoutIfPartial, ConsistencyLevel consistencyLevel)
throws WriteTimeoutException
{
    ProposeCallback callback = new ProposeCallback(endpoints.size(), requiredParticipants, !timeoutIfPartial, consistencyLevel);
    MessageOut<Commit> message = new MessageOut<Commit>(MessagingService.Verb.PAXOS_PROPOSE, proposal, Commit.serializer);
    for (InetAddress target : endpoints)
        MessagingService.instance().sendRR(message, target, callback);

    callback.await();

    if (callback.isSuccessful())
        return true;

    if (timeoutIfPartial && !callback.isFullyRefused())
        throw new WriteTimeoutException(WriteType.CAS, consistencyLevel, callback.getAcceptCount(), requiredParticipants);

    return false;
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static PaxosState loadPaxosState(DecoratedKey key, CFMetaData metadata, int nowInSec)
{
    String req = "SELECT * FROM system.%s WHERE row_key = ? AND cf_id = ?";
    UntypedResultSet results = QueryProcessor.executeInternalWithNow(nowInSec, String.format(req, PAXOS), key.getKey(), metadata.cfId);
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = row.has("in_progress_ballot")
                    ? new Commit(row.getUUID("in_progress_ballot"), new PartitionUpdate(metadata, key, metadata.partitionColumns(), 1))
                    : Commit.emptyCommit(key, metadata);
    // either we have both a recently accepted ballot and update or we have neither
    int proposalVersion = row.has("proposal_version") ? row.getInt("proposal_version") : MessagingService.VERSION_21;
    Commit accepted = row.has("proposal")
                    ? new Commit(row.getUUID("proposal_ballot"), PartitionUpdate.fromBytes(row.getBytes("proposal"), proposalVersion, key))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    int mostRecentVersion = row.has("most_recent_commit_version") ? row.getInt("most_recent_commit_version") : MessagingService.VERSION_21;
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(row.getUUID("most_recent_commit_at"), PartitionUpdate.fromBytes(row.getBytes("most_recent_commit"), mostRecentVersion, key))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
private void testShouldHint(Verb verb, ConsistencyLevel cl, boolean allowHints, boolean expectHint) throws Exception
{
    Object payload = verb == Verb.PAXOS_COMMIT
                     ? new Commit(UUID.randomUUID(), new PartitionUpdate(MockSchema.newCFMetaData("", ""), ByteBufferUtil.EMPTY_BYTE_BUFFER, PartitionColumns.NONE, 1))
                     : new Mutation("", new BufferDecoratedKey(new Murmur3Partitioner.LongToken(0), ByteBufferUtil.EMPTY_BYTE_BUFFER));

    WriteCallbackInfo wcbi = new WriteCallbackInfo(InetAddress.getByName("192.168.1.1"), null, new MessageOut(verb, payload, null), null, cl, allowHints);
    Assert.assertEquals(expectHint, wcbi.shouldHint());
    if (expectHint)
    {
        Assert.assertNotNull(wcbi.mutation());
    }
    else
    {
        boolean fail = false;
        try
        {
            wcbi.mutation();
        }
        catch (Throwable t)
        {
            fail = true;
        }
        Assert.assertTrue(fail);
    }
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata)
{
    String req = "SELECT * FROM system.%s WHERE row_key = 0x%s AND cf_id = %s";
    UntypedResultSet results = processInternal(String.format(req, PAXOS_CF, ByteBufferUtil.bytesToHex(key), metadata.cfId));
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = row.has("in_progress_ballot")
                    ? new Commit(key, row.getUUID("in_progress_ballot"), EmptyColumns.factory.create(metadata))
                    : Commit.emptyCommit(key, metadata);
    // either we have both a recently accepted ballot and update or we have neither
    Commit accepted = row.has("proposal")
                    ? new Commit(key, row.getUUID("proposal_ballot"), ColumnFamily.fromBytes(row.getBytes("proposal")))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(key, row.getUUID("most_recent_commit_at"), ColumnFamily.fromBytes(row.getBytes("most_recent_commit")))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata)
{
    String req = "SELECT * FROM system.%s WHERE row_key = 0x%s AND cf_id = %s";
    UntypedResultSet results = processInternal(String.format(req, PAXOS_CF, ByteBufferUtil.bytesToHex(key), metadata.cfId));
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = row.has("in_progress_ballot")
                    ? new Commit(key, row.getUUID("in_progress_ballot"), ArrayBackedSortedColumns.factory.create(metadata))
                    : Commit.emptyCommit(key, metadata);
    // either we have both a recently accepted ballot and update or we have neither
    Commit accepted = row.has("proposal")
                    ? new Commit(key, row.getUUID("proposal_ballot"), ColumnFamily.fromBytes(row.getBytes("proposal")))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(key, row.getUUID("most_recent_commit_at"), ColumnFamily.fromBytes(row.getBytes("most_recent_commit")))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static void savePaxosPromise(Commit promise)
{
    String req = "UPDATE system.%s USING TIMESTAMP ? AND TTL ? SET in_progress_ballot = ? WHERE row_key = ? AND cf_id = ?";
    executeInternal(String.format(req, PAXOS_CF),
                    UUIDGen.microsTimestamp(promise.ballot),
                    paxosTtl(promise.update.metadata),
                    promise.ballot,
                    promise.key,
                    promise.update.id());
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static void savePaxosProposal(Commit proposal)
{
    executeInternal(String.format("UPDATE system.%s USING TIMESTAMP ? AND TTL ? SET proposal_ballot = ?, proposal = ? WHERE row_key = ? AND cf_id = ?", PAXOS_CF),
                    UUIDGen.microsTimestamp(proposal.ballot),
                    paxosTtl(proposal.update.metadata),
                    proposal.ballot,
                    proposal.update.toBytes(),
                    proposal.key,
                    proposal.update.id());
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static void savePaxosCommit(Commit commit)
{
    // We always erase the last proposal (with the commit timestamp to no erase more recent proposal in case the commit is old)
    // even though that's really just an optimization  since SP.beginAndRepairPaxos will exclude accepted proposal older than the mrc.
    String cql = "UPDATE system.%s USING TIMESTAMP ? AND TTL ? SET proposal_ballot = null, proposal = null, most_recent_commit_at = ?, most_recent_commit = ? WHERE row_key = ? AND cf_id = ?";
    executeInternal(String.format(cql, PAXOS_CF),
                    UUIDGen.microsTimestamp(commit.ballot),
                    paxosTtl(commit.update.metadata),
                    commit.ballot,
                    commit.update.toBytes(),
                    commit.key,
                    commit.update.id());
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
@Test
public void testCommittingAfterTruncation() throws Exception
{
    ColumnFamilyStore cfs = Keyspace.open("Keyspace1").getColumnFamilyStore("Standard1");
    DecoratedKey key = Util.dk("key" + System.nanoTime());
    CellName name = Util.cellname("col");
    ByteBuffer value = ByteBufferUtil.bytes(0);
    ColumnFamily update = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    update.addColumn(name, value, FBUtilities.timestampMicros());

    // CFS should be empty initially
    assertNoDataPresent(cfs, key);

    // Commit the proposal & verify the data is present
    Commit beforeTruncate = newProposal(0, key.getKey(), update);
    PaxosState.commit(beforeTruncate);
    assertDataPresent(cfs, key, name, value);

    // Truncate then attempt to commit again, mutation should
    // be ignored as the proposal predates the truncation
    cfs.truncateBlocking();
    PaxosState.commit(beforeTruncate);
    assertNoDataPresent(cfs, key);

    // Now try again with a ballot created after the truncation
    long timestamp = SystemKeyspace.getTruncatedAt(update.metadata().cfId) + 1;
    Commit afterTruncate = newProposal(timestamp, key.getKey(), update);
    PaxosState.commit(afterTruncate);
    assertDataPresent(cfs, key, name, value);
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static void savePaxosPromise(Commit promise)
{
    String req = "UPDATE %s USING TIMESTAMP %d AND TTL %d SET in_progress_ballot = %s WHERE row_key = 0x%s AND cf_id = %s";
    processInternal(String.format(req,
                                  PAXOS_CF,
                                  UUIDGen.microsTimestamp(promise.ballot),
                                  paxosTtl(promise.update.metadata),
                                  promise.ballot,
                                  ByteBufferUtil.bytesToHex(promise.key),
                                  promise.update.id()));
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static void savePaxosProposal(Commit proposal)
{
    processInternal(String.format("UPDATE %s USING TIMESTAMP %d AND TTL %d SET proposal_ballot = %s, proposal = 0x%s WHERE row_key = 0x%s AND cf_id = %s",
                                  PAXOS_CF,
                                  UUIDGen.microsTimestamp(proposal.ballot),
                                  paxosTtl(proposal.update.metadata),
                                  proposal.ballot,
                                  ByteBufferUtil.bytesToHex(proposal.update.toBytes()),
                                  ByteBufferUtil.bytesToHex(proposal.key),
                                  proposal.update.id()));
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static void savePaxosCommit(Commit commit)
{
    // We always erase the last proposal (with the commit timestamp to no erase more recent proposal in case the commit is old)
    // even though that's really just an optimization  since SP.beginAndRepairPaxos will exclude accepted proposal older than the mrc.
    String cql = "UPDATE %s USING TIMESTAMP %d AND TTL %d SET proposal_ballot = null, proposal = null, most_recent_commit_at = %s, most_recent_commit = 0x%s WHERE row_key = 0x%s AND cf_id = %s";
    processInternal(String.format(cql,
                                  PAXOS_CF,
                                  UUIDGen.microsTimestamp(commit.ballot),
                                  paxosTtl(commit.update.metadata),
                                  commit.ballot,
                                  ByteBufferUtil.bytesToHex(commit.update.toBytes()),
                                  ByteBufferUtil.bytesToHex(commit.key),
                                  commit.update.id()));
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
/**
 * Unlike commitPaxos, this does not wait for replies
 */
private static void sendCommit(Commit commit, Iterable<InetAddress> replicas)
{
    MessageOut<Commit> message = new MessageOut<Commit>(MessagingService.Verb.PAXOS_COMMIT, commit, Commit.serializer);
    for (InetAddress target : replicas)
        MessagingService.instance().sendOneWay(message, target);
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
private static PrepareCallback preparePaxos(Commit toPrepare, List<InetAddress> endpoints, int requiredParticipants, ConsistencyLevel consistencyForPaxos)
throws WriteTimeoutException
{
    PrepareCallback callback = new PrepareCallback(toPrepare.update.partitionKey(), toPrepare.update.metadata(), requiredParticipants, consistencyForPaxos);
    MessageOut<Commit> message = new MessageOut<Commit>(MessagingService.Verb.PAXOS_PREPARE, toPrepare, Commit.serializer);
    for (InetAddress target : endpoints)
        MessagingService.instance().sendRR(message, target, callback);
    callback.await();
    return callback;
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
/**
 * Commit a PAXOS task locally, and if the task times out rather then submitting a real hint
 * submit a fake one that executes immediately on the mutation stage, but generates the necessary backpressure
 * signal for hints
 */
private static void commitPaxosLocal(final MessageOut<Commit> message, final AbstractWriteResponseHandler<?> responseHandler)
{
    StageManager.getStage(MessagingService.verbStages.get(MessagingService.Verb.PAXOS_COMMIT)).maybeExecuteImmediately(new LocalMutationRunnable()
    {
        public void runMayThrow()
        {
            try
            {
                PaxosState.commit(message.payload);
                if (responseHandler != null)
                    responseHandler.response(null);
            }
            catch (Exception ex)
            {
                if (!(ex instanceof WriteTimeoutException))
                    logger.error("Failed to apply paxos commit locally : {}", ex);
                responseHandler.onFailure(FBUtilities.getBroadcastAddress());
            }
        }

        @Override
        protected Verb verb()
        {
            return MessagingService.Verb.PAXOS_COMMIT;
        }
    });
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static void savePaxosPromise(Commit promise)
{
    String req = "UPDATE system.%s USING TIMESTAMP ? AND TTL ? SET in_progress_ballot = ? WHERE row_key = ? AND cf_id = ?";
    executeInternal(String.format(req, PAXOS),
                    UUIDGen.microsTimestamp(promise.ballot),
                    paxosTtlSec(promise.update.metadata()),
                    promise.ballot,
                    promise.update.partitionKey().getKey(),
                    promise.update.metadata().cfId);
}
 

import org.apache.cassandra.service.paxos.Commit; //导入依赖的package包/类
public static void savePaxosProposal(Commit proposal)
{
    executeInternal(String.format("UPDATE system.%s USING TIMESTAMP ? AND TTL ? SET proposal_ballot = ?, proposal = ?, proposal_version = ? WHERE row_key = ? AND cf_id = ?", PAXOS),
                    UUIDGen.microsTimestamp(proposal.ballot),
                    paxosTtlSec(proposal.update.metadata()),
                    proposal.ballot,
                    PartitionUpdate.toBytes(proposal.update, MessagingService.current_version),
                    MessagingService.current_version,
                    proposal.update.partitionKey().getKey(),
                    proposal.update.metadata().cfId);
}