import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
@Override protected void before() throws Throwable {
  if (server != null) return;

  DatabaseDescriptor.daemonInitialization();

  // Cleanup first
  try {
    cleanupAndLeaveDirs();
  } catch (IOException e) {
    throw new RuntimeException("Failed to cleanup and recreate directories.", e);
  }

  Keyspace.setInitialized();
  SystemKeyspace.persistLocalMetadata();
  SystemKeyspace.finishStartup();
  StorageService.instance.initServer();

  server = new Server.Builder().withHost(nativeAddr).withPort(nativePort).build();
  server.start();
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
/**
 * Return the data center for which an endpoint resides in
 *
 * @param endpoint the endpoint to process
 * @return string of data center
 */
public String getDatacenter(InetAddress endpoint)
{
    if (endpoint.equals(FBUtilities.getBroadcastAddress()))
        return myDC;

    EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
    if (epState == null || epState.getApplicationState(ApplicationState.DC) == null)
    {
        if (psnitch == null)
        {
            if (savedEndpoints == null)
                savedEndpoints = SystemKeyspace.loadDcRackInfo();
            if (savedEndpoints.containsKey(endpoint))
                return savedEndpoints.get(endpoint).get("data_center");
            return DEFAULT_DC;
        }
        else
            return psnitch.getDatacenter(endpoint);
    }
    return epState.getApplicationState(ApplicationState.DC).value;
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
/**
 * Return the rack for which an endpoint resides in
 *
 * @param endpoint the endpoint to process
 * @return string of rack
 */
public String getRack(InetAddress endpoint)
{
    if (endpoint.equals(FBUtilities.getBroadcastAddress()))
        return myRack;

    EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
    if (epState == null || epState.getApplicationState(ApplicationState.RACK) == null)
    {
        if (psnitch == null)
        {
            if (savedEndpoints == null)
                savedEndpoints = SystemKeyspace.loadDcRackInfo();
            if (savedEndpoints.containsKey(endpoint))
                return savedEndpoints.get(endpoint).get("rack");
            return DEFAULT_RACK;
        }
        else
            return psnitch.getRack(endpoint);
    }
    return epState.getApplicationState(ApplicationState.RACK).value;
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
public StreamResultFuture fetchAsync()
{
    for (Map.Entry<String, Map.Entry<InetAddress, Collection<Range<Token>>>> entry : toFetch.entries())
    {
        String keyspace = entry.getKey();
        InetAddress source = entry.getValue().getKey();
        InetAddress preferred = SystemKeyspace.getPreferredIP(source);
        Collection<Range<Token>> ranges = entry.getValue().getValue();
        /* Send messages to respective folks to stream data over to me */
        if (logger.isDebugEnabled())
            logger.debug("{}ing from {} ranges {}", description, source, StringUtils.join(ranges, ", "));
        streamPlan.requestRanges(source, preferred, keyspace, ranges);
    }

    return streamPlan.execute();
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
private void bootstrap(Collection<Token> tokens)
{
    isBootstrapMode = true;
    SystemKeyspace.updateTokens(tokens); // DON'T use setToken, that makes us part of the ring locally which is incorrect until we are done bootstrapping
    if (!DatabaseDescriptor.isReplacing())
    {
        // if not an existing token then bootstrap
        List<Pair<ApplicationState, VersionedValue>> states = new ArrayList<Pair<ApplicationState, VersionedValue>>();
        states.add(Pair.create(ApplicationState.TOKENS, valueFactory.tokens(tokens)));
        states.add(Pair.create(ApplicationState.STATUS, valueFactory.bootstrapping(tokens)));
        Gossiper.instance.addLocalApplicationStates(states);
        setMode(Mode.JOINING, "sleeping " + RING_DELAY + " ms for pending range setup", true);
        Uninterruptibles.sleepUninterruptibly(RING_DELAY, TimeUnit.MILLISECONDS);
    }
    else
    {
        // Dont set any state for the node which is bootstrapping the existing token...
        tokenMetadata.updateNormalTokens(tokens, FBUtilities.getBroadcastAddress());
        SystemKeyspace.removeEndpoint(DatabaseDescriptor.getReplaceAddress());
    }
    if (!Gossiper.instance.seenAnySeed())
        throw new IllegalStateException("Unable to contact any seeds!");
    setMode(Mode.JOINING, "Starting to bootstrap...", true);
    new BootStrapper(FBUtilities.getBroadcastAddress(), tokens, tokenMetadata).bootstrap(); // handles token update
    logger.info("Bootstrap completed! for the tokens {}", tokens);
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
private void initiateStreaming()
{
    long repairedAt = ActiveRepairService.UNREPAIRED_SSTABLE;
    InetAddress dest = request.dst;
    InetAddress preferred = SystemKeyspace.getPreferredIP(dest);
    if (desc.parentSessionId != null && ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId) != null)
        repairedAt = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).repairedAt;
    logger.info(String.format("[streaming task #%s] Performing streaming repair of %d ranges with %s", desc.sessionId, request.ranges.size(), request.dst));
    StreamResultFuture op = new StreamPlan("Repair", repairedAt, 1)
                                .flushBeforeTransfer(true)
                                // request ranges from the remote node
                                .requestRanges(dest, preferred, desc.keyspace, request.ranges, desc.columnFamily)
                                // send ranges to the remote node
                                .transferRanges(dest, preferred, desc.keyspace, request.ranges, desc.columnFamily)
                                .execute();
    op.addEventListener(this);
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
/**
 * Load schema definitions.
 *
 * @param updateVersion true if schema version needs to be updated
 */
public static void loadSchemas(boolean updateVersion)
{
    ColumnFamilyStore schemaCFS = SystemKeyspace.schemaCFS(SystemKeyspace.SCHEMA_KEYSPACES_CF);

    // if keyspace with definitions is empty try loading the old way
    if (schemaCFS.estimateKeys() == 0)
    {
        logger.info("Couldn't detect any schema definitions in local storage.");
        // peek around the data directories to see if anything is there.
        if (hasExistingNoSystemTables())
            logger.info("Found keyspace data in data directories. Consider using cqlsh to define your schema.");
        else
            logger.info("To create keyspaces and column families, see 'help create' in cqlsh.");
    }
    else
    {
        Schema.instance.load(DefsTables.loadFromKeyspace());
    }

    if (updateVersion)
        Schema.instance.updateVersion();
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
LocalCounterIdHistory()
{
    CounterId id = SystemKeyspace.getCurrentLocalCounterId();
    if (id == null)
    {
        // no recorded local counter id, generating a new one and saving it
        id = generate();
        logger.info("No saved local counter id, using newly generated: {}", id);
        SystemKeyspace.writeCurrentLocalCounterId(id, FBUtilities.timestampMicros());
        current = new AtomicReference<>(id);
        olds = new CopyOnWriteArrayList<>();
    }
    else
    {
        logger.info("Saved local counter id: {}", id);
        current = new AtomicReference<>(id);
        olds = new CopyOnWriteArrayList<>(SystemKeyspace.getOldLocalCounterIds());
    }
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的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.db.SystemKeyspace; //导入依赖的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.db.SystemKeyspace; //导入依赖的package包/类
/** load keyspace (keyspace) definitions, but do not initialize the keyspace instances. */
public static void loadSchemas()
{
    ColumnFamilyStore schemaCFS = SystemKeyspace.schemaCFS(SystemKeyspace.SCHEMA_KEYSPACES_CF);

    // if keyspace with definitions is empty try loading the old way
    if (schemaCFS.estimateKeys() == 0)
    {
        logger.info("Couldn't detect any schema definitions in local storage.");
        // peek around the data directories to see if anything is there.
        if (hasExistingNoSystemTables())
            logger.info("Found keyspace data in data directories. Consider using cqlsh to define your schema.");
        else
            logger.info("To create keyspaces and column families, see 'help create' in cqlsh.");
    }
    else
    {
        Schema.instance.load(DefsTables.loadFromKeyspace());
    }

    Schema.instance.updateVersion();
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
public void run()
{
    if (tracker != null)
        tracker.notifyDeleting(referent);

    if (referent.readMeter != null)
        SystemKeyspace.clearSSTableReadMeter(referent.getKeyspaceName(), referent.getColumnFamilyName(), referent.descriptor.generation);

    // If we can't successfully delete the DATA component, set the task to be retried later: see above
    File datafile = new File(desc.filenameFor(Component.DATA));
    if (!datafile.delete())
    {
        logger.error("Unable to delete " + datafile + " (it will be removed on server restart; we'll also retry after GC)");
        failedTasks.add(this);
        return;
    }
    // let the remainder be cleaned up by delete
    SSTable.delete(desc, Sets.difference(components, Collections.singleton(Component.DATA)));
    if (tracker != null)
        tracker.spaceReclaimed(size);
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
public void validate(ClientState state) throws RequestValidationException
{
    // validate login here before checkAccess to avoid leaking user existence to anonymous users.
    state.ensureNotAnonymous();

    if (!DatabaseDescriptor.getRoleManager().isExistingRole(grantee))
        throw new InvalidRequestException(String.format("Role %s doesn't exist", grantee.getRoleName()));

    // if a keyspace is omitted when GRANT/REVOKE ON TABLE <table>, we need to correct the resource.
    resource = maybeCorrectResource(resource, state);

    // altering permissions on builtin functions is not supported
    if (resource instanceof FunctionResource
        && SystemKeyspace.NAME.equals(((FunctionResource)resource).getKeyspace()))
    {
        throw new InvalidRequestException("Altering permissions on builtin functions is not supported");
    }

    if (!resource.exists())
        throw new InvalidRequestException(String.format("Resource %s doesn't exist", resource));
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
public StreamResultFuture fetchAsync()
{
    for (Map.Entry<String, Map.Entry<InetAddress, Collection<Range<Token>>>> entry : toFetch.entries())
    {
        String keyspace = entry.getKey();
        InetAddress source = entry.getValue().getKey();
        InetAddress preferred = SystemKeyspace.getPreferredIP(source);
        Collection<Range<Token>> ranges = entry.getValue().getValue();

        // filter out already streamed ranges
        Set<Range<Token>> availableRanges = stateStore.getAvailableRanges(keyspace, StorageService.instance.getTokenMetadata().partitioner);
        if (ranges.removeAll(availableRanges))
        {
            logger.info("Some ranges of {} are already available. Skipping streaming those ranges.", availableRanges);
        }

        if (logger.isTraceEnabled())
            logger.trace("{}ing from {} ranges {}", description, source, StringUtils.join(ranges, ", "));
        /* Send messages to respective folks to stream data over to me */
        streamPlan.requestRanges(source, preferred, keyspace, ranges);
    }

    return streamPlan.execute();
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
/**
 * When StreamSession completes, make all keyspaces/ranges in session available to be served.
 *
 * @param event Stream event.
 */
@Override
public void handleStreamEvent(StreamEvent event)
{
    if (event.eventType == StreamEvent.Type.STREAM_COMPLETE)
    {
        StreamEvent.SessionCompleteEvent se = (StreamEvent.SessionCompleteEvent) event;
        if (se.success)
        {
            for (StreamRequest request : se.requests)
            {
                SystemKeyspace.updateAvailableRanges(request.keyspace, request.ranges);
            }
        }
    }
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
public Iterable<InetAddress> replicasMissingMostRecentCommit(CFMetaData metadata, int nowInSec)
{
    // In general, we need every replicas that have answered to the prepare (a quorum) to agree on the MRC (see
    // coment in StorageProxy.beginAndRepairPaxos(), but basically we need to make sure at least a quorum of nodes
    // have learn a commit before commit a new one otherwise that previous commit is not guaranteed to have reach a
    // quorum and further commit may proceed on incomplete information).
    // However, if that commit is too hold, it may have been expired from some of the replicas paxos table (we don't
    // keep the paxos state forever or that could grow unchecked), and we could end up in some infinite loop as
    // explained on CASSANDRA-12043. To avoid that, we ignore a MRC that is too old, i.e. older than the TTL we set
    // on paxos tables. For such old commit, we rely on hints and repair to ensure the commit has indeed be
    // propagated to all nodes.
    long paxosTtlSec = SystemKeyspace.paxosTtlSec(metadata);
    if (UUIDGen.unixTimestampInSec(mostRecentCommit.ballot) + paxosTtlSec < nowInSec)
        return Collections.emptySet();

    return Iterables.filter(commitsByReplica.keySet(), new Predicate<InetAddress>()
    {
        public boolean apply(InetAddress inetAddress)
        {
            return (!commitsByReplica.get(inetAddress).ballot.equals(mostRecentCommit.ballot));
        }
    });
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
/**
 * Starts sending/receiving our list of differences to/from the remote endpoint: creates a callback
 * that will be called out of band once the streams complete.
 */
protected void startSync(List<Range<Token>> differences)
{
    InetAddress local = FBUtilities.getBroadcastAddress();
    // We can take anyone of the node as source or destination, however if one is localhost, we put at source to avoid a forwarding
    InetAddress dst = r2.endpoint.equals(local) ? r1.endpoint : r2.endpoint;
    InetAddress preferred = SystemKeyspace.getPreferredIP(dst);

    String message = String.format("Performing streaming repair of %d ranges with %s", differences.size(), dst);
    logger.info("[repair #{}] {}", desc.sessionId, message);
    boolean isIncremental = false;
    if (desc.parentSessionId != null)
    {
        ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId);
        isIncremental = prs.isIncremental;
    }
    Tracing.traceRepair(message);
    new StreamPlan("Repair", repairedAt, 1, false, isIncremental).listeners(this)
                                        .flushBeforeTransfer(true)
                                        // request ranges from the remote node
                                        .requestRanges(dst, preferred, desc.keyspace, differences, desc.columnFamily)
                                        // send ranges to the remote node
                                        .transferRanges(dst, preferred, desc.keyspace, differences, desc.columnFamily)
                                        .execute();
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
public void run()
{
    if (FailureDetector.instance.isAlive(neighbor))
    {
        AnticompactionRequest acr = new AnticompactionRequest(parentSession, successfulRanges);
        CassandraVersion peerVersion = SystemKeyspace.getReleaseVersion(neighbor);
        if (peerVersion != null && peerVersion.compareTo(VERSION_CHECKER) > 0)
        {
            MessagingService.instance().sendRR(acr.createMessage(), neighbor, new AnticompactionCallback(this), TimeUnit.DAYS.toMillis(1), true);
        }
        else
        {
            MessagingService.instance().sendOneWay(acr.createMessage(), neighbor);
            // immediately return after sending request
            set(neighbor);
        }
    }
    else
    {
        setException(new IOException(neighbor + " is down"));
    }
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
public void run()
{
    InetAddress dest = request.dst;
    InetAddress preferred = SystemKeyspace.getPreferredIP(dest);
    logger.info(String.format("[streaming task #%s] Performing streaming repair of %d ranges with %s", desc.sessionId, request.ranges.size(), request.dst));
    boolean isIncremental = false;
    if (desc.parentSessionId != null)
    {
        ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId);
        isIncremental = prs.isIncremental;
    }
    new StreamPlan("Repair", repairedAt, 1, false, isIncremental).listeners(this)
                                        .flushBeforeTransfer(true)
                                        // request ranges from the remote node
                                        .requestRanges(dest, preferred, desc.keyspace, request.ranges, desc.columnFamily)
                                        // send ranges to the remote node
                                        .transferRanges(dest, preferred, desc.keyspace, request.ranges, desc.columnFamily)
                                        .execute();
}
 

import org.apache.cassandra.db.SystemKeyspace; //导入依赖的package包/类
public void run()
{
    SystemKeyspace.clearSSTableReadMeter(desc.ksname, desc.cfname, desc.generation);

    try
    {
        // If we can't successfully delete the DATA component, set the task to be retried later: see TransactionTidier
        File datafile = new File(desc.filenameFor(Component.DATA));

        delete(datafile);
        // let the remainder be cleaned up by delete
        SSTable.delete(desc, SSTable.discoverComponentsFor(desc));
    }
    catch (Throwable t)
    {
        logger.error("Failed deletion for {}, we'll retry after GC and on server restart", desc);
        failedDeletions.add(this);
        return;
    }

    if (tracker != null && tracker.cfstore != null && !wasNew)
        tracker.cfstore.metric.totalDiskSpaceUsed.dec(sizeOnDisk);

    // release the referent to the parent so that the all transaction files can be released
    parentRef.release();
}