<commit_before>/*
 Licensed to <NAME>. under one
 or more contributor license agreements. See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership. <NAME>. licenses this file
 to you 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 majordodo.worker;

import java.util.ArrayList;
import majordodo.network.BrokerRejectedConnectionException;
import majordodo.network.BrokerNotAvailableException;
import majordodo.network.BrokerLocator;
import majordodo.network.ConnectionRequestInfo;
import majordodo.executors.TaskExecutor;
import majordodo.executors.TaskExecutorFactory;
import majordodo.executors.TaskExecutorStatus;
import majordodo.network.Channel;
import majordodo.network.ChannelEventListener;
import majordodo.network.Message;
import majordodo.network.SendResultCallback;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;
import majordodo.utils.ErrorUtils;

/**
 * Core of the worker inside the JVM
 *
 * @author enrico.olivelli
 */
public class WorkerCore implements ChannelEventListener, ConnectionRequestInfo, AutoCloseable {

    private static final Logger LOGGER = Logger.getLogger(WorkerCore.class.getName());

    private final ExecutorService threadpool;
    private final String processId;
    private final String workerId;
    private final String location;
    private final Map<Long, String> runningTasks = new ConcurrentHashMap<>();
    private final BrokerLocator brokerLocator;
    private final Thread coreThread;
    private volatile boolean stopped = false;
    private Channel channel;
    private WorkerStatusListener listener;
    private KillWorkerHandler killWorkerHandler = KillWorkerHandler.GRACEFULL_STOP;
    private Callable<Void> externalProcessChecker; // PIDFILECHECKER

    public KillWorkerHandler getKillWorkerHandler() {
        return killWorkerHandler;
    }

    public void setKillWorkerHandler(KillWorkerHandler killWorkerHandler) {
        this.killWorkerHandler = killWorkerHandler;
    }

    public Callable<Void> getExternalProcessChecker() {
        return externalProcessChecker;
    }

    public void setExternalProcessChecker(Callable<Void> externalProcessChecker) {
        this.externalProcessChecker = externalProcessChecker;
    }

    public void die() {
        LOGGER.log(Level.SEVERE, "Die!");
        if (coreThread != null) {
            coreThread.interrupt();
        }
        this.stop();
    }

    private static final class NotImplementedTaskExecutorFactory implements TaskExecutorFactory {

        @Override
        public TaskExecutor createTaskExecutor(String taskType, Map<String, Object> parameters) {
            return new TaskExecutor();
        }

    };
    private TaskExecutorFactory executorFactory;

    public Map<Long, String> getRunningTasks() {
        return runningTasks;
    }

    @Override
    public Set<Long> getRunningTaskIds() {
        return runningTasks.keySet();
    }

    public TaskExecutorFactory getExecutorFactory() {
        return executorFactory;
    }

    public void setExecutorFactory(TaskExecutorFactory executorFactory) {
        this.executorFactory = executorFactory;
    }

    private void requestNewTasks() {
        Channel _channel = channel;
        if (_channel != null) {
            Map<String, Integer> availableSpace = new HashMap<>(config.getMaxThreadsByTaskType());
            runningTasks.values().forEach(tasktype -> {
                Integer count = availableSpace.get(tasktype);
                if (count != null && count > 1) {
                    availableSpace.put(tasktype, count - 1);
                } else {
                    availableSpace.remove(tasktype);
                }
            });

            int maxnewthreads = config.getMaxThreads() - runningTasks.size();
            long _start = System.currentTimeMillis();
            LOGGER.log(Level.FINER, "requestNewTasks maxnewthreads:" + maxnewthreads + ", availableSpace:" + availableSpace + " groups:" + config.getGroups() + " excludedGroups" + config.getExcludedGroups());
            if (availableSpace.isEmpty() || maxnewthreads <= 0) {
                return;
            }<commit_msg>fix reconnection with finished tasks<commit_after>/*
 Licensed to <NAME>. under one
 or more contributor license agreements. See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership. <NAME>. licenses this file
 to you 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 majordodo.worker;

import java.util.ArrayList;
import majordodo.network.BrokerRejectedConnectionException;
import majordodo.network.BrokerNotAvailableException;
import majordodo.network.BrokerLocator;
import majordodo.network.ConnectionRequestInfo;
import majordodo.executors.TaskExecutor;
import majordodo.executors.TaskExecutorFactory;
import majordodo.executors.TaskExecutorStatus;
import majordodo.network.Channel;
import majordodo.network.ChannelEventListener;
import majordodo.network.Message;
import majordodo.network.SendResultCallback;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;
import majordodo.utils.ErrorUtils;

/**
 * Core of the worker inside the JVM
 *
 * @author enrico.olivelli
 */
public class WorkerCore implements ChannelEventListener, ConnectionRequestInfo, AutoCloseable {

    private static final Logger LOGGER = Logger.getLogger(WorkerCore.class.getName());

    private final ExecutorService threadpool;
    private final String processId;
    private final String workerId;
    private final String location;
    private final Map<Long, String> runningTasks = new ConcurrentHashMap<>();
    private final BrokerLocator brokerLocator;
    private final Thread coreThread;
    private volatile boolean stopped = false;
    private Channel channel;
    private WorkerStatusListener listener;
    private KillWorkerHandler killWorkerHandler = KillWorkerHandler.GRACEFULL_STOP;
    private Callable<Void> externalProcessChecker; // PIDFILECHECKER

    public KillWorkerHandler getKillWorkerHandler() {
        return killWorkerHandler;
    }

    public void setKillWorkerHandler(KillWorkerHandler killWorkerHandler) {
        this.killWorkerHandler = killWorkerHandler;
    }

    public Callable<Void> getExternalProcessChecker() {
        return externalProcessChecker;
    }

    public void setExternalProcessChecker(Callable<Void> externalProcessChecker) {
        this.externalProcessChecker = externalProcessChecker;
    }

    public void die() {
        LOGGER.log(Level.SEVERE, "Die!");
        if (coreThread != null) {
            coreThread.interrupt();
        }
        this.stop();
    }

    private static final class NotImplementedTaskExecutorFactory implements TaskExecutorFactory {

        @Override
        public TaskExecutor createTaskExecutor(String taskType, Map<String, Object> parameters) {
            return new TaskExecutor();
        }

    };
    private TaskExecutorFactory executorFactory;

    public Map<Long, String> getRunningTasks() {
        return runningTasks;
    }

    @Override
    public Set<Long> getRunningTaskIds() {
        Set<Long> res = new HashSet<>();
        res.addAll(runningTasks.keySet());
        for (FinishedTaskNotification f : this.pendingFinishedTaskNotifications) {
            res.add(f.taskId);
        }
        return res;
    }

    public TaskExecutorFactory getExecutorFactory() {
        return executorFactory;
    }

    public void setExecutorFactory(TaskExecutorFactory executorFactory) {
        this.executorFactory = executorFactory;
    }

    private void requestNewTasks() {
        Channel _channel = channel;
        if (_channel != null) {
            Map<String, Integer> availableSpace = new HashMap<>(config.getMaxThreadsByTaskType());
            runningTasks.values().forEach(tasktype -> {
                Integer count = availableSpace.get(tasktype);
                if (count != null && count > 1) {
                    availableSpace.put(tasktype, count - 1);
                } else {
                    availableSpace.remove(tasktype);
                }
            });

            int maxnewthreads = config.getMaxThreads() - runningTasks.size();
            long _start = System.currentTimeMillis();
            LOGGER.log(Level.FINER, "requestNewTasks maxnewthreads:" + maxnewthreads + ", availableSpace:" + availableSpace + " groups:" + config.getGroups() + " excludedGroups" + config.getExcludedGroups());
            if (availableSpace.isEmpty() || maxnewthreads <= 0) {
                return;
            }
{'fraction_non_alphanumeric': 0.07923054947147891, 'fraction_numerical': 0.001333206361298924, 'mean_word_length': 3.548289302728454, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 33, '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': '1461483', 'n_tokens_mistral': 2646, 'n_tokens_neox': 2612, 'n_words': 744}