text:
package org.apache.cassandra.service;

import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import javax.management.MBeanServer;
import javax.management.ObjectName;

import com.google.common.base.Function;
import com.google.common.collect.*;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.filter.ColumnSlice;
import org.apache.cassandra.db.filter.IDiskAtomFilter;
import org.apache.cassandra.db.filter.SliceQueryFilter;
import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.RingPosition;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.util.FastByteArrayOutputStream;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.metrics.ClientRequestMetrics;
import org.apache.cassandra.net.*;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;

public class StorageProxy implements StorageProxyMBean
{
    public static final String MBEAN_NAME = "org.apache.cassandra.db:type=StorageProxy";
    private static final Logger logger = LoggerFactory.getLogger(StorageProxy.class);
    static final boolean OPTIMIZE_LOCAL_REQUESTS = true; // set to false to test messagingservice path on single node

    public static final String UNREACHABLE = "UNREACHABLE";

    private static final WritePerformer standardWritePerformer;
    private static final WritePerformer counterWritePerformer;
    private static final WritePerformer counterWriteOnCoordinatorPerformer;

    public static final StorageProxy instance = new StorageProxy();

    private static volatile boolean hintedHandoffEnabled = DatabaseDescriptor.hintedHandoffEnabled();
    private static volatile int maxHintWindow = DatabaseDescriptor.getMaxHintWindow();
    private static volatile int maxHintsInProgress = 1024 * FBUtilities.getAvailableProcessors();
    private static final AtomicInteger totalHintsInProgress = new AtomicInteger();
    private static final Map<InetAddress, AtomicInteger> hintsInProgress = new MapMaker().concurrencyLevel(1).makeComputingMap(new Function<InetAddress, AtomicInteger>()
    {
        public AtomicInteger apply(InetAddress inetAddress)
        {
            return new AtomicInteger(0);
        }
    });
    private static final AtomicLong totalHints = new AtomicLong();
    private static final ClientRequestMetrics readMetrics = new ClientRequestMetrics("Read");
    private static final ClientRequestMetrics rangeMetrics = new ClientRequestMetrics("RangeSlice");
    private static final ClientRequestMetrics writeMetrics = new ClientRequestMetrics("Write");

    private StorageProxy() {}

    static
    {
        MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
        try
        {
            mbs.registerMBean(new StorageProxy(), new ObjectName(MBEAN_NAME));
        }
        catch (Exception e)
        {
            throw new RuntimeException(e);
        }

        standardWritePerformer = new WritePerformer()
        {
            public void apply(IMutation mutation,
                              Iterable<InetAddress> targets,
                              AbstractWriteResponseHandler responseHandler,
                              String localDataCenter,
                              ConsistencyLevel consistency_level)
            throws IOException, OverloadedException
            {
                assert mutation instanceof RowMutation;
                sendToHintedEndpoints((RowMutation) mutation, targets, responseHandler, localDataCenter, consistency_level);
            }
        };

        /*
         * We execute counter writes in 2 places: either directly in the coordinator node if it is a replica, or
         * in CounterMutationVerbHandler on a replica othewise. The write must be executed on the MUTATION stage
         * but on the latter case, the verb handler already run on the MUTATION stage, so we must not execute the
         * underlying on the stage otherwise we risk a deadlock. Hence two different performer.
         */
        counterWritePerformer = new WritePerformer()
        {
            public void apply(IMutation mutation,
                              Iterable<InetAddress> targets,
                              AbstractWriteResponseHandler responseHandler,
                              String localDataCenter,
                              ConsistencyLevel consistency_level)
            throws IOException
            {
                if (logger.isTraceEnabled())
                    logger.trace("insert writing local & replicate " + mutation.toString(true));

                Runnable runnable = counterWriteTask(mutation, targets, responseHandler, localDataCenter, consistency_level);
                runnable.run();
            }
        };

        counterWriteOnCoordinatorPerformer = new WritePerformer()
        {
            public void apply(IMutation mutation,
                              Iterable<InetAddress> targets,
                              AbstractWriteResponseHandler responseHandler,
                              String localDataCenter,
                              ConsistencyLevel consistency_level)
            throws IOException
            {
                if (logger.isTraceEnabled())
                    logger.trace("insert writing local & replicate " + mutation.toString(true));

                Runnable runnable = counterWriteTask(mutation, targets, responseHandler, localDataCenter, consistency_level);
                StageManager.getStage(Stage.MUTATION).execute(runnable);
            }
        };
    }

    /**
     * Use this method to have these Mutations applied
     * across all replicas. This method will take care
     * of the possibility of a replica being down and hint
     * the data across to some other replica.
     *
     * @param mutations the mutations to be applied across the replicas
     * @param consistency_level the consistency level for the operation
     */
    public static void mutate(Collection<? extends IMutation> mutations, ConsistencyLevel consistency_level)
    throws UnavailableException, OverloadedException, WriteTimeoutException
    {
        Tracing.trace("Determining replicas for mutation");
        logger.trace("Mutations/ConsistencyLevel are {}/{}", mutations, consistency_level);
        final String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());

        long startTime = System.nanoTime();
        List<AbstractWriteResponseHandler> responseHandlers = new ArrayList<AbstractWriteResponseHandler>(mutations.size());

        IMutation mostRecentMutation = null;
        try
        {
            for (IMutation mutation : mutations)
            {
                mostRecentMutation = mutation;
                if (mutation instanceof CounterMutation)
                {
                    responseHandlers.add(mutateCounter((CounterMutation)mutation, localDataCenter));
                }
                else
                {
                    WriteType wt = mutations.size() <= 1 ? WriteType.SIMPLE : WriteType.UNLOGGED_BATCH;
                    responseHandlers.add(performWrite(mutation, consistency_level, localDataCenter, standardWritePerformer, null, wt));
                }
            }

            // wait for writes.  throws TimeoutException if necessary
            for (AbstractWriteResponseHandler responseHandler : responseHandlers)
            {
                responseHandler.get();
            }

        }
        catch (WriteTimeoutException ex)
        {
            writeMetrics.timeouts.mark();
            ClientRequestMetrics.writeTimeouts.inc();
            if (logger.isDebugEnabled())
            {
                List<String> mstrings = new ArrayList<String>(mutations.size());
                for (IMutation mutation : mutations)
                    mstrings.add(mutation.toString(true));
                logger.debug("Write timeout {} for one (or more) of: {}", ex.toString(), mstrings);
            }
            Tracing.trace("Write timeout");
            throw ex;
        }
        catch (UnavailableException e)
        {
            writeMetrics.unavailables.mark();
            ClientRequestMetrics.writeUnavailables.inc();
            Tracing.trace("Unavailable");
            throw e;
        }
        catch (OverloadedException e)
        {
            ClientRequestMetrics.writeUnavailables.inc();
            Tracing.trace("Overloaded");
            throw e;
        }
        catch (IOException e)
        {
            assert mostRecentMutation != null;
            throw new RuntimeException("error writing key " + ByteBufferUtil.bytesToHex(mostRecentMutation.key()), e);
        }
        finally
        {
            writeMetrics.addNano(System.nanoTime() - startTime);
        }
    }

    /**
     * See mutate. Adds additional steps before and after writing a batch.
     * Before writing the batch (but after doing availability check against the FD for the row replicas):
     *      write the entire batch to a batchlog elsewhere in the cluster.
     * After: remove the batchlog entry (after writing hints for the batch rows, if necessary).
     *
     * @param mutations the RowMutations to be applied across the replicas
     * @param consistency_level the consistency level for the operation
     */
    public static void mutateAtomically(Collection<RowMutation> mutations, ConsistencyLevel consistency_level)
    throws UnavailableException, OverloadedException, WriteTimeoutException
    {
        Tracing.trace("Determining replicas for atomic batch");
        long startTime = System.nanoTime();
        logger.trace("Mutations/ConsistencyLevel are {}/{}", mutations, consistency_level);

        List<WriteResponseHandlerWrapper> wrappers = new ArrayList<WriteResponseHandlerWrapper>(mutations.size());
        String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());

        try
        {
            // add a handler for each mutation - includes checking availability, but doesn't initiate any writes, yet
            for (RowMutation mutation : mutations)
            {
                WriteResponseHandlerWrapper wrapper = wrapResponseHandler(mutation, consistency_level, WriteType.BATCH);
                // exit early if we can't fulfill the CL at this time.
                wrapper.handler.assureSufficientLiveNodes();
                wrappers.add(wrapper);
            }

            // write to the batchlog
            Collection<InetAddress> batchlogEndpoints = getBatchlogEndpoints(localDataCenter);
            UUID batchUUID = UUID.randomUUID();
            syncWriteToBatchlog(mutations, batchlogEndpoints, batchUUID);

            // now actually perform the writes and wait for them to complete
            syncWriteBatchedMutations(wrappers, localDataCenter, consistency_level);

            // remove the batchlog entries asynchronously
            asyncRemoveFromBatchlog(batchlogEndpoints, batchUUID);
        }
        catch (UnavailableException e)
        {
            writeMetrics.unavailables.mark();
            ClientRequestMetrics.writeUnavailables.inc();
            Tracing.trace("Unavailable");
            throw e;
        }
        catch (WriteTimeoutException e)
        {
            writeMetrics.timeouts.mark();
            ClientRequestMetrics.writeTimeouts.inc();
            Tracing.trace("Write timeout");
            throw e;
        }
        finally
        {
            writeMetrics.addNano(System.nanoTime() - startTime);
        }
    }

    private static void syncWriteToBatchlog(Collection<RowMutation> mutations, Collection<InetAddress> endpoints, UUID uuid)
    throws WriteTimeoutException
    {
        RowMutation rm = BatchlogManager.getBatchlogMutationFor(mutations, uuid);
        AbstractWriteResponseHandler handler = new WriteResponseHandler(endpoints,
                                                                        Collections.<InetAddress>emptyList(),
                                                                        ConsistencyLevel.ONE,
                                                                        Table.open(Table.SYSTEM_KS),
                                                                        null,
                                                                        WriteType.BATCH_LOG);
        updateBatchlog(rm, endpoints, handler);
        handler.get();
    }

    private static void asyncRemoveFromBatchlog(Collection<InetAddress> endpoints, UUID uuid)
    {
        ColumnFamily cf = EmptyColumns.factory.create(Schema.instance.getCFMetaData(Table.SYSTEM_KS, SystemTable.BATCHLOG_CF));
        cf.delete(new DeletionInfo(FBUtilities.timestampMicros(), (int) (System.currentTimeMillis() / 1000)));
        AbstractWriteResponseHandler handler = new WriteResponseHandler(endpoints,
                                                                        Collections.<InetAddress>emptyList(),
                                                                        ConsistencyLevel.ANY,
                                                                        Table.open(Table.SYSTEM_KS),
                                                                        null,
                                                                        WriteType.SIMPLE);
        RowMutation rm = new RowMutation(Table.SYSTEM_KS, UUIDType.instance.decompose(uuid), cf);
        updateBatchlog(rm, endpoints, handler);
    }

    private static void updateBatchlog(RowMutation rm, Collection<InetAddress> endpoints, AbstractWriteResponseHandler handler)
    {
        if (endpoints.contains(FBUtilities.getBroadcastAddress()))
        {
            assert endpoints.size() == 1;
            insertLocal(rm, handler);
        }
        else
        {
            sendMessagesToOneDC(rm.createMessage(), endpoints, true, handler);
        }
    }

    private static void syncWriteBatchedMutations(List<WriteResponseHandlerWrapper> wrappers,
                                                  String localDataCenter,
                                                  ConsistencyLevel consistencyLevel)
    throws WriteTimeoutException, OverloadedException
    {
        for (WriteResponseHandlerWrapper wrapper : wrappers)
        {
            Iterable<InetAddress> endpoints = Iterables.concat(wrapper.handler.naturalEndpoints, wrapper.handler.pendingEndpoints);
            sendToHintedEndpoints(wrapper.mutation, endpoints, wrapper.handler, localDataCenter, consistencyLevel);
        }

        for (WriteResponseHandlerWrapper wrapper : wrappers)
        {
            wrapper.handler.get();
        }
    }

    /**
     * Perform the write of a mutation given a WritePerformer.
     * Gather the list of write endpoints, apply locally and/or forward the mutation to
     * said write endpoint (deletaged to the actual WritePerformer) and wait for the
     * responses based on consistency level.
     *
     * @param mutation the mutation to be applied
     * @param consistency_level the consistency level for the write operation
     * @param performer the WritePerformer in charge of appliying the mutation
     * given the list of write endpoints (either standardWritePerformer for
     * standard writes or counterWritePerformer for counter writes).
     * @param callback an optional callback to be run if and when the write is
     * successful.
     */
    public static AbstractWriteResponseHandler performWrite(IMutation mutation,
                                                            ConsistencyLevel consistency_level,
                                                            String localDataCenter,
                                                            WritePerformer performer,
                                                            Runnable callback,
                                                            WriteType writeType)
    throws UnavailableException, OverloadedException, IOException
    {
        String table = mutation.getTable();
        AbstractReplicationStrategy rs = Table.open(table).getReplicationStrategy();

        Token tk = StorageService.getPartitioner().getToken(mutation.key());
        List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(table, tk);
        Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, table);

        AbstractWriteResponseHandler responseHandler = rs.getWriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, callback, writeType);

        // exit early if we can't fulfill the CL at this time
        responseHandler.assureSufficientLiveNodes();

        performer.apply(mutation, Iterables.concat(naturalEndpoints, pendingEndpoints), responseHandler, localDataCenter, consistency_level);
        return responseHandler;
    }

    // same as above except does not initiate writes (but does perfrom availability checks).
    private static WriteResponseHandlerWrapper wrapResponseHandler(RowMutation mutation, ConsistencyLevel consistency_level, WriteType writeType)
    {
        AbstractReplicationStrategy rs = Table.open(mutation.getTable()).getReplicationStrategy();
        String table = mutation.getTable();
        Token tk = StorageService.getPartitioner().getToken(mutation.key());
        List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(table, tk);
        Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, table);
        AbstractWriteResponseHandler responseHandler = rs.getWriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, null, writeType);
        return new WriteResponseHandlerWrapper(responseHandler, mutation);
    }

    // used by atomic_batch_mutate to decouple availability check from the write itself, caches consistency level and endpoints.
    private static class WriteResponseHandlerWrapper
    {
        final AbstractWriteResponseHandler handler;
        final RowMutation mutation;

        WriteResponseHandlerWrapper(AbstractWriteResponseHandler handler, RowMutation mutation)
        {
            this.handler = handler;
            this.mutation = mutation;
        }
    }

    /*
     * Replicas are picked manually:
     * - replicas should be alive according to the failure detector
     * - replicas should be in the local datacenter
     * - choose min(2, number of qualifying candiates above)
     * - allow the local node to be the only replica only if it's a single-node cluster
     */
    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;
    }

    /**
     * Send the mutations to the right targets, write it locally if it corresponds or writes a hint when the node
     * is not available.
     *
     * Note about hints:
     *
     * | Hinted Handoff | Consist. Level |
     * | on             |       >=1      | --> wait for hints. We DO NOT notify the handler with handler.response() for hints;
     * | on             |       ANY      | --> wait for hints. Responses count towards consistency.
     * | off            |       >=1      | --> DO NOT fire hints. And DO NOT wait for them to complete.
     * | off            |       ANY      | --> DO NOT fire hints. And DO NOT wait for them to complete.
     *
     * @throws TimeoutException if the hints cannot be written/enqueued
     */
    public static void sendToHintedEndpoints(final RowMutation rm,
                                             Iterable<InetAddress> targets,
                                             AbstractWriteResponseHandler responseHandler,
                                             String localDataCenter,
                                             ConsistencyLevel consistency_level)
    throws OverloadedException
    {
        // Multimap that holds onto all the messages and addresses meant for a specific datacenter
        Map<String, Multimap<MessageOut, InetAddress>> dcMessages = new HashMap<String, Multimap<MessageOut, InetAddress>>();

        for (InetAddress destination : targets)
        {
            // avoid OOMing due to excess hints.  we need to do this check even for "live" nodes, since we can
            // still generate hints for those if it's overloaded or simply dead but not yet known-to-be-dead.
            // The idea is that if we have over maxHintsInProgress hints in flight, this is probably due to
            // a small number of nodes causing problems, so we should avoid shutting down writes completely to
            // healthy nodes.  Any node with no hintsInProgress is considered healthy.
            if (totalHintsInProgress.get() > maxHintsInProgress
                && (hintsInProgress.get(destination).get() > 0 && shouldHint(destination)))
            {
                throw new OverloadedException("Too many in flight hints: " + totalHintsInProgress.get());
            }

            if (FailureDetector.instance.isAlive(destination))
            {
                if (destination.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS)
                {
                    insertLocal(rm, responseHandler);
                }
                else
                {
                    // belongs on a different server
                    if (logger.isTraceEnabled())
                        logger.trace("insert writing key " + ByteBufferUtil.bytesToHex(rm.key()) + " to " + destination);

                    String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(destination);
                    Multimap<MessageOut, InetAddress> messages = dcMessages.get(dc);
                    if (messages == null)
                    {
                        messages = HashMultimap.create();
                        dcMessages.put(dc, messages);
                    }

                    messages.put(rm.createMessage(), destination);
                }
            }
            else
            {
                if (!shouldHint(destination))
                    continue;

                // Schedule a local hint
                submitHint(rm, destination, responseHandler, consistency_level);
            }
        }

        sendMessages(localDataCenter, dcMessages, responseHandler);
    }

    public static Future<Void> submitHint(final RowMutation mutation,
                                          final InetAddress target,
                                          final AbstractWriteResponseHandler responseHandler,
                                          final ConsistencyLevel consistencyLevel)
    {
        // local write that time out should be handled by LocalMutationRunnable
        assert !target.equals(FBUtilities.getBroadcastAddress()) : target;

        HintRunnable runnable = new HintRunnable(target)
        {
            public void runMayThrow() throws IOException
            {
                logger.debug("Adding hint for {}", target);

                writeHintForMutation(mutation, target);
                // Notify the handler only for CL == ANY
                if (responseHandler != null && consistencyLevel == ConsistencyLevel.ANY)
                    responseHandler.response(null);
            }
        };

        return submitHint(runnable);
    }

    private static Future<Void> submitHint(HintRunnable runnable)
    {
        totalHintsInProgress.incrementAndGet();
        hintsInProgress.get(runnable.target).incrementAndGet();
        return (Future<Void>) StageManager.getStage(Stage.MUTATION).submit(runnable);
    }

    public static void writeHintForMutation(RowMutation mutation, InetAddress target) throws IOException
    {
        UUID hostId = StorageService.instance.getTokenMetadata().getHostId(target);
        if ((hostId == null) && (MessagingService.instance().getVersion(target) < MessagingService.VERSION_12))
        {
            logger.warn("Unable to store hint for host with missing ID, {} (old node?)", target.toString());
            return;
        }
        assert hostId != null : "Missing host ID for " + target.getHostAddress();
        RowMutation hintedMutation = HintedHandOffManager.hintFor(mutation, hostId);
        hintedMutation.apply();

        totalHints.incrementAndGet();
    }

    /**
     * for each datacenter, send a message to one node to relay the write to other replicas
     */
    private static void sendMessages(String localDataCenter, Map<String, Multimap<MessageOut, InetAddress>> dcMessages, AbstractWriteResponseHandler handler)
    {
        for (Map.Entry<String, Multimap<MessageOut, InetAddress>> entry: dcMessages.entrySet())
        {
            boolean isLocalDC = entry.getKey().equals(localDataCenter);
            for (Map.Entry<MessageOut, Collection<InetAddress>> messages: entry.getValue().asMap().entrySet())
            {
                MessageOut message = messages.getKey();
                Collection<InetAddress> targets = messages.getValue();
                // a single message object is used for unhinted writes, so clean out any forwards
                // from previous loop iterations
                message = message.withHeaderRemoved(RowMutation.FORWARD_TO);
                sendMessagesToOneDC(message, targets, isLocalDC, handler);
            }
        }
    }

    private static void sendMessagesToOneDC(MessageOut message, Collection<InetAddress> targets, boolean localDC, AbstractWriteResponseHandler handler)
    {
        try
        {
            sendMessagesToOneDCInternal(message, targets, localDC, handler);
        }
        catch (IOException e)
        {
            throw new RuntimeException(e);
        }
    }

    private static void sendMessagesToOneDCInternal(MessageOut message, Collection<InetAddress> targets, boolean localDC, AbstractWriteResponseHandler handler) throws IOException
    {
        Iterator<InetAddress> iter = targets.iterator();
        InetAddress target = iter.next();

        // direct writes to local DC or old Cassandra versions
        // (1.1 knows how to forward old-style String message IDs; updated to int in 2.0)
        if (localDC || MessagingService.instance().getVersion(target) < MessagingService.VERSION_20)
        {
            // yes, the loop and non-loop code here are the same; this is clunky but we want to avoid
            // creating a second iterator since we already have a perfectly good one
            MessagingService.instance().sendRR(message, target, handler);
            while (iter.hasNext())
            {
                target = iter.next();
                MessagingService.instance().sendRR(message, target, handler);
            }
            return;
        }

        // Add all the other destinations of the same message as a FORWARD_HEADER entry
        FastByteArrayOutputStream bos = new FastByteArrayOutputStream();
        DataOutputStream out = new DataOutputStream(bos);
        out.writeInt(targets.size() - 1);
        while (iter.hasNext())
        {
            InetAddress destination = iter.next();
            CompactEndpointSerializationHelper.serialize(destination, out);
            int id = MessagingService.instance().addCallback(handler, message, destination, message.getTimeout());
            out.writeInt(id);
            logger.trace("Adding FWD message to {}@{}", id, destination);
        }
        message = message.withParameter(RowMutation.FORWARD_TO, bos.toByteArray());
        // send the combined message + forward headers
        int id = MessagingService.instance().sendRR(message, target, handler);
        logger.trace("Sending message to {}@{}", id, target);
    }

    private static void insertLocal(final RowMutation rm, final AbstractWriteResponseHandler responseHandler)
    {
        if (logger.isTraceEnabled())
            logger.trace("insert writing local " + rm.toString(true));

        Runnable runnable = new DroppableRunnable(MessagingService.Verb.MUTATION)
        {
            public void runMayThrow() throws IOException
            {
                rm.apply();
                responseHandler.response(null);
            }
        };
        StageManager.getStage(Stage.MUTATION).execute(runnable);
    }

    /**
     * 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, IOException
    {
        InetAddress endpoint = findSuitableEndpoint(cm.getTable(), 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 table = cm.getTable();
            AbstractReplicationStrategy rs = Table.open(table).getReplicationStrategy();
            Token tk = StorageService.getPartitioner().getToken(cm.key());
            List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(table, tk);
            Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, table);

            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);

            if (logger.isTraceEnabled())
                logger.trace("forwarding counter update of key " + ByteBufferUtil.bytesToHex(cm.key()) + " to " + endpoint);
            MessagingService.instance().sendRR(cm.makeMutationMessage(), endpoint, responseHandler);
            return responseHandler;
        }
    }

    /**
     * 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()));
        }
    }

    // Must be called on a replica of the mutation. This replica becomes the
    // leader of this mutation.
    public static AbstractWriteResponseHandler applyCounterMutationOnLeader(CounterMutation cm, String localDataCenter, Runnable callback)
    throws UnavailableException, IOException, OverloadedException
    {
        return performWrite(cm, cm.consistency(), localDataCenter, counterWritePerformer, callback, WriteType.COUNTER);
    }

    // Same as applyCounterMutationOnLeader but must with the difference that it use the MUTATION stage to execute the write (while
    // applyCounterMutationOnLeader assumes it is on the MUTATION stage already)
    public static AbstractWriteResponseHandler applyCounterMutationOnCoordinator(CounterMutation cm, String localDataCenter)
    throws UnavailableException, IOException, OverloadedException
    {
        return performWrite(cm, cm.consistency(), localDataCenter, counterWriteOnCoordinatorPerformer, null, WriteType.COUNTER);
    }

    private static Runnable counterWriteTask(final IMutation mutation,
                                             final Iterable<InetAddress> targets,
                                             final AbstractWriteResponseHandler responseHandler,
                                             final String localDataCenter,
                                             final ConsistencyLevel consistency_level)
    {
        return new LocalMutationRunnable()
        {
            public void runMayThrow() throws IOException
            {
                assert mutation instanceof CounterMutation;
                final CounterMutation cm = (CounterMutation) mutation;

                // apply mutation
                cm.apply();
                responseHandler.response(null);

                // then send to replicas, if any
                final Set<InetAddress> remotes = Sets.difference(ImmutableSet.copyOf(targets), ImmutableSet.of(FBUtilities.getBroadcastAddress()));
                if (cm.shouldReplicateOnWrite() && !remotes.isEmpty())
                {
                    // We do the replication on another stage because it involves a read (see CM.makeReplicationMutation)
                    // and we want to avoid blocking too much the MUTATION stage
                    StageManager.getStage(Stage.REPLICATE_ON_WRITE).execute(new DroppableRunnable(MessagingService.Verb.READ)
                    {
                        public void runMayThrow() throws OverloadedException
                        {
                            // send mutation to other replica
                            sendToHintedEndpoints(cm.makeReplicationMutation(), remotes, responseHandler, localDataCenter, consistency_level);
                        }
                    });
                }
            }
        };
    }

    private static boolean systemTableQuery(List<ReadCommand> cmds)
    {
        for (ReadCommand cmd : cmds)
            if (!cmd.table.equals(Table.SYSTEM_KS))
                return false;
        return true;
    }

    /**
     * Performs the actual reading of a row out of the StorageService, fetching
     * a specific set of column names from a given column family.
     */
    public static List<Row> read(List<ReadCommand> commands, ConsistencyLevel consistency_level)
    throws UnavailableException, IsBootstrappingException, ReadTimeoutException
    {
        if (StorageService.instance.isBootstrapMode() && !systemTableQuery(commands))
        {
            readMetrics.unavailables.mark();
            ClientRequestMetrics.readUnavailables.inc();
            throw new IsBootstrappingException();
        }
        long startTime = System.nanoTime();
        List<Row> rows = null;
        try
        {
            rows = fetchRows(commands, consistency_level);
        }
        catch (UnavailableException e)
        {
            readMetrics.unavailables.mark();
            ClientRequestMetrics.readUnavailables.inc();
            throw e;
        }
        catch (ReadTimeoutException e)
        {
            readMetrics.timeouts.mark();
            ClientRequestMetrics.readTimeouts.inc();
            throw e;
        }
        finally
        {
            readMetrics.addNano(System.nanoTime() - startTime);
        }
        return rows;
    }

    /**
     * This function executes local and remote reads, and blocks for the results:
     *
     * 1. Get the replica locations, sorted by response time according to the snitch
     * 2. Send a data request to the closest replica, and digest requests to either
     *    a) all the replicas, if read repair is enabled
     *    b) the closest R-1 replicas, where R is the number required to satisfy the ConsistencyLevel
     * 3. Wait for a response from R replicas
     * 4. If the digests (if any) match the data return the data
     * 5. else carry out read repair by getting data from all the nodes.
     */
    private static List<Row> fetchRows(List<ReadCommand> initialCommands, ConsistencyLevel consistency_level)
    throws UnavailableException, ReadTimeoutException
    {
        List<Row> rows = new ArrayList<Row>(initialCommands.size());
        List<ReadCommand> commandsToRetry = Collections.emptyList();

        do
        {
            List<ReadCommand> commands = commandsToRetry.isEmpty() ? initialCommands : commandsToRetry;
            AbstractReadExecutor[] readExecutors = new AbstractReadExecutor[commands.size()];

            if (!commandsToRetry.isEmpty())
                logger.debug("Retrying {} commands", commandsToRetry.size());

            // send out read requests
            for (int i = 0; i < commands.size(); i++)
            {
                ReadCommand command = commands.get(i);
                assert !command.isDigestQuery();
                logger.trace("Command/ConsistencyLevel is {}/{}", command, consistency_level);

                AbstractReadExecutor exec = AbstractReadExecutor.getReadExecutor(command, consistency_level);
                exec.executeAsync();
                readExecutors[i] = exec;
            }

            for (AbstractReadExecutor exec: readExecutors)
                exec.speculate();

            // read results and make a second pass for any digest mismatches
            List<ReadCommand> repairCommands = null;
            List<ReadCallback<ReadResponse, Row>> repairResponseHandlers = null;
            for (AbstractReadExecutor exec: readExecutors)
            {
                try
                {
                    Row row = exec.get();
                    if (row != null)
                    {
                        exec.command.maybeTrim(row);
                        rows.add(row);
                    }
                    if (logger.isDebugEnabled())
                        logger.debug("Read: " + (System.currentTimeMillis() - exec.handler.startTime) + " ms.");
                }
                catch (DigestMismatchException ex)
                {
                    logger.trace("Digest mismatch: {}", ex);
                    // Do a full data read to resolve the correct response (and repair node that need be)
                    RowDataResolver resolver = new RowDataResolver(exec.command.table, exec.command.key, exec.command.filter());
                    ReadCallback<ReadResponse, Row> repairHandler = exec.handler.withNewResolver(resolver);

                    if (repairCommands == null)
                    {
                        repairCommands = new ArrayList<ReadCommand>();
                        repairResponseHandlers = new ArrayList<ReadCallback<ReadResponse, Row>>();
                    }
                    repairCommands.add(exec.command);
                    repairResponseHandlers.add(repairHandler);

                    MessageOut<ReadCommand> message = exec.command.createMessage();
                    for (InetAddress endpoint : exec.handler.endpoints)
                        MessagingService.instance().sendRR(message, endpoint, repairHandler);
                }
            }

            if (commandsToRetry != Collections.EMPTY_LIST)
                commandsToRetry.clear();

            // read the results for the digest mismatch retries
            if (repairResponseHandlers != null)
            {
                for (int i = 0; i < repairCommands.size(); i++)
                {
                    ReadCommand command = repairCommands.get(i);
                    ReadCallback<ReadResponse, Row> handler = repairResponseHandlers.get(i);

                    Row row;
                    try
                    {
                        row = handler.get();
                    }
                    catch (DigestMismatchException e)
                    {
                        throw new AssertionError(e); // full data requested from each node here, no digests should be sent
                    }

                    RowDataResolver resolver = (RowDataResolver)handler.resolver;
                    try
                    {
                        // wait for the repair writes to be acknowledged, to minimize impact on any replica that's
                        // behind on writes in case the out-of-sync row is read multiple times in quick succession
                        FBUtilities.waitOnFutures(resolver.repairResults, DatabaseDescriptor.getWriteRpcTimeout());
                    }
                    catch (TimeoutException e)
                    {
                        int blockFor = consistency_level.blockFor(Table.open(command.getKeyspace()));
                        throw new ReadTimeoutException(consistency_level, blockFor, blockFor, true);
                    }

                    // retry any potential short reads
                    ReadCommand retryCommand = command.maybeGenerateRetryCommand(resolver, row);
                    if (retryCommand != null)
                    {
                        logger.debug("Issuing retry for read command");
                        if (commandsToRetry == Collections.EMPTY_LIST)
                            commandsToRetry = new ArrayList<ReadCommand>();
                        commandsToRetry.add(retryCommand);
                        continue;
                    }

                    if (row != null)
                    {
                        command.maybeTrim(row);
                        rows.add(row);
                    }
                }
            }
        } while (!commandsToRetry.isEmpty());

        return rows;
    }

    static class LocalReadRunnable extends DroppableRunnable
    {
        private final ReadCommand command;
        private final ReadCallback<ReadResponse, Row> handler;
        private final long start = System.currentTimeMillis();

        LocalReadRunnable(ReadCommand command, ReadCallback<ReadResponse, Row> handler)
        {
            super(MessagingService.Verb.READ);
            this.command = command;
            this.handler = handler;
        }

        protected void runMayThrow()
        {
            logger.trace("LocalReadRunnable reading {}", command);

            Table table = Table.open(command.table);
            Row r = command.getRow(table);
            ReadResponse result = ReadVerbHandler.getResponse(command, r);
            MessagingService.instance().addLatency(FBUtilities.getBroadcastAddress(), System.currentTimeMillis() - start);
            handler.response(result);
        }
    }

    static class LocalRangeSliceRunnable extends DroppableRunnable
    {
        private final RangeSliceCommand command;
        private final ReadCallback<RangeSliceReply, Iterable<Row>> handler;
        private final long start = System.currentTimeMillis();

        LocalRangeSliceRunnable(RangeSliceCommand command, ReadCallback<RangeSliceReply, Iterable<Row>> handler)
        {
            super(MessagingService.Verb.READ);
            this.command = command;
            this.handler = handler;
        }

        protected void runMayThrow() throws ExecutionException, InterruptedException
        {
            logger.trace("LocalReadRunnable reading {}", command);

            RangeSliceReply result = new RangeSliceReply(RangeSliceVerbHandler.executeLocally(command));
            MessagingService.instance().addLatency(FBUtilities.getBroadcastAddress(), System.currentTimeMillis() - start);
            handler.response(result);
        }
    }

    public static List<InetAddress> getLiveSortedEndpoints(Table table, ByteBuffer key)
    {
        return getLiveSortedEndpoints(table, StorageService.instance.getPartitioner().decorateKey(key));
    }

    private static List<InetAddress> getLiveSortedEndpoints(Table table, RingPosition pos)
    {
        List<InetAddress> liveEndpoints = StorageService.instance.getLiveNaturalEndpoints(table, pos);
        DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getBroadcastAddress(), liveEndpoints);
        return liveEndpoints;
    }

    private static List<InetAddress> intersection(List<InetAddress> l1, List<InetAddress> l2)
    {
        // Note: we don't use Guava Sets.intersection() for 3 reasons:
        //   1) retainAll would be inefficient if l1 and l2 are large but in practice both are the replicas for a range and
        //   so will be very small (< RF). In that case, retainAll is in fact more efficient.
        //   2) we do ultimately need a list so converting everything to sets don't make sense
        //   3) l1 and l2 are sorted by proximity. The use of retainAll  maintain that sorting in the result, while using sets wouldn't.
        List<InetAddress> inter = new ArrayList<InetAddress>(l1);
        inter.retainAll(l2);
        return inter;
    }

    public static List<Row> getRangeSlice(RangeSliceCommand command, ConsistencyLevel consistency_level)
    throws UnavailableException, ReadTimeoutException
    {
        Tracing.trace("Determining replicas to query");
        logger.trace("Command/ConsistencyLevel is {}/{}", command.toString(), consistency_level);
        long startTime = System.nanoTime();

        Table table = Table.open(command.keyspace);
        List<Row> rows;
        // now scan until we have enough results
        try
        {
            IDiskAtomFilter commandPredicate = command.predicate;

            int cql3RowCount = 0;
            rows = new ArrayList<Row>();
            List<AbstractBounds<RowPosition>> ranges = getRestrictedRanges(command.range);
            int i = 0;
            AbstractBounds<RowPosition> nextRange = null;
            List<InetAddress> nextEndpoints = null;
            List<InetAddress> nextFilteredEndpoints = null;
            while (i < ranges.size())
            {
                AbstractBounds<RowPosition> range = nextRange == null
                                                  ? ranges.get(i)
                                                  : nextRange;
                List<InetAddress> liveEndpoints = nextEndpoints == null
                                                ? getLiveSortedEndpoints(table, range.right)
                                                : nextEndpoints;
                List<InetAddress> filteredEndpoints = nextFilteredEndpoints == null
                                                    ? consistency_level.filterForQuery(table, liveEndpoints)
                                                    : nextFilteredEndpoints;
                ++i;

                // getRestrictedRange has broken the queried range into per-[vnode] token ranges, but this doesn't take
                // the replication factor into account. If the intersection of live endpoints for 2 consecutive ranges
                // still meets the CL requirements, then we can merge both ranges into the same RangeSliceCommand.
                while (i < ranges.size())
                {
                    nextRange = ranges.get(i);
                    nextEndpoints = getLiveSortedEndpoints(table, nextRange.right);
                    nextFilteredEndpoints = consistency_level.filterForQuery(table, nextEndpoints);

                    /*
                     * If the current range right is the min token, we should stop merging because CFS.getRangeSlice
                     * don't know how to deal with a wrapping range.
                     * Note: it would be slightly more efficient to have CFS.getRangeSlice on the destination nodes unwraps
                     * the range if necessary and deal with it. However, we can't start sending wrapped range without breaking
                     * wire compatibility, so It's likely easier not to bother;
                     */
                    if (range.right.isMinimum())
                        break;

                    List<InetAddress> merged = intersection(liveEndpoints, nextEndpoints);

                    // Check if there is enough endpoint for the merge to be possible.
                    if (!consistency_level.isSufficientLiveNodes(table, merged))
                        break;

                    List<InetAddress> filteredMerged = consistency_level.filterForQuery(table, merged);

                    // Estimate whether merging will be a win or not
                    if (!DatabaseDescriptor.getEndpointSnitch().isWorthMergingForRangeQuery(filteredMerged, filteredEndpoints, nextFilteredEndpoints))
                        break;

                    // If we get there, merge this range and the next one
                    range = range.withNewRight(nextRange.right);
                    liveEndpoints = merged;
                    filteredEndpoints = filteredMerged;
                    ++i;
                }

                RangeSliceCommand nodeCmd = new RangeSliceCommand(command.keyspace,
                                                                  command.column_family,
                                                                  commandPredicate,
                                                                  range,
                                                                  command.row_filter,
                                                                  command.maxResults,
                                                                  command.countCQL3Rows,
                                                                  command.isPaging);

                // collect replies and resolve according to consistency level
                RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(nodeCmd.keyspace);
                ReadCallback<RangeSliceReply, Iterable<Row>> handler = new ReadCallback(resolver, consistency_level, nodeCmd, filteredEndpoints);
                handler.assureSufficientLiveNodes();
                resolver.setSources(filteredEndpoints);
                if (filteredEndpoints.size() == 1
                    && filteredEndpoints.get(0).equals(FBUtilities.getBroadcastAddress())
                    && OPTIMIZE_LOCAL_REQUESTS)
                {
                    logger.trace("reading data locally");
                    StageManager.getStage(Stage.READ).execute(new LocalRangeSliceRunnable(nodeCmd, handler));
                }
                else
                {
                    MessageOut<RangeSliceCommand> message = nodeCmd.createMessage();
                    for (InetAddress endpoint : filteredEndpoints)
                    {
                        MessagingService.instance().sendRR(message, endpoint, handler);
                        logger.trace("reading {} from {}", nodeCmd, endpoint);
                    }
                }

                try
                {
                    for (Row row : handler.get())
                    {
                        rows.add(row);
                        if (nodeCmd.countCQL3Rows)
                            cql3RowCount += row.getLiveCount(commandPredicate);
                        logger.trace("range slices read {}", row.key);
                    }
                    FBUtilities.waitOnFutures(resolver.repairResults, DatabaseDescriptor.getWriteRpcTimeout());
                }
                catch (TimeoutException ex)
                {
                    logger.debug("Range slice timeout: {}", ex.toString());
                    // We actually got all response at that point
                    int blockFor = consistency_level.blockFor(table);
                    throw new ReadTimeoutException(consistency_level, blockFor, blockFor, true);
                }
                catch (DigestMismatchException e)
                {
                    throw new AssertionError(e); // no digests in range slices yet
                }

                // if we're done, great, otherwise, move to the next range
                int count = nodeCmd.countCQL3Rows ? cql3RowCount : rows.size();
                if (count >= nodeCmd.maxResults)
                    break;

                // if we are paging and already got some rows, reset the column filter predicate,
                // so we start iterating the next row from the first column
                if (!rows.isEmpty() && command.isPaging)
                {
                    // We only allow paging with a slice filter (doesn't make sense otherwise anyway)
                    assert commandPredicate instanceof SliceQueryFilter;
                    commandPredicate = ((SliceQueryFilter)commandPredicate).withUpdatedSlices(ColumnSlice.ALL_COLUMNS_ARRAY);
                }
            }
        }
        finally
        {
            rangeMetrics.addNano(System.nanoTime() - startTime);
        }
        return trim(command, rows);
    }

    private static List<Row> trim(RangeSliceCommand command, List<Row> rows)
    {
        // When countCQL3Rows, we let the caller trim the result.
        if (command.countCQL3Rows)
            return rows;
        else
            return rows.size() > command.maxResults ? rows.subList(0, command.maxResults) : rows;
    }

    /**
     * initiate a request/response session with each live node to check whether or not everybody is using the same
     * migration id. This is useful for determining if a schema change has propagated through the cluster. Disagreement
     * is assumed if any node fails to respond.
     */
    public static Map<String, List<String>> describeSchemaVersions()
    {
        final String myVersion = Schema.instance.getVersion().toString();
        final Map<InetAddress, UUID> versions = new ConcurrentHashMap<InetAddress, UUID>();
        final Set<InetAddress> liveHosts = Gossiper.instance.getLiveMembers();
        final CountDownLatch latch = new CountDownLatch(liveHosts.size());

        IAsyncCallback<UUID> cb = new IAsyncCallback<UUID>()
        {
            public void response(MessageIn<UUID> message)
            {
                // record the response from the remote node.
                logger.trace("Received schema check response from {}", message.from.getHostAddress());
                versions.put(message.from, message.payload);
                latch.countDown();
            }

            public boolean isLatencyForSnitch()
            {
                return false;
            }
        };
        // an empty message acts as a request to the SchemaCheckVerbHandler.
        MessageOut message = new MessageOut(MessagingService.Verb.SCHEMA_CHECK);
        for (InetAddress endpoint : liveHosts)
            MessagingService.instance().sendRR(message, endpoint, cb);

        try
        {
            // wait for as long as possible. timeout-1s if possible.
            latch.await(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
        }
        catch (InterruptedException ex)
        {
            throw new AssertionError("This latch shouldn't have been interrupted.");
        }

        logger.trace("My version is {}", myVersion);

        // maps versions to hosts that are on that version.
        Map<String, List<String>> results = new HashMap<String, List<String>>();
        Iterable<InetAddress> allHosts = Iterables.concat(Gossiper.instance.getLiveMembers(), Gossiper.instance.getUnreachableMembers());
        for (InetAddress host : allHosts)
        {
            UUID version = versions.get(host);
            String stringVersion = version == null ? UNREACHABLE : version.toString();
            List<String> hosts = results.get(stringVersion);
            if (hosts == null)
            {
                hosts = new ArrayList<String>();
                results.put(stringVersion, hosts);
            }
            hosts.add(host.getHostAddress());
        }

        // we're done: the results map is ready to return to the client.  the rest is just debug logging:
        if (results.get(UNREACHABLE) != null)
            logger.debug("Hosts not in agreement. Didn't get a response from everybody: {}", StringUtils.join(results.get(UNREACHABLE), ","));
        for (Map.Entry<String, List<String>> entry : results.entrySet())
        {
            // check for version disagreement. log the hosts that don't agree.
            if (entry.getKey().equals(UNREACHABLE) || entry.getKey().equals(myVersion))
                continue;
            for (String host : entry.getValue())
                logger.debug("{} disagrees ({})", host, entry.getKey());
        }
        if (results.size() == 1)
            logger.debug("Schemas are in agreement.");

        return results;
    }

    /**
     * Compute all ranges we're going to query, in sorted order. Nodes can be replica destinations for many ranges,
     * so we need to restrict each scan to the specific range we want, or else we'd get duplicate results.
     */
    static <T extends RingPosition> List<AbstractBounds<T>> getRestrictedRanges(final AbstractBounds<T> queryRange)
    {
        // special case for bounds containing exactly 1 (non-minimum) token
        if (queryRange instanceof Bounds && queryRange.left.equals(queryRange.right) && !queryRange.left.isMinimum(StorageService.getPartitioner()))
        {
            logger.trace("restricted single token match for query {}", queryRange);
            return Collections.singletonList(queryRange);
        }

        TokenMetadata tokenMetadata = StorageService.instance.getTokenMetadata();

        List<AbstractBounds<T>> ranges = new ArrayList<AbstractBounds<T>>();
        // divide the queryRange into pieces delimited by the ring and minimum tokens
        Iterator<Token> ringIter = TokenMetadata.ringIterator(tokenMetadata.sortedTokens(), queryRange.left.getToken(), true);
        AbstractBounds<T> remainder = queryRange;
        while (ringIter.hasNext())
        {
            /*
             * remainder can be a range/bounds of token _or_ keys and we want to split it with a token:
             *   - if remainder is tokens, then we'll just split using the provided token.
             *   - if remainder is keys, we want to split using token.upperBoundKey. For instance, if remainder
             *     is [DK(10, 'foo'), DK(20, 'bar')], and we have 3 nodes with tokens 0, 15, 30. We want to
             *     split remainder to A=[DK(10, 'foo'), 15] and B=(15, DK(20, 'bar')]. But since we can't mix
             *     tokens and keys at the same time in a range, we uses 15.upperBoundKey() to have A include all
             *     keys having 15 as token and B include none of those (since that is what our node owns).
             * asSplitValue() abstracts that choice.
             */
            Token upperBoundToken = ringIter.next();
            T upperBound = (T)upperBoundToken.upperBound(queryRange.left.getClass());
            if (!remainder.left.equals(upperBound) && !remainder.contains(upperBound))
                // no more splits
                break;
            Pair<AbstractBounds<T>,AbstractBounds<T>> splits = remainder.split(upperBound);
            if (splits == null)
                continue;

            ranges.add(splits.left);
            remainder = splits.right;
        }
        ranges.add(remainder);
        if (logger.isDebugEnabled())
            logger.trace("restricted ranges for query {} are {}", queryRange, ranges);

        return ranges;
    }

    public long getReadOperations()
    {
        return readMetrics.latency.count();
    }

    public long getTotalReadLatencyMicros()
    {
        return readMetrics.totalLatency.count();
    }

    public double getRecentReadLatencyMicros()
    {
        return readMetrics.getRecentLatency();
    }

    public long[] getTotalReadLatencyHistogramMicros()
    {
        return readMetrics.totalLatencyHistogram.getBuckets(false);
    }

    public long[] getRecentReadLatencyHistogramMicros()
    {
        return readMetrics.recentLatencyHistogram.getBuckets(true);
    }

    public long getRangeOperations()
    {
        return rangeMetrics.latency.count();
    }

    public long getTotalRangeLatencyMicros()
    {
        return rangeMetrics.totalLatency.count();
    }

    public double getRecentRangeLatencyMicros()
    {
        return rangeMetrics.getRecentLatency();
    }

    public long[] getTotalRangeLatencyHistogramMicros()
    {
        return rangeMetrics.totalLatencyHistogram.getBuckets(false);
    }

    public long[] getRecentRangeLatencyHistogramMicros()
    {
        return rangeMetrics.recentLatencyHistogram.getBuckets(true);
    }

    public long getWriteOperations()
    {
        return writeMetrics.latency.count();
    }

    public long getTotalWriteLatencyMicros()
    {
        return writeMetrics.totalLatency.count();
    }

    public double getRecentWriteLatencyMicros()
    {
        return writeMetrics.getRecentLatency();
    }

    public long[] getTotalWriteLatencyHistogramMicros()
    {
        return writeMetrics.totalLatencyHistogram.getBuckets(false);
    }

    public long[] getRecentWriteLatencyHistogramMicros()
    {
        return writeMetrics.recentLatencyHistogram.getBuckets(true);
    }

    public boolean getHintedHandoffEnabled()
    {
        return DatabaseDescriptor.hintedHandoffEnabled();
    }

    public void setHintedHandoffEnabled(boolean b)
    {
        DatabaseDescriptor.setHintedHandoffEnabled(b);
    }

    public int getMaxHintWindow()
    {
        return DatabaseDescriptor.getMaxHintWindow();
    }

    public void setMaxHintWindow(int ms)
    {
        DatabaseDescriptor.setMaxHintWindow(ms);
    }

    public static boolean shouldHint(InetAddress ep)
    {
        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);
            logger.trace("not hinting {} which has been down {}ms", ep, Gossiper.instance.getEndpointDowntime(ep));
        }
        return !hintWindowExpired;
    }

    /**
     * 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 (isAnyHostDown())
        {
            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.getLiveMembers();
        int blockFor = allEndpoints.size();
        final TruncateResponseHandler responseHandler = new TruncateResponseHandler(blockFor);

        // Send out the truncate calls and track the responses with the callbacks.
        logger.trace("Starting to send 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
        logger.trace("Sent all truncate messages, now waiting for {} responses", blockFor);
        responseHandler.get();
    }

    /**
     * Asks the gossiper if there are any nodes that are currently down.
     * @return true if the gossiper thinks all nodes are up.
     */
    private static boolean isAnyHostDown()
    {
        return !Gossiper.instance.getUnreachableMembers().isEmpty();
    }

    public interface WritePerformer
    {
        public void apply(IMutation mutation, Iterable<InetAddress> targets, AbstractWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException, OverloadedException;
    }

    /**
     * A Runnable that aborts if it doesn't start running before it times out
     */
    private static abstract class DroppableRunnable implements Runnable
    {
        private final long constructionTime = System.currentTimeMillis();
        private final MessagingService.Verb verb;

        public DroppableRunnable(MessagingService.Verb verb)
        {
            this.verb = verb;
        }

        public final void run()
        {
            if (System.currentTimeMillis() > constructionTime + DatabaseDescriptor.getTimeout(verb))
            {
                MessagingService.instance().incrementDroppedMessages(verb);
                return;
            }

            try
            {
                runMayThrow();
            }
            catch (Exception e)
            {
                throw new RuntimeException(e);
            }
        }

        abstract protected void runMayThrow() throws Exception;
    }

    /**
     * Like DroppableRunnable, but if it aborts, it will rerun (on the mutation stage) after
     * marking itself as a hint in progress so that the hint backpressure mechanism can function.
     */
    private static abstract class LocalMutationRunnable implements Runnable
    {
        private final long constructionTime = System.currentTimeMillis();

        public final void run()
        {
            if (System.currentTimeMillis() > constructionTime + DatabaseDescriptor.getTimeout(MessagingService.Verb.MUTATION))
            {
                MessagingService.instance().incrementDroppedMessages(MessagingService.Verb.MUTATION);
                HintRunnable runnable = new HintRunnable(FBUtilities.getBroadcastAddress())
                {
                    protected void runMayThrow() throws Exception
                    {
                        LocalMutationRunnable.this.runMayThrow();
                    }
                };
                submitHint(runnable);
                return;
            }

            try
            {
                runMayThrow();
            }
            catch (Exception e)
            {
                throw new RuntimeException(e);
            }
        }

        abstract protected void runMayThrow() throws Exception;
    }

    /**
     * HintRunnable will decrease totalHintsInProgress and targetHints when finished.
     * It is the caller's responsibility to increment them initially.
     */
    private abstract static class HintRunnable implements Runnable
    {
        public final InetAddress target;

        protected HintRunnable(InetAddress target)
        {
            this.target = target;
        }

        public void run()
        {
            try
            {
                runMayThrow();
            }
            catch (Exception e)
            {
                throw new RuntimeException(e);
            }
            finally
            {
                totalHintsInProgress.decrementAndGet();
                hintsInProgress.get(target).decrementAndGet();
            }
        }

        abstract protected void runMayThrow() throws Exception;
    }

    public long getTotalHints()
    {
        return totalHints.get();
    }

    public int getMaxHintsInProgress()
    {
        return maxHintsInProgress;
    }

    public void setMaxHintsInProgress(int qs)
    {
        maxHintsInProgress = qs;
    }

    public int getHintsInProgress()
    {
        return totalHintsInProgress.get();
    }

    public void verifyNoHintsInProgress()
    {
        if (getHintsInProgress() > 0)
            logger.warn("Some hints were not written before shutdown.  This is not supposed to happen.  You should (a) run repair, and (b) file a bug report");
    }

    public Long getRpcTimeout() { return DatabaseDescriptor.getRpcTimeout(); }
    public void setRpcTimeout(Long timeoutInMillis) { DatabaseDescriptor.setRpcTimeout(timeoutInMillis); }

    public Long getReadRpcTimeout() { return DatabaseDescriptor.getReadRpcTimeout(); }
    public void setReadRpcTimeout(Long timeoutInMillis) { DatabaseDescriptor.setReadRpcTimeout(timeoutInMillis); }

    public Long getWriteRpcTimeout() { return DatabaseDescriptor.getWriteRpcTimeout(); }
    public void setWriteRpcTimeout(Long timeoutInMillis) { DatabaseDescriptor.setWriteRpcTimeout(timeoutInMillis); }

    public Long getRangeRpcTimeout() { return DatabaseDescriptor.getRangeRpcTimeout(); }
    public void setRangeRpcTimeout(Long timeoutInMillis) { DatabaseDescriptor.setRangeRpcTimeout(timeoutInMillis); }

    public Long getTruncateRpcTimeout() { return DatabaseDescriptor.getTruncateRpcTimeout(); }
    public void setTruncateRpcTimeout(Long timeoutInMillis) { DatabaseDescriptor.setTruncateRpcTimeout(timeoutInMillis); }
}

redpajama_set_name: RedPajamaGithub


The American Athletic Conference football divisions were officially announced today by conference commissioner Mike Aresco.
Beginning with the 2015 season after Navy joins the conference, the American will have East and West Divisions with six teams in each.
The conference will continue to play an eight-game football schedule. Each team will play five games against the other teams in their division plus three from the opposite division.
The division winners will then meet in the American Athletic Conference Football Championship Game, which will be played at the home site of the higher ranked team.
"This is an exciting step in the continued development of our conference," said Aresco. "With our performance on the field and the exposure we received from our media partners, I couldn't be more satisfied with our first year of American Athletic Conference football. Creating this divisional format will allow us to build upon that success. Our fans will enjoy our divisional play and the championship game that will be established in 2015.
I would like to know why Navy is in the West. Cincinnati would have made more sense there.
They reportedly asked to be there so they could play more games in Texas.
Navy came out and said they wanted to be in the west.
Switching Navy and Cincinnati would make more sense geographically. I bet Cincinnati demanded to be in the East to tie into Florida recruiting.
I think both schools are fine with that. Cincy heavily recruits Florida, Navy recruits Texas.
As a Navy fan, I really hope we don't have to put up with this conference experiment for more than a season or two. No offense to any of the teams in the conference, but not a single one of them excites me. I'm happy to play any of them occasionally, but Tulane, Tulsa, Memphis, SMU and Houston on the schedule every single year… no thanks. Plus with 8 conference games and Army, Air Force and ND on the schedule every year, we have almost zero schedule flexibility.
I agree Dan. No need for Navy to do this. They're Navy and would be just fine as an independent, much better than BYU. Patriotism alone guarantees them the ability to schedule well. Same for Army and Air Force to a certain extent. No need for Navy to join a conference. Next to ND, the most attractive independent.
Navy joined the conference because it means more $. Get used to it. They are not leaving.
Navy is very happy to be in the West Division so they can play more games in Texas.
I wish the American would add UMASS and Rice, and stay at an 8 game Conference schedule, and require one game with the Big Group of 5 every year as well?
The American would be WAY better off to add Air Force and Army. Bring the Commander-In-Chief trophy in conference. Perfecting branding move.
Air Force declined to join conference previously.
Navy has done what they could to try and stay relevant. Army has seemingly given up scheduling down to get a few wins while waiting for the Navy game. Air Force is perfectly happy in the MWC, thank you very much.
All I know is I'm pretty excited about this AAC divisional breakdown, especially looking at the East. Those who think Navy won't have real competitive conference play year in and year out need to think twice and worse case scenario Navy can always stack up its out of conference schedule by adding a p5 team every year. I feel every team in the AAC should play 2-3 p5 games every season and just do what they can to get talent in the conference and even try to persuade BYU and Boise St. to join. If that happened, oh man! Don't underestimate this conference people, if you do you'll probably look as dumb as ESPN annalists looked when they made their predictions for the 2013 Tostitos Fiesta Bowl.
No way one would split ECU and UCF. That's one of the better new rivalries.
Tough slate for the eastern division teams. Temple is really going to struggle in this set up.
UH doesn't cover the Houston market. They typically draw 20,000 in a city of 4 million.
UTSA would be a great add if UConn left or the conference wanted to expand further. San Antonio is a football crazy area with no other competition in the city (no NFL or major college team in the city). To the guy that said Houston only draws 20k in a city of 4 million – It's not just about attendance, it's about television markets. I would venture to say most of the people in Houston are probably aTm, Texas, Baylor, Texas Tech, etc fans first – then Houston fans after that. A lot may not spend money to go to the game, but would watch if Houston was playing SMU on a cable channel. I live in Tulsa & can tell you that first hand. A ton of people here are Oklahoma State or Oklahoma fans, but also want Tulsa to win (except when they play each other). For instance, I will watch Tulsa games on tv, but rarely go to them in person (unless it's a Thursday or Friday night game). Just saying it's not a simple – if I'm a fan of school x (big 5 school) then I won't be interested in following school y's season (non AQ smaller school).
UH alumni only funded 50% of the new stadium. Increased student fees were slapped on future students or there would have been no stadium. Naming rights have been cut in half and still no takers. Texas, aTm, Baylor, Texas Tech and UTSA fans do not care about UH football.
Well my larger point is it's about television markets (Why Houston, SMU were added specifically). And even if only 5-10% or so of the residents of Dallas or Houston are "fans" – it still represents a larger population versus the metropolitan area surrounding Marshall or Louisiana Tech. You said UH doesn't cover the Houston market – what market would it cover then? The American conference isn't the SEC, nor will it ever be. I'm merely stating it's not as black and white as you think – and that the mere presence of a university in a large city (even if it's playing 5th fiddle) is worth something to the conference.
Yes, I think Navy and Cinncinnati are in the wrong divisions but we will get used to it. What I see is a major discrepancy in the balance of power, at least initially, with the East being a lot better overall than the west.
The West is two urban schools and four private. Total imbalance when compared to the East.
Navy grad and long time season ticket holder…not super excited about the move because we are virtually locked in with 8 conference games plus, AFA, ND and Army pretty much permanently scheduled. Leaves room for one game and I am imagining that would be an FCS team someplace near the East Coast. No games – or only one game – west of the Rockies every year seems to hamper our ability to get exposure out west. We're not recruiting for football as much as we are for kids who will serve in the Navy and Marine Corps after graduation. We might get better football recruit exposure in Texas where we recruit specifically for that, but not for future grads west of Colorado Springs. I get that it's about the do-re-mi, but I am a little bummed that we are so locked down. Hopefully, as more teams join the AAC, we can drop the three out of division games and get a little more flexibility to play on the West Coast or even a Big ten Team.
BYU might join the AAC at some point if offered. After going indy, they are hesitant to join any conference, but their preference would be the Big 12, as it is closer to them. The PAC took Utah but did not want BYU, presumably because they are a church owned school (PAC said the reason is that BYU is not a research school). BYU kind of likes being indy because of their broadcast facility, which is first rate, and can broadcast not only nation-wide but world-wide, on BYUtv. However, they only have broadcast rights on some of their games, not all. The only game that was broadcast live this year was the Savannah State game. Others were broadcast on a delayed basis (ESPN games). I'm still looking to see if the BYU – Cal game (which was broadcast on the PAC 12 network by way of Dish) will be re-broadcast on BYUtv. Hope so, as it appears to have been a great game. Tried to watch it at a BWW sports bar, but they did not have Dish where we were at the time (Farmington, NM). If the Big 12 doesn't offer BYU a chance to join, would love to see the AAC offer BYU an opportunity to join, as a 2nd option. I think it would work for BYU and I feel it would enhance the footprint of the ACC and may entice other teams in the western part of the US to be offered and perhaps also join. The next two years should be interesting to see what happens to the college FB playoff situation. Will it go to an 8 team playoff? Perhaps.


"Profound in its simplicity and clarity, this roadmap to business success is packed with insightful and practical tools you can begin to use and profit from immediately."
Just Focus on Two or Three Things?
Should a company, which aims to achieve the best performance and acquire the right customers, focus on two or three things?
This is the question that guided Sandy Richardson as she began the Business Results Revolution journey. According to her, yes, there exists a short list of industry secrets that all successful business follows. However, those secrets are not for everyone to understand.
Business Results Revolution presents those secrets in such a way that businesses may apply them easily to achieve their goals. It is a simplified guide to not only all the proven effective business strategies but also to the most critical elements that successful businesses focus on.
Why do our customers do business with us?
Why do our employees think our customers do business with us?


Free 6000 MB, free 4500 Zong mins&sms, free 450 other network mins. No promo code is needed here.
Get A Free Device At Zong!
Enter the landing page and check the offer out. No promo code is needed here.
Super Weekly Max For Rs. 200!
Subscribe now to the Super Weekly Max from Zong and get the highest mobile internet volume in any weekly bundle! Get 5GB available between 4AM to 4PM!
Zong Offer: 4GB Data Only For Rs. 130!
Enjoy 4GB data only for Rs 130+tax by subscribing to Super Weekly Plus bundle. For activation of bundle please click on the link below!
Simply SMS "dto" to 6464 and enjoy 1GB (valid from 4am to 4pm) for as low as R12 + tax. Head to the landing page to get more details.
Exclusive Discocunt on Bolt+ Device from Zong: Save Rs. 450!
Get the Bolt+ Device from Zong Rs. 450 cheaper! No promo code is required to get this offer! Just follow the instructions on the landing page.
Subscribe to one of the MBB Bundles to get Internet Volume, Voice minutes and SMS on their PLUS ONE SIM for free. Click to get this Zong special offer.
Get free 2GB Internet and 200 Zong to Zong minutes by upgrading your SIM to Zong 4G. Click to start now!
Simply dial this code on your mobile phone to get 4000 free WhatsApp MBs valid for 30 days. Click to get this Zong promo offer.
Simply dial this Zong code on your phone to get Super Student Bundle including 30MB mobile internet valid for 2 hours and unlimited free calls (Zong to Zong) for as little as R5 + tax. Don't miss out this Zong special offer!
Get Zong Super Weekly Offer and pay as low as R100 for 2000MB for a week. Click to get the deal!
Check Zong prepaid packages on the landing page and save up to 50% off. Click to start saving while shopping at Zong.
Check Zong postpaid packages on the landing page and take even to 50% off. Don't miss out!
Choose one of the All-In-1 bundles available at Zong to save your money. Head to the landing page to search offers.
Now you can SURPRISE your ZONG Friends and Family by gifting then AMAZING Zong Products. GIFT A BUNDLE is a unique service that enables prepaid customers to activate Top Products on other Prepaid Zong numbers. Use the dial Zong code below!
Zong New Sim Offer 2018!
Buy a Zong Prepaid SIM & enjoy 2,000 MB free for 3 days on Pakistan's No. 1 data network! To avail the offer simply dial the zong code below!
Zong 4G Device Packages 2018 from Rs. 2,500!
Get 4G devices from Zong and avail the lowest prices online. No Zong promo code is required.
Head to the ladning page to check all 4G Devices available at Zong now. Click to check the latest offers.
Archive offer Spring Sale At Zong! Save Rs. 400!
Avail an exclusive online discount and get your fastest Internet device. No promo code is required.
Archive offer Reactivate Your MBB Devic! Get 50% Price Discount & Get 108 GBs For 3 Months In Just Rs. 2,000!
Go to the landing page and get special offer at Zong. The discount is time-limited, hurry up.
Recharge your Zong number at Daraz and get 23% off thanks to this coupon code. Provide the code on the landing page to avail the discount.
Reactivate your device/MBB SIM now with Zong to enjoy a price discount of 50% and get 108 GBs for 3 months in just R2000 Click to get more details.
Archive offer Zong Promo Code: Get 20% Off Careem Rides If You Are A Zong User!
Enjoy 20% off your Careem ride when you use the Zong promo code when you book your ride! You only need to be a Zong user!
Jazz Weekly Mega For Only Rs. 70!
Dial the Jazz code to get 4000MB for only Rs. 70!
Would you like to be informed as soon as new Zong discount arrives?
The research indicates that for the most of Pakistanis it is very import to keep in touch with the closest family and friends. A quick phone to a friend, exchanging text messages with sister and everyday chat with mum – unfortunately it is all that lead us to receive large bills. But it doesn't have to be this way. There are several networks which keep offering you promotions for cheap phone calls or really interesting discount coupons on phones. One of the operators of this kind is Zong which operates in Pakistan.
Zong is a Chinese operator which wants to touch the Pakistan market. They serve to inspire and empower the people living in this country with innovative technology which is taking place in today's fast evolving culture. The company tries to expand their wings of coverage to all corners and maintaining perfection in connectivity no matter what the localisation is. The short definition of the company might be Affordability and Innovation with a diverse set of informative and entertaining Value Added Services. Zong enables the users to avail the best possible mobile Internet bundles which are not available in the other stores. The idea of the company is to allow people to communicate at their free will, in a stress-free environment where the clients are not worried about tariffs, capacity issues and congestion.
On the website you can select from many categories such as Prepaid (SMS and Data Bundles, Packages, Promotions, power pack SIM), postpaid (packages, e-care, data calculator, payments), Internet (Internet packages, Internet products, My data calculator, 3G, G), MBB, Internet SIM and kid's watch service, International IDD &IR. All the Zong offers also available on the website.
The operator is famous for the fact that you can get there the newest mobile phones at extremely low prices. In the online store, you can buy phones with a dozen of discounts, paying the price that is only a fraction of its real value.
Starting from today the offer of the Zong can be even more attractive because of Zong Discount coupons. Long conversations almost for free? Packages of free SMS for nothing? All of these are possible. There are so many promotions that sometimes they can skip our attention. Now you have a quick and easy access to them in one place.
If you want to take an advantage of the promotion you do not need to call the consultant, or go to the retail store. All formalities you can get in Zong online store. You decide which offer suits you and in which discount you are interested in. What is also important in the online store you can use discount codes which can be also used in retail stories. If you have problems with selecting the right offer you can use chat that connects you with the consultant who will help you to choose the best deal.
All discount coupons all available no only for a chosen group. You should spend only a few minutes to find the interesting promotions which will make your order much cheaper or your order will be enriched with the additional package. The book of knowledge about discounts and promotions are the profiles of the company in social media. You will also be able to find there many competitions with interesting prizes.
On this website from the list above you have to choose an interesting promotion, and all you need is to click the link. The system will automatically redirect you to Zong website. At the store, you can complete your order and clarify all details of the offer. In the preview of the basket, you will find the box where you should insert the code. The discount coupon will be automatically added to your order. After following this easy procedure you will gain more paying less.
Every year Zong prepares special offers for Black Friday and Cyber Monday.


text:
futurist forum
Can Science Fiction Writers Inspire The World To Save Itself?
The Hieroglyph project asks sci-fi writers to stop creating dark dystopias, and instead show us visions of a better future, so that we work harder to get there.
By Ariel Schwartz 4 minute Read
Science fiction has often predicted scientific advances, but what happens when scientists and sci-fi authors collaborate on visions of the future? Hieroglyph, a new project from Arizona State University's Center for Science and the Imagination and science fiction author Neal Stephenson, is working on an answer.
The modern science fiction landscape is clogged with dystopian fantasies: the battles of The Hunger Games, the eerie rent-a-human business of Dollhouse, the zombies of the Walking Dead, the (admittedly visionary) cyberpunk novels of William Gibson. At the same time, the rosy 1950s visions of a future with moon colonies and flying cars are nowhere close to reality. It almost seems as though science fiction writers–and the general public–have given up on the future as a happy, technologically enhanced place to be.
Maybe it's the science fiction writers that should help us dream better dreams.
Humanity's lack of a positive vision for the future can be blamed in part on an engineering culture that's more focused on incrementalism (and VC funding) than big ideas. But maybe science fiction writers should share some of the blame. That's the idea that came out of a conversation in 2011 between science fiction author Neal Stephenson and Michael Crow, the president of Arizona State University.
Stephenson was giving a talk on stage at a conference, lamenting the proliferation of apocalyptic books and movies and complaining that humanity has lost its ambition to do big things. Crow's response, according to Ed Finn, the founding director of the Center for Science and the Imagination at ASU: "Maybe it's all your fault. Maybe it's the science fiction writers that should help us dream better dreams."
In an article in the World Policy Journal, Stephenson lays out the problem with sci-fi today:
Speaking broadly, the techno-optimism of the Golden Age of SF has given way to fiction written in a generally darker, more skeptical and ambiguous tone. I myself have tended to write a lot about hackers–trickster archetypes who exploit the arcane capabilities of complex systems devised by faceless others.
Believing we have all the technology we'll ever need, we seek to draw attention to its destructive side effects. This seems foolish now that we find ourselves saddled with technologies like Japan's ramshackle 1960's-vintage reactors at Fukushima when we have the possibility of clean nuclear fusion on the horizon. The imperative to develop new technologies and implement them on a heroic scale no longer seems like the childish preoccupation of a few nerds with slide rules. It's the only way for the human race to escape from its current predicaments. Too bad we've forgotten how to do it.
If science fiction inspires scientists and engineers to create new things–Stephenson believes it can–then more visionary, realistic sci-fi stories can help create a better future. Hence the Hieroglyph experiment, launched this month as a collaborative website for researchers and writers. Many of the stories created on the platform will go into a HarperCollins anthology of fiction and non-fiction, set to be published in 2014.
Hieroglyph has already drafted writers like Bruce Sterling, Cory Doctorow, and Madeline Ashby, to share ideas on the platform, where they're discussing things like Russian-style space cities, artificial spider silk, and a shadow drone Internet (a "decentralized, drone-based wireless Internet commons").
The ground rules, as per Stephenson (relayed to us by Crow): "Let's avoid the easy dystopian story which is lots of fun to write and pretty easy to write these days. Let's also avoid stories that rely on magic, where it's impossible and implausible to see how we could get there from today. Let's avoid stories that fall too far down the well of humans reverse-engineering technology, where technology becomes the obstacle to happier human existence."
The first concrete project to emerge from Hieroglyph is the Tall Tower, which comes from the minds of Stephenson and ASU structural engineering professor Keith Hjelmstad. The pair (and other Hieroglyph participants) are discussing what it would take to build an incredibly tall tower made out of high-grade steel. Crow explains: "The basic question is, how tall can we build something? The way Neal is thinking about the idea of a tall tower is that it's a framework or platform for all sorts of different things."
According to Io9, Stephenson envisions the 20-kilometer high tower as a refueling station for docked planes or a place to launch space missions. It would be a feat of structural engineering on par with some of humanity's biggest achievements–and it will make a compelling story, knowing Stephenson.
Doctorow is moving along with his own creative idea: a story about using 3D printers on the moon to create a lunar colony. Here's an excerpt from his premise:
A group of Burning Man hackers mod a powder-deposition printer to use solar energy and some kind of locally derived binder to print playa dust (gypsum sand), powered by photovoltaic arrays, possibly using COTS blue lasers from surplus DVD players.
They learn so much doing this that they throw in their lot with Long-Now-style space colonization utopians and raise the money to put one (or more) of these printers on the moon, modded to print in regolith (possibly autosorting the regloith for a trace-element to use as a binder).
This printer is directed from a ground-station that is controlled by an Architecture for Humanity-style wiki on which arguments and technical challenges reign for a generation. The wiki periodically squirts the printer with updates for its operational plan (something like the new firmware periodically loaded onto the Mars rovers).
There are no concrete plans for Hieroglyph beyond the upcoming anthology, but the project will live on. "I think the collaborations are going to be just as interesting as the stories that come out in the end," says Crow. "Already, people are interested in this as a model for other kinds of publications and collaborations."
Ariel Schwartz is a Senior Editor at Co.Exist. She has contributed to SF Weekly, Popular Science, Inhabitat, Greenbiz, NBC Bay Area, GOOD Magazine and more
Abigail Disney, other millionaires, slam Davos and urge governments to tax the rich

redpajama_set_name: RedPajamaCommonCrawl


, we all selects the very best selections having greatest quality simply for you, and this photographs is among photos selections in this ideal pictures gallery with regards to Best Of Birthday Cake Designs for Kids. I am hoping you can enjoy it.
submitted simply by koyaz on 2019-02-10 18:02:19. To see almost all photos inside Best Of Birthday Cake Designs for Kids graphics gallery you should follow this link.
Birthday Cake Designs For Kids Best Of Birthday Cakes Pinterest Beautiful Beautiful Potty Birthday Cake Uploaded by koyaz on Sunday, February 10th, 2019 in category Modern Wedding cakes.
See also Birthday Cake Designs For Kids Beautiful Birthday Cake Ideas How To Make A Fun Turtle Cupcake Cake from Modern Wedding cakes Topic.
Here we have another image Birthday Cake Designs For Kids Awesome Diy 1st Birthday Smash Cake For Baby 3 Ways To Decorate featured under Birthday Cake Designs For Kids Best Of Birthday Cakes Pinterest Beautiful Beautiful Potty Birthday Cake. We hope you enjoyed it and if you want to download the pictures in high quality, simply right click the image and choose "Save As". Thanks for reading Birthday Cake Designs For Kids Best Of Birthday Cakes Pinterest Beautiful Beautiful Potty Birthday Cake.