<commit_before>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.documentapi.messagebus.protocol;

import com.yahoo.concurrent.CopyOnWriteHashMap;
import com.yahoo.document.BucketId;
import com.yahoo.document.BucketIdFactory;
import com.yahoo.jrt.slobrok.api.IMirror;
import com.yahoo.jrt.slobrok.api.Mirror;
import com.yahoo.messagebus.EmptyReply;
import com.yahoo.messagebus.Error;
import com.yahoo.messagebus.ErrorCode;
import com.yahoo.messagebus.Message;
import com.yahoo.messagebus.Reply;
import com.yahoo.messagebus.routing.Hop;
import com.yahoo.messagebus.routing.Route;
import com.yahoo.messagebus.routing.RoutingContext;
import com.yahoo.messagebus.routing.RoutingNodeIterator;
import com.yahoo.messagebus.routing.VerbatimDirective;
import com.yahoo.vdslib.distribution.Distribution;
import com.yahoo.vdslib.state.ClusterState;
import com.yahoo.vdslib.state.Node;
import com.yahoo.vdslib.state.NodeType;
import com.yahoo.vdslib.state.State;
import com.yahoo.vespa.config.content.DistributionConfig;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Routing policy to determine which distributor in a content cluster to send data to.
 * Using different key=value parameters separated by semicolon (";"), the user can control which cluster to send to.
 *
 * cluster=[clusterName] (Mandatory, determines the cluster name)
 * config=[config] (Optional, a comma separated list of config servers to use. Used to talk to clusters not defined in this vespa application)
 * clusterconfigid=[id] (Optional, use given config id for distribution instead of default)
 *
 * @author <NAME>
 */
public class ContentPolicy extends SlobrokPolicy {

    private static final Logger log = Logger.getLogger(ContentPolicy.class.getName());
    public static final String owningBucketStates = "uim";
    private static final String upStates = "ui";

    /** This class merely generates a slobrok host pattern for a given distributor. */
    static class SlobrokHostPatternGenerator {

        private final String base;

        SlobrokHostPatternGenerator(String clusterName) {
            this.base = "storage/cluster." + clusterName + "/distributor/";
        }

        /**
         * Find host pattern of the hosts that are valid targets for this request.
         *
         * @param distributor Set to null if any distributor is valid target.
         */
        String getDistributorHostPattern(Integer distributor) {
            return base + (distributor == null ? "*" : distributor) + "/default";
        }

    }

    /** Helper class to match a host pattern with node to use. */
    public abstract static class HostFetcher {

        private static class Targets {
            private final List<Integer> list;
            private final int total;
            Targets() {
                this(Collections.emptyList(), 1);
            }
            Targets(List<Integer> list, int total) {
                this.list = list;
                this.total = total;
            }
        }

        private final int requiredUpPercentageToSendToKnownGoodNodes;
        private final AtomicReference<Targets> validTargets = new AtomicReference<>(new Targets());
        protected final Random randomizer = new Random(12345); // Use same randomizer each time to make unit testing easy.

        protected HostFetcher(int percent) {
            requiredUpPercentageToSendToKnownGoodNodes = percent;
        }

        void updateValidTargets(ClusterState state) {
            List<Integer> validRandomTargets = new ArrayList<>();
            for (int i=0; i<state.getNodeCount(NodeType.DISTRIBUTOR); ++i) {
                if (state.getNodeState(new Node(NodeType.DISTRIBUTOR, i)).getState().oneOf(upStates)) validRandomTargets.add(i);
            }
            validTargets.set(new Targets(new CopyOnWriteArrayList<>(validRandomTargets), state.getNodeCount(NodeType.DISTRIBUTOR)));
        }
        public abstract String getTargetSpec(Integer distributor, RoutingContext context);
        String getRandomTargetSpec(RoutingContext context) {
            Targets targets = validTargets.get();
            // Try to use list of random targets, if at least X % of the nodes are up
            while ((targets.total != 0) &&
                   (100 * targets.list.size() / targets.total >= requiredUpPercentageToSendToKnownGoodNodes))
            {
                int randIndex = randomizer.nextInt(targets.list.size());
                String targetSpec = getTargetSpec(targets.list.get(randIndex), context);
                if (targetSpec != null) {
                    context.trace(3, "Sending to random node seen up in cluster state");
                    return targetSpec;
                }
                targets.list.remove(randIndex);
            }
            context.trace(3, "Too few nodes seen up in state. Sending totally random.");
            return getTargetSpec(null, context);
        }
        public void close() {}
    }

    /** Host fetcher using a slobrok mirror to find the hosts. */
    public static class SlobrokHostFetcher extends HostFetcher {
        private final SlobrokHostPatternGenerator patternGenerator;
        private final SlobrokPolicy policy;

        SlobrokHostFetcher(SlobrokHostPatternGenerator patternGenerator, SlobrokPolicy policy, int percent) {
            super(percent);
            this.patternGenerator = patternGenerator;
            this.policy = policy;
        }
<commit_msg>Remove by index is not thread safe, even if the list is a CopyOnWriteArrayList. iOne thing is that it might throw if index is invalid. The other thing is that the index might point to something completely different. But remove by value should be fine.<commit_after>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.documentapi.messagebus.protocol;

import com.yahoo.concurrent.CopyOnWriteHashMap;
import com.yahoo.document.BucketId;
import com.yahoo.document.BucketIdFactory;
import com.yahoo.jrt.slobrok.api.IMirror;
import com.yahoo.jrt.slobrok.api.Mirror;
import com.yahoo.messagebus.EmptyReply;
import com.yahoo.messagebus.Error;
import com.yahoo.messagebus.ErrorCode;
import com.yahoo.messagebus.Message;
import com.yahoo.messagebus.Reply;
import com.yahoo.messagebus.routing.Hop;
import com.yahoo.messagebus.routing.Route;
import com.yahoo.messagebus.routing.RoutingContext;
import com.yahoo.messagebus.routing.RoutingNodeIterator;
import com.yahoo.messagebus.routing.VerbatimDirective;
import com.yahoo.vdslib.distribution.Distribution;
import com.yahoo.vdslib.state.ClusterState;
import com.yahoo.vdslib.state.Node;
import com.yahoo.vdslib.state.NodeType;
import com.yahoo.vdslib.state.State;
import com.yahoo.vespa.config.content.DistributionConfig;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Routing policy to determine which distributor in a content cluster to send data to.
 * Using different key=value parameters separated by semicolon (";"), the user can control which cluster to send to.
 *
 * cluster=[clusterName] (Mandatory, determines the cluster name)
 * config=[config] (Optional, a comma separated list of config servers to use. Used to talk to clusters not defined in this vespa application)
 * clusterconfigid=[id] (Optional, use given config id for distribution instead of default)
 *
 * @author <NAME>
 */
public class ContentPolicy extends SlobrokPolicy {

    private static final Logger log = Logger.getLogger(ContentPolicy.class.getName());
    public static final String owningBucketStates = "uim";
    private static final String upStates = "ui";

    /** This class merely generates a slobrok host pattern for a given distributor. */
    static class SlobrokHostPatternGenerator {

        private final String base;

        SlobrokHostPatternGenerator(String clusterName) {
            this.base = "storage/cluster." + clusterName + "/distributor/";
        }

        /**
         * Find host pattern of the hosts that are valid targets for this request.
         *
         * @param distributor Set to null if any distributor is valid target.
         */
        String getDistributorHostPattern(Integer distributor) {
            return base + (distributor == null ? "*" : distributor) + "/default";
        }

    }

    /** Helper class to match a host pattern with node to use. */
    public abstract static class HostFetcher {

        private static class Targets {
            private final List<Integer> list;
            private final int total;
            Targets() {
                this(Collections.emptyList(), 1);
            }
            Targets(List<Integer> list, int total) {
                this.list = list;
                this.total = total;
            }
        }

        private final int requiredUpPercentageToSendToKnownGoodNodes;
        private final AtomicReference<Targets> validTargets = new AtomicReference<>(new Targets());
        protected final Random randomizer = new Random(12345); // Use same randomizer each time to make unit testing easy.

        protected HostFetcher(int percent) {
            requiredUpPercentageToSendToKnownGoodNodes = percent;
        }

        void updateValidTargets(ClusterState state) {
            List<Integer> validRandomTargets = new ArrayList<>();
            for (int i=0; i<state.getNodeCount(NodeType.DISTRIBUTOR); ++i) {
                if (state.getNodeState(new Node(NodeType.DISTRIBUTOR, i)).getState().oneOf(upStates)) validRandomTargets.add(i);
            }
            validTargets.set(new Targets(new CopyOnWriteArrayList<>(validRandomTargets), state.getNodeCount(NodeType.DISTRIBUTOR)));
        }
        public abstract String getTargetSpec(Integer distributor, RoutingContext context);
        String getRandomTargetSpec(RoutingContext context) {
            Targets targets = validTargets.get();
            // Try to use list of random targets, if at least X % of the nodes are up
            while ((targets.total != 0) &&
                   (100 * targets.list.size() / targets.total >= requiredUpPercentageToSendToKnownGoodNodes))
            {
                Integer distributor = targets.list.get(randomizer.nextInt(targets.list.size()));
                String targetSpec = getTargetSpec(distributor, context);
                if (targetSpec != null) {
                    context.trace(3, "Sending to random node seen up in cluster state");
                    return targetSpec;
                }
                targets.list.remove(distributor);
            }
            context.trace(3, "Too few nodes seen up in state. Sending totally random.");
            return getTargetSpec(null, context);
        }
        public void close() {}
    }

    /** Host fetcher using a slobrok mirror to find the hosts. */
    public static class SlobrokHostFetcher extends HostFetcher {
        private final SlobrokHostPatternGenerator patternGenerator;
        private final SlobrokPolicy policy;

        SlobrokHostFetcher(SlobrokHostPatternGenerator patternGenerator, SlobrokPolicy policy, int percent) {
            super(percent);
            this.patternGenerator = patternGenerator;
            this.policy = policy;
        }

{'fraction_non_alphanumeric': 0.08599899514319209, 'fraction_numerical': 0.003182046558365433, 'mean_word_length': 2.979673442185938, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2019924', 'n_tokens_mistral': 3194, 'n_tokens_neox': 3079, 'n_words': 926}