<commit_before>package org.jgroups.demos;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

import org.jgroups.JChannel;
import org.jgroups.blocks.executor.ExecutionCompletionService;
import org.jgroups.blocks.executor.ExecutionRunner;
import org.jgroups.blocks.executor.ExecutionService;
import org.jgroups.jmx.JmxConfigurator;
import org.jgroups.protocols.CENTRAL_EXECUTOR;
import org.jgroups.util.Streamable;
import org.jgroups.util.Util;

public class ExecutionServiceDemo {
    protected String props;
    protected JChannel ch;
    protected ExecutionService execution_service;
    protected String name;
    protected ExecutionRunner runner;
    protected int size;
    protected boolean printValues;
    protected Random random;
    
    protected ExecutorService executor;
    protected Queue<Future<?>> queue;

    public ExecutionServiceDemo(String props, String name, int size) {
        this.props=props;
        this.name=name;
        queue=new ArrayDeque<Future<?>>();
        executor = Executors.newCachedThreadPool(new ThreadFactory() {
            
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "Consumer-" + 
                    poolNumber.getAndIncrement());
                thread.setDaemon(true);
                return thread;
            }
            AtomicInteger poolNumber = new AtomicInteger();
        });
        this.size=size;
    }
    
    public static void main(String[] args) throws Exception {
        String props=null;
        String name=null;
        String size="1000";
        

        for(int i=0; i < args.length; i++) {
            if(args[i].equals("-props")) {
                props=args[++i];
                continue;
            }
            if(args[i].equals("-name")) {
                name=args[++i];
                continue;
            }


            help();
            return;
        }

        ExecutionServiceDemo demo=new ExecutionServiceDemo(props, name, 
            Integer.valueOf(size));
        demo.start();
    }
    
    public void start() throws Exception {
        ch=new JChannel(props);
        if(name != null)
            ch.setName(name);
        ch.getProtocolStack().insertProtocolAtTop(new CENTRAL_EXECUTOR());
        execution_service=new ExecutionService(ch);
        runner=new ExecutionRunner(ch);
        ch.connect("executing-cluster");
        JmxConfigurator.registerChannel(ch, Util.getMBeanServer(), 
            "execution-service", ch.getClusterName(), true);
        
        // Start a consumer
        queue.add(executor.submit(runner));
        random = new Random();
        printValues = false;

        try {
            loop();
        }
        catch(Exception e) {
            e.printStackTrace();<commit_msg>Forgot to remove adding of CENTRAL_EXECUTOR to the stack.<commit_after>package org.jgroups.demos;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

import org.jgroups.JChannel;
import org.jgroups.blocks.executor.ExecutionCompletionService;
import org.jgroups.blocks.executor.ExecutionRunner;
import org.jgroups.blocks.executor.ExecutionService;
import org.jgroups.jmx.JmxConfigurator;
import org.jgroups.util.Streamable;
import org.jgroups.util.Util;

public class ExecutionServiceDemo {
    protected String props;
    protected JChannel ch;
    protected ExecutionService execution_service;
    protected String name;
    protected ExecutionRunner runner;
    protected int size;
    protected boolean printValues;
    protected Random random;
    
    protected ExecutorService executor;
    protected Queue<Future<?>> queue;

    public ExecutionServiceDemo(String props, String name, int size) {
        this.props=props;
        this.name=name;
        queue=new ArrayDeque<Future<?>>();
        executor = Executors.newCachedThreadPool(new ThreadFactory() {
            
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "Consumer-" + 
                    poolNumber.getAndIncrement());
                thread.setDaemon(true);
                return thread;
            }
            AtomicInteger poolNumber = new AtomicInteger();
        });
        this.size=size;
    }
    
    public static void main(String[] args) throws Exception {
        String props=null;
        String name=null;
        String size="1000";
        

        for(int i=0; i < args.length; i++) {
            if(args[i].equals("-props")) {
                props=args[++i];
                continue;
            }
            if(args[i].equals("-name")) {
                name=args[++i];
                continue;
            }


            help();
            return;
        }

        ExecutionServiceDemo demo=new ExecutionServiceDemo(props, name, 
            Integer.valueOf(size));
        demo.start();
    }
    
    public void start() throws Exception {
        ch=new JChannel(props);
        if(name != null)
            ch.setName(name);
        execution_service=new ExecutionService(ch);
        runner=new ExecutionRunner(ch);
        ch.connect("executing-cluster");
        JmxConfigurator.registerChannel(ch, Util.getMBeanServer(), 
            "execution-service", ch.getClusterName(), true);
        
        // Start a consumer
        queue.add(executor.submit(runner));
        random = new Random();
        printValues = false;

        try {
            loop();
        }
        catch(Exception e) {
            e.printStackTrace();
{'fraction_non_alphanumeric': 0.10082435003170577, 'fraction_numerical': 0.0015852885225110971, 'mean_word_length': 2.3683929524826484, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 11, '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': '5555890', 'n_tokens_mistral': 1740, 'n_tokens_neox': 1682, 'n_words': 359}