<commit_before>/*
 * Copyright (c) 2011-2019, PCJ Library, <NAME>
 * All rights reserved.
 *
 * Licensed under New BSD License (3-clause license).
 *
 * See the file "LICENSE" for the full license governing this code.
 */
package org.pcj.internal;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.nio.channels.SocketChannel;
import java.util.ArrayDeque;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Objects;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import org.pcj.NodesDescription;
import org.pcj.PcjFuture;
import org.pcj.PcjRuntimeException;
import org.pcj.StartPoint;
import org.pcj.internal.message.bye.ByeState;
import org.pcj.internal.message.hello.HelloMessage;
import org.pcj.internal.message.hello.HelloState;
import org.pcj.internal.message.join.GroupJoinRequestMessage;
import org.pcj.internal.message.join.GroupJoinStates;
import org.pcj.internal.message.join.GroupQueryMessage;
import org.pcj.internal.message.join.GroupQueryStates;
import org.pcj.internal.network.LoopbackSocketChannel;

/**
 * Internal class for external PCJ class.
 *
 * @author <NAME> (<EMAIL>)
 */
public abstract class InternalPCJ {

    private static final Logger LOGGER = Logger.getLogger(InternalPCJ.class.getName());
    private static final String PCJ_VERSION;
    private static Networker networker;
    private static NodeData nodeData;

    static {
        Package p = InternalPCJ.class.getPackage();
        PCJ_VERSION = p.getImplementationVersion() == null ? "UNKNOWN" : p.getImplementationVersion();
    }

    /**
     * Suppress default constructor for noninstantiability.
     * <p>
     * Have to be protected to allow inheritance - to give rights to its protected methods.
     */
    protected InternalPCJ() {
        throw new AssertionError();
    }

    protected static void start(Class<? extends StartPoint> startPoint,
                                NodesDescription nodesFile) {
        NodeInfo node0 = nodesFile.getNode0();
        NodeInfo currentJvm = nodesFile.getCurrentJvm();
        int allNodesThreadCount = nodesFile.getAllNodesThreadCount();
        start(startPoint, node0, currentJvm, allNodesThreadCount);
    }

    protected static void start(Class<? extends StartPoint> startPointClass,
                                NodeInfo node0, NodeInfo currentJvm, int allNodesThreadCount) {
        if (currentJvm == null) {
            throw new IllegalArgumentException("There is no entry for PCJ threads for current JVM");
        }
        if (node0 == null) {
            throw new IllegalArgumentException("There is no node0 entry in nodes description file.");
        }

        boolean isCurrentJvmNode0 = node0.getPort() == currentJvm.getPort() && node0.isLocalAddress();

        if (isCurrentJvmNode0) {
            LOGGER.log(Level.INFO, "PCJ version {0}", PCJ_VERSION);
        }

        networker = prepareNetworker(currentJvm);
        networker.startup();
        try {
            /* Prepare initial data */
            nodeData = new NodeData();
            nodeData.setHelloState(new HelloState(allNodesThreadCount));
            if (isCurrentJvmNode0) {
                nodeData.setNode0Data(new NodeData.Node0Data());
            }

            /* Getting all interfaces */
            Queue<InetAddress> inetAddresses = getHostAllNetworkInferfaces();

            /* Binding on all interfaces */
            bindOnAllNetworkInterfaces(inetAddresses, currentJvm.getPort());

            ScheduledFuture<?> helloTimeoutTimer = scheduleInterruptTimer(Thread.currentThread(), Configuration.INIT_MAXTIME, TimeUnit.SECONDS);

            /* connecting to node0 */
            SocketChannel node0Socket = connectToNode0(node0, isCurrentJvmNode0);
            nodeData.setNode0Socket(node0Socket);

            /* send HELLO message */
            helloPhase(currentJvm.getPort(), currentJvm.getThreadIds());

            helloTimeoutTimer.cancel(true);

            nodeData.setHelloState(null);

            long nanoTime = Long.MIN_VALUE;
            /* Starting execution */
            if (isCurrentJvmNode0) {
                nanoTime = System.nanoTime();
                LOGGER.log(Level.INFO, "Starting {0}"
                                               + " with {1,number,#} {1,choice,1#thread|1<threads}"
                                               + " (on {2,number,#} {2,choice,1#node|1<nodes})...",
                        new Object[]{startPointClass.getName(),
                                nodeData.getCommonGroupById(InternalCommonGroup.GLOBAL_GROUP_ID).threadCount(),
                                nodeData.getTotalNodeCount(),});
            }

            /* Preparing PcjThreads */
            Semaphore notificationObject = new Semaphore(0);

            Set<PcjThread> pcjThreads = preparePcjThreads(startPointClass, currentJvm.getThreadIds(), notificationObject);
            pcjThreads.forEach(nodeData::putPcjThread);

            /* Starting PcjThreads */
            pcjThreads.forEach(PcjThread::start);

            /* Waiting for all threads complete */
            waitForPcjThreadsComplete(pcjThreads, notificationObject);

            /* Finishing asyncTaskWorkers */
            pcjThreads.forEach(PcjThread::shutdownAsyncTasksWorkers);

            if (isCurrentJvmNode0) {
                long timer = (System.nanoTime() - nanoTime) / 1_000_000_000L;
                long h = timer / (60 * 60);
                long m = (timer / 60) % 60;
                long s = (timer % 60);

                LOGGER.log(Level.INFO, "Completed {0}"
                                               + " with {1,number,#} {1,choice,1#thread|1<threads}"
                                               + " (on {2,number,#} {2,choice,1#node|1<nodes})"
                                               + " after {3,number,#}h {4,number,#}m {5,number,#}s.",
                        new Object[]{
                                startPointClass.getName(),
                                nodeData.getCommonGroupById(InternalCommonGroup.GLOBAL_GROUP_ID).threadCount(),
                                nodeData.getTotalNodeCount(),
                                h, m, s
                        });
            }

            /* finishing */
            byePhase();
        } finally {
            networker.shutdown();
        }
    }

    private static Networker prepareNetworker(NodeInfo currentJvm) {
        int minWorkerCount = currentJvm.getThreadIds().size() + 1;
        int maxWorkerCount = currentJvm.getThreadIds().size() + 1;
        if (Configuration.WORKERS_MIN_COUNT >= 0) {
            minWorkerCount = Configuration.WORKERS_MIN_COUNT;
        }
        if (Configuration.WORKERS_MAX_COUNT > 0) {
            maxWorkerCount = Configuration.WORKERS_MAX_COUNT;
        }
        return new Networker(minWorkerCount, Math.max(minWorkerCount, maxWorkerCount));<commit_msg>Moved byePhase before finishing async task workers (wait for all normal threads to complete before shutting down)<commit_after>/*
 * Copyright (c) 2011-2019, PCJ Library, <NAME>
 * All rights reserved.
 *
 * Licensed under New BSD License (3-clause license).
 *
 * See the file "LICENSE" for the full license governing this code.
 */
package org.pcj.internal;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.nio.channels.SocketChannel;
import java.util.ArrayDeque;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Objects;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import org.pcj.NodesDescription;
import org.pcj.PcjFuture;
import org.pcj.PcjRuntimeException;
import org.pcj.StartPoint;
import org.pcj.internal.message.bye.ByeState;
import org.pcj.internal.message.hello.HelloMessage;
import org.pcj.internal.message.hello.HelloState;
import org.pcj.internal.message.join.GroupJoinRequestMessage;
import org.pcj.internal.message.join.GroupJoinStates;
import org.pcj.internal.message.join.GroupQueryMessage;
import org.pcj.internal.message.join.GroupQueryStates;
import org.pcj.internal.network.LoopbackSocketChannel;

/**
 * Internal class for external PCJ class.
 *
 * @author <NAME> (<EMAIL>)
 */
public abstract class InternalPCJ {

    private static final Logger LOGGER = Logger.getLogger(InternalPCJ.class.getName());
    private static final String PCJ_VERSION;
    private static Networker networker;
    private static NodeData nodeData;

    static {
        Package p = InternalPCJ.class.getPackage();
        PCJ_VERSION = p.getImplementationVersion() == null ? "UNKNOWN" : p.getImplementationVersion();
    }

    /**
     * Suppress default constructor for noninstantiability.
     * <p>
     * Have to be protected to allow inheritance - to give rights to its protected methods.
     */
    protected InternalPCJ() {
        throw new AssertionError();
    }

    protected static void start(Class<? extends StartPoint> startPoint,
                                NodesDescription nodesFile) {
        NodeInfo node0 = nodesFile.getNode0();
        NodeInfo currentJvm = nodesFile.getCurrentJvm();
        int allNodesThreadCount = nodesFile.getAllNodesThreadCount();
        start(startPoint, node0, currentJvm, allNodesThreadCount);
    }

    protected static void start(Class<? extends StartPoint> startPointClass,
                                NodeInfo node0, NodeInfo currentJvm, int allNodesThreadCount) {
        if (currentJvm == null) {
            throw new IllegalArgumentException("There is no entry for PCJ threads for current JVM");
        }
        if (node0 == null) {
            throw new IllegalArgumentException("There is no node0 entry in nodes description file.");
        }

        boolean isCurrentJvmNode0 = node0.getPort() == currentJvm.getPort() && node0.isLocalAddress();

        if (isCurrentJvmNode0) {
            LOGGER.log(Level.INFO, "PCJ version {0}", PCJ_VERSION);
        }

        networker = prepareNetworker(currentJvm);
        networker.startup();
        try {
            /* Prepare initial data */
            nodeData = new NodeData();
            nodeData.setHelloState(new HelloState(allNodesThreadCount));
            if (isCurrentJvmNode0) {
                nodeData.setNode0Data(new NodeData.Node0Data());
            }

            /* Getting all interfaces */
            Queue<InetAddress> inetAddresses = getHostAllNetworkInferfaces();

            /* Binding on all interfaces */
            bindOnAllNetworkInterfaces(inetAddresses, currentJvm.getPort());

            ScheduledFuture<?> helloTimeoutTimer = scheduleInterruptTimer(Thread.currentThread(), Configuration.INIT_MAXTIME, TimeUnit.SECONDS);

            /* connecting to node0 */
            SocketChannel node0Socket = connectToNode0(node0, isCurrentJvmNode0);
            nodeData.setNode0Socket(node0Socket);

            /* send HELLO message */
            helloPhase(currentJvm.getPort(), currentJvm.getThreadIds());

            helloTimeoutTimer.cancel(true);

            nodeData.setHelloState(null);

            long nanoTime = Long.MIN_VALUE;
            /* Starting execution */
            if (isCurrentJvmNode0) {
                nanoTime = System.nanoTime();
                LOGGER.log(Level.INFO, "Starting {0}"
                                               + " with {1,number,#} {1,choice,1#thread|1<threads}"
                                               + " (on {2,number,#} {2,choice,1#node|1<nodes})...",
                        new Object[]{startPointClass.getName(),
                                nodeData.getCommonGroupById(InternalCommonGroup.GLOBAL_GROUP_ID).threadCount(),
                                nodeData.getTotalNodeCount(),});
            }

            /* Preparing PcjThreads */
            Semaphore notificationObject = new Semaphore(0);

            Set<PcjThread> pcjThreads = preparePcjThreads(startPointClass, currentJvm.getThreadIds(), notificationObject);
            pcjThreads.forEach(nodeData::putPcjThread);

            /* Starting PcjThreads */
            pcjThreads.forEach(PcjThread::start);

            /* Waiting for all threads complete */
            waitForPcjThreadsComplete(pcjThreads, notificationObject);

            /* finishing */
            byePhase();

            /* Finishing asyncTaskWorkers */
            pcjThreads.forEach(PcjThread::shutdownAsyncTasksWorkers);

            if (isCurrentJvmNode0) {
                long timer = (System.nanoTime() - nanoTime) / 1_000_000_000L;
                long h = timer / (60 * 60);
                long m = (timer / 60) % 60;
                long s = (timer % 60);

                LOGGER.log(Level.INFO, "Completed {0}"
                                               + " with {1,number,#} {1,choice,1#thread|1<threads}"
                                               + " (on {2,number,#} {2,choice,1#node|1<nodes})"
                                               + " after {3,number,#}h {4,number,#}m {5,number,#}s.",
                        new Object[]{
                                startPointClass.getName(),
                                nodeData.getCommonGroupById(InternalCommonGroup.GLOBAL_GROUP_ID).threadCount(),
                                nodeData.getTotalNodeCount(),
                                h, m, s
                        });
            }
        } finally {
            networker.shutdown();
        }
    }

    private static Networker prepareNetworker(NodeInfo currentJvm) {
        int minWorkerCount = currentJvm.getThreadIds().size() + 1;
        int maxWorkerCount = currentJvm.getThreadIds().size() + 1;
        if (Configuration.WORKERS_MIN_COUNT >= 0) {
            minWorkerCount = Configuration.WORKERS_MIN_COUNT;
        }
        if (Configuration.WORKERS_MAX_COUNT > 0) {
            maxWorkerCount = Configuration.WORKERS_MAX_COUNT;
        }
        return new Networker(minWorkerCount, Math.max(minWorkerCount, maxWorkerCount));
{'fraction_non_alphanumeric': 0.10019032082653616, 'fraction_numerical': 0.010603588907014683, 'mean_word_length': 2.2804905239687847, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7051655', 'n_tokens_mistral': 4055, 'n_tokens_neox': 3859, 'n_words': 825}