import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
public static boolean isEncryptedChannel(InetAddress address)
{
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    switch (DatabaseDescriptor.getServerEncryptionOptions().internode_encryption)
    {
        case none:
            return false; // if nothing needs to be encrypted then return immediately.
        case all:
            break;
        case dc:
            if (snitch.getDatacenter(address).equals(snitch.getDatacenter(FBUtilities.getBroadcastAddress())))
                return false;
            break;
        case rack:
            // for rack then check if the DC's are the same.
            if (snitch.getRack(address).equals(snitch.getRack(FBUtilities.getBroadcastAddress()))
                    && snitch.getDatacenter(address).equals(snitch.getDatacenter(FBUtilities.getBroadcastAddress())))
                return false;
            break;
    }
    return true;
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
public RangeStreamer(TokenMetadata metadata,
                     Collection<Token> tokens,
                     InetAddress address,
                     String description,
                     boolean useStrictConsistency,
                     IEndpointSnitch snitch,
                     StreamStateStore stateStore)
{
    this.metadata = metadata;
    this.tokens = tokens;
    this.address = address;
    this.description = description;
    this.streamPlan = new StreamPlan(description, true);
    this.useStrictConsistency = useStrictConsistency;
    this.snitch = snitch;
    this.stateStore = stateStore;
    streamPlan.listeners(this.stateStore);
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
public void testAllocateTokensNetworkStrategy(int rackCount, int replicas) throws UnknownHostException
{
    IEndpointSnitch oldSnitch = DatabaseDescriptor.getEndpointSnitch();
    try
    {
        DatabaseDescriptor.setEndpointSnitch(new RackInferringSnitch());
        int vn = 16;
        String ks = "BootStrapperTestNTSKeyspace" + rackCount + replicas;
        String dc = "1";
        SchemaLoader.createKeyspace(ks, KeyspaceParams.nts(dc, replicas, "15", 15), SchemaLoader.standardCFMD(ks, "Standard1"));
        TokenMetadata tm = new TokenMetadata();
        tm.clearUnsafe();
        for (int i = 0; i < rackCount; ++i)
            generateFakeEndpoints(tm, 10, vn, dc, Integer.toString(i));
        InetAddress addr = InetAddress.getByName("127." + dc + ".0.99");
        allocateTokensForNode(vn, ks, tm, addr);
        // Note: Not matching replication factor in second datacentre, but this should not affect us.
    } finally {
        DatabaseDescriptor.setEndpointSnitch(oldSnitch);
    }
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
/**
 * Find a suitable replica as leader for counter update.
 * For now, we pick a random replica in the local DC (or ask the snitch if
 * there is no replica alive in the local DC).
 * TODO: if we track the latency of the counter writes (which makes sense
 * contrarily to standard writes since there is a read involved), we could
 * trust the dynamic snitch entirely, which may be a better solution. It
 * is unclear we want to mix those latencies with read latencies, so this
 * may be a bit involved.
 */
private static InetAddress findSuitableEndpoint(String table, ByteBuffer key, String localDataCenter) throws UnavailableException
{
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(table, key);
    if (endpoints.isEmpty())
        throw new UnavailableException();

    List<InetAddress> localEndpoints = new ArrayList<InetAddress>();
    for (InetAddress endpoint : endpoints)
    {
        if (snitch.getDatacenter(endpoint).equals(localDataCenter))
            localEndpoints.add(endpoint);
    }
    if (localEndpoints.isEmpty())
    {
        // No endpoint in local DC, pick the closest endpoint according to the snitch
        snitch.sortByProximity(FBUtilities.getLocalAddress(), endpoints);
        return endpoints.get(0);
    }
    else
    {
        return localEndpoints.get(FBUtilities.threadLocalRandom().nextInt(localEndpoints.size()));
    }
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
public void gossipSnitchInfo()
{
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    String dc = snitch.getDatacenter(FBUtilities.getBroadcastAddress());
    String rack = snitch.getRack(FBUtilities.getBroadcastAddress());
    Gossiper.instance.addLocalApplicationState(ApplicationState.DC, StorageService.instance.valueFactory.datacenter(dc));
    Gossiper.instance.addLocalApplicationState(ApplicationState.RACK, StorageService.instance.valueFactory.rack(rack));
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
/**
 * Finds living endpoints responsible for the given ranges
 *
 * @param keyspaceName the keyspace ranges belong to
 * @param ranges the ranges to find sources for
 * @return multimap of addresses to ranges the address is responsible for
 */
private Multimap<InetAddress, Range<Token>> getNewSourceRanges(String keyspaceName, Set<Range<Token>> ranges)
{
    InetAddress myAddress = FBUtilities.getBroadcastAddress();
    Multimap<Range<Token>, InetAddress> rangeAddresses = Keyspace.open(keyspaceName).getReplicationStrategy().getRangeAddresses(tokenMetadata.cloneOnlyTokenMap());
    Multimap<InetAddress, Range<Token>> sourceRanges = HashMultimap.create();
    IFailureDetector failureDetector = FailureDetector.instance;

    // find alive sources for our new ranges
    for (Range<Token> range : ranges)
    {
        Collection<InetAddress> possibleRanges = rangeAddresses.get(range);
        IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
        List<InetAddress> sources = snitch.getSortedListByProximity(myAddress, possibleRanges);

        assert (!sources.contains(myAddress));

        for (InetAddress source : sources)
        {
            if (failureDetector.isAlive(source))
            {
                sourceRanges.put(source, range);
                break;
            }
        }
    }
    return sourceRanges;
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
private static Predicate<InetAddress> sameDCPredicateFor(final String dc)
{
    final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    return new Predicate<InetAddress>()
    {
        public boolean apply(InetAddress host)
        {
            return dc.equals(snitch.getDatacenter(host));
        }
    };
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
/**
 * Find a suitable replica as leader for counter update.
 * For now, we pick a random replica in the local DC (or ask the snitch if
 * there is no replica alive in the local DC).
 * TODO: if we track the latency of the counter writes (which makes sense
 * contrarily to standard writes since there is a read involved), we could
 * trust the dynamic snitch entirely, which may be a better solution. It
 * is unclear we want to mix those latencies with read latencies, so this
 * may be a bit involved.
 */
private static InetAddress findSuitableEndpoint(String keyspaceName, ByteBuffer key, String localDataCenter, ConsistencyLevel cl) throws UnavailableException
{
    Keyspace keyspace = Keyspace.open(keyspaceName);
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(keyspace, key);
    if (endpoints.isEmpty())
        // TODO have a way to compute the consistency level
        throw new UnavailableException(cl, cl.blockFor(keyspace), 0);

    List<InetAddress> localEndpoints = new ArrayList<InetAddress>();
    for (InetAddress endpoint : endpoints)
    {
        if (snitch.getDatacenter(endpoint).equals(localDataCenter))
            localEndpoints.add(endpoint);
    }
    if (localEndpoints.isEmpty())
    {
        // No endpoint in local DC, pick the closest endpoint according to the snitch
        snitch.sortByProximity(FBUtilities.getBroadcastAddress(), endpoints);
        return endpoints.get(0);
    }
    else
    {
        return localEndpoints.get(ThreadLocalRandom.current().nextInt(localEndpoints.size()));
    }
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
private static void setupVersion()
{
    String req = "INSERT INTO system.%s (key, release_version, cql_version, thrift_version, native_protocol_version, data_center, rack, partitioner) VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    executeOnceInternal(String.format(req, LOCAL_CF),
                        LOCAL_KEY,
                        FBUtilities.getReleaseVersionString(),
                        QueryProcessor.CQL_VERSION.toString(),
                        cassandraConstants.VERSION,
                        String.valueOf(Server.CURRENT_VERSION),
                        snitch.getDatacenter(FBUtilities.getBroadcastAddress()),
                        snitch.getRack(FBUtilities.getBroadcastAddress()),
                        DatabaseDescriptor.getPartitioner().getClass().getName());
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
private static IEndpointSnitch createEndpointSnitch(String snitchClassName) throws ConfigurationException
{
    if (!snitchClassName.contains("."))
        snitchClassName = "org.apache.cassandra.locator." + snitchClassName;
    IEndpointSnitch snitch = FBUtilities.construct(snitchClassName, "snitch");
    return conf.dynamic_snitch ? new DynamicEndpointSnitch(snitch) : snitch;
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
@BeforeClass
public static void setUp() throws ConfigurationException
{
    IEndpointSnitch snitch = new PropertyFileSnitch();
    DatabaseDescriptor.setEndpointSnitch(snitch);
    Keyspace.setInitialized();
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
private static Collection<InetAddress> getBatchlogEndpoints(String localDataCenter, ConsistencyLevel consistencyLevel) throws UnavailableException
{
    // will include every known node in the DC, including localhost.
    TokenMetadata.Topology topology = StorageService.instance.getTokenMetadata().cloneOnlyTokenMap().getTopology();
    Collection<InetAddress> localMembers = topology.getDatacenterEndpoints().get(localDataCenter);

    // special case for single-node datacenters
    if (localMembers.size() == 1)
        return localMembers;

    // not a single-node cluster - don't count the local node.
    localMembers.remove(FBUtilities.getBroadcastAddress());

    // include only alive nodes
    List<InetAddress> candidates = new ArrayList<InetAddress>(localMembers.size());
    for (InetAddress member : localMembers)
    {
        if (FailureDetector.instance.isAlive(member))
            candidates.add(member);
    }

    if (candidates.isEmpty())
    {
        if (consistencyLevel == ConsistencyLevel.ANY)
            return Collections.singleton(FBUtilities.getBroadcastAddress());

        throw new UnavailableException(ConsistencyLevel.ONE, 1, 0);
    }

    if (candidates.size() > 2)
    {
        IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
        snitch.sortByProximity(FBUtilities.getBroadcastAddress(), candidates);
        candidates = candidates.subList(0, 2);
    }

    return candidates;
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
/**
 * Find a suitable replica as leader for counter update.
 * For now, we pick a random replica in the local DC (or ask the snitch if
 * there is no replica alive in the local DC).
 * TODO: if we track the latency of the counter writes (which makes sense
 * contrarily to standard writes since there is a read involved), we could
 * trust the dynamic snitch entirely, which may be a better solution. It
 * is unclear we want to mix those latencies with read latencies, so this
 * may be a bit involved.
 */
private static InetAddress findSuitableEndpoint(String keyspaceName, ByteBuffer key, String localDataCenter, ConsistencyLevel cl) throws UnavailableException
{
    Keyspace keyspace = Keyspace.open(keyspaceName);
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(keyspace, key);
    if (endpoints.isEmpty())
        // TODO have a way to compute the consistency level
        throw new UnavailableException(cl, cl.blockFor(keyspace), 0);

    List<InetAddress> localEndpoints = new ArrayList<InetAddress>();
    for (InetAddress endpoint : endpoints)
    {
        if (snitch.getDatacenter(endpoint).equals(localDataCenter))
            localEndpoints.add(endpoint);
    }
    if (localEndpoints.isEmpty())
    {
        // No endpoint in local DC, pick the closest endpoint according to the snitch
        snitch.sortByProximity(FBUtilities.getBroadcastAddress(), endpoints);
        return endpoints.get(0);
    }
    else
    {
        return localEndpoints.get(FBUtilities.threadLocalRandom().nextInt(localEndpoints.size()));
    }
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
private static void setupVersion()
{
    String req = "INSERT INTO system.%s (key, release_version, cql_version, thrift_version, native_protocol_version, data_center, rack, partitioner) VALUES ('%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s')";
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    processInternal(String.format(req, LOCAL_CF,
                                     LOCAL_KEY,
                                     FBUtilities.getReleaseVersionString(),
                                     QueryProcessor.CQL_VERSION.toString(),
                                     cassandraConstants.VERSION,
                                     Server.CURRENT_VERSION,
                                     snitch.getDatacenter(FBUtilities.getBroadcastAddress()),
                                     snitch.getRack(FBUtilities.getBroadcastAddress()),
                                     DatabaseDescriptor.getPartitioner().getClass().getName()));
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
public static void persistLocalMetadata()
{
    String req = "INSERT INTO system.%s (" +
                 "key," +
                 "cluster_name," +
                 "release_version," +
                 "cql_version," +
                 "thrift_version," +
                 "native_protocol_version," +
                 "data_center," +
                 "rack," +
                 "partitioner," +
                 "rpc_address," +
                 "broadcast_address," +
                 "listen_address" +
                 ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    executeOnceInternal(String.format(req, LOCAL),
                        LOCAL,
                        DatabaseDescriptor.getClusterName(),
                        FBUtilities.getReleaseVersionString(),
                        QueryProcessor.CQL_VERSION.toString(),
                        cassandraConstants.VERSION,
                        String.valueOf(Server.CURRENT_VERSION),
                        snitch.getDatacenter(FBUtilities.getBroadcastAddress()),
                        snitch.getRack(FBUtilities.getBroadcastAddress()),
                        DatabaseDescriptor.getPartitioner().getClass().getName(),
                        DatabaseDescriptor.getRpcAddress(),
                        FBUtilities.getBroadcastAddress(),
                        FBUtilities.getLocalAddress());
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
private static Collection<InetAddress> getBatchlogEndpoints(String localDataCenter) throws UnavailableException
{
    // will include every known node in the DC, including localhost.
    TokenMetadata.Topology topology = StorageService.instance.getTokenMetadata().cloneOnlyTokenMap().getTopology();
    Collection<InetAddress> localMembers = topology.getDatacenterEndpoints().get(localDataCenter);

    // special case for single-node datacenters
    if (localMembers.size() == 1)
        return localMembers;

    // not a single-node cluster - don't count the local node.
    localMembers.remove(FBUtilities.getBroadcastAddress());

    // include only alive nodes
    List<InetAddress> candidates = new ArrayList<InetAddress>(localMembers.size());
    for (InetAddress member : localMembers)
    {
        if (FailureDetector.instance.isAlive(member))
            candidates.add(member);
    }

    if (candidates.isEmpty())
        throw new UnavailableException(ConsistencyLevel.ONE, 1, 0);

    if (candidates.size() > 2)
    {
        IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
        snitch.sortByProximity(FBUtilities.getBroadcastAddress(), candidates);
        candidates = candidates.subList(0, 2);
    }

    return candidates;
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
/**
 * Find a suitable replica as leader for counter update.
 * For now, we pick a random replica in the local DC (or ask the snitch if
 * there is no replica alive in the local DC).
 * TODO: if we track the latency of the counter writes (which makes sense
 * contrarily to standard writes since there is a read involved), we could
 * trust the dynamic snitch entirely, which may be a better solution. It
 * is unclear we want to mix those latencies with read latencies, so this
 * may be a bit involved.
 */
private static InetAddress findSuitableEndpoint(String tableName, ByteBuffer key, String localDataCenter, ConsistencyLevel cl) throws UnavailableException
{
    Table table = Table.open(tableName);
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(table, key);
    if (endpoints.isEmpty())
        // TODO have a way to compute the consistency level
        throw new UnavailableException(cl, cl.blockFor(table), 0);

    List<InetAddress> localEndpoints = new ArrayList<InetAddress>();
    for (InetAddress endpoint : endpoints)
    {
        if (snitch.getDatacenter(endpoint).equals(localDataCenter))
            localEndpoints.add(endpoint);
    }
    if (localEndpoints.isEmpty())
    {
        // No endpoint in local DC, pick the closest endpoint according to the snitch
        snitch.sortByProximity(FBUtilities.getBroadcastAddress(), endpoints);
        return endpoints.get(0);
    }
    else
    {
        return localEndpoints.get(FBUtilities.threadLocalRandom().nextInt(localEndpoints.size()));
    }
}
 

import org.apache.cassandra.locator.IEndpointSnitch; //导入依赖的package包/类
private static void setupVersion()
{
    String req = "INSERT INTO system.%s (key, release_version, cql_version, thrift_version, data_center, rack, partitioner) VALUES ('%s', '%s', '%s', '%s', '%s', '%s', '%s')";
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    processInternal(String.format(req, LOCAL_CF,
                                     LOCAL_KEY,
                                     FBUtilities.getReleaseVersionString(),
                                     QueryProcessor.CQL_VERSION.toString(),
                                     Constants.VERSION,
                                     snitch.getDatacenter(FBUtilities.getBroadcastAddress()),
                                     snitch.getRack(FBUtilities.getBroadcastAddress()),
                                     DatabaseDescriptor.getPartitioner().getClass().getName()));
}