Itoyori: Reconciling Global Address Space andGlobal Fork-Join Task Parallelism
SC23
Shumpei Shiina, Kenjiro Taura
The University of Tokyo
2023.11.14

What We Really Want to Reconcile: Productivity and Performance in HPC

Low-level programming models
that can achieve
the highest performance

High-level programming models
that can shortly achievesufficiently good performance

Two different programming models
for shared/distributed memory(MPI+X model)

Desired properties:

X = Pthreads, OpenMP, TBB, ...

Require much effort by HPC experts
Lower productivity

A single, unified programming model
for shared/distributed memory
General enough to easily express
dynamic and irregular parallelism

More is needed on this front
1 / 19

Fork-Join Task Parallelism on Distributed Memory?
Parallel execution based on dynamically forked tasks
Well suited for dynamic and irregular applications
Programmers can focus on logical parallelism without
considering hardware details (processor-obliviousness)
Popular shared-memory programming models forfork-join task parallelism:

OpenMP(#pragma omp task)... any systems for
distributed memory?

(oneTBB)2 / 19Itoyori: A Distributed Task-Parallel Runtime SystemA C++17 library for fork-join task parallelism on distributed memory
It depends only on MPI (capable of MPI-3 RMA) → Good Portability“Itoyori” is the Japanese name of the fish “Threadfin Breams”
Shared-memory-like simple global-view programming
Yet highly scalable and efficient
GitHub:https://github.com/itoyori/itoyori

3 / 19What Itoyori Offers
Work-stealing scheduler for fine-grained, global fork-join task parallelism
Tasks (user-level threads) can be scheduled across different nodes
Based on the uni-address scheme for inter-node dynamic thread migration– [Akiyama & Taura, HPDC ’15], scalability to > 100k cores [Shiina & Taura, Cluster ’22]

4 / 19What Itoyori Offers
Work-stealing scheduler for fine-grained, global fork-join task parallelism
Tasks (user-level threads) can be scheduled across different nodes
Based on the uni-address scheme for inter-node dynamic thread migration– [Akiyama & Taura, HPDC ’15], scalability to > 100k cores [Shiina & Taura, Cluster ’22]

Global address space, a view of shared memory over distributed memory
More specifically, Partitioned Global Address Space (PGAS)4 / 19

What Itoyori Offers
Work-stealing scheduler for fine-grained, global fork-join task parallelism
Tasks (user-level threads) can be scheduled across different nodes
Based on the uni-address scheme for inter-node dynamic thread migration– [Akiyama & Taura, HPDC ’15], scalability to > 100k cores [Shiina & Taura, Cluster ’22]

Global address space, a view of shared memory over distributed memory
More specifically, Partitioned Global Address Space (PGAS)

High-level C++ parallel STL-like interfacese.g., transform(), reduce()
They automatically call fork-join and global memory access APIs internally

4 / 19What Itoyori Does NOT Offer
Explicit point-to-point communication
Communication is implicitly issued when accessing the global address space

5 / 19What Itoyori Does NOT Offer
Explicit point-to-point communication
Communication is implicitly issued when accessing the global address space

Distributed shared memory (DSM) that allows transparent global memory access
Explicit API calls are required for global memory access in Itoyori (PGAS)

5 / 19What Itoyori Does NOT Offer
Explicit point-to-point communication
Communication is implicitly issued when accessing the global address space

Distributed shared memory (DSM) that allows transparent global memory access
Explicit API calls are required for global memory access in Itoyori (PGAS)

APIs to distinguish between inter- and intra-node processes
No need for two-level parallelization (e.g., MPI+X)

5 / 19What Itoyori Does NOT Offer
Explicit point-to-point communication
Communication is implicitly issued when accessing the global address space

Distributed shared memory (DSM) that allows transparent global memory access
Explicit API calls are required for global memory access in Itoyori (PGAS)APIs to distinguish between inter- and intra-node processes
No need for two-level parallelization (e.g., MPI+X)

Complicated APIs for task-parallel execution

5 / 19What Itoyori Does NOT Offer
Explicit point-to-point communication
Communication is implicitly issued when accessing the global address space

Distributed shared memory (DSM) that allows transparent global memory access
Explicit API calls are required for global memory access in Itoyori (PGAS)

APIs to distinguish between inter- and intra-node processes
No need for two-level parallelization (e.g., MPI+X)

Complicated APIs for task-parallel execution
Special compilers other than ordinary C++17 compilers
5 / 19

Key Contributions of Our Research
Proposing Itoyori, a distributed fork-join task-parallel runtime system
Itoyori reconciles PGAS and fine-grained fork-join task parallelism by introducing asoftware cache for global memory access

Demonstrating high productivity and performance of Itoyori through a real-world
application ExaFMM
7.5× speedup when scaled from a single node to 12 nodes
comparable performance to a hand-optimized MPI implementation

Itoyori is expected to strike a good balance
between productivity and performance!
6 / 19Outline

Itoyori’s Programming Model
Software Caching for Global Memory Access
Evaluation
Summary

Outline

Itoyori’s Programming Model
Software Caching for Global Memory Access
Evaluation
SummaryItoyori’s Programming Model ▷ An Example of Parallel Merge Sort
Sequential C++ code:
void msort(int* a, size_t n) {if (n < CUTOFF) {sort_small(a, n);} else {msort(a, n/2);
msort(a + n/2, n/2);
merge(a, n, n/2);}}

7 / 19Itoyori’s Programming Model ▷ An Example of Parallel Merge Sort
Sequential C++ code:
void msort(int* a, size_t n) {if (n < CUTOFF) {sort_small(a, n);} else {msort(a, n/2);
msort(a + n/2, n/2);

Divide the input array into two sub-arrays andsort them recursively (divide-and-conquer)

merge(a, n, n/2);}}

7 / 19Itoyori’s Programming Model ▷ An Example of Parallel Merge Sort
Sequential C++ code:
void msort(int* a, size_t n) {if (n < CUTOFF) {sort_small(a, n);} else {msort(a, n/2);
msort(a + n/2, n/2);

Divide the input array into two sub-arrays andsort them recursively (divide-and-conquer)

merge(a, n, n/2);}}

Merge the two sorted arrays
7 / 19Itoyori’s Programming Model ▷ An Example of Parallel Merge Sort
Sequential C++ code:
void msort(int* a, size_t n) {if (n < CUTOFF) {sort_small(a, n);} else {msort(a, n/2);
msort(a + n/2, n/2);

Switch to a fast sequential algorithm for small arrays
Divide the input array into two sub-arrays andsort them recursively (divide-and-conquer)

merge(a, n, n/2);}}

Merge the two sorted arrays
7 / 19Itoyori’s Programming Model ▷ An Example of Parallel Merge Sort
Sequential C++ code:

Distributed parallel code in Itoyori:

void msort(int* a, size_t n) {if (n < CUTOFF) {void msort(int* a, size_t n) {if (n < CUTOFF) {checkout(a, n, mode::read_write);
sort_small(a, n);checkin(a, n, mode::read_write);} else {thread th = fork([=]{ msort(a, n/2); });msort(a + n/2, n/2);
th.join();merge(a, n, n/2);}}

sort_small(a, n);} else {msort(a, n/2);
msort(a + n/2, n/2);
merge(a, n, n/2);}}

7 / 19Itoyori’s Programming Model ▷ An Example of Parallel Merge Sort
Sequential C++ code:

Distributed parallel code in Itoyori:

void msort(int* a, size_t n) {if (n < CUTOFF) {void msort(int* a, size_t n) {if (n < CUTOFF) {checkout(a, n, mode::read_write);
sort_small(a, n);checkin(a, n, mode::read_write);} else {thread th = fork([=]{ msort(a, n/2); });msort(a + n/2, n/2);
th.join();merge(a, n, n/2);}}

sort_small(a, n);} else {msort(a, n/2);
msort(a + n/2, n/2);
merge(a, n, n/2);}}

Parallel tasks can be dynamically forked and joined,even recursively (Nested fork-join parallelism)7 / 19Itoyori’s Programming Model ▷ An Example of Parallel Merge Sort
In order to access global memory, programmers
call checkout/checkin API
SequentialneedC++ tocode:
Distributed parallel code in Itoyori:
void msort(int* a, size_t n) {if (n < CUTOFF) {sort_small(a, n);} else {msort(a, n/2);
msort(a + n/2, n/2);
merge(a, n, n/2);}}

void msort(int* a, size_t n) {if (n < CUTOFF) {checkout(a, n, mode::read_write);
sort_small(a, n);checkin(a, n, mode::read_write);} else {thread th = fork([=]{ msort(a, n/2); });msort(a + n/2, n/2);
th.join();merge(a, n, n/2);}}

Parallel tasks can be dynamically forked and joined,even recursively (Nested fork-join parallelism)7 / 19

Checkout/Checkin APIs
Raw virtual addresses can be used for
global memory accessvoid msort(int* a, size_t n) {if (n < CUTOFF) {checkout(a, n, mode::read_write);
sort_small(a, n);checkin(a, n, mode::read_write);} else {thread th = fork([=]{ msort(a, n/2); });msort(a + n/2, n/2);
th.join();merge(a, n, n/2);}}

8 / 19Checkout/Checkin APIs
Raw virtual addresses can be used for
global memory accessvoid msort(int* a, size_t n) {if (n < CUTOFF) {checkout(a, n, mode::read_write);
sort_small(a, n);checkin(a, n, mode::read_write);} else {thread th = fork([=]{ msort(a, n/2); });msort(a + n/2, n/2);
th.join();merge(a, n, n/2);}}

Requests local access to global
memory region [𝑎, 𝑎 + 𝑛)Specifies the access mode
(read, read_write, or write)
If read or read_write, the latest data
may be fetched from owners

8 / 19

Checkout/Checkin APIs
Raw virtual addresses can be used for
global memory accessvoid msort(int* a, size_t n) {if (n < CUTOFF) {checkout(a, n, mode::read_write);
sort_small(a, n);checkin(a, n, mode::read_write);} else {thread th = fork([=]{ msort(a, n/2); });msort(a + n/2, n/2);
th.join();merge(a, n, n/2);}}

Requests local access to global
memory region [𝑎, 𝑎 + 𝑛)Specifies the access mode(read, read_write, or write)
If read or read_write, the latest data
may be fetched from owners

Claims the completion of memory access
Passes the same arguments as thecorresponding checkout call
If read_write or write, this region isconsidered modified

8 / 19Outline

Itoyori’s Programming Model
Software Caching for Global Memory Access
Evaluation
SummaryGlobal Address Space + Global Task Parallelism = ?
Partitioned Global Address Space(PGAS) model:Inter-node dynamic load balancing(global task parallelism):

Programmers optimize data
movement by explicitly distinguishing
between global and local data

The runtime system can dynamicallymove tasks across nodes for loadbalancing

We want to aggregate communication
for different tasks working on thesame data

Requiring each task independently
issue communication for its own data

If we naively combine these two...
⇨ Redundant, fine-grained communication
9 / 19Redundant, Fine-Grained Communication in Parallel Merge Sort

void msort(int* a, size_t n) {if (n < CUTOFF) {checkout(a, n, mode::read_write);
sort_small(a, n);checkin(a, n, mode::read_write);} else {thread th = fork([=]{ msort(a, n/2); });msort(a + n/2, n/2);
th.join();At merge, we want to reuse remote data fetched in themerge(a, n, n/2);
previous sort functions}}However, it is difficult for programmers to do sobecause these tasks may run on different nodes

10 / 19

Redundant, Fine-Grained Communication in Parallel Merge Sort
As a result, global memory accesses

void msort(int* a, size_t n) {are issued for each task
if (n < CUTOFF) {checkout(a, n, mode::read_write);
More fine-grained tasks
sort_small(a, n);⇨ More fine-grained communication
checkin(a, n, mode::read_write);} else {thread th = fork([=]{ msort(a, n/2); });msort(a + n/2, n/2);
th.join();At merge, we want to reuse remote data fetched in themerge(a, n, n/2);
previous sort functions}}

However, it is difficult for programmers to do so
because these tasks may run on different nodes

10 / 19

Execution Time (s)Performance of Naive Combination of PGAS and Dynamic Load Balancing

Recursive parallel merge sort10

called Cilksort

Ran on 12 nodes (576 cores)
More fine-grained tasks⇨ More fine-grained communication⇨ Worse performance

1100

1000

10k

100k

Cutoff count (# of elements)

11 / 19

Reconciling Them by Software Caching!

By introducing a software cache, the runtime system, rather than programmers,
can aggregate communication for tasks that are scheduled on the same node
Exploit spatial locality by fetching larger data than requested
Exploit temporal locality by reusing fetched data across different tasks
We designed checkout/checkin APIs for efficient software caching
Avoid unnecessary copy overhead that would occur in traditional PGAS APIs (GET/PUT)
See our paper for more details!

12 / 19

Execution Time (s)Performance Improvement by Software Caching
No CacheItoyori (with Cache)

10By software caching, Itoyori becomes
more robust to fine-grained parallelism

11001000

10k

100k

Cutoff count (# of elements)

13 / 19

Virtual Memory Mappings for Software Cache

Process 1

Global view
(virtual address space)

Cache blocks
Process 1

Process 2

Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Process 1

Global view
(virtual address space)

Global memory consists of memory blocks of fixed sizeHome blocks are physical memory blocks owned by eachprocessblocks
Cache

Process 1

Process 2

Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Process 1

Global view
(virtual address space)

Cache blocks
Process 1

Home blocks
aredirectly3 mappedProcess
to the globalview4
Process2
Process

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Checkout/CheckinProcess 1Global view
(virtual address space)

Cache blocks
Process 1

Process 2

Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Checkout/CheckinProcess 1

Global view
(virtual address space)

Cache blocks
Cache blocks are mapped to the global view on demand
Process 1

Process 2

Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Checkout/CheckinProcess 1

Global view
(virtual address space)

Cache blocks

AR DM
Process 1

Process 2

Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Checkout/CheckinProcess 1

Global view
(virtual address space)

Cache blocks
Process 1

Process 2

Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Checkout/CheckinProcess 1

Global view
(virtual address space)

Cache blocks
Cache blocks can be dynamically evictedfrom the global view
Process 1
Process 2
Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Checkout/CheckinProcess 1

Global view
(virtual address space)

Cache blocks
Cache blocks can be dynamically evictedfrom the global view
Process 1
Process 2
Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Virtual Memory Mappings for Software Cache

Checkout/CheckinProcess 1

Global view
(virtual address space)

Cache blocks
Process 1

Process 2

Process 3

Process 4

Home blocks
Memory distribution
14 / 19

Memory Consistency and Cache Coherence

Itoyori employs a relaxed memory consistency model that assumes that theprogram is data-race-free
No data race is allowed in Itoyori programs

Caches can be invalidated and written back to their home at fork-join points
but only when work-stealing events happen

RDMA-based efficient cache management for work stealing
Please check out the paper for more details!

15 / 19

Outline

Itoyori’s Programming Model
Software Caching for Global Memory Access
Evaluation
SummaryPerformance Evaluation of Itoyori
We evaluated Itoyori with three fork-join applications
Cilksort, UTS-Mem, and ExaFMM

In this talk, we show the result for ExaFMM only
Experimental environment:
Wisteria/BDEC-01 Odyssey supercomputer at The University of Tokyo
Similar configuration to Fugaku Supercomputer
CPU: Fujitsu A64FX (48 cores/node)
Memory: HBM2 (32 GiB/node)
Network: Fujitsu MPI over Tofu Interconnect D

16 / 19

ExaFMM (High-Performance FMM Library) [Yokota+, CPC ’13]
ExaFMM approximates interactions between
far-enough particles by using a global tree
Highly dynamic and irregular parallelism

We ported a shared-memory fork-jointask-parallel implementation of ExaFMM[Taura+, ScalA ’12] to Itoyori
The overall parallel algorithm was notchanged from the original shared-memory
code, except for microscopic changes

Tree-based computation in ExaFMM
from [Yokota+, CPC ’13]

If we were to use MPI, we would have to redesign
the parallel algorithm itself

17 / 19

ExaFMM ▷ Strong Scaling
No CacheItoyori (with Cache)
MPI

Execution time (s)1000

10M bodies
Software caching improved
performance by up to 6.0×

Li100ne(v ar ss. pe
se edria ul) p

6.0×

7.5× speedup on 12 nodes (vs. 1 node)

10100# of cores

1000
18 / 19

ExaFMM ▷ Strong Scaling
No CacheItoyori (with Cache)
MPI

Execution time (s)1000

An existing MPI implementation of ExaFMM
Hybrid of MPI and fork-join task parallelism
Inter-node: static load balancing using MPI
Intra-node: task parallelism (the same)

10M bodies

Software caching improved
performance by up to 6.0×

Li100ne(v ar ss. pese edria ul) p

6.0×

7.5× speedup on 12 nodes (vs. 1 node)

10100# of cores

1000
18 / 19

ExaFMM ▷ Strong Scaling
No CacheItoyori (with Cache)
MPI

Execution time (s)1000

An existing MPI implementation of ExaFMM
Hybrid of MPI and fork-join task parallelism
Inter-node: static load balancing using MPI
Intra-node: task parallelism (the same)

10M bodies

Software caching improvedperformance by up to 6.0×

Li100ne(v ar ss. pese edria ul) p

6.0×

10100

# of cores

1000

7.5× speedup on 12 nodes (vs. 1 node)

Itoyori performs competitively to thehand-optimized MPI version, while maintaining high productivity
18 / 19

Outline

Itoyori’s Programming Model
Software Caching for Global Memory Access
Evaluation
SummarySummary
Itoyori is a C++ global-view programming framework for fork-join task parallelism
Software caching is a key to scale fork-join parallelism to distributed memory
We designed efficient software cache with checkout/checkin APIs
Our experiments suggested that Itoyori could achieve a good balance between
productivity and performance
GitHub:https://github.com/itoyori/itoyori

19 / 19