public final class ControllerCoreModule extends AbstractModule {
    @Override
    protected void configure() {
        bind(WorkerControlFactory.class);
        bind(JobFarmer.class);

        bind(ControllerJobResource.class);

        bind(ControllerJerseyApp.class);
    }
} public class ControllerJerseyApp extends JerseyResourceConfigBase {

    @Inject
    ControllerJerseyApp(@Ipc ObjectMapperContextResolver objectMapperContextResolver) {
        super(objectMapperContextResolver);

        register(ControllerJobResource.class);
    }
} @Singleton
@ThreadSafe
public final class JobFarmer {
    private static final Logger logger = LoggerFactory.getLogger(JobFarmer.class);

    private final WorkerControlFactory workerControlFactory;
    private final ServiceProvider<WorkerMetadata> serviceProvider;

    private final UUID controllerId = UUID.randomUUID();

    @GuardedBy("this")
    private final Map<UUID, JobStatus> jobs = new HashMap<>();

    private final JobTypePluginRegistry jobTypePluginRegistry;
    private final ObjectReader objectReader;

    private final ObjectWriter objectWriter;
    // todo make final
    @GuardedBy("this")
    private String httpListenHost;
    @GuardedBy("this")
    private int httpListenPort;

    private static final String REPORT_PATH = "/report";
    private static final String PROGRESS_PATH = REPORT_PATH + "/progress";
    private static final String FINISHED_PATH = REPORT_PATH + "/finished";

    @Inject
    JobFarmer(WorkerControlFactory workerControlFactory,
            ServiceProvider<WorkerMetadata> serviceProvider,
            JobTypePluginRegistry jobTypePluginRegistry, @Ipc ObjectReader objectReader,
            @Ipc ObjectWriter objectWriter) {
        this.workerControlFactory = workerControlFactory;
        this.serviceProvider = serviceProvider;
        this.jobTypePluginRegistry = jobTypePluginRegistry;
        this.objectReader = objectReader;
        this.objectWriter = objectWriter;
    }

    /**
     * Farm out a job to the available workers.
     *
     * @param job The job to cultivate
     * @return 202 on success with Job in the body, 412 on failure
     */
    public Response submitJob(Job job) {

        // Create a set of workers we can lock
        Set<WorkerMetadata> lockedWorkers = new HashSet<>();
        Collection<ServiceInstance<WorkerMetadata>> registeredWorkers;
        try {
            registeredWorkers = serviceProvider.getAllInstances();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        for (ServiceInstance<WorkerMetadata> instance : registeredWorkers) {
            WorkerMetadata workerMetadata = instance.getPayload();
            WorkerControl workerControl = workerControlFactory.getWorkerControl(workerMetadata);
            if (!workerControl.acquireLock(controllerId)) {
                logger.warn("Unable to lock worker <" + workerControl.getMetadata().getWorkerId() + ">");
                continue;
            }

            lockedWorkers.add(workerMetadata);
        }

        if (lockedWorkers.isEmpty()) {
            logger.warn("No unlocked workers");
            return Response.status(Response.Status.PRECONDITION_FAILED).entity("No unlocked workers found").build();
        }

        // TODO unlock locked workers if job slicing, etc. fails to avoid losing workers permanently

        List<Task> tasks;
        try {
            JobSlicer jobSlicer =
                    jobTypePluginRegistry.get(job.getTask().getTaskType()).getControllerComponentFactory(
                            objectReader, job.getTask().getConfigNode()).getJobSlicer();

            tasks = jobSlicer
                    .slice(job.getJobId(), lockedWorkers.size(), getProgressUrl(job.getJobId()),
                            getFinishedUrl(job.getJobId()), objectReader, objectWriter);
        } catch (IOException e) {
            logger.warn("Failed to slice job", e);
            return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
        }

        if (tasks.isEmpty()) {
            logger.warn("No slices created");
            return Response.status(Response.Status.PRECONDITION_FAILED).entity("No slices created").build();
        }

        TandemIterator<Task, WorkerMetadata> titerator = new TandemIterator<>(tasks.iterator(), lockedWorkers.iterator());

        // Submit the slice to the worker
        List<SliceMetadata> jobSlices = new ArrayList<>();
        int sliceId = 0;
        while (titerator.hasNext()) {
            TandemIterator<Task, WorkerMetadata>.Pair pair = titerator.next();
            JobSlice slice = new JobSlice(job.getJobId(), sliceId, pair.first, getProgressUrl(job.getJobId()),
                    getFinishedUrl(job.getJobId()));

            workerControlFactory.getWorkerControl(pair.second).submitSlice(job.getJobId(), slice);
            jobSlices.add(new SliceMetadata(pair.first, pair.second));
        }

        JobStatus jobStatus = new JobStatus(job, jobSlices);

        // Save the JobStatus for our accounting
        synchronized (this) {
            jobs.put(job.getJobId(), jobStatus);
        }

        logger.info("Cultivating job");
        return Response.status(Response.Status.ACCEPTED).entity(job).build();
    }

    /**
     * Get info about a job.
     *
     * @param jobId The job to retrieve
     * @return A Job object corresponding to the given jobId
     */
    public synchronized JobStatusResponse getJobStatus(UUID jobId) {
        return jobs.get(jobId).buildStatusResponse();
    }

    /**
     * Get the jobs that this farmer is cultivating.
     *
     * @return A Set of job IDs
     */
    public synchronized Set<UUID> getJobIds() {
        return jobs.keySet();
    }

    /**
     * Handle a completed slice
     *
     * @param jobId               The jobId that this taskProgressReport is for
     * @param sliceFinishedReport The results data
     * @return ACCEPTED if we handled the taskProgressReport, NOT_FOUND if this farmer doesn't know the given jobId
     */
    public Response handleSliceFinishedReport(UUID jobId, SliceFinishedReport sliceFinishedReport) {

        WorkerControl workerControl;
        synchronized (this) {
            if (!jobs.containsKey(jobId)) {
                logger.warn("Couldn't find job <" + jobId + ">");
                return Response.status(Response.Status.NOT_FOUND).build();
            }

            JobStatus jobStatus = jobs.get(jobId);
            int sliceId = sliceFinishedReport.getSliceId();
            jobStatus.sliceFinished(sliceId, sliceFinishedReport.getDuration());
            logger.info("Slice <" + sliceId + "> finished");

            workerControl = workerControlFactory.getWorkerControl(jobStatus.getWorkerMetadata(sliceId));
        }

        workerControl.releaseLock(controllerId);

        return Response.status(Response.Status.ACCEPTED).build();
    }

    public Response handleSliceProgressReport(UUID jobId, SliceProgressReport sliceProgressReport) {
        synchronized (this) {
            if (!jobs.containsKey(jobId)) {
                logger.warn("Couldn't find job <" + jobId + ">");
                return Response.status(Response.Status.NOT_FOUND).build();
            }

            JobStatus jobStatus = jobs.get(jobId);
            int sliceId = sliceProgressReport.getSliceId();
            jobStatus.sliceProgress(sliceId, sliceProgressReport.getData());
            logger.info("Slice <" + sliceId + "> reported progress");
        }

        return Response.ok().build();
    }

    public synchronized void setListenAddress(String httpListenHost) {
        this.httpListenHost = httpListenHost;
    }

    public synchronized void setListenPort(int httpListenPort) {
        this.httpListenPort = httpListenPort;
    }

    public UUID getControllerId() {
        return controllerId;
    }

    private String getProgressUrl(UUID jobId) {
        return "http://" + httpListenHost + ":" + httpListenPort + "/controller/job/" + jobId + PROGRESS_PATH;
    }

    private String getFinishedUrl(UUID jobId) {
        return "http://" + httpListenHost + ":" + httpListenPort + "/controller/job/" + jobId + FINISHED_PATH;
    }

    final class TandemIterator<S, T> implements Iterator<TandemIterator<S, T>.Pair> {
        private final Iterator<S> iter1;
        private final Iterator<T> iter2;

        TandemIterator(Iterator<S> iter1, Iterator<T> iter2) {
            this.iter1 = iter1;
            this.iter2 = iter2;
        }

        @Override
        public boolean hasNext() {
            return iter1.hasNext() && iter2.hasNext();
        }

        @Override
        public Pair next() {
            return new Pair(iter1.next(), iter2.next());
        }

        @Override
        public void remove() {
            throw new UnsupportedOperationException();
        }

        class Pair {
            final S first;
            final T second;

            Pair(S first, T second) {
                this.first = first;
                this.second = second;
            }
        }
    }
} @Singleton
public final class ZKServer implements Runnable, AutoCloseable {

    private static final Logger logger = LoggerFactory.getLogger(ZKServer.class);

    private final ZKServerConfig zkServerConfig;
    private final ZKServerWithShutdown server = new ZKServerWithShutdown();

    @Inject
    ZKServer(ZKServerConfig zkServerConfig) {
        this.zkServerConfig = zkServerConfig;
    }

    public boolean isEnabled() {
        return zkServerConfig.isEnabled();
    }

    @Override
    public void run() {
        if (!zkServerConfig.isEnabled()) {
            logger.warn("Got run even though disabled");
            return;
        }

        Properties zkProp = new Properties();
        zkProp.put("dataDir", zkServerConfig.getTmpDir());
        zkProp.put("clientPort", zkServerConfig.getListenPort());
        zkProp.put("clientPortAddress", zkServerConfig.getListenHost());
        zkProp.put("tickTime", zkServerConfig.getTickTime());
        zkProp.put("maxClientCnxns", zkServerConfig.getMaxClientConnections());

        QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig();

        try {
            quorumPeerConfig.parseProperties(zkProp);
        } catch (IOException e) {
            logger.warn("Couldn't parse config", e);
            throw Throwables.propagate(e);
        } catch (QuorumPeerConfig.ConfigException e) {
            logger.warn("Bad config", e);
            throw Throwables.propagate(e);
        }

        ServerConfig serverConfig = new ServerConfig();
        serverConfig.readFrom(quorumPeerConfig);

        try {
            logger.info("Running on " + zkServerConfig.getListenHost() + ":" + zkServerConfig.getListenPort());
            server.runFromConfig(serverConfig);
        } catch (IOException e) {
            logger.warn("ZooKeeper server failed", e);
        }
    }

    @Override
    public void close() {
        server.shutdown();
    }

    /**
     * Exposes shutdown method.
     */
    private static class ZKServerWithShutdown extends ZooKeeperServerMain {
        @Override
        protected void shutdown() {
            super.shutdown();
        }
    }
} public final class ZKServerModule extends AbstractModule {
    @Override
    protected void configure() {
        ConfigModule.bindConfigBean(binder(), ZKServerConfig.class);
        bind(ZKServer.class);
    }
} public final class IpcJsonModule extends AbstractModule {
    @Override
    protected void configure() {
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        objectMapper.registerModule(new KotlinModule());

        SerializationConfig config = objectMapper.getSerializationConfig()
                .withoutFeatures(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
        objectMapper.setConfig(config);

        bind(ObjectReader.class).annotatedWith(Ipc.class).toInstance(objectMapper.reader());
        bind(ObjectWriter.class).annotatedWith(Ipc.class).toInstance(objectMapper.writer());
        // have to have ObjectMapper for Jersey integration
        bind(ObjectMapper.class).annotatedWith(Ipc.class).toInstance(objectMapper);
    }
} public class GuiceServiceLocatorGenerator implements ServiceLocatorGenerator {

    private final Injector injector;

    public GuiceServiceLocatorGenerator(Injector injector) {
        this.injector = injector;
    }

    @Override
    public ServiceLocator create(String name, ServiceLocator parent) {
        if (!name.startsWith("__HK2_Generated_")) {
            return null;
        }

        Injector childInjector = injector.createChildInjector(new JerseyGuiceModule(name));

        return childInjector.getInstance(ServiceLocator.class);
    }
} public class JerseySupportModule extends ServletModule {
    @Override
    protected void configureServlets() {
        install(new HttpServerWrapperModule());

        serve("/*").with(ServletContainer.class);
    }

    /**
     * Provide a ContextResolver to be used in a ResourceConfig (like {@link JerseyResourceConfigBase} that uses {@link
     * Ipc} object mapper in Jersey
     */
    @Ipc
    @Provides
    @Singleton
    ObjectMapperContextResolver getOMCR(@Ipc ObjectMapper objectMapper) {
        return new ObjectMapperContextResolver(objectMapper);
    }
} public final class TaskPluginRegistryModule extends AbstractModule {
    @Override
    protected void configure() {
        bind(JobTypePluginRegistry.class);
        // provide a (possibly empty) set of bindings
        Multibinder.newSetBinder(binder(), JobTypePlugin.class);
    }
} public final class CuratorModule extends AbstractModule {

    @Override
    protected void configure() {
        ConfigModule.bindConfigBean(binder(), ZookeeperConfig.class);
        bind(CuratorLifecycleHook.class);

        ObjectMapper objectMapper = new ObjectMapper();
        bind(InstanceSerializerFactory.class)
                .toInstance(new InstanceSerializerFactory(objectMapper.reader(), objectMapper.writer()));
    }

    @Provides
    @Singleton
    public CuratorFramework getCuratorFramework(ZookeeperConfig zookeeperConfig) {
        return CuratorFrameworkFactory.builder()
                .connectionTimeoutMs(1000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 10))
                .connectString(zookeeperConfig.getConnectionString())
                .build();
    }

    @Provides
    @Singleton
    public ServiceDiscovery<WorkerMetadata> getServiceDiscovery(ZookeeperConfig zookeeperConfig,
            CuratorFramework curatorFramework, InstanceSerializerFactory instanceSerializerFactory) {
        return ServiceDiscoveryBuilder.builder(WorkerMetadata.class)
                .basePath(zookeeperConfig.getBasePath())
                .client(curatorFramework)
                .serializer(instanceSerializerFactory
                        .getInstanceSerializer(new TypeReference<ServiceInstance<WorkerMetadata>>() {}))
                .build();
    }

    @Provides
    @Singleton
    public ServiceProvider<WorkerMetadata> getServiceProvider(ServiceDiscovery<WorkerMetadata> serviceDiscovery,
            ZookeeperConfig zookeeperConfig) {
        return serviceDiscovery.serviceProviderBuilder()
                .serviceName(zookeeperConfig.getWorkerServiceName())
                .build();
    }

    /**
     * Encapsulates Curator startup.
     */
    @Singleton
    public static class CuratorLifecycleHook implements AutoCloseable {

        private final CuratorFramework curatorFramework;
        private final ServiceDiscovery<WorkerMetadata> serviceDiscovery;
        private final ServiceProvider<WorkerMetadata> serviceProvider;

        @Inject
        CuratorLifecycleHook(CuratorFramework curatorFramework, ServiceDiscovery<WorkerMetadata> serviceDiscovery,
                ServiceProvider<WorkerMetadata> serviceProvider) {
            this.curatorFramework = curatorFramework;
            this.serviceDiscovery = serviceDiscovery;
            this.serviceProvider = serviceProvider;
        }

        /**
         * Start up the curator instance and service discovery system.
         *
         * @throws Exception because Curator throws Exception
         */
        public void start() throws Exception {
            curatorFramework.start();
            serviceDiscovery.start();
            serviceProvider.start();
        }

        @Override
        public void close() throws IOException {
            serviceProvider.close();
            serviceDiscovery.close();
            curatorFramework.close();
        }
    }
} @Nonnull
public static Module getModuleForModuleNamesString(@Nullable String moduleNames) {
    if (moduleNames != null) {
        return new ReflectiveModuleInstantiationModule(
            Lists.newArrayList(moduleNames.split(",")));
    }

    // otherwise, no op module
    return new AbstractModule() {
        @Override
        protected void configure() {
        }
    };
}
import com.google.inject.AbstractModule;
import com.google.inject.Guice;
import com.google.inject.Inject;
import com.google.inject.Injector;
import com.google.inject.Provides;
import com.google.inject.Singleton;
import com.google.inject.Stage;
import com.palominolabs.benchpress.controller.ControllerCoreModule;
import com.palominolabs.benchpress.controller.ControllerJerseyApp;
import com.palominolabs.benchpress.controller.JobFarmer;
import com.palominolabs.benchpress.controller.zookeeper.ZKServer;
import com.palominolabs.benchpress.controller.zookeeper.ZKServerModule;
import com.palominolabs.benchpress.ipc.IpcJsonModule;
import com.palominolabs.benchpress.jersey.GuiceServiceLocatorGenerator;
import com.palominolabs.benchpress.jersey.JerseySupportModule;
import com.palominolabs.benchpress.job.task.TaskPluginRegistryModule;
import com.palominolabs.benchpress.curator.CuratorModule;
import com.palominolabs.config.ConfigModule;
import com.palominolabs.config.ConfigModuleBuilder;
import com.palominolabs.http.server.HttpServerWrapper;
import com.palominolabs.http.server.HttpServerWrapperConfig;
import com.palominolabs.http.server.HttpServerWrapperFactory;
import com.squarespace.jersey2.guice.JerseyGuiceUtils;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.logging.LogManager;
import org.apache.commons.configuration.SystemConfiguration;
import org.glassfish.jersey.servlet.ServletContainer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.bridge.SLF4JBridgeHandler;
import static com.palominolabs.benchpress.guice.ReflectiveModuleInstantiationModule.getModuleForModuleNamesString;
import static com.palominolabs.http.server.HttpServerConnectorConfig.forHttp;
package com.palominolabs.benchpress.controller.svc;



final class ControllerMain {
    private static final Logger logger = LoggerFactory.getLogger(ControllerMain.class);

    private final HttpServerWrapperFactory httpServerFactory;
    private final JobFarmer jobFarmer;
    private final ControllerConfig controllerConfig;
    private final ZKServer zkServer;
    private final CuratorModule.CuratorLifecycleHook curatorLifecycleHook;

    private final ExecutorService zkServerExService = Executors.newCachedThreadPool();

    @Inject
    ControllerMain(HttpServerWrapperFactory httpServerFactory, JobFarmer jobFarmer, ControllerConfig controllerConfig,
            ZKServer zkServer, CuratorModule.CuratorLifecycleHook curatorLifecycleHook) {
        this.httpServerFactory = httpServerFactory;
        this.jobFarmer = jobFarmer;
        this.controllerConfig = controllerConfig;
        this.zkServer = zkServer;
        this.curatorLifecycleHook = curatorLifecycleHook;
    }

    public static void main(String[] args) throws Exception {
        // Disable Zookeeper's annoying attempt to start its own out-of-date Jetty
        System.setProperty("zookeeper.admin.enableServer", "false");

        LogManager.getLogManager().reset();
        SLF4JBridgeHandler.install();
        Injector injector = Guice.createInjector(Stage.PRODUCTION, new ControllerMainModule(),
                getModuleForModuleNamesString(System.getProperty("benchpress.plugin.module-names")));

        JerseyGuiceUtils.install(new GuiceServiceLocatorGenerator(injector));

        injector.getInstance(ControllerMain.class).go();
    }

    void go() throws Exception {
        if (zkServer.isEnabled()) {
            Future<?> zkFuture = zkServerExService.submit(zkServer);
            zkServerExService.submit(new FutureWatcher(zkFuture));
        } else {
            logger.info("Not running embedded ZooKeeper server");
        }

        HttpServerWrapperConfig config = new HttpServerWrapperConfig().withHttpServerConnectorConfig(
                forHttp(controllerConfig.getHttpServerIp(), controllerConfig.getHttpServerPort()));

        HttpServerWrapper httpServer = httpServerFactory.getHttpServerWrapper(config);
        httpServer.start();

        curatorLifecycleHook.start();

        logger.info("Controller started listening on port " + controllerConfig.getHttpServerPort());

        jobFarmer.setListenAddress(controllerConfig.getHttpServerIp());
        jobFarmer.setListenPort(controllerConfig.getHttpServerPort());
    }

    /**
     * Runnable that logs if a specified Future completes
     */
    private static class FutureWatcher implements Runnable {
        private static final Logger logger = LoggerFactory.getLogger(FutureWatcher.class);
        private final Future<?> future;

        /**
         * @param future future that should never complete
         */
        private FutureWatcher(Future<?> future) {
            this.future = future;
        }

        @Override
        public void run() {
            try {
                future.get();
                logger.error("Future completed normally");
            } catch (InterruptedException e) {
                logger.info("Interrupted; aborting");
            } catch (ExecutionException e) {
                logger.error("Future failed", e);
            }
        }
    }

    public static final class ControllerMainModule extends AbstractModule {
        @Override
        protected void configure() {
            binder().requireExplicitBindings();
            binder().requireAtInjectOnConstructors();
            binder().requireExactBindingAnnotations();

            bind(ControllerMain.class);

            install(new ConfigModuleBuilder().addConfiguration(new SystemConfiguration()).build());
            ConfigModule.bindConfigBean(binder(), ControllerConfig.class);

            install(new ControllerCoreModule());
            install(new IpcJsonModule());