source: src/Fragmentation/Automation/FragmentScheduler.hpp@ 270364

/*
 * FragmentScheduler.hpp
 *
 *  Created on: Oct 19, 2011
 *      Author: heber
 */

#ifndef FRAGMENTSCHEDULER_HPP_
#define FRAGMENTSCHEDULER_HPP_

// include config.h
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <boost/asio.hpp>
#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <deque>
#include <vector>

#include "CodePatterns/Observer/Observer.hpp"
#include "Connection.hpp"
#include "ControllerChoices.hpp"
#include "Operations/AsyncOperation.hpp"
#include "Operations/OperationQueue.hpp"
#include "Operations/Servers/ShutdownWorkerOperation.hpp"
#include "ControllerChoices.hpp"
#include "FragmentQueue.hpp"
#include "GlobalJobId.hpp"
#include "Jobs/FragmentJob.hpp"
#include "Listener.hpp"
#include "Results/FragmentResult.hpp"
#include "types.hpp"
#include "Pool/WorkerPool.hpp"
#include "WorkerAddress.hpp"
#include "WorkerChoices.hpp"

/** FragmentScheduler serves FragmentJobs to Workers and accepts commands from
 * a Controller.
 *
 */
class FragmentScheduler : public Observer
{
public:
  /// Constructor opens the acceptor and starts waiting for the first incoming
  /// Connection.
  FragmentScheduler(boost::asio::io_service& io_service, unsigned short workerport, unsigned short controllerport);
  ~FragmentScheduler();

private:
  void sendJobToWorker(const WorkerAddress &address, FragmentJob::ptr &job);
  void sendAvailableJobToNextIdleWorker();
  bool shutdown();
  void shutdownWorker(const WorkerAddress &address);
  void removeAllWorkers();
  void cleanupOperationQueue(AsyncOperation *op);

  void update(Observable *publisher);
  void recieveNotification(Observable *publisher, Notification_ptr notification);
  void subjectKilled(Observable *publisher);


  class WorkerListener_t : public Listener
  {
  public:
    WorkerListener_t(
        boost::asio::io_service& io_service,
        unsigned short port,
        FragmentQueue &_JobsQueue,
        WorkerPool &_pool,
        boost::function<void (const WorkerAddress&, FragmentJob::ptr&)> _callback) :
      Listener(io_service, port),
      address("127.0.0.1", "0"),
      JobsQueue(_JobsQueue),
      pool(_pool),
      result( new FragmentResult(JobId::NoJob) ),
      callback_sendJobToWorker(_callback),
      choice(NoWorkerOperation)
    {}
    virtual ~WorkerListener_t() {}

  protected:
    /// Handle completion of a accept worker operation.
    void handle_Accept(const boost::system::error_code& e, connection_ptr conn);

    /// Handle completion of Worker operation to read choice
    void handle_ReadChoice(const boost::system::error_code& e, connection_ptr conn);

    /// Worker callback function when job has been sent.
    void handle_ReadAddress(const boost::system::error_code& e, connection_ptr conn);

    /// Worker callback function when new worker has enrolled.
    void handle_enrolled(const boost::system::error_code& e, connection_ptr conn);

    /// Worker callback function when result has been received.
    void handle_ReceiveResultFromWorker(const boost::system::error_code& e, connection_ptr conn);

    /// Worker callback function when invalid result has been received.
    void handle_RejectResultFromWorker(const boost::system::error_code& e, connection_ptr conn);
  private:
    //!> address of new Worker
    WorkerAddress address;

    //!> reference to Queue
    FragmentQueue &JobsQueue;

    //!> callback reference to container class
    WorkerPool &pool;

    //!> result that is received from the client.
    FragmentResult::ptr result;

    //!> callback function to access send job function
    boost::function<void (const WorkerAddress&, FragmentJob::ptr&)> callback_sendJobToWorker;

    //!> choice
    enum WorkerChoices choice;
  };

  class ControllerListener_t : public Listener
  {
  public:
    ControllerListener_t(
        boost::asio::io_service& io_service,
        unsigned short port,
        FragmentQueue &_JobsQueue,
        boost::function<void ()> _removeallWorkers,
        boost::function<bool ()> _shutdownAllSockets) :
      Listener(io_service, port),
      JobsQueue(_JobsQueue),
      jobInfo((size_t)2, 0),
      choice(NoControllerOperation),
      removeallWorkers(_removeallWorkers),
      shutdownAllSockets(_shutdownAllSockets),
      globalId(0)
    {}
    virtual ~ControllerListener_t() {}

  protected:
    /// Handle completion of a accept controller operation.
    void handle_Accept(const boost::system::error_code& e, connection_ptr conn);

    /// Handle completion of controller operation to read choice
    void handle_ReadChoice(const boost::system::error_code& e, connection_ptr conn);

    /// Controller callback function when job has been sent.
    void handle_ReceiveJobs(const boost::system::error_code& e, connection_ptr conn);

    /// Controller callback function when checking on state of results.
    void handle_CheckResultState(const boost::system::error_code& e, connection_ptr conn);

    /// Controller callback function when checking on state of results.
    void handle_GetNextJobIdState(const boost::system::error_code& e, connection_ptr conn);

    /// Controller callback function when free job ids have been sent.
    void handle_SendIds(const boost::system::error_code& e, connection_ptr conn);

    /// Controller callback function when result has been received.
    void handle_SendResults(const boost::system::error_code& e, connection_ptr conn);

  private:
    //!> reference to external FragmentQueue containing jobs to work on
    FragmentQueue & JobsQueue;

    //!> bunch of jobs received from controller before placed in JobsQueue
    std::vector<FragmentJob::ptr> jobs;

    //!> number of jobs that are waiting to be and are calculated, required for returning status
    std::vector<size_t> jobInfo;

    //!> number of job ids request from controller;
    size_t NumberIds;

    //!> choice
    enum ControllerChoices choice;

    //!> bound function to remove all workers
    boost::function<void ()> removeallWorkers;

    //!> bound function to shutdown all sockets
    boost::function<bool ()> shutdownAllSockets;

    // TODO: replace this instance by a IdPool.
    //!> global id to give next available job id
    GlobalJobId globalId;
  };

private:
  //!> static entity to indicate to clients that the queue is empty.
  static FragmentJob::ptr NoJob;

  //!> reference to the io_service which we use for connections
  boost::asio::io_service& io_service;

  //!> Queue with data to be sent to each client.
  FragmentQueue JobsQueue;

  //!> Pool of Workers
  WorkerPool pool;

  //!> Listener instance that waits for a worker
  WorkerListener_t  WorkerListener;

  //!> Listener instance that waits for a controller
  ControllerListener_t  ControllerListener;

public:
  /** Getter for Exitflag.
   *
   * @return Exitflag of operations
   */
  size_t getExitflag() const
  {
    if (WorkerListener.getExitflag() != 0)
      return WorkerListener.getExitflag();
    if (ControllerListener.getExitflag() != 0)
      return ControllerListener.getExitflag();
    return 0;
  }

private:
  //!> Connection for sending jobs to workers
  Connection connection;

  //!> internal queue for all asynchronous operations
  OperationQueue OpQueue;
};

#endif /* FRAGMENTSCHEDULER_HPP_ */