ndcctools.work_queue.WorkQueue Class Reference

Python Work Queue object. More...

Inheritance diagram for ndcctools.work_queue.WorkQueue:

## Public Member Functions

| __init__ (self, port=WORK_QUEUE_DEFAULT_PORT, name=None, shutdown=False, stats_log=None, transactions_log=None, debug_log=None, ssl=None, status_display_interval=None)
| Create a new work queue.
| name (self)
| Get the project name of the queue.
| port (self)
| Get the listening port of the queue.
| using_ssl (self)
| Whether the manager is using ssl to talk to workers.
| stats (self)
| Get queue statistics.
| stats_hierarchy (self)
| Get worker hierarchy statistics.
| stats_category (self, category)
| Get the task statistics for the given category.
| status (self, request)
| Get queue information as list of dictionaries.
| workers_summary (self)
| Get resource statistics of workers connected.
| specify_category_mode (self, category, mode)
| Turn on or off first-allocation labeling for a given category.
| specify_category_autolabel_resource (self, category, resource, autolabel)
| Turn on or off first-allocation labeling for a given category and resource.
| task_state (self, taskid)
| Get current task state.
| enable_monitoring (self, dirname=None, watchdog=True)
| Enables resource monitoring of tasks in the queue, and writes a summary per task to the directory given.
| enable_monitoring_full (self, dirname=None, watchdog=True)
| As ndcctools.work_queue.WorkQueue.enable_monitoring, but it also generates a time series and a debug file.
| activate_fast_abort (self, multiplier)
| Turn on or off fast abort functionality for a given queue for tasks in the "default" category, and for task which category does not set an explicit multiplier.
| activate_fast_abort_category (self, name, multiplier)
| Turn on or off fast abort functionality for a given queue.
| specify_draining_by_hostname (self, hostname, drain_mode=True)
| Turn on or off draining mode for workers at hostname.
| empty (self)
| Determine whether there are any known tasks queued, running, or waiting to be collected.
| hungry (self)
| Determine whether the queue can support more tasks.
| specify_algorithm (self, algorithm)
| Set the worker selection algorithm for queue.
| specify_task_order (self, order)
| Set the order for dispatching submitted tasks in the queue.
| specify_name (self, name)
| Change the project name for the given queue.
| specify_manager_preferred_connection (self, mode)
| Set the preference for using hostname over IP address to connect.
| specify_master_preferred_connection (self, mode)
| See ndcctools.work_queue.WorkQueue.specify_manager_preferred_connection.
| specify_min_taskid (self, minid)
| Set the minimum taskid of future submitted tasks.
| specify_priority (self, priority)
| Change the project priority for the given queue.
| specify_num_tasks_left (self, ntasks)
| Specify the number of tasks not yet submitted to the queue.
| specify_manager_mode (self, mode)
| Specify the manager mode for the given queue.
| specify_master_mode (self, mode)
| specify_catalog_server (self, hostname, port)
| Specify the catalog server the manager should report to.
| specify_log (self, logfile)
| Specify a log file that records the cummulative stats of connected workers and submitted tasks.
| specify_transactions_log (self, logfile)
| Specify a log file that records the states of tasks.
| specify_password (self, password)
| Add a mandatory password that each worker must present.
| specify_password_file (self, file)
| Add a mandatory password file that each worker must present.
| specify_max_resources (self, rmd)
| Specifies the maximum resources allowed for the default category.
| specify_min_resources (self, rmd)
| Specifies the minimum resources allowed for the default category.
| specify_category_max_resources (self, category, rmd)
| Specifies the maximum resources allowed for the given category.
| specify_category_min_resources (self, category, rmd)
| Specifies the minimum resources allowed for the given category.
| specify_category_first_allocation_guess (self, category, rmd)
| Specifies the first-allocation guess for the given category.
| specify_category_max_concurrent (self, category, max_concurrent)
| Specifies the maximum resources allowed for the given category.
| initialize_categories (self, filename, rm)
| Initialize first value of categories.
| cancel_by_taskid (self, id)
| Cancel task identified by its taskid and remove from the given queue.
| cancel_by_tasktag (self, tag)
| Cancel task identified by its tag and remove from the given queue.
| cancel_by_category (self, category)
| Cancel all tasks of the given category and remove them from the queue.
| shutdown_workers (self, n)
| Shutdown workers connected to queue.
| block_host (self, host)
| Block workers running on host from working for the manager.
| blacklist (self, host)
| Replaced by ndcctools.work_queue.WorkQueue.block_host.
| block_host_with_timeout (self, host, timeout)
| Block workers running on host for the duration of the given timeout.
| blacklist_with_timeout (self, host, timeout)
| See ndcctools.work_queue.WorkQueue.block_host_with_timeout.
| unblock_host (self, host=None)
| Unblock given host, of all hosts if host not given.
| blacklist_clear (self, host=None)
| See ndcctools.work_queue.WorkQueue.unblock_host.
| invalidate_cache_file (self, local_name)
| Delete file from workers's caches.
| specify_keepalive_interval (self, interval)
| Change keepalive interval for a given queue.
| specify_keepalive_timeout (self, timeout)
| Change keepalive timeout for a given queue.
| estimate_capacity (self)
| Turn on manager capacity measurements.
| tune (self, name, value)
| Tune advanced parameters for work queue.
| submit (self, task)
| Submit a task to the queue.
| wait (self, timeout=WORK_QUEUE_WAITFORTASK)
| Wait for tasks to complete.
| wait_for_tag (self, tag, timeout=WORK_QUEUE_WAITFORTASK)
| Similar to ndcctools.work_queue.WorkQueue.wait, but guarantees that the returned task has the specified tag.
| application_info (self)
| Should return a dictionary with information for the status display.
| map (self, fn, seq, chunksize=1)
| Maps a function to elements in a sequence using work_queue.
| pair (self, fn, seq1, seq2, chunksize=1, env=None)
| Returns the values for a function of each pair from 2 sequences.
| tree_reduce (self, fn, seq, chunksize=2)
| Reduces a sequence until only one value is left, and then returns that value.
| remote_map (self, fn, seq, coprocess, name, chunksize=1)
| Maps a function to elements in a sequence using work_queue remote task.
| remote_pair (self, fn, seq1, seq2, coprocess, name, chunksize=1)
| Returns the values for a function of each pair from 2 sequences using remote task.
| remote_tree_reduce (self, fn, seq, coprocess, name, chunksize=2)
| Reduces a sequence until only one value is left, and then returns that value.

## Detailed Description

Python Work Queue object.

## Constructor & Destructor Documentation

## __init__()

ndcctools.work_queue.WorkQueue.__init__ | ( |  | self,
|  |  | port = WORK_QUEUE_DEFAULT_PORT,
|  |  | name = None,
|  |  | shutdown = False,
|  |  | stats_log = None,
|  |  | transactions_log = None,
|  |  | debug_log = None,
|  |  | ssl = None,
|  |  | status_display_interval = None )

Create a new work queue.

Parameters

  self | Reference to the current work queue object.
  port | The port number to listen on. If zero, then a random port is chosen. A range of possible ports (low, hight) can be also specified instead of a single integer.
  name | The project name to use.
  stats_log | The name of a file to write the queue's statistics log.
  transactions_log | The name of a file to write the queue's transactions log.
  debug_log | The name of a file to write the queue's debug log.
  shutdown | Automatically shutdown workers when queue is finished. Disabled by default.
  ssl | A tuple of filenames (ssl_key, ssl_cert) in pem format, or True. If not given, then TSL is not activated. If True, a self-signed temporary key and cert are generated.
  status_display_interval | Number of seconds between updates to the jupyter status display. None, or less than 1 disables it.

See also
  work_queue_create - For more information about environmental variables that affect the behavior this method.

## Member Function Documentation

## name()

ndcctools.work_queue.WorkQueue.name | ( |  | self | ) |

Get the project name of the queue.

>>> print(q.name)

## port()

ndcctools.work_queue.WorkQueue.port | ( |  | self | ) |

Get the listening port of the queue.

>>> print(q.port)

## using_ssl()

ndcctools.work_queue.WorkQueue.using_ssl | ( |  | self | ) |

Whether the manager is using ssl to talk to workers.

>>> print(q.using_ssl)

## stats()

ndcctools.work_queue.WorkQueue.stats | ( |  | self | ) |

Get queue statistics.

>>> print(q.stats)

The fields in ndcctools.work_queue.WorkQueue.stats can also be individually accessed through this call. For example:

>>> print(q.stats.workers_busy)

## stats_hierarchy()

ndcctools.work_queue.WorkQueue.stats_hierarchy | ( |  | self | ) |

Get worker hierarchy statistics.

>>> print(q.stats_hierarchy)

The fields in ndcctools.work_queue.WorkQueue.stats_hierarchy can also be individually accessed through this call. For example:

>>> print(q.stats_hierarchy.workers_busy)

## stats_category()

ndcctools.work_queue.WorkQueue.stats_category | ( |  | self,
|  |  | category )

Get the task statistics for the given category.

Parameters

  self | Reference to the current work queue object.
  category | A category name. For example: s = q.stats_category("my_category") >>> print(s) The fields in work_queue_stats can also be individually accessed through this call. For example: >>> print(s.tasks_waiting)

## status()

ndcctools.work_queue.WorkQueue.status | ( |  | self,
|  |  | request )

Get queue information as list of dictionaries.

Parameters

  self | Reference to the current work queue object
  request | One of: "queue", "tasks", "workers", or "categories" For example: import json tasks_info = q.status("tasks")

## workers_summary()

ndcctools.work_queue.WorkQueue.workers_summary | ( |  | self | ) |

Get resource statistics of workers connected.

Parameters

  self | Reference to the current work queue object.

Returns
  A list of dictionaries that indicate how many .workers connected with a certain number of .cores, .memory, and disk. For example:
  workers = q.worker_summary()
  >>> for w in workers:
  >>> print("{} workers with: {} cores, {} MB memory, {} MB disk".format(w.workers, w.cores, w.memory, w.disk)

## specify_category_mode()

ndcctools.work_queue.WorkQueue.specify_category_mode | ( |  | self,
|  |  | category,
|  |  | mode )

Turn on or off first-allocation labeling for a given category.

By default, only cores, memory, and disk are labeled, and gpus are unlabeled. NOTE: autolabeling is only meaningfull when task monitoring is enabled (ndcctools.work_queue.WorkQueue.enable_monitoring). When monitoring is enabled and a task exhausts resources in a worker, mode dictates how work queue handles the exhaustion:

Parameters

  self | Reference to the current work queue object.
  category | A category name. If None, sets the mode by default for newly created categories.

  mode

  One of:
  - WORK_QUEUE_ALLOCATION_MODE_FIXED Task fails (default).
  - WORK_QUEUE_ALLOCATION_MODE_MAX If maximum values are specified for cores, memory, disk, and gpus (e.g. via ndcctools.work_queue.WorkQueue.specify_category_max_resources or ndcctools.work_queue.Task.specify_memory), and one of those resources is exceeded, the task fails. Otherwise it is retried until a large enough worker connects to the manager, using the maximum values specified, and the maximum values so far seen for resources not specified. Use ndcctools.work_queue.Task.specify_max_retries to set a limit on the number of times work queue attemps to complete the task.
  - WORK_QUEUE_ALLOCATION_MODE_MIN_WASTE As above, but work queue tries allocations to minimize resource waste.
  - WORK_QUEUE_ALLOCATION_MODE_MAX_THROUGHPUT As above, but work queue tries allocations to maximize throughput.

## specify_category_autolabel_resource()

ndcctools.work_queue.WorkQueue.specify_category_autolabel_resource | ( |  | self,
|  |  | category,
|  |  | resource,
|  |  | autolabel )

Turn on or off first-allocation labeling for a given category and resource.

This function should be use to fine-tune the defaults from ndcctools.work_queue.WorkQueue.specify_category_mode.

Parameters

  self | Reference to the current work queue object.
  category | A category name.
  resource | A resource name.
  autolabel | True/False for on/off.

Returns
  1 if resource is valid, 0 otherwise.

## task_state()

ndcctools.work_queue.WorkQueue.task_state | ( |  | self,
|  |  | taskid )

Get current task state.

See work_queue_task_state_t for possible values.

>>> print(q.task_state(taskid))

## enable_monitoring()

ndcctools.work_queue.WorkQueue.enable_monitoring | ( |  | self,
|  |  | dirname = None,
|  |  | watchdog = True )

Enables resource monitoring of tasks in the queue, and writes a summary per task to the directory given.

Additionally, all summaries are consolidate into the file all_summaries-PID.log

Returns 1 on success, 0 on failure (i.e., monitoring was not enabled).

Parameters

  self | Reference to the current work queue object.
  dirname | Directory name for the monitor output.
  watchdog | If True (default), kill tasks that exhaust their declared resources.

## enable_monitoring_full()

ndcctools.work_queue.WorkQueue.enable_monitoring_full | ( |  | self,
|  |  | dirname = None,
|  |  | watchdog = True )

As ndcctools.work_queue.WorkQueue.enable_monitoring, but it also generates a time series and a debug file.

WARNING: Such files may reach gigabyte sizes for long running tasks.

Returns 1 on success, 0 on failure (i.e., monitoring was not enabled).

Parameters

  self | Reference to the current work queue object.
  dirname | Directory name for the monitor output.
  watchdog | If True (default), kill tasks that exhaust their declared resources.

## activate_fast_abort()

ndcctools.work_queue.WorkQueue.activate_fast_abort | ( |  | self,
|  |  | multiplier )

Turn on or off fast abort functionality for a given queue for tasks in the "default" category, and for task which category does not set an explicit multiplier.

Parameters

  self | Reference to the current work queue object.
  multiplier | The multiplier of the average task time at which point to abort; if negative (the default) fast_abort is deactivated.

## activate_fast_abort_category()

ndcctools.work_queue.WorkQueue.activate_fast_abort_category | ( |  | self,
|  |  | name,
|  |  | multiplier )

Turn on or off fast abort functionality for a given queue.

Parameters

  self | Reference to the current work queue object.
  name | Name of the category.
  multiplier | The multiplier of the average task time at which point to abort; if zero, deacticate for the category, negative (the default), use the one for the "default" category (see ndcctools.work_queue.WorkQueue.activate_fast_abort)

## specify_draining_by_hostname()

ndcctools.work_queue.WorkQueue.specify_draining_by_hostname | ( |  | self,
|  |  | hostname,
|  |  | drain_mode = True )

Turn on or off draining mode for workers at hostname.

Parameters

  self | Reference to the current work queue object.
  hostname | The hostname the host running the workers.
  drain_mode | If True, no new tasks are dispatched to workers at hostname, and empty workers are shutdown. Else, workers works as usual.

## empty()

ndcctools.work_queue.WorkQueue.empty | ( |  | self | ) |

Determine whether there are any known tasks queued, running, or waiting to be collected.

Returns 0 if there are tasks remaining in the system, 1 if the system is "empty".

Parameters

  self | Reference to the current work queue object.

## hungry()

ndcctools.work_queue.WorkQueue.hungry | ( |  | self | ) |

Determine whether the queue can support more tasks.

Returns the number of additional tasks it can support if "hungry" and 0 if "sated".

Parameters

  self | Reference to the current work queue object.

## specify_algorithm()

ndcctools.work_queue.WorkQueue.specify_algorithm | ( |  | self,
|  |  | algorithm )

Set the worker selection algorithm for queue.

Parameters

  self | Reference to the current work queue object.
  algorithm | One of the following algorithms to use in assigning a task to a worker. See work_queue_schedule_t for possible values.

## specify_task_order()

ndcctools.work_queue.WorkQueue.specify_task_order | ( |  | self,
|  |  | order )

Set the order for dispatching submitted tasks in the queue.

Parameters

  self | Reference to the current work queue object.

  order

  One of the following algorithms to use in dispatching submitted tasks to workers:
  - WORK_QUEUE_TASK_ORDER_FIFO
  - WORK_QUEUE_TASK_ORDER_LIFO

## specify_name()

ndcctools.work_queue.WorkQueue.specify_name | ( |  | self,
|  |  | name )

Change the project name for the given queue.

Parameters

  self | Reference to the current work queue object.
  name | The new project name.

## specify_manager_preferred_connection()

ndcctools.work_queue.WorkQueue.specify_manager_preferred_connection | ( |  | self,
|  |  | mode )

Set the preference for using hostname over IP address to connect.

'by_ip' uses IP addresses from the network interfaces of the manager (standard behavior), 'by_hostname' to use the hostname at the manager, or 'by_apparent_ip' to use the address of the manager as seen by the catalog server.

Parameters

  self | Reference to the current work queue object.
  mode | An string to indicate using 'by_ip', 'by_hostname' or 'by_apparent_ip'.

## specify_master_preferred_connection()

ndcctools.work_queue.WorkQueue.specify_master_preferred_connection | ( |  | self,
|  |  | mode )

See ndcctools.work_queue.WorkQueue.specify_manager_preferred_connection.

## specify_min_taskid()

ndcctools.work_queue.WorkQueue.specify_min_taskid | ( |  | self,
|  |  | minid )

Set the minimum taskid of future submitted tasks.

Further submitted tasks are guaranteed to have a taskid larger or equal to minid. This function is useful to make taskids consistent in a workflow that consists of sequential managers. (Note: This function is rarely used). If the minimum id provided is smaller than the last taskid computed, the minimum id provided is ignored.

Parameters

  self | Reference to the current work queue object.
  minid | Minimum desired taskid

Returns
  Returns the actual minimum taskid for future tasks.

## specify_priority()

ndcctools.work_queue.WorkQueue.specify_priority | ( |  | self,
|  |  | priority )

Change the project priority for the given queue.

Parameters

  self | Reference to the current work queue object.
  priority | An integer that presents the priorty of this work queue manager. The higher the value, the higher the priority.

## specify_num_tasks_left()

ndcctools.work_queue.WorkQueue.specify_num_tasks_left | ( |  | self,
|  |  | ntasks )

Specify the number of tasks not yet submitted to the queue.

It is used by work_queue_factory to determine the number of workers to launch. If not specified, it defaults to 0. work_queue_factory considers the number of tasks as: num tasks left + num tasks running + num tasks read.

Parameters

  self | Reference to the current work queue object.
  ntasks | Number of tasks yet to be submitted.

## specify_manager_mode()

ndcctools.work_queue.WorkQueue.specify_manager_mode | ( |  | self,
|  |  | mode )

Specify the manager mode for the given queue.

(Kept for compatibility. It is no-op.)

Parameters

  self | Reference to the current work queue object.
  mode | This may be one of the following values: WORK_QUEUE_MASTER_MODE_STANDALONE or WORK_QUEUE_MASTER_MODE_CATALOG.

## specify_master_mode()

ndcctools.work_queue.WorkQueue.specify_master_mode | ( |  | self,
|  |  | mode )

See also
  ndcctools.work_queue.WorkQueue.specify_manager_mode

## specify_catalog_server()

ndcctools.work_queue.WorkQueue.specify_catalog_server | ( |  | self,
|  |  | hostname,
|  |  | port )

Specify the catalog server the manager should report to.

Parameters

  self | Reference to the current work queue object.
  hostname | The hostname of the catalog server.
  port | The port the catalog server is listening on.

## specify_log()

ndcctools.work_queue.WorkQueue.specify_log | ( |  | self,
|  |  | logfile )

Specify a log file that records the cummulative stats of connected workers and submitted tasks.

Parameters

  self | Reference to the current work queue object.
  logfile | Filename.

## specify_transactions_log()

ndcctools.work_queue.WorkQueue.specify_transactions_log | ( |  | self,
|  |  | logfile )

Specify a log file that records the states of tasks.

Parameters

  self | Reference to the current work queue object.
  logfile | Filename.

## specify_password()

ndcctools.work_queue.WorkQueue.specify_password | ( |  | self,
|  |  | password )

Add a mandatory password that each worker must present.

Parameters

  self | Reference to the current work queue object.
  password | The password.

## specify_password_file()

ndcctools.work_queue.WorkQueue.specify_password_file | ( |  | self,
|  |  | file )

Add a mandatory password file that each worker must present.

Parameters

  self | Reference to the current work queue object.
  file | Name of the file containing the password.

## specify_max_resources()

ndcctools.work_queue.WorkQueue.specify_max_resources | ( |  | self,
|  |  | rmd )

Specifies the maximum resources allowed for the default category.

Parameters

  self | Reference to the current work queue object.
  rmd | Dictionary indicating maximum values. See ndcctools.work_queue.Task.resources_measured for possible fields. For example: >>> # A maximum of 4 cores is found on any worker: >>> q.specify_max_resources({'cores': 4}) >>> # A maximum of 8 cores, 1GB of memory, and 10GB disk are found on any worker: >>> q.specify_max_resources({'cores': 8, 'memory': 1024, 'disk': 10240})

## specify_min_resources()

ndcctools.work_queue.WorkQueue.specify_min_resources | ( |  | self,
|  |  | rmd )

Specifies the minimum resources allowed for the default category.

Parameters

  self | Reference to the current work queue object.
  rmd | Dictionary indicating minimum values. See ndcctools.work_queue.Task.resources_measured for possible fields. For example: >>> # A minimum of 2 cores is found on any worker: >>> q.specify_min_resources({'cores': 2}) >>> # A minimum of 4 cores, 512MB of memory, and 1GB disk are found on any worker: >>> q.specify_min_resources({'cores': 4, 'memory': 512, 'disk': 1024})

## specify_category_max_resources()

ndcctools.work_queue.WorkQueue.specify_category_max_resources | ( |  | self,
|  |  | category,
|  |  | rmd )

Specifies the maximum resources allowed for the given category.

Parameters

  self | Reference to the current work queue object.
  category | Name of the category.

  rmd

  Dictionary indicating maximum values. See ndcctools.work_queue.Task.resources_measured for possible fields. For example:
  >>> # A maximum of 4 cores may be used by a task in the category:
  >>> q.specify_category_max_resources("my_category", {'cores': 4})
  >>> # A maximum of 8 cores, 1GB of memory, and 10GB may be used by a task:
  >>> q.specify_category_max_resources("my_category", {'cores': 8, 'memory': 1024, 'disk': 10240})

## specify_category_min_resources()

ndcctools.work_queue.WorkQueue.specify_category_min_resources | ( |  | self,
|  |  | category,
|  |  | rmd )

Specifies the minimum resources allowed for the given category.

Parameters

  self | Reference to the current work queue object.
  category | Name of the category.

  rmd

  Dictionary indicating minimum values. See ndcctools.work_queue.Task.resources_measured for possible fields. For example:
  >>> # A minimum of 2 cores is found on any worker:
  >>> q.specify_category_min_resources("my_category", {'cores': 2})
  >>> # A minimum of 4 cores, 512MB of memory, and 1GB disk are found on any worker:
  >>> q.specify_category_min_resources("my_category", {'cores': 4, 'memory': 512, 'disk': 1024})

## specify_category_first_allocation_guess()

ndcctools.work_queue.WorkQueue.specify_category_first_allocation_guess | ( |  | self,
|  |  | category,
|  |  | rmd )

Specifies the first-allocation guess for the given category.

Parameters

  self | Reference to the current work queue object.
  category | Name of the category.

  rmd

  Dictionary indicating maximum values. See ndcctools.work_queue.Task.resources_measured for possible fields. For example:
  >>> # Tasks are first tried with 4 cores:
  >>> q.specify_category_first_allocation_guess("my_category", {'cores': 4})
  >>> # Tasks are first tried with 8 cores, 1GB of memory, and 10GB:
  >>> q.specify_category_first_allocation_guess("my_category", {'cores': 8, 'memory': 1024, 'disk': 10240})

## specify_category_max_concurrent()

ndcctools.work_queue.WorkQueue.specify_category_max_concurrent | ( |  | self,
|  |  | category,
|  |  | max_concurrent )

Specifies the maximum resources allowed for the given category.

Parameters

  self | Reference to the current work queue object.
  category | Name of the category.
  max_concurrent | Number of maximum concurrent tasks. Less then 0 means unlimited (this is the default). For example: >>> # Do not run more than 5 tasks of "my_category" concurrently: >>> q.specify_category_max_concurrent("my_category", 5)

## initialize_categories()

ndcctools.work_queue.WorkQueue.initialize_categories | ( |  | self,
|  |  | filename,
|  |  | rm )

Initialize first value of categories.

Parameters

  self | Reference to the current work queue object.
  rm | Dictionary indicating maximum values. See ndcctools.work_queue.Task.resources_measured for possible fields.
  filename | JSON file with resource summaries.

## cancel_by_taskid()

ndcctools.work_queue.WorkQueue.cancel_by_taskid | ( |  | self,
|  |  | id )

Cancel task identified by its taskid and remove from the given queue.

Parameters

  self | Reference to the current work queue object.
  id | The taskid returned from ndcctools.work_queue.WorkQueue.submit.

## cancel_by_tasktag()

ndcctools.work_queue.WorkQueue.cancel_by_tasktag | ( |  | self,
|  |  | tag )

Cancel task identified by its tag and remove from the given queue.

Parameters

  self | Reference to the current work queue object.
  tag | The tag assigned to task using ndcctools.work_queue.Task.specify_tag.

## cancel_by_category()

ndcctools.work_queue.WorkQueue.cancel_by_category | ( |  | self,
|  |  | category )

Cancel all tasks of the given category and remove them from the queue.

Parameters

  self | Reference to the current work queue object.
  category | The name of the category to cancel.

## shutdown_workers()

ndcctools.work_queue.WorkQueue.shutdown_workers | ( |  | self,
|  |  | n )

Shutdown workers connected to queue.

Gives a best effort and then returns the number of workers given the shutdown order.

Parameters

  self | Reference to the current work queue object.
  n | The number to shutdown. To shut down all workers, specify "0".

## block_host()

ndcctools.work_queue.WorkQueue.block_host | ( |  | self,
|  |  | host )

Block workers running on host from working for the manager.

Parameters

  self | Reference to the current work queue object.
  host | The hostname the host running the workers.

## blacklist()

ndcctools.work_queue.WorkQueue.blacklist | ( |  | self,
|  |  | host )

Replaced by ndcctools.work_queue.WorkQueue.block_host.

## block_host_with_timeout()

ndcctools.work_queue.WorkQueue.block_host_with_timeout | ( |  | self,
|  |  | host,
|  |  | timeout )

Block workers running on host for the duration of the given timeout.

Parameters

  self | Reference to the current work queue object.
  host | The hostname the host running the workers.
  timeout | How long this block entry lasts (in seconds). If less than 1, block indefinitely.

## blacklist_with_timeout()

ndcctools.work_queue.WorkQueue.blacklist_with_timeout | ( |  | self,
|  |  | host,
|  |  | timeout )

See ndcctools.work_queue.WorkQueue.block_host_with_timeout.

## unblock_host()

ndcctools.work_queue.WorkQueue.unblock_host | ( |  | self,
|  |  | host = None )

Unblock given host, of all hosts if host not given.

Parameters

  self | Reference to the current work queue object.
  host | The of the hostname the host.

## blacklist_clear()

ndcctools.work_queue.WorkQueue.blacklist_clear | ( |  | self,
|  |  | host = None )

See ndcctools.work_queue.WorkQueue.unblock_host.

## invalidate_cache_file()

ndcctools.work_queue.WorkQueue.invalidate_cache_file | ( |  | self,
|  |  | local_name )

Delete file from workers's caches.

Parameters

  self | Reference to the current work queue object.
  local_name | Name of the file as seen by the manager.

## specify_keepalive_interval()

ndcctools.work_queue.WorkQueue.specify_keepalive_interval | ( |  | self,
|  |  | interval )

Change keepalive interval for a given queue.

Parameters

  self | Reference to the current work queue object.
  interval | Minimum number of seconds to wait before sending new keepalive checks to workers.

## specify_keepalive_timeout()

ndcctools.work_queue.WorkQueue.specify_keepalive_timeout | ( |  | self,
|  |  | timeout )

Change keepalive timeout for a given queue.

Parameters

  self | Reference to the current work queue object.
  timeout | Minimum number of seconds to wait for a keepalive response from worker before marking it as dead.

## estimate_capacity()

ndcctools.work_queue.WorkQueue.estimate_capacity | ( |  | self | ) |

Turn on manager capacity measurements.

Parameters

  self | Reference to the current work queue object.

## tune()

ndcctools.work_queue.WorkQueue.tune | ( |  | self,
|  |  | name,
|  |  | value )

Tune advanced parameters for work queue.

Parameters

  self | Reference to the current work queue object.

  name

  The name fo the parameter to tune. Can be one of following:
  - "resource-submit-multiplier" Treat each worker as having ({cores,memory,gpus} * multiplier) when submitting tasks. This allows for tasks to wait at a worker rather than the manager. (default = 1.0)
  - "min-transfer-timeout" Set the minimum number of seconds to wait for files to be transferred to or from a worker. (default=10)
  - "foreman-transfer-timeout" Set the minimum number of seconds to wait for files to be transferred to or from a foreman. (default=3600)
  - "transfer-outlier-factor" Transfer that are this many times slower than the average will be aborted. (default=10x)
  - "default-transfer-rate" The assumed network bandwidth used until sufficient data has been collected. (1MB/s)
  - "fast-abort-multiplier" Set the multiplier of the average task time at which point to abort; if negative or zero fast_abort is deactivated. (default=0)
  - "keepalive-interval" Set the minimum number of seconds to wait before sending new keepalive checks to workers. (default=300)
  - "keepalive-timeout" Set the minimum number of seconds to wait for a keepalive response from worker before marking it as dead. (default=30)
  - "short-timeout" Set the minimum timeout when sending a brief message to a single worker. (default=5s)
  - "long-timeout" Set the minimum timeout when sending a brief message to a foreman. (default=1h)
  - "category-steady-n-tasks" Set the number of tasks considered when computing category buckets.
  - "hungry-minimum" Mimimum number of tasks to consider queue not hungry. (default=10)
  - "wait-for-workers" Mimimum number of workers to connect before starting dispatching tasks. (default=0)
  - "attempt-schedule-depth" The amount of tasks to attempt scheduling on each pass of send_one_task in the main loop. (default=100)
  - "wait_retrieve_many" Parameter to alter how work_queue_wait works. If set to 0, work_queue_wait breaks out of the while loop whenever a task changes to WORK_QUEUE_TASK_DONE (wait_retrieve_one mode). If set to 1, work_queue_wait does not break, but continues recieving and dispatching tasks. This occurs until no task is sent or recieved, at which case it breaks out of the while loop (wait_retrieve_many mode). (default=0)
  - "monitor-interval" Parameter to change how frequently the resource monitor records resource consumption of a task in a times series, if this feature is enabled. See enable_monitoring_full.

  value | The value to set the parameter to.

Returns
  0 on succes, -1 on failure.

## submit()

ndcctools.work_queue.WorkQueue.submit | ( |  | self,
|  |  | task )

Submit a task to the queue.

It is safe to re-submit a task returned by ndcctools.work_queue.WorkQueue.wait.

Parameters

  self | Reference to the current work queue object.
  task | A task description created from ndcctools.work_queue.Task.

## wait()

ndcctools.work_queue.WorkQueue.wait | ( |  | self,
|  |  | timeout = WORK_QUEUE_WAITFORTASK )

Wait for tasks to complete.

This call will block until the timeout has elapsed

Parameters

  self | Reference to the current work queue object.
  timeout | The number of seconds to wait for a completed task before returning. Use an integer to set the timeout or the constant WORK_QUEUE_WAITFORTASK to block until a task has completed.

## wait_for_tag()

ndcctools.work_queue.WorkQueue.wait_for_tag | ( |  | self,
|  |  | tag,
|  |  | timeout = WORK_QUEUE_WAITFORTASK )

Similar to ndcctools.work_queue.WorkQueue.wait, but guarantees that the returned task has the specified tag.

This call will block until the timeout has elapsed.

Parameters

  self | Reference to the current work queue object.
  tag | Desired tag. If None, then it is equivalent to self.wait(timeout)
  timeout | The number of seconds to wait for a completed task before returning.

## application_info()

ndcctools.work_queue.WorkQueue.application_info | ( |  | self | ) |

Should return a dictionary with information for the status display.

This method is meant to be overriden by custom applications.

The dictionary should be of the form:

{ "application_info" : {"values" : dict, "units" : dict} }

where "units" is an optional dictionary that indicates the units of the corresponding key in "values".

Parameters

  self | Reference to the current work queue object.

For example:

>>> myapp.application_info()
{'application_info': {'values': {'size_max_output': 0.361962, 'current_chunksize': 65536}, 'units': {'size_max_output': 'MB'}}}

## map()

ndcctools.work_queue.WorkQueue.map | ( |  | self,
|  |  | fn,
|  |  | seq,
|  |  | chunksize = 1 )

Maps a function to elements in a sequence using work_queue.

Similar to regular map function in python

Parameters

  self | Reference to the current work queue object.
  fn | The function that will be called on each element
  seq | The sequence that will call the function
  chunksize | The number of elements to process at once

## pair()

ndcctools.work_queue.WorkQueue.pair | ( |  | self,
|  |  | fn,
|  |  | seq1,
|  |  | seq2,
|  |  | chunksize = 1,
|  |  | env = None )

Returns the values for a function of each pair from 2 sequences.

The pairs that are passed into the function are generated by itertools

Parameters

  self | Reference to the current work queue object.
  fn | The function that will be called on each element
  seq1 | The first seq that will be used to generate pairs
  seq2 | The second seq that will be used to generate pairs
  chunksize | The number of pairs to process at once
  env | Poncho or conda environment tarball filename

## tree_reduce()

ndcctools.work_queue.WorkQueue.tree_reduce | ( |  | self,
|  |  | fn,
|  |  | seq,
|  |  | chunksize = 2 )

Reduces a sequence until only one value is left, and then returns that value.

The sequence is reduced by passing a pair of elements into a function and then stores the result. It then makes a sequence from the results, and reduces again until one value is left.

If the sequence has an odd length, the last element gets reduced at the end.

Parameters

  self | Reference to the current work queue object.
  fn | The function that will be called on each element
  seq | The seq that will be reduced
  chunksize | The number of elements per Task (for tree reduc, must be greater than 1)

## remote_map()

ndcctools.work_queue.WorkQueue.remote_map | ( |  | self,
|  |  | fn,
|  |  | seq,
|  |  | coprocess,
|  |  | name,
|  |  | chunksize = 1 )

Maps a function to elements in a sequence using work_queue remote task.

Similar to regular map function in python, but creates a task to execute each function on a worker running a coprocess

Parameters

  self | Reference to the current work queue object.
  fn | The function that will be called on each element. This function exists in coprocess.
  seq | The sequence that will call the function
  coprocess | The name of the coprocess that contains the function fn.
  name | This defines the key in the event json that wraps the data sent to the coprocess.
  chunksize | The number of elements to process at once

## remote_pair()

ndcctools.work_queue.WorkQueue.remote_pair | ( |  | self,
|  |  | fn,
|  |  | seq1,
|  |  | seq2,
|  |  | coprocess,
|  |  | name,
|  |  | chunksize = 1 )

Returns the values for a function of each pair from 2 sequences using remote task.

The pairs that are passed into the function are generated by itertools

Parameters

  self | Reference to the current work queue object.
  fn | The function that will be called on each element. This function exists in coprocess.
  seq1 | The first seq that will be used to generate pairs
  seq2 | The second seq that will be used to generate pairs
  coprocess | The name of the coprocess that contains the function fn.
  name | This defines the key in the event json that wraps the data sent to the coprocess.
  chunksize | The number of elements to process at once

## remote_tree_reduce()

ndcctools.work_queue.WorkQueue.remote_tree_reduce | ( |  | self,
|  |  | fn,
|  |  | seq,
|  |  | coprocess,
|  |  | name,
|  |  | chunksize = 2 )

Reduces a sequence until only one value is left, and then returns that value.

The sequence is reduced by passing a pair of elements into a function and then stores the result. It then makes a sequence from the results, and reduces again until one value is left. Executes on coprocess

If the sequence has an odd length, the last element gets reduced at the end.

Parameters

  self | Reference to the current work queue object.
  fn | The function that will be called on each element. Exists on the coprocess
  seq | The seq that will be reduced
  coprocess | The name of the coprocess that contains the function fn.
  name | This defines the key in the event json that wraps the data sent to the coprocess.
  chunksize | The number of elements per Task (for tree reduc, must be greater than 1)

The documentation for this class was generated from the following file:

- work_queue.py