EntityRetriever.h

#ifndef __ARC_ENTITYRETRIEVER_H__
#define __ARC_ENTITYRETRIEVER_H__

#include <list>
#include <map>
#include <string>

#include <arc/ArcConfig.h>
#include <arc/compute/Endpoint.h>
#include <arc/UserConfig.h>
#include <arc/Utils.h>
#include <arc/compute/EndpointQueryingStatus.h>
#include <arc/compute/ExecutionTarget.h>
#include <arc/compute/EntityRetrieverPlugin.h>

namespace Arc {

class Logger;
class SharedMutex;
class SimpleCondition;
class SimpleCounter;
class ThreadedPointer<class Endpoint>;
class UserConfig;


template<typename T>
class EntityConsumer {
public:
  virtual ~EntityConsumer() {}

  virtual void addEntity(const T&) = 0;
};


template<typename T>
class EntityContainer : public EntityConsumer<T>, public std::list<T> {
public:
  virtual ~EntityContainer() {}

  virtual void addEntity(const T& t) { this->push_back(t); }
};


template<typename T>
class EntityRetriever : public EntityConsumer<T> {
public:

  EntityRetriever(const UserConfig& uc, const EndpointQueryOptions<T>& options = EndpointQueryOptions<T>());
  ~EntityRetriever() { common->deactivate(); }

  void wait() const { result.wait(); };
  
  //void waitForAll() const; // TODO: Make it possible to be nice and wait for all threads to finish.
  
  bool isDone() const { return result.wait(0); };

  void addConsumer(EntityConsumer<T>& consumer) { consumerLock.lock(); consumers.push_back(&consumer); consumerLock.unlock(); };
    
  void removeConsumer(const EntityConsumer<T>& consumer);


  EndpointQueryingStatus getStatusOfEndpoint(const Endpoint& endpoint) const;


  EndpointStatusMap getAllStatuses() const { statusLock.lock(); EndpointStatusMap s = statuses; statusLock.unlock(); return s; }
    

  bool setStatusOfEndpoint(const Endpoint& endpoint, const EndpointQueryingStatus& status, bool overwrite = true);


  void getServicesWithStatus(const EndpointQueryingStatus& status, std::set<std::string>& result);


  void clearEndpointStatuses() { statusLock.lock(); statuses.clear(); statusLock.unlock(); return; }


  bool removeEndpoint(const Endpoint& e) { statusLock.lock(); EndpointStatusMap::iterator it = statuses.find(e); if (it != statuses.end()) { statuses.erase(it); statusLock.unlock(); return true; }; statusLock.unlock(); return false; }

  virtual void addEntity(const T& entity);
  

  virtual void addEndpoint(const Endpoint& endpoint);


  void needAllResults(bool all_results =  true) { need_all_results = all_results; }

protected:
  static void queryEndpoint(void *arg_);

  void checkSuspendedAndStart(const Endpoint& e);

  // Common configuration part
  class Common : public EntityRetrieverPluginLoader<T> {
  public:
    Common(EntityRetriever* t, const UserConfig& u) : EntityRetrieverPluginLoader<T>(),
      mutex(), t(t), uc(u) {};
    void deactivate(void) {
      mutex.lockExclusive();
      t = NULL;
      mutex.unlockExclusive();
    }
    bool lockExclusiveIfValid(void) {
      mutex.lockExclusive();
      if(t) return true;
      mutex.unlockExclusive();
      return false;
    }
    void unlockExclusive(void) { mutex.unlockExclusive(); }
    bool lockSharedIfValid(void) {
      mutex.lockShared();
      if(t) return true;
      mutex.unlockShared();
      return false;
    }
    void unlockShared(void) { mutex.unlockShared(); }

    operator const UserConfig&(void) const { return uc; }
    const std::list<std::string>& getAvailablePlugins(void) const { return availablePlugins; }
    void setAvailablePlugins(const std::list<std::string>& newAvailablePlugins) { availablePlugins = newAvailablePlugins; }
    EntityRetriever* operator->(void) { return t; }
    EntityRetriever* operator*(void) { return t; }
  private:
    SharedMutex mutex;
    EntityRetriever* t;
    const UserConfig uc;
    std::list<std::string> availablePlugins;
  };
  ThreadedPointer<Common> common;

  // Represents completeness of queriies run in threads.
  // Different implementations are meant for waiting for either one or all threads.
  // TODO: counter is duplicate in this implimentation. It may be simplified
  //       either by using counter of ThreadedPointer or implementing part of
  //       ThreadedPointer directly.
  class Result: private ThreadedPointer<SimpleCounter>  {
  public:
    // Creates initial instance
    Result(bool one_success = false):
      ThreadedPointer<SimpleCounter>(new SimpleCounter),
      success(false),need_one_success(one_success) { };
    // Creates new reference representing query - increments counter
    Result(const Result& r):
      ThreadedPointer<SimpleCounter>(r),
      success(false),need_one_success(r.need_one_success) {
      Ptr()->inc();
    };
    // Query finished - decrement or reset counter (if one result is enough)
    ~Result(void) {
      if(need_one_success && success) {
        Ptr()->set(0);
      } else {
        Ptr()->dec();
      };
    };
    // Mark this result as successful (failure by default)
    void setSuccess(void) { success = true; };
    // Wait for queries to finish
    bool wait(int t = -1) const { return Ptr()->wait(t); };
  private:
    bool success;
    bool need_one_success;
  };
  Result result;

  class ThreadArg {
  public:
    ThreadArg(const ThreadedPointer<Common>& common, Result& result, const Endpoint& endpoint, const EndpointQueryOptions<T>& options) : common(common), result(result), endpoint(endpoint), options(options) {};
    ThreadArg(const ThreadArg& v, Result& result) : common(v.common), result(result), endpoint(v.endpoint), pluginName(v.pluginName), options(options) {};
    // Objects for communication with caller
    ThreadedPointer<Common> common;
    Result result;
    // Per-thread parameters
    Endpoint endpoint;
    std::string pluginName;
    EndpointQueryOptions<T> options;
  };

  EndpointStatusMap statuses;

  static Logger logger;
  const UserConfig& uc;
  std::list< EntityConsumer<T>* > consumers;
  const EndpointQueryOptions<T> options;

  mutable SimpleCondition consumerLock;
  mutable SimpleCondition statusLock;
  std::map<std::string, std::string> interfacePluginMap;
  bool need_all_results;
};


typedef EntityRetriever<Endpoint>             ServiceEndpointRetriever;


typedef EntityRetriever<ComputingServiceType>             TargetInformationRetriever;


typedef EntityRetriever<Job>             JobListRetriever;

} // namespace Arc

#endif // __ARC_ENTITYRETRIEVER_H__