DTR.h

// Summary page of data staging for doxygen
namespace DataStaging {
} // namespace DataStaging

#ifndef DTR_H_
#define DTR_H_

#include <arc/data/DataHandle.h>
#include <arc/CheckSum.h>
#include <arc/data/URLMap.h>
#include <arc/DateTime.h>
#include <arc/Logger.h>
#include <arc/User.h>
#include <arc/UserConfig.h>
#include <arc/Thread.h>
#include "DTRStatus.h"

#ifdef WIN32
#ifndef uid_t
#define uid_t int
#endif
#ifndef gid_t
#define gid_t int
#endif
#endif

namespace DataStaging {

  class DTR;


  typedef Arc::ThreadedPointer<DTR> DTR_ptr;


  typedef Arc::ThreadedPointer<Arc::Logger> DTRLogger;


  enum StagingProcesses {
    GENERATOR,     
    SCHEDULER,     
    PRE_PROCESSOR, 
    DELIVERY,      
    POST_PROCESSOR 
  };
  

  enum ProcessState {
    INITIATED, 
    RUNNING,   
    TO_STOP,   
    STOPPED    
  };


  class TransferParameters {
    public:

    unsigned long long int min_average_bandwidth;

    unsigned int max_inactivity_time;

    unsigned long long int min_current_bandwidth;
    unsigned int averaging_time;
    TransferParameters() : min_average_bandwidth(0), max_inactivity_time(0),
                           min_current_bandwidth(0), averaging_time(0) {};
  };


  class DTRCacheParameters {
    public:
    std::vector<std::string> cache_dirs;
    std::vector<std::string> remote_cache_dirs;
    std::vector<std::string> drain_cache_dirs;
    DTRCacheParameters(void) {};
    DTRCacheParameters(std::vector<std::string> caches,
                       std::vector<std::string> remote_caches,
                       std::vector<std::string> drain_caches);
  };


  enum CacheState {
    CACHEABLE,             
    NON_CACHEABLE,         
    CACHE_ALREADY_PRESENT, 
    CACHE_DOWNLOADED,      
    CACHE_LOCKED,          
    CACHE_SKIP,            
    CACHE_NOT_USED         
  };
        

  class DTRCallback {
    public:
      virtual ~DTRCallback() {};

      virtual void receiveDTR(DTR_ptr dtr) = 0;
      // TODO
      //virtual void suspendDTR(DTR& dtr) = 0;
      //virtual void cancelDTR(DTR& dtr) = 0;
  };


  class DTR {
        
  private:
    std::string DTR_ID;

    Arc::URL source_url;
    Arc::URL destination_url;
    Arc::UserConfig cfg;

    Arc::DataHandle source_endpoint;
    Arc::DataHandle destination_endpoint;

    std::string source_url_str;
    std::string destination_url_str;

    /* Kept as string so we don't need to duplicate DataHandle properties
     * of destination. Delivery should check if this is set and if so use
     * it as destination. */
    std::string cache_file;

    DTRCacheParameters cache_parameters;

    CacheState cache_state;

    Arc::User user;

    bool rfc_proxy;

    std::string parent_job_id;

    int priority;

    std::string transfershare;


    std::string sub_share;

    unsigned int tries_left;

    unsigned int initial_tries;

    bool replication;


    bool force_registration;


    std::string mapped_source;

    DTRStatus status;

    DTRErrorStatus error_status;

    unsigned long long int bytes_transferred; // TODO and/or offset?


    Arc::Time timeout;
    Arc::Time created;
    Arc::Time last_modified;
    Arc::Time next_process_time;

    bool cancel_request;

    bool bulk_start;
    bool bulk_end;
    bool source_supports_bulk;


    Arc::URL delivery_endpoint;

    std::vector<Arc::URL> problematic_delivery_endpoints;

    bool use_host_cert_for_remote_delivery;

    StagingProcesses current_owner;


    DTRLogger logger;


    std::list<Arc::LogDestination*> log_destinations;

    std::map<StagingProcesses,std::list<DTRCallback*> > proc_callback;

    Arc::SimpleCondition lock;



    /* Methods */
    void mark_modification () { last_modified.SetTime(time(NULL)); };

    std::list<DTRCallback*> get_callbacks(const std::map<StagingProcesses, std::list<DTRCallback*> >& proc_callback,
                                          StagingProcesses owner);

    DTR& operator=(const DTR& dtr);
    DTR(const DTR& dtr);
    DTR();


  public:

    static const Arc::URL LOCAL_DELIVERY;

    static Arc::LogLevel LOG_LEVEL;


    DTR(const std::string& source,
        const std::string& destination,
        const Arc::UserConfig& usercfg,
        const std::string& jobid,
        const uid_t& uid,
        DTRLogger log);

    ~DTR() {};
      
    operator bool() const {
      return (!DTR_ID.empty());
    }
    bool operator!() const {
      return (DTR_ID.empty());
    }


    void registerCallback(DTRCallback* cb, StagingProcesses owner);


    void reset();

    void set_id(const std::string& id);
    std::string get_id() const { return DTR_ID; };
    std::string get_short_id() const;
     
    Arc::DataHandle& get_source() { return source_endpoint; };
    Arc::DataHandle& get_destination() { return destination_endpoint; };

    std::string get_source_str() const { return source_url_str; };
    std::string get_destination_str() const { return destination_url_str; };

    const Arc::UserConfig& get_usercfg() const { return cfg; };

    void set_timeout(time_t value) { timeout.SetTime(Arc::Time().GetTime() + value); };
    Arc::Time get_timeout() const { return timeout; };
     
    void set_process_time(const Arc::Period& process_time);
    Arc::Time get_process_time() const { return next_process_time; };

    Arc::Time get_creation_time() const { return created; };
     
    Arc::Time get_modification_time() const { return last_modified; };

    std::string get_parent_job_id() const { return parent_job_id; };
     
    void set_priority(int pri);
    int get_priority() const { return priority; };
     
    void set_rfc_proxy(bool rfc) { rfc_proxy = rfc; };
    bool is_rfc_proxy() const { return rfc_proxy; };

    void set_transfer_share(const std::string& share_name);
    std::string get_transfer_share() const { return transfershare; };
     
    void set_sub_share(const std::string& share) { sub_share = share; };
    std::string get_sub_share() const { return sub_share; };

    void set_tries_left(unsigned int tries);
    unsigned int get_tries_left() const { return tries_left; };
    unsigned int get_initial_tries() const { return initial_tries; }
    void decrease_tries_left();

    void set_status(DTRStatus stat);
    DTRStatus get_status();
     

    void set_error_status(DTRErrorStatus::DTRErrorStatusType error_stat,
                          DTRErrorStatus::DTRErrorLocation error_loc,
                          const std::string& desc="");
    void reset_error_status();
    DTRErrorStatus get_error_status();

    void set_bytes_transferred(unsigned long long int bytes);
    unsigned long long int get_bytes_transferred() const { return bytes_transferred; };

    void set_cancel_request();
    bool cancel_requested() const { return cancel_request; };
     
    void set_delivery_endpoint(const Arc::URL& endpoint) { delivery_endpoint = endpoint; };
    const Arc::URL& get_delivery_endpoint() const { return delivery_endpoint; };


    void add_problematic_delivery_service(const Arc::URL& endpoint) { problematic_delivery_endpoints.push_back(endpoint); };
    const std::vector<Arc::URL>& get_problematic_delivery_services() const { return problematic_delivery_endpoints; };

    void host_cert_for_remote_delivery(bool host) { use_host_cert_for_remote_delivery = host; };
    bool host_cert_for_remote_delivery() const { return use_host_cert_for_remote_delivery; };

    void set_cache_file(const std::string& filename);
    std::string get_cache_file() const { return cache_file; };

    void set_cache_parameters(const DTRCacheParameters& param) { cache_parameters = param; };
    const DTRCacheParameters& get_cache_parameters() const { return cache_parameters; };

    void set_cache_state(CacheState state);
    CacheState get_cache_state() const { return cache_state; };

    void set_mapped_source(const std::string& file = "") { mapped_source = file; };
    std::string get_mapped_source() const { return mapped_source; };

    StagingProcesses get_owner() const { return current_owner; };
     
    Arc::User get_local_user() const { return user; };

    void set_replication(bool rep) { replication = rep; };
    bool is_replication() const { return replication; };
    void set_force_registration(bool force) { force_registration = force; };
    bool is_force_registration() const { return force_registration; };

    void set_bulk_start(bool value) { bulk_start = value; };
    bool get_bulk_start() const { return bulk_start; };
    void set_bulk_end(bool value) { bulk_end = value; };
    bool get_bulk_end() const { return bulk_end; };
    bool bulk_possible();

    const DTRLogger& get_logger() const { return logger; };

    void connect_logger() { if (logger) logger->setDestinations(log_destinations); };
    void disconnect_logger() { if (logger) logger->removeDestinations(); };

    static void push(DTR_ptr dtr, StagingProcesses new_owner);
     
    bool suspend();
     
    bool error() const { return (error_status != DTRErrorStatus::NONE_ERROR); }
     
    bool is_destined_for_pre_processor() const;
    bool is_destined_for_post_processor() const;
    bool is_destined_for_delivery() const;
     
    bool came_from_pre_processor() const;
    bool came_from_post_processor() const;
    bool came_from_delivery() const;
    bool came_from_generator() const;
    bool is_in_final_state() const;
  };
  
  DTR_ptr createDTRPtr(const std::string& source,
                       const std::string& destination,
                       const Arc::UserConfig& usercfg,
                       const std::string& jobid,
                       const uid_t& uid,
                       DTRLogger log);

  DTRLogger createDTRLogger(Arc::Logger& parent,
                            const std::string& subdomain);

} // namespace DataStaging
#endif /*DTR_H_*/