DagScheduler

DagScheduler

#include <xalgospp/scheduling/dag_scheduler.hh>

class DagScheduler

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:175

The primary scheduler of processing work defined by a DAG.

The scheduler has a hierarchy of preferences for selecting the next piece of work to be run. It starts by searching its local queues (in the sense of NUMA locality) for work, before moving on to the items which do not have preferential locations, and finally to queues on remote NUMA nodes. This set of preferences is currently implemented solely for Linux, with a simplified fallback set of queues for other operating systems.

Schematically, the process can be viewed like:

                              [ Pipeline Tasks ]
                                      |
                              +-------v-------+
                              | Locality Hint |
                              +-------+-------+
                                      |
                +---------------------+---------------------+
                | (Affinity: Node 0)  | (Affinity: Node 1)  | (No Affinity)
                v                     v                     v
        +---------------+     +---------------+     +---------------+
        |  Node 0 Queue |     |  Node 1 Queue |     |  Global Queue |
        +-------+-------+     +-------+-------+     +-------+-------+
                |                     |                     |
   +------------+------------+        |                     |
   |                         |        |                     |

+-—v–—+ +-—v–—+ | | | Worker 0 | | Worker 1 | | | | (Node 0) | | (Node 0) | | | +—––—+ +—––—+ | | | (Steal Option 1) | | | +–––––––––––—v—–—+ | | (Steal Option 2) | +—————————————————–—+

Note

To try to remain generic, the scheduler relies on locality hints that may not be fully determined at construction. This allows for Tasks to generate data using mechanisms that are fully under their control - i.e., the Task and scheduling infrastructure does not enforce the use of specific buffers or memory spaces, so the generator can control this themselves. From a scheduling perspective this makes things more complex as the locality must be resolved lazily after preceeding parent Tasks have executed.

On top of the base hierarchy, the scheduler uses two context-aware throttling metrics to attempt to optimize the overall throughput of the system.

1. A concurrency limit is used for memory-bound steps to avoid negative impacts from oversubscription of available bandwidth.

2. There furthermore is a dynamic backpressure mechanism to avoid excessive in flight Tasks from steps which are NOT memory-bound themselves.

Beginning from a simple pipeline consisting of:

      [ Input Data Generator ] -----> [ Memory Bound Processing Step]

The overall behaviour can be modelled as:

                     [ Input Generator ]         [ High Mem Processing ]
                             |                             |
                          (Submit)                      (Submit)
                             |                             |
                          +-----------------------------------+
                          |             enqueue()             |
                          +-----------------------------------+
                             |                             |
                          +--v------------+                |
                          |  Overloaded?  |                |
                          +-------+-------+                |
                                  |                        |
                                  |                        |
                  +-----+         |       +----+           |
            +-----| Yes |<--------+------>| No |>-----+    |
            |     +-----+                 +----+      |    |
            |                                         |    |
            v                                         |    |
    +---------------+                           +-----v----v----+
    | Suspend Queue |             +------------>|  Queues Above |<-----------+
    +-------+-------+             |             +-----v----v----+            |
            |                     |                   |    |                 |
            |                     |                   |    |                 |
            |                     |                   | +--v--+    +-----+   |
            |                     |                   | | Hm? |--->| Yes |---+
            |                     |                   | +-----+    +-----+
            |                     |                   |    |
            |                     |                   |    |
            |                     |                   | +--v--+
            |                     |                   | | No. |
            |         (Pull from suspendeded)         | +--v--+
            |          (Upon any completion)          |    |
            |                     |                   |    |
            |                     |             +-----v----v----+
            +-------------------->+<------------|  On Complete  |
                                                +---------------+

Complete control of the queueing and throttling systems is exposed via the Config which can be configured at construction. There is additionally, in the Config, an option to perform “autotuning”. The DagScheduler will then attempt to optimize the parameters to maximize the throughput of the above systems. There is currently only a single simple algorithm available using basic hints from the system, the wrapped Tasks, and Little’s law.

The algorithm works as follows:

1. Determine the system resources.

2. Start from the input data size.

3. Using the input data size, and the memory multipliers from wrapped Task steps, determine an average memory footprint.

4. Using the resources and footprint, find the maximum number of steps that could be run concurrently.

5. Based on Little’s law determine the minimum number of steps to hide the latency of the input generation. A profiling step is used after a few submissions to get estimates of step latencies.

6. Calculate bandwidths from the total traffic and latencies.

7. From 5 and 6 calculate a limit based on memory bandiwdth.

8. From 7, determine the remaining concurrency numbers.

List of all members

Name	Kind	Owner
DagScheduler	function	Declared here
submit_dag	function	Declared here
enqueue	function	Declared here
wait_all	function	Declared here
acquire_buffer	function	Declared here
check_memory_bandwidth	function	Declared here
stage_algorithm	function	Declared here
m_config	variable	Declared here
m_topology	variable	Declared here
m_workers	variable	Declared here
m_node_queues	variable	Declared here
m_global_queue	variable	Declared here
m_running	variable	Declared here
m_unfinished_tasks	variable	Declared here
m_wait_mutex	variable	Declared here
m_wait_cv	variable	Declared here
m_cv_mutex	variable	Declared here
m_job_cv	variable	Declared here
m_hm_mutex	variable	Declared here
m_hm_cv	variable	Declared here
m_active_high_mem_tasks	variable	Declared here
m_pool_mutex	variable	Declared here
m_pools	variable	Declared here
m_suspension_mutex	variable	Declared here
m_num_suspended_generators	variable	Declared here
m_suspended_generators	variable	Declared here
m_world_comm	variable	Declared here
m_world_rank	variable	Declared here
m_world_size	variable	Declared here
m_shmem_comms	variable	Declared here
m_algo_windows	variable	Declared here
m_submissions_count	variable	Declared here
m_profiling_phase	variable	Declared here
m_io_profile	variable	Declared here
m_compute_profile	variable	Declared here
m_logger	variable	Declared here
worker_loop	function	Declared here
resolve_locality	function	Declared here
pin_thread_to_cores	function	Declared here
detect_numa_topology	function	Declared here
should_throttle_generators	function	Declared here
get_system_ram_bytes	function	Declared here
perform_autotune	function	Declared here
record_task_metrics	function	Declared here
on_task_complete	function	Declared here

Public Methods

Return	Name	Description
	DagScheduler explicit
void	submit_dag	Submit a complete DAG for execution.
void	enqueue	Enqueue a single task directly for execution.
void	wait_all	Wait on all submitted work to complete.
std::shared_ptr< ncarray::SOArray >	acquire_buffer	Retrieve an array buffer of the requested size from the buffer pool.
void	check_memory_bandwidth	Run a test timing routine for estimation of memory bandwidth.
void	stage_algorithm inline	Setup an Algorithm using a shared-memory MPI strategy.

---

DagScheduler

explicit

explicit DagScheduler(Config cfg = Config { 0, 0, true, 2, 2, true, true, 0, 5, 50.0, 0.25 })

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:226

---

submit_dag

void submit_dag(std::vector< std::shared_ptr< Task > > tasks)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:247

Submit a complete DAG for execution.

Parameters

Parameter	Type	Description
tasks	std::vector< std::shared_ptr< [Task](api-xalgospp-scheduling-Task.md#task) > >	The complete DAG - the root task contains the edges for its dependent tasks.

---

enqueue

void enqueue(std::shared_ptr< Task > task)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:254

Enqueue a single task directly for execution.

Parameters

Parameter	Type	Description
task	std::shared_ptr< [Task](api-xalgospp-scheduling-Task.md#task) >	The task to be enqueued for execution.

---

wait_all

void wait_all()

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:259

Wait on all submitted work to complete.

---

acquire_buffer

std::shared_ptr< ncarray::SOArray > acquire_buffer(numa_node_t node, ssize_t ndim, const ssize_t * shape, ncarray::DType dtype)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:270

Retrieve an array buffer of the requested size from the buffer pool.

Returns

A shared pointer to the array in the pool.

Parameters

Parameter	Type	Description
node	[numa_node_t](api-xalgospp-scheduling.md#numa_node_t)	The NUMA node for the buffer pool.
ndim	ssize_t	The dimensions for the requested array.
shape	const ssize_t *	The shape for the requested array.
dtype	ncarray::DType	The datatype for the requested array.

---

check_memory_bandwidth

void check_memory_bandwidth(std::size_t test_bytes = 32ULL *1024 *1024, std::size_t niter = 10)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:295

Run a test timing routine for estimation of memory bandwidth.

This routine will estimate the total bandwidth using test cases of a copy, scalar multiplier (buffer * scalar), addition of two buffers, and a multiply add routine. The input test buffer size should be of the order ~4-5x the total cache memory size. After that point, additionally increases in the size wont affect the measurement much, but will increase the total memory footprint.

Note

Currently, this is a very simple function, and it is not thread-safe. It should be called from the main thread only. In the future it will be improved to account for multi-thread behaviour. In the meantime, the best measurements will be collected when run from an isolated NUMA domain (I.e., the process is the only one on the domain).

Parameters

Parameter	Type	Description
test_bytes	std::size_t	The size in bytes of the array buffer to use. Should be on the order of 4-5x the cache size (L3 cache).
niter	std::size_t	The number of iterations to use. The timings will be taken from the average across iterations.

---

stage_algorithm

inline

template<class Algo> inline void stage_algorithm(Algo & algo, ShmemType shmem_type = ShmemType::MACHINE)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:314

Setup an Algorithm using a shared-memory MPI strategy.

Algorithms which require some staged data (e.g. constant matrices) can be setup to use various MPI shared communication strategies. Doing so, however, requires preparing the memory backing this data in a particular fashion. This function simplifies this process, setting up any needed communicators as well as ensuring proper synchronization so the memory will be valid for use by an rank running the Algorithm.

Parameters

Parameter	Type	Description
algo	Algo &	The Algorithm to run the data staging for.
shmem_type	[ShmemType](api-xalgospp-scheduling.md#shmemtype)	The granularity of the shared memory backing strategy. The enumerator specifies from machine/node down to cache level.

Private Attributes

Return	Name	Description
Config	m_config
std::map< numa_node_t, std::vector< int > >	m_topology	Mapping of node topology.
std::vector< std::thread >	m_workers	Complete set of workers.
std::vector< std::unique_ptr< WorkQueue > >	m_node_queues	Node-specific queues.
WorkQueue	m_global_queue	Global work queue.
std::atomic< bool >	m_running	Whether still running.
std::atomic< std::size_t >	m_unfinished_tasks	Number of pending tasks.
std::mutex	m_wait_mutex	Mutex for waitixng on all DAG work.
std::condition_variable	m_wait_cv	CV for waiting on all DAG work.
std::mutex	m_cv_mutex	Mutex for alerting a worker.
std::condition_variable	m_job_cv	CV for alerting a worker.
std::mutex	m_hm_mutex	Mutex for high-memory throttling.
std::condition_variable	m_hm_cv	CV for alerting when the HM status releases (token)
std::atomic< std::size_t >	m_active_high_mem_tasks
std::mutex	m_pool_mutex
std::map< PoolKey, std::shared_ptr< ArrayBufferPool > >	m_pools
std::mutex	m_suspension_mutex
std::atomic< std::size_t >	m_num_suspended_generators
std::vector< std::shared_ptr< Task > >	m_suspended_generators
MPI_Comm	m_world_comm
int	m_world_rank
int	m_world_size
std::vector< MPI_Comm >	m_shmem_comms	Any communicators used for Algorithms.
std::vector< RCWindow >	m_algo_windows	Backing windows for staged Algorithm data.
std::atomic< std::size_t >	m_submissions_count	Number of DAGs submitted.
std::atomic< bool >	m_profiling_phase	Whether currently profiling.
TaskProfileData	m_io_profile	Profile for IO-type Tasks.
TaskProfileData	m_compute_profile	Profiel for compute/processing Tasks.
std::shared_ptr< spdlog::logger >	m_logger

---

m_config

Config m_config

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:410

---

m_topology

std::map< numa_node_t, std::vector< int > > m_topology

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:411

Mapping of node topology.

---

m_workers

std::vector< std::thread > m_workers

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:412

Complete set of workers.

---

m_node_queues

std::vector< std::unique_ptr< WorkQueue > > m_node_queues

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:414

Node-specific queues.

---

m_global_queue

WorkQueue m_global_queue

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:415

Global work queue.

---

m_running

std::atomic< bool > m_running { true }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:417

Whether still running.

---

m_unfinished_tasks

std::atomic< std::size_t > m_unfinished_tasks { 0 }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:418

Number of pending tasks.

---

m_wait_mutex

std::mutex m_wait_mutex

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:421

Mutex for waitixng on all DAG work.

---

m_wait_cv

std::condition_variable m_wait_cv

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:422

CV for waiting on all DAG work.

---

m_cv_mutex

std::mutex m_cv_mutex

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:425

Mutex for alerting a worker.

---

m_job_cv

std::condition_variable m_job_cv

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:426

CV for alerting a worker.

---

m_hm_mutex

std::mutex m_hm_mutex

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:429

Mutex for high-memory throttling.

---

m_hm_cv

std::condition_variable m_hm_cv

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:430

CV for alerting when the HM status releases (token)

---

m_active_high_mem_tasks

std::atomic< std::size_t > m_active_high_mem_tasks { 0 }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:433

---

m_pool_mutex

std::mutex m_pool_mutex

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:438

---

m_pools

std::map< PoolKey, std::shared_ptr< ArrayBufferPool > > m_pools

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:439

---

m_suspension_mutex

std::mutex m_suspension_mutex

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:442

---

m_num_suspended_generators

std::atomic< std::size_t > m_num_suspended_generators { 0 }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:443

---

m_suspended_generators

std::vector< std::shared_ptr< Task > > m_suspended_generators

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:444

---

m_world_comm

MPI_Comm m_world_comm { MPI_COMM_NULL }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:447

---

m_world_rank

int m_world_rank { -1 }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:448

---

m_world_size

int m_world_size { -1 }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:449

---

m_shmem_comms

std::vector< MPI_Comm > m_shmem_comms

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:451

Any communicators used for Algorithms.

---

m_algo_windows

std::vector< RCWindow > m_algo_windows

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:452

Backing windows for staged Algorithm data.

---

m_submissions_count

std::atomic< std::size_t > m_submissions_count { 0 }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:455

Number of DAGs submitted.

---

m_profiling_phase

std::atomic< bool > m_profiling_phase { true }

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:456

Whether currently profiling.

---

m_io_profile

TaskProfileData m_io_profile

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:458

Profile for IO-type Tasks.

---

m_compute_profile

TaskProfileData m_compute_profile

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:459

Profiel for compute/processing Tasks.

---

m_logger

std::shared_ptr< spdlog::logger > m_logger

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:461

Private Methods

Return	Name	Description
void	worker_loop	Enter the main work queue.
numa_node_t	resolve_locality	Try to determine NUMA support using __NR_get_mempolicy syscall.
bool	pin_thread_to_cores
std::map< numa_node_t, std::vector< int > >	detect_numa_topology	Try to auto-determine the NUMA topology (Linux only).
bool	should_throttle_generators	When dynamic back-pressure monitoring is enabled, decide whether to throttle.
std::size_t	get_system_ram_bytes	Determine system memory resources.
void	perform_autotune	Run the automatic parameter tuning.
void	record_task_metrics
void	on_task_complete

---

worker_loop

void worker_loop(std::size_t thread_id, numa_node_t home_node)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:355

Enter the main work queue.

The work loop is as follows:

1. First check the local queue for work. The local queue will be configured to provide work that should preferrentially be performed within the NUMA node. If work is found, go to 4.

2. If no work is found check the global queue. This queue is configured to hold work that is not preferrentially localized to a NUMA node. If work is found, go to 4.

3. If both the local and global queues do not currently have work to do, then see if there is work to be stolen from remote NUMA node queues. If found go to 4.

4. Once a Task been found, first check thresholds for throttling. Currently this scheduler only has a throttling threshold on memory pressure. If too high and the concurrency limit has been reached, then sleep or go back to check the queues for other work.

5. Finally, execute the Task.

Parameters

Parameter	Type	Description
thread_id	std::size_t	This worker’s thread id.
home_node	[numa_node_t](api-xalgospp-scheduling.md#numa_node_t)	The indicator for which is the local queue.

---

resolve_locality

numa_node_t resolve_locality(const LocalityHint & hint)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:371

Try to determine NUMA support using __NR_get_mempolicy syscall.

This is linux-only and will only try to resolve the locality if the provided hint includes a memory affinity set with physical page info, and the hint does NOT include a preferred node already. If it does, that is returned. This routine will also return ANY_NODE in the case that a node was found, but there does not currently exist a work queue for it.

As fallback it will return ANY_NODE as the locality.

Returns

The NUMA node, or ANY_NODE.

Parameters

Parameter	Type	Description
hint	const [LocalityHint](api-xalgospp-scheduling-LocalityHint.md#localityhint) &	The locality hint.

---

pin_thread_to_cores

bool pin_thread_to_cores(const std::vector< int > & core_ids)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:373

---

detect_numa_topology

std::map< numa_node_t, std::vector< int > > detect_numa_topology()

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:385

Try to auto-determine the NUMA topology (Linux only).

This routine will try to use the exposed data from the filesystem as provided by the kernel.

The fallback currently just returns a dumb 4-thread “node”.

Returns

The determined NUMA topology.

---

should_throttle_generators

bool should_throttle_generators()

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:393

When dynamic back-pressure monitoring is enabled, decide whether to throttle.

Returns

When using the back-pressure system, returns true if throttling should begin for back pressure. Otherwise, false.

---

get_system_ram_bytes

std::size_t get_system_ram_bytes()

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:401

Determine system memory resources.

Returns

If on Linux, and it could be determined, the available ram. Otherwise, it just returns 16 GB for now.

---

perform_autotune

void perform_autotune()

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:406

Run the automatic parameter tuning.

---

record_task_metrics

void record_task_metrics(std::shared_ptr< Task > task, std::uint64_t elapsed_ns)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:408

---

on_task_complete

void on_task_complete(std::shared_ptr< Task > completed_task)

Defined in src/lib/xalgospp/scheduling/dag_scheduler.hh:435