<commit_before>package org.jgroups.tests.perf;

import org.jgroups.util.Util;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.*;

/**  You start the test by running this class.
 * Use parameters -Xbatch -Xconcurrentio (Solaris specific)
 * @author <NAME> (<EMAIL>)

 */
public class Test implements Receiver {
    String          props=null;
    Properties      config;
    boolean         sender=false;
    Transport       transport=null;
    Object          local_addr=null;

    /** HashMap<Object,MemberInfo> members. Keys=member addresses, value=MemberInfo */
    HashMap         senders=new HashMap();

    /** Keeps track of members */
    ArrayList       members=new ArrayList();

    /** Set when first message is received */
    long            start=0;

    /** Set when last message is received */
    long            stop=0;

    int             num_members=0;

    Log             log=LogFactory.getLog(getClass());

    boolean         all_received=false;

    /** HashMap<Object, HashMap>. A hashmap of senders, each value is the 'senders' hashmap */
    HashMap         results=new HashMap();

    /** Dump info in gnuplot format */
    boolean         gnuplot_output=false;

    /** Log every n msgs received (if gnuplot_output == true) */
    long            log_interval=1000;

    /** Last time we dumped info */
    long            last_dump=0;

    long            counter=1;
    long            msg_size=1000;
    boolean         jmx=false;



    public void start(Properties c, boolean verbose) throws Exception {
        String          config_file="config.txt";
        BufferedReader  fileReader;
        String          line;
        String          key, val;
        StringTokenizer st;
        Properties      tmp=new Properties();

        config_file=c.getProperty("config");
        fileReader=new BufferedReader(new FileReader(config_file));
        while((line=fileReader.readLine()) != null) {
            if(line.startsWith("#"))
                continue;
            line=line.trim();
            if(line.length() == 0)
                continue;
            st=new StringTokenizer(line, "=", false);
            key=st.nextToken().toLowerCase();
            val=st.nextToken();
            tmp.put(key, val);
        }
        fileReader.close();

        // 'tmp' now contains all properties from the file, now we need to override the ones
        // passed to us by 'c'
        tmp.putAll(c);
        this.config=tmp;

        StringBuffer sb=new StringBuffer();
        sb.append("\n\n----------------------- TEST -----------------------\n");
        sb.append("Date: ").append(new Date()).append('\n');
        sb.append("Run by: ").append(System.getProperty("user.name")).append("\n\n");
        if(verbose)
            sb.append("Properties: ").append(printProperties()).append("\n-------------------------\n\n");

        for(Iterator it=this.config.entrySet().iterator(); it.hasNext();) {
            Map.Entry entry=(Map.Entry)it.next();
            sb.append(entry.getKey()).append(":\t").append(entry.getValue()).append('\n');
        }
        sb.append('\n');
        System.out.println("Configuration is: " + sb);

        log.info(sb.toString());

        props=this.config.getProperty("props");
        num_members=Integer.parseInt(this.config.getProperty("num_members"));
        sender=Boolean.valueOf(this.config.getProperty("sender")).booleanValue();
        msg_size=Long.parseLong(this.config.getProperty("msg_size"));
        String tmp2=this.config.getProperty("gnuplot_output", "false");
        if(Boolean.valueOf(tmp2).booleanValue())
            this.gnuplot_output=true;
        tmp2=this.config.getProperty("log_interval");
        if(tmp2 != null)
            log_interval=Long.parseLong(tmp2);

        if(gnuplot_output) {
            sb=new StringBuffer();
            sb.append("\n##### msgs_received");
            sb.append(", free_mem [KB] ");
            sb.append(", total_mem [KB] ");
            sb.append(", total_msgs_sec [msgs/sec] ");
            sb.append(", total_throughput [KB/sec] ");
            sb.append(", rolling_msgs_sec (last ").append(log_interval).append(" msgs) ");
            sb.append(" [msgs/sec] ");
            sb.append(", rolling_throughput (last ").append(log_interval).append(" msgs) ");
            sb.append(" [KB/sec]\n");
            if(log.isInfoEnabled()) log.info(sb.toString());
        }
        jmx=new Boolean(this.config.getProperty("jmx")).booleanValue();

        String transport_name=this.config.getProperty("transport");
        transport=(Transport)Thread.currentThread().getContextClassLoader().loadClass(transport_name).newInstance();
        transport.create(this.config);
        transport.setReceiver(this);
        transport.start();
        local_addr=transport.getLocalAddress();
    }

    private String printProperties() {
        StringBuffer sb=new StringBuffer();
        Properties p=System.getProperties();
        for(Iterator it=p.entrySet().iterator(); it.hasNext();) {
            Map.Entry entry=(Map.Entry)it.next();
            sb.append(entry.getKey()).append(": ").append(entry.getValue()).append('\n');
        }
        return sb.toString();
    }

    public void stop() {
        if(transport != null) {
            transport.stop();
            transport.destroy();
        }
    }

    public void receive(Object sender, byte[] payload) {
        Data d;

        try {
            int type=payload[0];
            if(type == 1) { // DATA
                int len=payload.length -1;
                handleData(sender, len);
                return;
            }

            byte[] tmp=new byte[payload.length-1];
            System.arraycopy(payload, 1, tmp, 0, tmp.length);
            d=(Data)Util.objectFromByteBuffer(tmp);

            switch(d.getType()) {
                case Data.DISCOVERY_REQ:
                    sendDiscoveryResponse();
                    break;
                case Data.DISCOVERY_RSP:
                    synchronized(this.members) {
                        if(!this.members.contains(sender)) {
                            this.members.add(sender);
                            System.out.println("-- " + sender + " joined");
                            if(d.sender) {
                                synchronized(this.members) {
                                    if(!this.senders.containsKey(sender)) {
                                        this.senders.put(sender, new MemberInfo(d.num_msgs));
                                    }
                                }
                            }
                            this.members.notify();
                        }
                    }
                    break;

                case Data.DONE:
                    if(all_received)
                        return;
                    MemberInfo mi=(MemberInfo)this.senders.get(sender);
                    if(mi != null) {
                        mi.done=true;
                        if(mi.stop == 0)
                            mi.stop=System.currentTimeMillis();
                        if(allReceived()) {
                            all_received=true;
                            if(stop == 0)
                                stop=System.currentTimeMillis();
                            sendResults();
                            if(!this.sender)
                                dumpSenders();
                            synchronized(this) {
                                this.notify();
                            }
                        }
                    }
                    else {
                        log.error("-- sender " + sender + " not found in senders hashmap");
                    }
                    break;

                case Data.RESULTS:
                    synchronized(results) {
                        if(!results.containsKey(sender)) {
                            results.put(sender, d.results);
                            results.notify();
                        }
                    }
                    break;

                default:
                    log.error("received invalid data type: " + payload[0]);
                    break;
            }
        }
        catch(Exception e) {
            e.printStackTrace();
        }
    }

    private void handleData(Object sender, int num_bytes) {
        if(all_received)
            return;
        if(start == 0) {
            start=System.currentTimeMillis();
            last_dump=start;
        }
        MemberInfo info=(MemberInfo)this.senders.get(sender);
        if(info != null) {
            if(info.start == 0)
                info.start=System.currentTimeMillis();
            info.num_msgs_received++;
            counter++;
            info.total_bytes_received+=num_bytes;
            if(info.num_msgs_received % 1000 == 0)
                System.out.println("-- received " + info.num_msgs_received +
                                   " messages from " + sender);

            if(counter % log_interval == 0) {
                if(log.isInfoEnabled()) log.info(dumpStats(counter));
            }

            if(info.num_msgs_received >= info.num_msgs_expected) {
                info.done=true;
                if(info.stop == 0)
                    info.stop=System.currentTimeMillis();
                if(allReceived()) {
                    all_received=true;
                    if(stop == 0)
                        stop=System.currentTimeMillis();
                    try {
                        sendResults();
                    }
                    catch(Exception e) {
                        e.printStackTrace();
                    }
                    if(!this.sender)
                        dumpSenders();
                    synchronized(this) {
                        this.notify();
                    }
                }
            }
        }
        else {
            log.error("-- sender " + sender + " not found in senders hashmap");
        }
    }

    void sendResults() throws Exception {
        Data d=new Data(Data.RESULTS);
        byte[] buf;
        d.results=(HashMap)this.senders.clone();
        buf=generatePayload(d, null);
        transport.send(null, buf);
    }

    boolean allReceived() {
        MemberInfo mi;

        for(Iterator it=this.senders.values().iterator(); it.hasNext();) {
            mi=(MemberInfo)it.next();
            if(mi.done == false)
                return false;
        }
        return true;
    }


    void sendMessages() throws Exception {
        long total_msgs=0;
        int msgSize=Integer.parseInt(config.getProperty("msg_size"));
        int num_msgs=Integer.parseInt(config.getProperty("num_msgs"));
        int logInterval=Integer.parseInt(config.getProperty("log_interval"));
        // boolean gnuplot_output=Boolean.getBoolean(config.getProperty("gnuplot_output", "false"));
        byte[] buf=new byte[msgSize];
        for(int k=0; k < msgSize; k++)
            buf[k]='.';
        Data d=new Data(Data.DATA);
        byte[] payload=generatePayload(d, buf);
        for(int i=0; i < num_msgs; i++) {
            transport.send(null, payload);
            total_msgs++;
            if(total_msgs % 1000 == 0) {
                System.out.println("++ sent " + total_msgs);
            }
            if(total_msgs % logInterval == 0) {
                //if(gnuplot_output == false)
                //  if(log.isInfoEnabled()) log.info(dumpStats(total_msgs));
            }
        }
    }


    byte[] generatePayload(Data d, byte[] buf) throws Exception {
        byte[] tmp=buf != null? buf : Util.objectToByteBuffer(d);
        byte[] payload=new byte[tmp.length +1];
        payload[0]=intToByte(d.getType());
        System.arraycopy(tmp, 0, payload, 1, tmp.length);
        return payload;
    }

    private byte intToByte(int type) {
        switch(type) {
            case Data.DATA: return 1;
            case Data.DISCOVERY_REQ: return 2;
            case Data.DISCOVERY_RSP: return 3;
            case Data.DONE: return 4;
            case Data.RESULTS: return 5;<commit_msg>no message<commit_after>package org.jgroups.tests.perf;

import org.jgroups.util.Util;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.*;

/**  You start the test by running this class.
 * Use parameters -Xbatch -Xconcurrentio (Solaris specific)
 * @author <NAME> (<EMAIL>)

 */
public class Test implements Receiver {
    String          props=null;
    Properties      config;
    boolean         sender=false;
    Transport       transport=null;
    Object          local_addr=null;

    /** HashMap<Object,MemberInfo> members. Keys=member addresses, value=MemberInfo */
    HashMap         senders=new HashMap();

    /** Keeps track of members */
    ArrayList       members=new ArrayList();

    /** Set when first message is received */
    long            start=0;

    /** Set when last message is received */
    long            stop=0;

    int             num_members=0;

    Log             log=LogFactory.getLog(getClass());

    boolean         all_received=false;

    /** HashMap<Object, HashMap>. A hashmap of senders, each value is the 'senders' hashmap */
    HashMap         results=new HashMap();

    /** Dump info in gnuplot format */
    boolean         gnuplot_output=false;

    /** Log every n msgs received (if gnuplot_output == true) */
    long            log_interval=1000;

    /** Last time we dumped info */
    long            last_dump=0;

    long            counter=1;
    long            msg_size=1000;
    boolean         jmx=false;




    public void start(Properties c, boolean verbose) throws Exception {
        String          config_file="config.txt";
        BufferedReader  fileReader;
        String          line;
        String          key, val;
        StringTokenizer st;
        Properties      tmp=new Properties();

        config_file=c.getProperty("config");
        fileReader=new BufferedReader(new FileReader(config_file));
        while((line=fileReader.readLine()) != null) {
            if(line.startsWith("#"))
                continue;
            line=line.trim();
            if(line.length() == 0)
                continue;
            st=new StringTokenizer(line, "=", false);
            key=st.nextToken().toLowerCase();
            val=st.nextToken();
            tmp.put(key, val);
        }
        fileReader.close();

        // 'tmp' now contains all properties from the file, now we need to override the ones
        // passed to us by 'c'
        tmp.putAll(c);
        this.config=tmp;

        StringBuffer sb=new StringBuffer();
        sb.append("\n\n----------------------- TEST -----------------------\n");
        sb.append("Date: ").append(new Date()).append('\n');
        sb.append("Run by: ").append(System.getProperty("user.name")).append("\n\n");
        if(verbose)
            sb.append("Properties: ").append(printProperties()).append("\n-------------------------\n\n");

        for(Iterator it=this.config.entrySet().iterator(); it.hasNext();) {
            Map.Entry entry=(Map.Entry)it.next();
            sb.append(entry.getKey()).append(":\t").append(entry.getValue()).append('\n');
        }
        sb.append('\n');
        System.out.println("Configuration is: " + sb);

        log.info(sb.toString());

        props=this.config.getProperty("props");
        num_members=Integer.parseInt(this.config.getProperty("num_members"));
        sender=Boolean.valueOf(this.config.getProperty("sender")).booleanValue();
        msg_size=Long.parseLong(this.config.getProperty("msg_size"));
        String tmp2=this.config.getProperty("gnuplot_output", "false");
        if(Boolean.valueOf(tmp2).booleanValue())
            this.gnuplot_output=true;
        tmp2=this.config.getProperty("log_interval");
        if(tmp2 != null)
            log_interval=Long.parseLong(tmp2);

        if(gnuplot_output) {
            sb=new StringBuffer();
            sb.append("\n##### msgs_received");
            sb.append(", free_mem [KB] ");
            sb.append(", total_mem [KB] ");
            sb.append(", total_msgs_sec [msgs/sec] ");
            sb.append(", total_throughput [KB/sec] ");
            sb.append(", rolling_msgs_sec (last ").append(log_interval).append(" msgs) ");
            sb.append(" [msgs/sec] ");
            sb.append(", rolling_throughput (last ").append(log_interval).append(" msgs) ");
            sb.append(" [KB/sec]\n");
            if(log.isInfoEnabled()) log.info(sb.toString());
        }
        jmx=new Boolean(this.config.getProperty("jmx")).booleanValue();

        String transport_name=this.config.getProperty("transport");
        transport=(Transport)Thread.currentThread().getContextClassLoader().loadClass(transport_name).newInstance();
        transport.create(this.config);
        transport.setReceiver(this);
        transport.start();
        local_addr=transport.getLocalAddress();
    }

    private String printProperties() {
        StringBuffer sb=new StringBuffer();
        Properties p=System.getProperties();
        for(Iterator it=p.entrySet().iterator(); it.hasNext();) {
            Map.Entry entry=(Map.Entry)it.next();
            sb.append(entry.getKey()).append(": ").append(entry.getValue()).append('\n');
        }
        return sb.toString();
    }

    public void stop() {
        if(transport != null) {
            transport.stop();
            transport.destroy();
        }
    }

    public void receive(Object sender, byte[] payload) {
        Data d;

        try {
            int type=payload[0];
            if(type == 1) { // DATA
                int len=payload.length -1;
                handleData(sender, len);
                return;
            }

            byte[] tmp=new byte[payload.length-1];
            System.arraycopy(payload, 1, tmp, 0, tmp.length);
            d=(Data)Util.objectFromByteBuffer(tmp);

            switch(d.getType()) {
                case Data.DISCOVERY_REQ:
                    sendDiscoveryResponse();
                    break;
                case Data.DISCOVERY_RSP:
                    synchronized(this.members) {
                        if(!this.members.contains(sender)) {
                            this.members.add(sender);
                            System.out.println("-- " + sender + " joined");
                            if(d.sender) {
                                synchronized(this.members) {
                                    if(!this.senders.containsKey(sender)) {
                                        this.senders.put(sender, new MemberInfo(d.num_msgs));
                                    }
                                }
                            }
                            this.members.notify();
                        }
                    }
                    break;

                case Data.DONE:
                    if(all_received)
                        return;
                    MemberInfo mi=(MemberInfo)this.senders.get(sender);
                    if(mi != null) {
                        mi.done=true;
                        if(mi.stop == 0)
                            mi.stop=System.currentTimeMillis();
                        if(allReceived()) {
                            all_received=true;
                            if(stop == 0)
                                stop=System.currentTimeMillis();
                            sendResults();
                            if(!this.sender)
                                dumpSenders();
                            synchronized(this) {
                                this.notify();
                            }
                        }
                    }
                    else {
                        log.error("-- sender " + sender + " not found in senders hashmap");
                    }
                    break;

                case Data.RESULTS:
                    synchronized(results) {
                        if(!results.containsKey(sender)) {
                            results.put(sender, d.results);
                            results.notify();
                        }
                    }
                    break;

                default:
                    log.error("received invalid data type: " + payload[0]);
                    break;
            }
        }
        catch(Exception e) {
            e.printStackTrace();
        }
    }

    private void handleData(Object sender, int num_bytes) {
        if(all_received)
            return;
        if(start == 0) {
            start=System.currentTimeMillis();
            last_dump=start;
        }
        MemberInfo info=(MemberInfo)this.senders.get(sender);
        if(info != null) {
            if(info.start == 0)
                info.start=System.currentTimeMillis();
            info.num_msgs_received++;
            counter++;
            info.total_bytes_received+=num_bytes;
            if(info.num_msgs_received % log_interval == 0)
                System.out.println("-- received " + info.num_msgs_received +
                                   " messages from " + sender);

            if(counter % log_interval == 0) {
                if(log.isInfoEnabled()) log.info(dumpStats(counter));
            }

            if(info.num_msgs_received >= info.num_msgs_expected) {
                info.done=true;
                if(info.stop == 0)
                    info.stop=System.currentTimeMillis();
                if(allReceived()) {
                    all_received=true;
                    if(stop == 0)
                        stop=System.currentTimeMillis();
                    try {
                        sendResults();
                    }
                    catch(Exception e) {
                        e.printStackTrace();
                    }
                    if(!this.sender)
                        dumpSenders();
                    synchronized(this) {
                        this.notify();
                    }
                }
            }
        }
        else {
            log.error("-- sender " + sender + " not found in senders hashmap");
        }
    }

    void sendResults() throws Exception {
        Data d=new Data(Data.RESULTS);
        byte[] buf;
        d.results=(HashMap)this.senders.clone();
        buf=generatePayload(d, null);
        transport.send(null, buf);
    }

    boolean allReceived() {
        MemberInfo mi;

        for(Iterator it=this.senders.values().iterator(); it.hasNext();) {
            mi=(MemberInfo)it.next();
            if(mi.done == false)
                return false;
        }
        return true;
    }


    void sendMessages() throws Exception {
        long total_msgs=0;
        int msgSize=Integer.parseInt(config.getProperty("msg_size"));
        int num_msgs=Integer.parseInt(config.getProperty("num_msgs"));
        // int logInterval=Integer.parseInt(config.getProperty("log_interval"));
        // boolean gnuplot_output=Boolean.getBoolean(config.getProperty("gnuplot_output", "false"));
        byte[] buf=new byte[msgSize];
        for(int k=0; k < msgSize; k++)
            buf[k]='.';
        Data d=new Data(Data.DATA);
        byte[] payload=generatePayload(d, buf);
        for(int i=0; i < num_msgs; i++) {
            transport.send(null, payload);
            total_msgs++;
            if(total_msgs % log_interval == 0) {
                System.out.println("++ sent " + total_msgs);
            }
            if(total_msgs % log_interval == 0) {
                //if(gnuplot_output == false)
                //  if(log.isInfoEnabled()) log.info(dumpStats(total_msgs));
            }
        }
    }


    byte[] generatePayload(Data d, byte[] buf) throws Exception {
        byte[] tmp=buf != null? buf : Util.objectToByteBuffer(d);
        byte[] payload=new byte[tmp.length +1];
        payload[0]=intToByte(d.getType());
        System.arraycopy(tmp, 0, payload, 1, tmp.length);
        return payload;
    }

    private byte intToByte(int type) {
        switch(type) {
            case Data.DATA: return 1;
            case Data.DISCOVERY_REQ: return 2;
            case Data.DISCOVERY_RSP: return 3;
            case Data.DONE: return 4;
            case Data.RESULTS: return 5;
{'fraction_non_alphanumeric': 0.12755519215044972, 'fraction_numerical': 0.004251839738348324, 'mean_word_length': 1.4608651911468813, 'pattern_counts': {'":': 4, '<': 15, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4544696', 'n_tokens_mistral': 6507, 'n_tokens_neox': 6276, 'n_words': 1180}