work_queue.h

Go to the documentation of this file.

/*
Copyright (C) 2008- The University of Notre Dame
This software is distributed under the GNU General Public License.
See the file COPYING for details.
*/

#ifndef WORK_QUEUE_H
#define WORK_QUEUE_H

#include <sys/types.h>
#include "timestamp.h"
#include "category.h"
#include "rmsummary.h"

#define WORK_QUEUE_DEFAULT_PORT 9123               
#define WORK_QUEUE_RANDOM_PORT  0                  
#define WORK_QUEUE_WAITFORTASK  -1                 
#define WORK_QUEUE_DEFAULT_KEEPALIVE_INTERVAL 120  
#define WORK_QUEUE_DEFAULT_KEEPALIVE_TIMEOUT  30   
typedef enum {
        WORK_QUEUE_INPUT  = 0,                         
        WORK_QUEUE_OUTPUT = 1                          
} work_queue_file_type_t;

typedef enum {
        WORK_QUEUE_NOCACHE  = 0, 
        WORK_QUEUE_CACHE    = 1, 
        WORK_QUEUE_SYMLINK  = 2, 
        WORK_QUEUE_THIRDGET = 8, 
        WORK_QUEUE_THIRDPUT = 8, 
        WORK_QUEUE_WATCH    = 16, 
        WORK_QUEUE_FAILURE_ONLY = 32,
        WORK_QUEUE_SUCCESS_ONLY = 64, 
        WORK_QUEUE_PREEXIST = 4 
} work_queue_file_flags_t;

typedef enum {
        WORK_QUEUE_SCHEDULE_UNSET = 0,
        WORK_QUEUE_SCHEDULE_FCFS,      
        WORK_QUEUE_SCHEDULE_FILES,     
        WORK_QUEUE_SCHEDULE_TIME,      
        WORK_QUEUE_SCHEDULE_RAND,      
        WORK_QUEUE_SCHEDULE_WORST      
} work_queue_schedule_t;


typedef enum {
        WORK_QUEUE_RESULT_SUCCESS             = 0,      
        WORK_QUEUE_RESULT_INPUT_MISSING       = 1,      
        WORK_QUEUE_RESULT_OUTPUT_MISSING      = 2,      
        WORK_QUEUE_RESULT_STDOUT_MISSING      = 4,      
        WORK_QUEUE_RESULT_SIGNAL              = 1 << 3, 
        WORK_QUEUE_RESULT_RESOURCE_EXHAUSTION = 2 << 3, 
        WORK_QUEUE_RESULT_TASK_TIMEOUT        = 3 << 3, 
        WORK_QUEUE_RESULT_UNKNOWN             = 4 << 3, 
        WORK_QUEUE_RESULT_FORSAKEN            = 5 << 3, 
        WORK_QUEUE_RESULT_MAX_RETRIES         = 6 << 3, 
        WORK_QUEUE_RESULT_TASK_MAX_RUN_TIME   = 7 << 3, 
        WORK_QUEUE_RESULT_DISK_ALLOC_FULL     = 8 << 3, 
        WORK_QUEUE_RESULT_RMONITOR_ERROR      = 9 << 3, 
        WORK_QUEUE_RESULT_OUTPUT_TRANSFER_ERROR = 10 << 3  
} work_queue_result_t;

typedef enum {
        WORK_QUEUE_TASK_UNKNOWN = 0,       
        WORK_QUEUE_TASK_READY,             
        WORK_QUEUE_TASK_RUNNING,           
        WORK_QUEUE_TASK_WAITING_RETRIEVAL, 
        WORK_QUEUE_TASK_RETRIEVED,         
        WORK_QUEUE_TASK_DONE,              
        WORK_QUEUE_TASK_CANCELED,           
} work_queue_task_state_t;

typedef enum {
        WORK_QUEUE_FILE = 1,              
        WORK_QUEUE_BUFFER,                
        WORK_QUEUE_REMOTECMD,             
        WORK_QUEUE_FILE_PIECE,            
        WORK_QUEUE_DIRECTORY,             
        WORK_QUEUE_URL                    
} work_queue_file_t;


// Here we repeat the category_mode_t declaration but with work_queue names.
// This is needed to generate uniform names in the API and bindings:

typedef enum {
    WORK_QUEUE_ALLOCATION_MODE_FIXED          = CATEGORY_ALLOCATION_MODE_FIXED, 
    WORK_QUEUE_ALLOCATION_MODE_MAX            = CATEGORY_ALLOCATION_MODE_MAX, 
    WORK_QUEUE_ALLOCATION_MODE_MIN_WASTE      = CATEGORY_ALLOCATION_MODE_MIN_WASTE, 
    WORK_QUEUE_ALLOCATION_MODE_MAX_THROUGHPUT = CATEGORY_ALLOCATION_MODE_MAX_THROUGHPUT 
} work_queue_category_mode_t;


extern int wq_option_scheduler;                
struct work_queue_task {
        char *tag;                                        
        char *command_line;                               
        work_queue_schedule_t worker_selection_algorithm; 
        char *output;                                     
        struct list *input_files;                         
        struct list *output_files;                        
        struct list *env_list;                            
        int taskid;                                       
        int return_status;                                
        work_queue_result_t result;                       
        char *host;                                       
        char *hostname;                                   
        char *category;                         
        category_allocation_t resource_request; 
        double priority;        
        int max_retries;        
        int try_count;          
        int exhausted_attempts; 
        int fast_abort_count; 
        /* All times in microseconds */
        /* A time_when_* refers to an instant in time, otherwise it refers to a length of time. */
        timestamp_t time_when_submitted;    
        timestamp_t time_when_done;         
        int disk_allocation_exhausted;                        
        int64_t min_running_time;           
        timestamp_t time_when_commit_start; 
        timestamp_t time_when_commit_end;   
        timestamp_t time_when_retrieval;    
        timestamp_t time_workers_execute_last;                 
        timestamp_t time_workers_execute_all;                  
        timestamp_t time_workers_execute_exhaustion;           
        timestamp_t time_workers_execute_failure;              
        int64_t bytes_received;                                
        int64_t bytes_sent;                                    
        int64_t bytes_transferred;                             
        struct rmsummary *resources_allocated;                 
        struct rmsummary *resources_measured;                  
        struct rmsummary *resources_requested;                 
        char *monitor_output_directory;                        
        char *monitor_snapshot_file;                          
        struct list *features;                                
        /* deprecated fields */
        //int total_submissions;                                 /**< @deprecated Use try_count. */

        timestamp_t time_task_submit;                          
        timestamp_t time_task_finish;                          
        timestamp_t time_committed;                            
        timestamp_t time_send_input_start;                     
        timestamp_t time_send_input_finish;                    
        timestamp_t time_receive_result_start;                 
        timestamp_t time_receive_result_finish;                
        timestamp_t time_receive_output_start;                 
        timestamp_t time_receive_output_finish;                
        timestamp_t time_execute_cmd_start;                    
        timestamp_t time_execute_cmd_finish;                   
        timestamp_t total_transfer_time;                       
        timestamp_t cmd_execution_time;                        
        timestamp_t total_cmd_execution_time;                  
        timestamp_t total_cmd_exhausted_execute_time;          
        timestamp_t total_time_until_worker_failure;           
        int64_t total_bytes_received;                          
        int64_t total_bytes_sent;                              
        int64_t total_bytes_transferred;                       
        timestamp_t time_app_delay;                            
};

struct work_queue_stats {
        /* Stats for the current state of workers: */
        int workers_connected;    
        int workers_init;         
        int workers_idle;         
        int workers_busy;         
        int workers_able;         
        /* Cumulative stats for workers: */
        int workers_joined;       
        int workers_removed;      
        int workers_released;     
        int workers_idled_out;    
        int workers_fast_aborted; 
        int workers_blocked ;     
        int workers_lost;         
        /* Stats for the current state of tasks: */
        int tasks_waiting;        
        int tasks_on_workers;     
        int tasks_running;        
        int tasks_with_results;   
        /* Cumulative stats for tasks: */
        int tasks_submitted;           
        int tasks_dispatched;          
        int tasks_done;                
        int tasks_failed;              
        int tasks_cancelled;           
        int tasks_exhausted_attempts;  
        /* All times in microseconds */
        /* A time_when_* refers to an instant in time, otherwise it refers to a length of time. */

        /* Master time statistics: */
        timestamp_t time_when_started; 
        timestamp_t time_send;         
        timestamp_t time_receive;      
        timestamp_t time_send_good;    
        timestamp_t time_receive_good; 
        timestamp_t time_status_msgs;  
        timestamp_t time_internal;     
        timestamp_t time_polling;      
        timestamp_t time_application;  
        /* Workers time statistics: */
        timestamp_t time_workers_execute;            
        timestamp_t time_workers_execute_good;       
        timestamp_t time_workers_execute_exhaustion; 
        /* BW statistics */
        int64_t bytes_sent;     
        int64_t bytes_received; 
        double  bandwidth;      
        /* resources statistics */
        int capacity_tasks;     
        int capacity_cores;     
        int capacity_memory;    
        int capacity_disk;      
        int capacity_gpus;      
        int capacity_instantaneous;      
        int capacity_weighted;  
        int64_t total_cores;      
        int64_t total_memory;     
        int64_t total_disk;           
        int64_t committed_cores;  
        int64_t committed_memory; 
        int64_t committed_disk;   
        int64_t max_cores;        
        int64_t max_memory;       
        int64_t max_disk;         
        int64_t min_cores;        
        int64_t min_memory;       
        int64_t min_disk;         
    double manager_load;      
        int total_workers_connected;    
        int total_workers_joined;       
        int total_workers_removed;      
        int total_workers_lost;         
        int total_workers_idled_out;    
        int total_workers_fast_aborted; 
        int tasks_complete;             
        int total_tasks_dispatched;     
        int total_tasks_complete;       
        int total_tasks_failed;         
        int total_tasks_cancelled;      
        int total_exhausted_attempts;   
        timestamp_t start_time;               
        timestamp_t total_send_time;          
        timestamp_t total_receive_time;       
        timestamp_t total_good_transfer_time; 
        timestamp_t total_execute_time;           
        timestamp_t total_good_execute_time;      
        timestamp_t total_exhausted_execute_time; 
        int64_t total_bytes_sent;     
        int64_t total_bytes_received; 
        double capacity; 
        double efficiency;      
        double idle_percentage; 
        int64_t total_gpus;       
        int64_t committed_gpus;   
        int64_t max_gpus;         
        int64_t min_gpus;         
        int port;                       
        int priority;                   
        int workers_ready;              
        int workers_full;               
        int total_worker_slots;         
        int avg_capacity;               
        int workers_blacklisted;         
};


/* Forward declare the queue's structure. This structure is opaque and defined in work_queue.c */
struct work_queue;



struct work_queue_task *work_queue_task_create(const char *full_command);

struct work_queue_task *work_queue_task_clone(const struct work_queue_task *task);

void work_queue_task_specify_command( struct work_queue_task *t, const char *cmd );

int work_queue_task_specify_file(struct work_queue_task *t, const char *local_name, const char *remote_name, work_queue_file_type_t type, work_queue_file_flags_t flags);

int work_queue_task_specify_file_piece(struct work_queue_task *t, const char *local_name, const char *remote_name, off_t start_byte, off_t end_byte, work_queue_file_type_t type, work_queue_file_flags_t flags);

int work_queue_task_specify_buffer(struct work_queue_task *t, const char *data, int length, const char *remote_name, work_queue_file_flags_t flags);

int work_queue_task_specify_directory(struct work_queue_task *t, const char *local_name, const char *remote_name, work_queue_file_type_t type, work_queue_file_flags_t flags, int recursive);

int work_queue_task_specify_file_command(struct work_queue_task *t, const char *remote_name, const char *cmd, work_queue_file_type_t type, work_queue_file_flags_t flags);

void work_queue_task_specify_max_retries( struct work_queue_task *t, int64_t max_retries );

void work_queue_task_specify_memory( struct work_queue_task *t, int64_t memory );

void work_queue_task_specify_disk( struct work_queue_task *t, int64_t disk );

void work_queue_task_specify_cores( struct work_queue_task *t, int cores );

void work_queue_task_specify_gpus( struct work_queue_task *t, int gpus );

void work_queue_task_specify_end_time( struct work_queue_task *t, int64_t useconds );

void work_queue_task_specify_start_time_min( struct work_queue_task *t, int64_t useconds );

void work_queue_task_specify_running_time( struct work_queue_task *t, int64_t useconds );

void work_queue_task_specify_running_time_max( struct work_queue_task *t, int64_t seconds );

void work_queue_task_specify_running_time_min( struct work_queue_task *t, int64_t seconds );

void work_queue_task_specify_tag(struct work_queue_task *t, const char *tag);

void work_queue_task_specify_category(struct work_queue_task *t, const char *category);

void work_queue_task_specify_feature(struct work_queue_task *t, const char *name);

void work_queue_task_specify_priority(struct work_queue_task *t, double priority );

void work_queue_task_specify_environment_variable( struct work_queue_task *t, const char *name, const char *value );

void work_queue_task_specify_algorithm(struct work_queue_task *t, work_queue_schedule_t algorithm);

void work_queue_task_specify_monitor_output(struct work_queue_task *t, const char *monitor_output);

void work_queue_task_delete(struct work_queue_task *t);


int work_queue_specify_snapshot_file(struct work_queue_task *t, const char *monitor_snapshot_file);




struct work_queue *work_queue_create(int port);


struct work_queue *work_queue_ssl_create(int port, const char *key, const char *cert);

int work_queue_enable_monitoring(struct work_queue *q, char *monitor_output_directory, int watchdog);

int work_queue_enable_monitoring_full(struct work_queue *q, char *monitor_output_directory, int watchdog);

int work_queue_submit(struct work_queue *q, struct work_queue_task *t);


int work_queue_specify_min_taskid(struct work_queue *q, int minid);

void work_queue_block_host(struct work_queue *q, const char *hostname);

void work_queue_block_host_with_timeout(struct work_queue *q, const char *hostname, time_t seconds);


void work_queue_unblock_host(struct work_queue *q, const char *hostname);

void work_queue_unblock_all(struct work_queue *q);

void work_queue_invalidate_cached_file(struct work_queue *q, const char *local_name, work_queue_file_t type);


struct work_queue_task *work_queue_wait(struct work_queue *q, int timeout);


struct work_queue_task *work_queue_wait_for_tag(struct work_queue *q, const char *tag, int timeout);

int work_queue_hungry(struct work_queue *q);

int work_queue_empty(struct work_queue *q);

int work_queue_port(struct work_queue *q);

void work_queue_get_stats(struct work_queue *q, struct work_queue_stats *s);

void work_queue_get_stats_hierarchy(struct work_queue *q, struct work_queue_stats *s);

void work_queue_get_stats_category(struct work_queue *q, const char *c, struct work_queue_stats *s);

char *work_queue_status(struct work_queue *q, const char *request);


struct rmsummary **work_queue_workers_summary(struct work_queue *q);

work_queue_task_state_t work_queue_task_state(struct work_queue *q, int taskid);

void work_queue_set_bandwidth_limit(struct work_queue *q, const char *bandwidth);

double work_queue_get_effective_bandwidth(struct work_queue *q);

char * work_queue_get_worker_summary( struct work_queue *q );

int work_queue_activate_fast_abort(struct work_queue *q, double multiplier);


int work_queue_activate_fast_abort_category(struct work_queue *q, const char *category, double multiplier);


int work_queue_specify_draining_by_hostname(struct work_queue *q, const char *hostname, int drain_flag);

int work_queue_specify_category_mode(struct work_queue *q, const char *category, work_queue_category_mode_t mode);

int work_queue_enable_category_resource(struct work_queue *q, const char *category, const char *resource, int autolabel);

void work_queue_specify_algorithm(struct work_queue *q, work_queue_schedule_t algorithm);

const char *work_queue_name(struct work_queue *q);

void work_queue_specify_name(struct work_queue *q, const char *name);

void work_queue_specify_debug_path(struct work_queue *q, const char *path);

void work_queue_specify_tlq_port(struct work_queue *q, int port);

void work_queue_specify_priority(struct work_queue *q, int priority);

void work_queue_specify_num_tasks_left(struct work_queue *q, int ntasks);

void work_queue_specify_catalog_server(struct work_queue *q, const char *hostname, int port);

void work_queue_specify_catalog_servers(struct work_queue *q, const char *hosts);

struct work_queue_task *work_queue_cancel_by_taskid(struct work_queue *q, int id);

struct work_queue_task *work_queue_cancel_by_tasktag(struct work_queue *q, const char *tag);

struct list * work_queue_cancel_all_tasks(struct work_queue *q);

int work_queue_shut_down_workers(struct work_queue *q, int n);

void work_queue_delete(struct work_queue *q);

int work_queue_specify_log(struct work_queue *q, const char *logfile);

int work_queue_specify_transactions_log(struct work_queue *q, const char *logfile);

void work_queue_specify_password( struct work_queue *q, const char *password );

int work_queue_specify_password_file( struct work_queue *q, const char *file );

void work_queue_specify_keepalive_interval(struct work_queue *q, int interval);

void work_queue_specify_keepalive_timeout(struct work_queue *q, int timeout);

void work_queue_manager_preferred_connection(struct work_queue *q, const char *preferred_connection);

int work_queue_tune(struct work_queue *q, const char *name, double value);

void work_queue_specify_max_resources(struct work_queue *q,  const struct rmsummary *rm);

void work_queue_specify_min_resources(struct work_queue *q,  const struct rmsummary *rm);

void work_queue_specify_category_max_resources(struct work_queue *q,  const char *category, const struct rmsummary *rm);

void work_queue_specify_category_min_resources(struct work_queue *q,  const char *category, const struct rmsummary *rm);

void work_queue_specify_category_first_allocation_guess(struct work_queue *q,  const char *category, const struct rmsummary *rm);

void work_queue_initialize_categories(struct work_queue *q, struct rmsummary *max, const char *summaries_file);


const char *work_queue_result_str(work_queue_result_t result);



#define WORK_QUEUE_TASK_ORDER_FIFO 0  
#define WORK_QUEUE_TASK_ORDER_LIFO 1  
void work_queue_specify_task_order(struct work_queue *q, int order);


#define WORK_QUEUE_MANAGER_MODE_STANDALONE 0 
#define WORK_QUEUE_MANAGER_MODE_CATALOG 1    
void work_queue_specify_manager_mode(struct work_queue *q, int mode);


void work_queue_specify_estimate_capacity_on(struct work_queue *q, int estimate_capacity_on);

int work_queue_task_specify_input_buf(struct work_queue_task *t, const char *buf, int length, const char *rname);

int work_queue_task_specify_input_file(struct work_queue_task *t, const char *fname, const char *rname);

int work_queue_task_specify_input_file_do_not_cache(struct work_queue_task *t, const char *fname, const char *rname);

int work_queue_task_specify_output_file(struct work_queue_task *t, const char *rname, const char *fname);

int work_queue_task_specify_output_file_do_not_cache(struct work_queue_task *t, const char *rname, const char *fname);

char *work_queue_generate_disk_alloc_full_filename(char *pwd, int taskid);


void work_queue_task_specify_enviroment_variable( struct work_queue_task *t, const char *name, const char *value );


// Renames for backwards compatibility
#define work_queue_master_preferred_connection work_queue_manager_preferred_connection
#define work_queue_specify_master_mode         work_queue_specify_manager_mode
#define work_queue_blacklist_add               work_queue_block_host
#define work_queue_blacklist_add_with_timeout  work_queue_block_host_with_timeout
#define work_queue_blacklist_remove            work_queue_unblock_host
#define work_queue_blacklist_clear             work_queue_unblock_all

#endif