api/html/work__queue_8h_source.html

/*

Copyright (C) 2022 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_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_ALLOCATION_MODE_GREEDY_BUCKETING = CATEGORY_ALLOCATION_MODE_GREEDY_BUCKETING,

    WORK_QUEUE_ALLOCATION_MODE_EXHAUSTIVE_BUCKETING = CATEGORY_ALLOCATION_MODE_EXHAUSTIVE_BUCKETING

} work_queue_category_mode_t;


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 *coprocess;

        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 );


void work_queue_task_specify_coprocess( struct work_queue_task *t, const char *coprocess_name );


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_url(struct work_queue_task *t, const char *url, const char *remote_name, work_queue_file_type_t type, work_queue_file_flags_t flags);


int work_queue_task_specify_file_command(struct work_queue_task *t, const char *cmd, const char *remote_name, 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);


void work_queue_specify_category_max_concurrent(struct work_queue *q, const char *category, int max_concurrent);


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