<commit_before>/**
 * Copyright 2014 <NAME>, <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.connection;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.redisson.misc.ReclosableLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lambdaworks.redis.RedisConnection;
import com.lambdaworks.redis.RedisConnectionException;
import com.lambdaworks.redis.codec.RedisCodec;
import com.lambdaworks.redis.pubsub.RedisPubSubConnection;

abstract class BaseLoadBalancer implements LoadBalancer {

    private final Logger log = LoggerFactory.getLogger(getClass());

    private RedisCodec codec;

    private String password;

    private final ReclosableLatch clientsEmpty = new ReclosableLatch();
    final Queue<ConnectionEntry> clients = new ConcurrentLinkedQueue<ConnectionEntry>();

    public void init(RedisCodec codec, String password) {
        this.codec = codec;
        this.password = password;
    }

    public void add(ConnectionEntry entry) {
        clients.add(entry);
        clientsEmpty.open();
    }

    public Queue<RedisPubSubConnection> remove(String host, int port) {
        InetSocketAddress addr = new InetSocketAddress(host, port);
        for (Iterator<ConnectionEntry> iterator = clients.iterator(); iterator.hasNext();) {
            ConnectionEntry entry = iterator.next();
            if (!entry.getClient().getAddr().equals(addr)) {
                continue;
            }

            log.info("slave {} removed", entry.getClient().getAddr());
            iterator.remove();
            if (clients.isEmpty()) {
                clientsEmpty.close();
            }
            entry.shutdown();
            log.info("slave {} shutdown", entry.getClient().getAddr());

            return entry.getSubscribeConnections();
        }
        throw new IllegalStateException("Can't find " + addr + " in slaves!");
    }

    @SuppressWarnings("unchecked")
    public RedisPubSubConnection nextPubSubConnection() {
        List<ConnectionEntry> clientsCopy = new ArrayList<ConnectionEntry>(clients);
        if (clientsCopy.isEmpty()) {
            clientsEmpty.awaitUninterruptibly();
        }
        while (true) {
            if (clientsCopy.isEmpty()) {
                // TODO refactor
                log.warn("Slave subscribe-connection pool gets exhausted! Trying to acquire connection again...");
                return nextPubSubConnection();
//                long time = System.currentTimeMillis();
//                long endTime = System.currentTimeMillis() - time;
//                log.warn("Connection acquired, time spended: {} ms", endTime);
            }

            int index = getIndex(clientsCopy);
            ConnectionEntry entry = clientsCopy.get(index);

            if (!entry.getSubscribeConnectionsSemaphore().tryAcquire()) {
                clientsCopy.remove(index);
            } else {
                try {
                    RedisPubSubConnection conn = entry.getSubscribeConnections().poll();
                    if (conn != null) {
                        return conn;
                    }
                    conn = entry.getClient().connectPubSub(codec);
                    if (password != null) {
                        conn.auth(password);
                    }
                    return conn;
                } catch (RedisConnectionException e) {
                    // TODO connection scoring
                    log.warn("Can't connect to {}, trying next connection!", entry.getClient().getAddr());
                    clientsCopy.remove(index);
                }
            }
        }
    }

    public RedisConnection nextConnection() {
        List<ConnectionEntry> clientsCopy = new ArrayList<ConnectionEntry>(clients);
        if (clientsCopy.isEmpty()) {
            clientsEmpty.awaitUninterruptibly();
        }
        while (true) {<commit_msg>Empty slaves handling fixed.  #30<commit_after>/**
 * Copyright 2014 <NAME>, <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.connection;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.redisson.misc.ReclosableLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lambdaworks.redis.RedisConnection;
import com.lambdaworks.redis.RedisConnectionException;
import com.lambdaworks.redis.codec.RedisCodec;
import com.lambdaworks.redis.pubsub.RedisPubSubConnection;

abstract class BaseLoadBalancer implements LoadBalancer {

    private final Logger log = LoggerFactory.getLogger(getClass());

    private RedisCodec codec;

    private String password;

    private final ReclosableLatch clientsEmpty = new ReclosableLatch();
    final Queue<ConnectionEntry> clients = new ConcurrentLinkedQueue<ConnectionEntry>();

    public void init(RedisCodec codec, String password) {
        this.codec = codec;
        this.password = password;
    }

    public void add(ConnectionEntry entry) {
        clients.add(entry);
        clientsEmpty.open();
    }

    public Queue<RedisPubSubConnection> remove(String host, int port) {
        InetSocketAddress addr = new InetSocketAddress(host, port);
        for (Iterator<ConnectionEntry> iterator = clients.iterator(); iterator.hasNext();) {
            ConnectionEntry entry = iterator.next();
            if (!entry.getClient().getAddr().equals(addr)) {
                continue;
            }

            log.info("slave {} removed", entry.getClient().getAddr());
            iterator.remove();
            if (clients.isEmpty()) {
                clientsEmpty.close();
            }
            entry.shutdown();
            log.info("slave {} shutdown", entry.getClient().getAddr());

            return entry.getSubscribeConnections();
        }
        throw new IllegalStateException("Can't find " + addr + " in slaves!");
    }

    @SuppressWarnings("unchecked")
    public RedisPubSubConnection nextPubSubConnection() {
        List<ConnectionEntry> clientsCopy = new ArrayList<ConnectionEntry>(clients);
        if (clientsCopy.isEmpty()) {
            clientsEmpty.awaitUninterruptibly();
            return nextPubSubConnection();
        }
        while (true) {
            if (clientsCopy.isEmpty()) {
                // TODO refactor
                log.warn("Slave subscribe-connection pool gets exhausted! Trying to acquire connection again...");
                return nextPubSubConnection();
//                long time = System.currentTimeMillis();
//                long endTime = System.currentTimeMillis() - time;
//                log.warn("Connection acquired, time spended: {} ms", endTime);
            }

            int index = getIndex(clientsCopy);
            ConnectionEntry entry = clientsCopy.get(index);

            if (!entry.getSubscribeConnectionsSemaphore().tryAcquire()) {
                clientsCopy.remove(index);
            } else {
                try {
                    RedisPubSubConnection conn = entry.getSubscribeConnections().poll();
                    if (conn != null) {
                        return conn;
                    }
                    conn = entry.getClient().connectPubSub(codec);
                    if (password != null) {
                        conn.auth(password);
                    }
                    return conn;
                } catch (RedisConnectionException e) {
                    // TODO connection scoring
                    log.warn("Can't connect to {}, trying next connection!", entry.getClient().getAddr());
                    clientsCopy.remove(index);
                }
            }
        }
    }

    public RedisConnection nextConnection() {
        List<ConnectionEntry> clientsCopy = new ArrayList<ConnectionEntry>(clients);
        if (clientsCopy.isEmpty()) {
            clientsEmpty.awaitUninterruptibly();
            return nextConnection();
        }
        while (true) {
{'fraction_non_alphanumeric': 0.09958686671015438, 'fraction_numerical': 0.0023918243096325287, 'mean_word_length': 2.298314808174973, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5105430', 'n_tokens_mistral': 2355, 'n_tokens_neox': 2271, 'n_words': 590}