Server/src/engine/job/JobManager.java

// • ▌ ▄ ·.  ▄▄▄·  ▄▄ • ▪   ▄▄· ▄▄▄▄·  ▄▄▄·  ▐▄▄▄  ▄▄▄ .
// ·██ ▐███▪▐█ ▀█ ▐█ ▀ ▪██ ▐█ ▌▪▐█ ▀█▪▐█ ▀█ •█▌ ▐█▐▌·
// ▐█ ▌▐▌▐█·▄█▀▀█ ▄█ ▀█▄▐█·██ ▄▄▐█▀▀█▄▄█▀▀█ ▐█▐ ▐▌▐▀▀▀
// ██ ██▌▐█▌▐█ ▪▐▌▐█▄▪▐█▐█▌▐███▌██▄▪▐█▐█ ▪▐▌██▐ █▌▐█▄▄▌
// ▀▀  █▪▀▀▀ ▀  ▀ ·▀▀▀▀ ▀▀▀·▀▀▀ ·▀▀▀▀  ▀  ▀ ▀▀  █▪ ▀▀▀
//      Magicbane Emulator Project © 2013 - 2022
//                www.magicbane.com


package engine.job;

import engine.core.ControlledRunnable;
import engine.server.MBServerStatics;
import engine.util.ThreadUtils;
import org.pmw.tinylog.Logger;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;


/**
 * Note to DEVs. When attempting to log something in this class, use logDIRECT
 * only.
 */
public class JobManager extends ControlledRunnable {

    /*
     * Singleton implementation.
     */
    private static volatile JobManager INSTANCE;
    private final ArrayList<JobPool> jobPoolList = new ArrayList<>();

    /*
     * Class implementation
     */
    private final ConcurrentHashMap<String, JobPool> jobQueueMapping = new ConcurrentHashMap<>(MBServerStatics.CHM_INIT_CAP, MBServerStatics.CHM_LOAD, MBServerStatics.CHM_THREAD_HIGH);
    private boolean shutdownNowFlag = false;

    private JobManager() {
        super("JobManager");

        // create the initial job pools with the correct sizes
        // based on the number of array elements in the initial_jo_workers
        // definition in Statisc

        if (MBServerStatics.INITIAL_JOBPOOL_WORKERS != null && MBServerStatics.INITIAL_JOBPOOL_WORKERS.length > 0) {
            for (int i = 0; i < MBServerStatics.INITIAL_JOBPOOL_WORKERS.length; i++) {
                JobPool jp = new JobPool(i, MBServerStatics.INITIAL_JOBPOOL_WORKERS[i]);
                this.jobPoolList.add(jp);
                Logger.info("Adding JobPool_" + jp.getJobPoolID() + " with " + MBServerStatics.INITIAL_JOBPOOL_WORKERS[i] + " workers");
            }
        } else {
            // not defined or empty in statics so just create 1 JobPool with 25 workers
            System.out.println("Creating 1 pool called 0");
            JobPool jp = new JobPool(0, 25);
            this.jobPoolList.add(jp);
            Logger.info("Adding JobPool_" + jp.getJobPoolID() + " with 25 workers");
        }

        JobScheduler.getInstance();

        // if you want any jobs to default onto a given queue put it here
        // otherwise everything goes on the P1 queue by default

    }

    public static JobManager getInstance() {
        if (JobManager.INSTANCE == null) {
            synchronized (JobManager.class) {
                if (JobManager.INSTANCE == null) {
                    JobManager.INSTANCE = new JobManager();
                    JobManager.INSTANCE.startup();
                }
            }
        }
        return JobManager.INSTANCE;
    }

    /**
     * Submit a job to be processed by the JobManager. There is no guarantee
     * that the job will be executed immediately.
     *
     * @param aj AbstractJob to be submitted.
     * @return boolean value indicating whether the job was submitted or not.
     */
    public boolean submitJob(AbstractJob aj) {

        if (jobQueueMapping.containsKey(aj.getClass().getSimpleName())) {
            return this.jobQueueMapping.get(aj.getClass().getSimpleName()).submitJob(aj);
        } else {
            return this.jobPoolList.get(0).submitJob(aj);
        }
    }


    private void auditAllWorkers() {

        for (JobPool jp : this.jobPoolList) {
            jp.auditWorkers();
        }
    }

    @Override
    protected boolean _Run() {
        // This thread is to be used to JM internal monitoring.

        // Startup workers:
        this.auditAllWorkers();
        ThreadUtils.sleep(MBServerStatics.JOBMANAGER_INTERNAL_MONITORING_INTERVAL_MS * 2);

        this.runStatus = true;

        // Monitoring loop
        while (this.runCmd) {
            this.auditAllWorkers();
            ThreadUtils.sleep(MBServerStatics.JOBMANAGER_INTERNAL_MONITORING_INTERVAL_MS);
        }

        // No new submissions
        for (JobPool jp : this.jobPoolList) {
            jp.setBlockNewSubmissions(true);
        }

        if (this.shutdownNowFlag == false) {
            // shutdown each pool
            for (JobPool jp : this.jobPoolList) {
                jp.shutdown();
            }
        } else {
            // Emergency Stop each pool
            for (JobPool jp : this.jobPoolList) {
                jp.emergencyStop();
            }
        }
        this.runStatus = false;
        Logger.info("JM Shutdown");
        return true;
    }

    @Override
    protected boolean _postRun() {
        return true;
    }

    @Override
    protected boolean _preRun() {
        return true;
    }

    @Override
    protected void _shutdown() {
        Logger.info("JM Shutdown Requested");
        JobScheduler.getInstance().shutdown();
    }

    @Override
    protected void _startup() {
        Logger.info("JM Starting up... ");
    }

    public void shutdownNow() {
        this.shutdownNowFlag = true;
        this.shutdown();
    }

    public String moveJob(String simpleClassName, String jobPoolID) {
        // moves a job to a different queue

        try {
            // parse string into an int
            Integer i = Integer.parseInt(jobPoolID);
            for (JobPool jp : this.jobPoolList) {
                if (jp.getJobPoolID() == i) {
                    // move the job to the new queue
                    this.jobQueueMapping.put(simpleClassName, jp);
                    return simpleClassName + " moved to JobPool ID " + jp.getJobPoolID();
                }
            }

        } catch (NumberFormatException e) {
            return "jobPoolID is not a valid number";
        }

        //Verify jobpool ID
        return "Invalid parameters <simpleClassName - case sensitive> <jobPoolID> required";
    }

    public String resetJobs() {
        // moves all jobs to the P1 queue
        this.jobQueueMapping.clear();
        return "All Jobs reset onto P1 queue";
    }

    public String showJobs() {
        String out = "";
        Iterator<String> jmi = this.jobQueueMapping.keySet().iterator();

        while (jmi.hasNext()) {
            String jmiKey = jmi.next();
            out += jmiKey + ' ' + this.jobQueueMapping.get(jmiKey).getJobPoolID() + '\n';
        }
        return out;
    }

    public ArrayList<JobPool> getJobPoolList() {

        return this.jobPoolList;
    }

    public String modifyJobPoolWorkers(String jobPoolID, String maxWorkers) {

        try {
            // parse string into an int
            Integer jid = Integer.parseInt(jobPoolID);
            Integer mw = Integer.parseInt(maxWorkers);
            for (JobPool jp : this.jobPoolList) {
                if (jp.getJobPoolID() == jid) {
                    // change the number of workers
                    return jp.setMaxWorkers(mw);
                }
            }

        } catch (NumberFormatException e) {
            return "Invalid parameters <jobPoolID> <maxWorkers> required";
        }

        return "Invalid parameters <jobPoolID> <maxWorkers> required";
    }
}