The INTERTWinE project and in-situ
data analytics
http://www.intertwine-project.eu
Nick Brown, EPCC – n.brown@epcc.ed.ac.uk
This project is funded from the European Union’s Horizon 2020 Research and Innovation programme
under Grant Agreement no. 671602.

INTERTWinE
Interoperability between programming models
for scalable performance on extreme-scale supercomputers• Computational Resource Management• Coordinated resource sharing for interoperability
between runtime systems, libraries

Linear Algebra

CFD

Big Data analytics

Space Plasma
Complex Fluids• Distributed Data Management• Scalable, transparent data sharing onheterogeneous, distributed memory hierarchies• Engagement with HPC community• Standards bodies: OpenMP, MPI, GASPI• Courses, workshops and Best Practice Guides
Follow INTERTWinE on Twitter: @intertwine_eu
http://www.intertwine-project.eu

Interoperable node-level resource sharing
Computational Resource Sharing• Multiple codes compete for CPU
cores, accelerator devices oncluster nodes• Application threads• Numerical libraries threads• Runtime systems threads• Communication library threads

Sequential
application
Sequential
Application

PLASMAlinear algebraon OpenMPPLASMAIntel MKL

CPU USAGEMKLdgemm()

dpotrf()

01

2

345

• Interference leads to resource over-subscription or under-subscription oncluster nodes• Interoperability?
• Need coordinated resource sharing:• Ability to express general resource needs• Ability to express dynamic resource requirements:• computational-heavy periods, idleness periods
http://www.intertwine-project.eu

Interoperable node-level resource sharing
Computational Resource Sharing• Multiple codes compete for CPU
cores, accelerator devices oncluster nodes• Application threads• Numerical libraries threads• Runtime systems threads• Communication library threads

ParallelapplicationOmpSs

Application

PLASMAlinearalgebraon OpenMP

Intel MKL

PLASMAdgemm()

CPU USAGE

MKL

dpotrf()

012

345• Interference leads to resource over-subscription or under-subscription oncluster nodes• Interoperability?
• Need coordinated resource sharing:• Ability to express general resource needs• Ability to express dynamic resource requirements:• computational-heavy periods, idleness periods
http://www.intertwine-project.eu

INTERTWinE — Resource Manager APIs
Dynamic Resource Sharing
Enable runtime systems to dynamically negotiate and adjust resource usage

Parallelapplication
OmpSs

Application

PLASMAlinear
algebra
PLASMA
on(OpenMP)
OpenMP

Resource
manager

DRS

lend ()
dgemm()Intel

MKL
MKL(TBB)CPU USAGE

lend ()

borrow()
enable()dpotrf()disablelend()

reclaim()borrow()

enable()lend()

012345

http://www.intertwine-project.eu

MPI and tasks deadlock issue
S1R1

S2R2

S3R3S4R4S5R5S6

R6S7

R7S8R8S9R9MPI and tasks deadlock issue
S1R2S3

S2 S7

R1 R3S4

S5 S9

R4 R6R5S6R7S8R8R9

Interoperability MPI for task-based communications
28/09/20
17Gauss-Seidel Strong Scaling with MPI+OmpSs
1684Speedup• Using the resource manager
we have extended the MPIlibrary• The resource manager willpause threads/tasks thatissue blocking
communication call (lend)• MPI library polls
outstanding requests anduses the resource manager
to rewake threads whencommunications complete(reclaim)

G nuplotOmpSs fork-join
OpenMP
fork-join
OmpSs
tasks
OpenMPtasks

2OmpSs tasks+interop
OpenMPtasks+interop10.50.25
12481632Nodes

Strong scaling in 2 dimensions,
global size is 64k by 64k.
100 iterations and 48 cores per node(MareNostrum4.)
http://www.intertwine-project.eu

Task multiple parallelism: 4 nodes (48 cores per node)

OmpSs
Fork /join

OmpSs
tasks

OmpSs
tasks +interop• Interoperability MPI,
built upon theresource manager
significantly cuts
down the amount ofsynchronisation
required• Lots more threads
can be active

Green arrow
represents
10 iterations
http://www.intertwine-project.eu

Implicit task and distributed memory interoperability• Schedules lots oftasks, which will notfor (i=0; i<x; i++) {run until their input
for (j=0; j<y; j++) {dependencies are#pragma omp task in(A[i,j], B[i,j]) out(A[j,i], B[j,i])my_task(A[i,j], B[i,j], A[j,i], B[j,i])met.
}}• Each produces outputs
that other tasks may
depend upon

• Task-based models tend to be limited to a single memory space• A significant limitation and hence we want to extend this to
distributed memory machines• For instance, the array A might be distributed amongst the nodes• When the task starts, the data it needs is available(irrespective of where that is located.) When the taskcompletes the resulting data is available to any other task.
http://www.intertwine-project.eu

Directory/cache: For distributed shared memory
0xFF0000

Node 2

Array A

Node 1
Programmer’sview ofmemory

Node 0Node 2

Array B

Node 10x000000

Array A

Node 0Array B

Array B

Array A
Array B
Node 0Physicalview ofmemory

Array A
Node 1

Node 2
http://www.intertwine-project.eu

Directory/cache integration and transport layers• It is intended to be integrated into the runtimes of task basedmodels (but can be used directly)• Such as OmpSs, StarPU, PaRSEC, GPI-Space• So its use (and existence!) is transparent to the programmer
OmpSs Runtime

StarPU Runtime

GPI-Space
RuntimePaRSEC
RuntimeDirectory/Cache (Virtual Memory Manager)

GASPI
Segment

MPI
Segment

BeeGFS
Segment…

• Transport layers are provided that implement the underlying
data movement• GASPI, MPI RMA, BeeGFS• Very easy to switch in and out
http://www.intertwine-project.eu

Early performance measures• Block Cholesky matrix factorisation• 16384 * 16384 elements, block size of 16 * 16
• BLAS for computation, concentrating here on thecost of data movement• On ARCHER, Cray XC30

http://www.intertwine-project.euLinear Algebra

Big Data analytics

CFDSpace Plasma

Complex Fluids

Big data analytics for atmospheric modelling• Separately we have worked with the Met
Office on a large scale atmospheric code
• The previous code was capable ofmaximum 20 million grid points over256 cores• We routinely run our model (MONC)
on 2.1 billion grid points, 32768 cores

Prognostics

Diagnostics

Analysing the data in-situ• Have many computational processes and a number of data analytics
cores• Typically one core per processor is dedicated to IO, serving the othercores running the computational model• Computational cores “fire and forget” their data
• In-situ as raw data is never written out• Would be too time consuming• Avoids blocking the computationalcores for analytics and IO• Dynamic time-stepping, checkpoint-restart of analytics,
bit reproducibility, easy configuration

Our analytics pipelineUnder the hood: Event handling• The federators and their sub activities are event handlers• Process events concurrently by assigning these to these from a pool
• Aids asynchronous data handling• As soon as data arrives process it• Internal state of event handlers needs protection (mutexes/rwlocks)• Challenge:• Bit reproducibility• For some handlers enforce a
predictable order ofprocessing events (based onmodel timestep.)• Queue up out of order events

Inter-IO communications challenge• We promote asynchronicity andprocessing of events out of orderwhere possible• Many inter IO communications
involve collective operations (such asa reduction)• We would like to use MPI, but issue order of collectives matters(i.e. if IO server 1 issues a reduce on field A and then B, then allother IO servers must issue reductions in that same order)• But ensuring this would require additional overhead and/orcoordination• Solution: Abstract through active messaging

Active messaging for inter-IO communications• These communication calls additionally provide• UniqueID: matching collectives even if they are issued out of order• Callback : Handling procedure called on the root when the dataarrives• inter_io_reduce(data, data_type, action, root, uniqueId,callback)• When this reduction is completed on the root, a thread is activatedfrom the pool and calls the handling function (typically in the eventhandler)call inter_io_reduce(data, type, 0, “sum”, fieldname//”_”//timestep, handler)subroutine handler(data, data_type, uniqueId)………
end subroutine handler

Performance and scalability• Standard MONC stratus cloud test case• Weak scaling on Cray XC30, 65536 local grid points• 232 diagnostic values every timestep, time averagedover 10 model seconds. File written every 100 modelseconds. Run terminates after 2000 model seconds.

What’s the link to INTERTWinE?
• This is (distributed) task-based parallelism• Active messaging addresses the interoperability
challenge between tasks (threads from the pool) andMPI communications• Our analytics pipeline is currently built on-top of ourown lower layer which provides active messaging via
MPI P2P

Analytics Pipeline
Task based modelResourcemanager

DirectorycacheAnalytics Pipeline
Event handlers
Active messagingMPIpthreads• Instead, can we keep our abstractionof the analytics pipeline but build thisupon existing task-based models(OmpSs, OpenMP etc)
• Build upon the resource manager anddirectory/cache of INTERTWinE• Ideally extend the notion of task
dependencies• The work of INTERTWinE is key here
http://www.intertwine-project.eu

Summary and collaboration opportunities• Best Practice Guides:• Writing GASPI-MPI Interoperable Programs• MPI + OpenMP Programming• MPI + OmpSs Interoperable Programs• Open MP/OmpSs/StarPU + Multi-threaded Libraries Interoperable Programs• “Developer Hub” of resources for developers & application users• We are very happy to make available our directory/cache andresource manager implementations• In-situ data analytics:• Keen to explore generalising these ideas to wide
variety of parallel codes generating lots of data
http://www.intertwine-project.eu