source: src/Fragmentation/Automation/Operations/OperationQueue.hpp@ 5fa1f0d

/*
 * OperationQueue.hpp
 *
 *  Created on: Apr 24, 2012
 *      Author: heber
 */

#ifndef OPERATIONQUEUE_HPP_
#define OPERATIONQUEUE_HPP_


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

#include <boost/shared_ptr.hpp>
#include <deque>

#include "CodePatterns/Observer/Observer.hpp"

class AsyncOperation;
class Observer;
class OperationQueueTest;
class WorkerAddress;

/** This class is a container for \ref AsyncOperation's that are kept as shared_ptr
 * and removed when the operation is done.
 */
class OperationQueue : public Observer
{
  //!> grant unit test access to private part
  friend class OperationQueueTest;
public:
  /** Default constructor for class OperationQueue.
   *
   */
  OperationQueue() :
    Observer("OperationQueue"),
    RunningOps(0),
    IsBlockedFlag(false)
  {}
  /** Default destructor for class OperationQueue.
   *
   */
  virtual ~OperationQueue()
  {}

  typedef boost::shared_ptr<AsyncOperation> AsyncOp_ptr;

  /** Add an operation to the internal queue and hand over memory responsibility to it,
   * also the operation is run.
   *
   * @param op operation to add, is NULL on return.
   */
  void push_back(AsyncOperation *&op, const WorkerAddress &address);

  /** States whether the queue is empty.
   *
   * @return true - queue is empty, false - operations are pending
   */
  bool empty() const {
    return queue.empty();
  }

  /** Blocks the queue for any further operation.
   *
   */
  void block()
  {
    IsBlockedFlag = true;
  }

  /** Unblocks the queue, operations may again be pushed.
   *
   */
  void unblock()
  {
    IsBlockedFlag = false;
  }

  /** Getter whether queue is currently blocked.
   *
   * @return OperationQueue::IsBlockedFlag
   */
  bool isBlocked() const
  {
    return IsBlockedFlag;
  }

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

private:
  /** Removes an operation from the queue.
   *
   * @param op operation to remove from queue
   * @param observer observer to sign off from operation, NULL if none to sign off
   */
  void remove(AsyncOperation *op, Observer *observer);

  /** Returns the number of currently running operations.
   *
   * @return Gives the difference between the entries in the queue and in the AddressMap.
   */
  size_t getNumberOfRunningOps() const
  {
    return queue.size() - AddressMap.size();
  }

  /** Helper to launch the next pending operation.
   *
   */
  void LaunchNextOp();

  //!> internal operation to send jobs to workers
  typedef std::deque<AsyncOp_ptr> OperationQueue_t;

  /** Tiny Helper function to find an operation inside FragmentScheduler::OperationQueue.
   *
   * @param op operation to remove from queue
   * @return iterator to element or to OperationQueue.end()
   */
  OperationQueue_t::iterator findOperation(AsyncOperation *op);

  //!> internal number stating how many operations are running
  size_t RunningOps;

  //!> giving the maximum number of connections
  static size_t max_connections;

private:
  //!> internal queue with operations
  OperationQueue_t queue;

  //!> typedef for the association for each operation to its address to connect to
  typedef std::map<AsyncOp_ptr, WorkerAddress> AddressMap_t;
  //!> Association for each operation to its address to connect to
  AddressMap_t AddressMap;

  //!> status flag whether queue is blocked or operations may be pushed.
  bool IsBlockedFlag;
};

#endif /* OPERATIONQUEUE_HPP_ */