<commit_before>package com.petuum.ps.server;

import com.petuum.ps.common.HostInfo;
import com.petuum.ps.common.NumberedMsg;
import com.petuum.ps.common.TableInfo;
import com.petuum.ps.common.comm.CommBus;
import com.petuum.ps.common.consistency.ConsistencyModel;
import com.petuum.ps.common.util.IntBox;
import com.petuum.ps.thread.*;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import zmq.Msg;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.Vector;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
 * Created by ZengJichuan on 2014/8/13.
 */

class ServerContext{
    Vector<Integer> bgThreadIds;
    Server serverObj;
    int numShutdownBgs;
}

public class ServerThreads {

    private static Logger log = LogManager.getLogger(ServerThread.class);
    private static CyclicBarrier initBarrier;
    private static Vector<Integer> threadIDs;
    private static Vector<ServerThread> threads;
    private static ThreadLocal<ServerContext> serverContext;
    private static Method commBusRecvAny;
    private static Method commBusRecvTimeOutAny;
    private static Method commBusSendAny;
    private static Method commBusRecvAsyncAny;
    private static Method commBusRecvAnyWrapper;
    private static Method commBusRecvAnyBusy;
    private static Method commBusRecvAnySleep;

    private static Method serverPushRow;
    private static Method rowSubscribe;

    private static CommBus comm_bus;

    private static class ConnectionResult {
        public int senderID;
        public boolean isClient;
        public int clientID;
    }

    private class ServerThread extends Thread {

        private int threadID;

        public void run() {
            ThreadContext.registerThread(threadID);
            //set up thread-specific server context
            setupServerContext();
            setupCommBus();
            try {
                initBarrier.await();
                initServer(threadID);

                IntBox senderID = new IntBox();
                Msg zmqMsg = new Msg();
                boolean destroy_mem = false;

                while(true) {
                    commBusRecvAnyWrapper.invoke(comm_bus, senderID, zmqMsg);
                    int msgType = new NumberedMsg(zmqMsg).getMsgType();

                    switch (msgType) {
                        case NumberedMsg.K_CLIENT_SHUT_DOWN:
                            log.info("get ClientShutDown from bg " + String.valueOf(senderID.intValue));
                            if(handleShutDownMsg()) {
                                log.info("Server shutdown");
                                comm_bus.threadDeregister();
                                return;
                            }
                        case NumberedMsg.K_CREATE_TABLE:
                            handleCreateTable(senderID.intValue, new CreateTableMsg(zmqMsg));
                            break;
                        case NumberedMsg.K_ROW_REQUEST:
                            handleRowRequest(senderID.intValue, new RowRequestMsg(zmqMsg));
                            break;
                        case NumberedMsg.K_CLIENT_SEND_OP_LOG:
                            handleOpLogMsg(senderID.intValue, new ClientSendOpLogMsg(zmqMsg));
                            break;
                        default:
                            log.error("Unrecognized message type " + String.valueOf(msgType));
                    }

                }

            } catch (InterruptedException e) {
                log.error(e.getMessage());
            } catch (BrokenBarrierException e) {
                log.error(e.getMessage());
            } catch (IllegalAccessException e) {
                log.error(e.getMessage());
            } catch (InvocationTargetException e) {
                log.error(e.getMessage());
            }

        }

        public void setThreadID(int id) {
            threadID = id;
        }
    }

    public static void init(int idST) throws NoSuchMethodException, BrokenBarrierException, InterruptedException {

        initBarrier = new CyclicBarrier(GlobalContext.getNumLocalServerThreads() + 1);
        threads = new Vector<ServerThread>(GlobalContext.getNumLocalServerThreads());
        threadIDs = new Vector<Integer>(GlobalContext.getNumLocalServerThreads());
        comm_bus = GlobalContext.commBus;

        if(GlobalContext.getNumClients() == 1) {
            commBusRecvAny = CommBus.class.getMethod("recvInproc", IntBox.class, Msg.class);
            commBusRecvAsyncAny = CommBus.class.getMethod("recvInprocAsync", IntBox.class, Msg.class);
            commBusRecvTimeOutAny = CommBus.class.getMethod("recvInprocTimeout", IntBox.class, Msg.class, long.class);
            commBusSendAny = CommBus.class.getMethod("sendInproc", int.class, ByteBuffer.class);
        }else {
            commBusRecvAny = CommBus.class.getMethod("recv", IntBox.class, Msg.class);
            commBusRecvAsyncAny = CommBus.class.getMethod("recvAsync", IntBox.class, Msg.class);
            commBusRecvTimeOutAny = CommBus.class.getMethod("recvTimeOut", IntBox.class, Msg.class, long.class);
            commBusSendAny = CommBus.class.getMethod("send", int.class, ByteBuffer.class);
        }

        //
        ConsistencyModel consistency_model = GlobalContext.getConsistencyModel();
        switch(consistency_model) {
            case SSP:
                serverPushRow = ServerThreads.class.getMethod("SSPPushServerPushRow");
                rowSubscribe = ServerThreads.class.getMethod("SSPRowSubscribe");
                break;
            case SSPPush:
                serverPushRow = ServerThreads.class.getMethod("SSPPushServerPushRow");
                rowSubscribe = ServerThreads.class.getMethod("SSPPushRowSubscribe");
                log.info("RowSubscribe = SSPPushRowSubscribe");
                break;
            default:
                log.error("Unrecognized consistency model " + consistency_model.toString());
        }

        if(GlobalContext.isAggressiveCpu()) {
            commBusRecvAnyWrapper = commBusRecvAnyBusy;
        } else {
            commBusRecvAnyWrapper = commBusRecvAnySleep;
        }

        for(int i = 0; i < GlobalContext.getNumLocalServerThreads(); i++) {
            threadIDs.set(i, idST + i);
            log.info("Create server thread " + String.valueOf(i));
            threads.get(i).setThreadID(idST + i);
            threads.get(i).start();
        }
        initBarrier.await();
    }

    private static void SSPPushServerPushRow() throws NoSuchMethodException {

        serverContext.get().serverObj.createSendServerPushRowMsgs(ServerThreads.class.getMethod("sendServerPushRowMsg"));

    }
    private static void SSPRowSubscribe(){

    }
    // communication function
    // assuming the caller is not name node
    private static void connectToNameNode(){
        int nameNodeID = GlobalContext.getNameNodeId();

        if(comm_bus.isLocalEntity(nameNodeID)) {
            log.info("Connect to local name node");
            comm_bus.connectTo(nameNodeID, new ServerConnectMsg(null).getByteBuffer());
        } else {
            log.info("Connect to remote name node");
            HostInfo nameNodeInfo = GlobalContext.getHostInfo(nameNodeID);
            String nameNodeAddr = nameNodeInfo.ip + ":" + nameNodeInfo.port;
            log.info("name_node_addr = " + String.valueOf(nameNodeAddr));
            comm_bus.connectTo(nameNodeID, nameNodeAddr, new ServerConnectMsg(null).getByteBuffer());
        }
    }
<commit_msg>debug at runnable ServerThread<commit_after>package com.petuum.ps.server;

import com.petuum.ps.common.HostInfo;
import com.petuum.ps.common.NumberedMsg;
import com.petuum.ps.common.TableInfo;
import com.petuum.ps.common.comm.CommBus;
import com.petuum.ps.common.consistency.ConsistencyModel;
import com.petuum.ps.common.util.IntBox;
import com.petuum.ps.thread.*;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import zmq.Msg;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Vector;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
 * Created by ZengJichuan on 2014/8/13.
 */

class ServerContext{
    Vector<Integer> bgThreadIds;
    Server serverObj;
    int numShutdownBgs;
}

public class ServerThreads {

    private static Logger log = LogManager.getLogger(ServerThread.class);
    private static CyclicBarrier initBarrier;
    private static int[] threadIDs;
    private static ArrayList<Thread> threads = new ArrayList<Thread>();
    private static ThreadLocal<ServerContext> serverContext;
    private static Method commBusRecvAny;
    private static Method commBusRecvTimeOutAny;
    private static Method commBusSendAny;
    private static Method commBusRecvAsyncAny;
    private static Method commBusRecvAnyWrapper;
    private static Method commBusRecvAnyBusy;
    private static Method commBusRecvAnySleep;

    private static Method serverPushRow;
    private static Method rowSubscribe;

    private static CommBus comm_bus;

    private static class ConnectionResult {
        public int senderID;
        public boolean isClient;
        public int clientID;
    }

    private static class ServerThread implements Runnable {

        public ServerThread(int threadID) {
            this.threadID = threadID;
        }

        private int threadID;

        public void run() {
            ThreadContext.registerThread(threadID);
            //set up thread-specific server context
            setupServerContext();
            setupCommBus();
            try {
                initBarrier.await();
                initServer(threadID);

                IntBox senderID = new IntBox();
                Msg zmqMsg = new Msg();
                boolean destroy_mem = false;

                while(true) {
                    commBusRecvAnyWrapper.invoke(comm_bus, senderID, zmqMsg);
                    int msgType = new NumberedMsg(zmqMsg).getMsgType();

                    switch (msgType) {
                        case NumberedMsg.K_CLIENT_SHUT_DOWN:
                            log.info("get ClientShutDown from bg " + String.valueOf(senderID.intValue));
                            if(handleShutDownMsg()) {
                                log.info("Server shutdown");
                                comm_bus.threadDeregister();
                                return;
                            }
                        case NumberedMsg.K_CREATE_TABLE:
                            handleCreateTable(senderID.intValue, new CreateTableMsg(zmqMsg));
                            break;
                        case NumberedMsg.K_ROW_REQUEST:
                            handleRowRequest(senderID.intValue, new RowRequestMsg(zmqMsg));
                            break;
                        case NumberedMsg.K_CLIENT_SEND_OP_LOG:
                            handleOpLogMsg(senderID.intValue, new ClientSendOpLogMsg(zmqMsg));
                            break;
                        default:
                            log.error("Unrecognized message type " + String.valueOf(msgType));
                    }

                }

            } catch (InterruptedException e) {
                log.error(e.getMessage());
            } catch (BrokenBarrierException e) {
                log.error(e.getMessage());
            } catch (IllegalAccessException e) {
                log.error(e.getMessage());
            } catch (InvocationTargetException e) {
                log.error(e.getMessage());
            }

        }
    }

    public static void init(int idST) throws NoSuchMethodException, BrokenBarrierException, InterruptedException {

        initBarrier = new CyclicBarrier(GlobalContext.getNumLocalServerThreads() + 1);
        threadIDs = new int[GlobalContext.getNumLocalServerThreads()];
        comm_bus = GlobalContext.commBus;

        if(GlobalContext.getNumClients() == 1) {
            commBusRecvAny = CommBus.class.getMethod("recvInproc", IntBox.class, Msg.class);
            commBusRecvAsyncAny = CommBus.class.getMethod("recvInprocAsync", IntBox.class, Msg.class);
            commBusRecvTimeOutAny = CommBus.class.getMethod("recvInprocTimeout", IntBox.class, Msg.class, long.class);
            commBusSendAny = CommBus.class.getMethod("sendInproc", int.class, ByteBuffer.class);
        }else {
            commBusRecvAny = CommBus.class.getMethod("recv", IntBox.class, Msg.class);
            commBusRecvAsyncAny = CommBus.class.getMethod("recvAsync", IntBox.class, Msg.class);
            commBusRecvTimeOutAny = CommBus.class.getMethod("recvTimeOut", IntBox.class, Msg.class, long.class);
            commBusSendAny = CommBus.class.getMethod("send", int.class, ByteBuffer.class);
        }

        //
        ConsistencyModel consistency_model = GlobalContext.getConsistencyModel();
        switch(consistency_model) {
            case SSP:
                serverPushRow = ServerThreads.class.getMethod("SSPPushServerPushRow");
                rowSubscribe = ServerThreads.class.getMethod("SSPRowSubscribe");
                break;
            case SSPPush:
                serverPushRow = ServerThreads.class.getMethod("SSPPushServerPushRow");
                rowSubscribe = ServerThreads.class.getMethod("SSPPushRowSubscribe");
                log.info("RowSubscribe = SSPPushRowSubscribe");
                break;
            default:
                log.error("Unrecognized consistency model " + consistency_model.toString());
        }

        if(GlobalContext.isAggressiveCpu()) {
            commBusRecvAnyWrapper = commBusRecvAnyBusy;
        } else {
            commBusRecvAnyWrapper = commBusRecvAnySleep;
        }

        for(int i = 0; i < GlobalContext.getNumLocalServerThreads(); i++) {
            threadIDs[i] = idST + i;
            log.info("Create server thread " + String.valueOf(i));
            threads.add(new Thread(new ServerThread(idST + i)));
            threads.get(i).start();
        }
        initBarrier.await();
    }

    public static void SSPPushServerPushRow() throws NoSuchMethodException {

        serverContext.get().serverObj.createSendServerPushRowMsgs(ServerThreads.class.getMethod("sendServerPushRowMsg"));

    }
    public static void SSPPushRowSubscribe(){

    }
    // communication function
    // assuming the caller is not name node
    private static void connectToNameNode(){
        int nameNodeID = GlobalContext.getNameNodeId();

        if(comm_bus.isLocalEntity(nameNodeID)) {
            log.info("Connect to local name node");
            comm_bus.connectTo(nameNodeID, new ServerConnectMsg(null).getByteBuffer());
        } else {
            log.info("Connect to remote name node");
            HostInfo nameNodeInfo = GlobalContext.getHostInfo(nameNodeID);
            String nameNodeAddr = nameNodeInfo.ip + ":" + nameNodeInfo.port;
            log.info("name_node_addr = " + String.valueOf(nameNodeAddr));
            comm_bus.connectTo(nameNodeID, nameNodeAddr, new ServerConnectMsg(null).getByteBuffer());
        }
    }

{'fraction_non_alphanumeric': 0.08909791625841008, 'fraction_numerical': 0.0015677052714089751, 'mean_word_length': 2.2340515420363327, 'pattern_counts': {'":': 2, '<': 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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3310240', 'n_tokens_mistral': 4164, 'n_tokens_neox': 4008, 'n_words': 776}