
package forager.client;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import forager.events.ForagerEventMap;
import forager.events.JoinEvent;
import forager.events.TaskRequest;
import forager.events.TaskSpec;

import galileo.event.EventContext;
import galileo.event.EventHandler;
import galileo.event.EventReactor;
import galileo.net.ClientMessageRouter;
import galileo.net.NetworkDestination;

public class Forager {

    private NetworkDestination server;
    private ClientMessageRouter messageRouter;

    private ForagerEventMap eventMap = new ForagerEventMap();
    private EventReactor eventReactor = new EventReactor(this, eventMap);

    private int maxTasks = 0;
    private int activeTasks = 0;
    private int pendingRequests = 0;
    protected ExecutorService threadPool;

    public Forager(NetworkDestination server) {
        this(server, 4);
    }

    public Forager(NetworkDestination server, int threads) {
        this.server = server;
        this.maxTasks = threads;
        this.threadPool = Executors.newFixedThreadPool(threads);
    }

    public void start()
    throws Exception {
        messageRouter = new ClientMessageRouter();
        messageRouter.addListener(eventReactor);

        /* Join the network */
        JoinEvent join = new JoinEvent();
        messageRouter.sendMessage(server, eventReactor.wrapEvent(join));

        /* Start the monitor thread */
        Thread monitor = new Thread(new StatusMonitor(this));
        monitor.start();

        while (true) {
            eventReactor.processNextEvent();
        }
    }

    protected synchronized void submitTaskRequest(int numTasks)
    throws IOException {
        TaskRequest tr = new TaskRequest(numTasks);
        pendingRequests += numTasks;
        messageRouter.sendMessage(server, eventReactor.wrapEvent(tr));
    }

    protected synchronized void finalizeTask() {
        System.out.println("Received task completion notification");
        activeTasks--;
    }

    protected synchronized int getNumActive() {
        return activeTasks;
    }

    protected synchronized int getNumPending() {
        return pendingRequests;
    }

    public int getMaxTasks() {
        return maxTasks;
    }

    @EventHandler
    public void processTaskSpec(TaskSpec taskSpec, EventContext context) {
        System.out.println("Starting job: " + taskSpec);
        Task task = new Task(taskSpec.command, this);
        pendingRequests--;
        activeTasks++;
        threadPool.submit(task);
    }

    public static void main(String[] args)
    throws Exception {
        if (args.length < 3) {
            System.out.println("Usage: forager.client.Forager <host> <port>");
            System.exit(1);
        }

        String host = args[0];
        int port = Integer.parseInt(args[1]);
        NetworkDestination overlord = new NetworkDestination(host, port);

        int threads = Integer.parseInt(args[2]);

        Forager forager = new Forager(overlord, threads);
        forager.start();
    }
}

@@ -1,8 +1,1326 @@
+/*%0ACopyright (c) 2014, Colorado State University%0AAll rights reserved.%0A%0ARedistribution and use in source and binary forms, with or without modification,%0Aare permitted provided that the following conditions are met:%0A%0A1. Redistributions of source code must retain the above copyright notice, this%0A   list of conditions and the following disclaimer.%0A2. Redistributions in binary form must reproduce the above copyright notice,%0A   this list of conditions and the following disclaimer in the documentation%0A   and/or other materials provided with the distribution.%0A%0AThis software is provided by the copyright holders and contributors %22as is%22 and%0Aany express or implied warranties, including, but not limited to, the implied%0Awarranties of merchantability and fitness for a particular purpose are%0Adisclaimed. In no event shall the copyright holder or contributors be liable for%0Aany direct, indirect, incidental, special, exemplary, or consequential damages%0A(including, but not limited to, procurement of substitute goods or services;%0Aloss of use, data, or profits; or business interruption) however caused and on%0Aany theory of liability, whether in contract, strict liability, or tort%0A(including negligence or otherwise) arising in any way out of the use of this%0Asoftware, even if advised of the possibility of such damage.%0A*/%0A
 %0Apackage

Add license