Skip to content

compiler: Make async handshake flag accesses atomic #3070

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ggorman wants to merge 1 commit into
base: fix-wait-timing
Choose a base branch
from
Draft

compiler: Make async handshake flag accesses atomic #3070

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 16 additions & 20 deletions devito/ir/iet/nodes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
AFFINE, INBOUND, PARALLEL, PARALLEL_IF_ATOMIC, PARALLEL_IF_PVT, SEQUENTIAL,
VECTORIZED, Forward, PrefetchUpdate, Property, WithLock
)
from devito.symbolics import CallFromPointer, ListInitializer, Macro
from devito.symbolics import Byref, CallFromPointer, DefFunction, ListInitializer, Macro
from devito.tools import (
Signer, as_tuple, ctypes_to_cstr, filter_ordered, filter_sorted, flatten
)
Expand Down Expand Up @@ -62,12 +62,13 @@
'Section',
'Switch',
'SyncSpot',
'ThreadFence',
'TimedList',
'Transfer',
'Using',
'UsingNamespace',
'While',
'atomic_load',
'atomic_store',
]

# First-class IET nodes
Expand Down Expand Up @@ -1304,29 +1305,24 @@ def size(self):
return self.ispace.size


class ThreadFence(Call):

def atomic_load(flag):
"""
A memory fence, ordering a thread's accesses either side of it.

Threads exchanging work through a shared flag need memory ordering on both
sides: a release before publishing the flag, and an acquire after observing it.
This ensures the state associated with the flag is visible before it is used.
Without this ordering, updates may be observed in the wrong order and lost.
An acquire load of a flag that threads synchronise on, for use in a wait
condition. Reads that follow it cannot be satisfied before it.

`__atomic_thread_fence` is a compiler builtin, so this renders the same in C
and in C++ and needs no header.
The `__atomic` builtins accept the existing `volatile int` flags and are
valid in C and C++ alike, so no declaration changes and no header is needed.
"""
return DefFunction('__atomic_load_n', (Byref(flag), Macro('__ATOMIC_ACQUIRE')))

def __init__(self, order):
assert order in ('acquire', 'release')
super().__init__('__atomic_thread_fence',
[Macro(f'__ATOMIC_{order.upper()}')])
self._order = order

@property
def order(self):
return self._order
def atomic_store(flag, value):
"""
A release store to a flag that threads synchronise on, the counterpart of
`atomic_load`: everything sequenced before it is visible to whoever then
acquires `flag`.
"""
return Call('__atomic_store_n', (Byref(flag), value, Macro('__ATOMIC_RELEASE')))


class Prodder(Call):
Expand Down
42 changes: 17 additions & 25 deletions devito/passes/iet/asynchrony.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,8 @@
from devito.ir import (
AsyncCall, AsyncCallable, BlankLine, BusyWait, BusyWaitCall, Call, Callable,
Conditional, DummyEq, DummyExpr, EntryFunction, FindNodes, FindSymbols, Increment,
Iteration, List, PointerCast, Return, ThreadCallable, ThreadFence, Transformer, While,
make_callable, maybe_alias
Iteration, List, PointerCast, Return, ThreadCallable, Transformer, While, atomic_load,
atomic_store, make_callable, maybe_alias
)
from devito.passes.iet.definitions import DataManager
from devito.passes.iet.engine import iet_pass
Expand Down Expand Up @@ -70,11 +70,13 @@ def _lower_async_objs(iet, tracker=None, sregistry=None, **kwargs):

if sdata.size == 1:
d = sdata.index
condition = CondNe(FieldFromComposite(sdata.symbolic_flag, sdata[d]), 1)
flag = FieldFromComposite(sdata.symbolic_flag, sdata[d])
condition = CondNe(atomic_load(flag), 1)
activation = [BusyWait(condition)]
else:
d = Temp(name=sregistry.make_name(prefix=sdata.index.name))
condition = CondNe(FieldFromComposite(sdata.symbolic_flag, sdata[d]), 1)
flag = FieldFromComposite(sdata.symbolic_flag, sdata[d])
condition = CondNe(atomic_load(flag), 1)
activation = [DummyExpr(d, 0),
BusyWait(condition, DummyExpr(d, (d + 1) % sdata.size))]
arguments = []
Expand All @@ -88,13 +90,8 @@ def _lower_async_objs(iet, tracker=None, sregistry=None, **kwargs):
activation.extend([DummyExpr(FieldFromComposite(i.base, sdata[d]), i)
for i in arguments])

# Publishing the request must happen last. Everything the thread depends on
# must be visible before this flag is set. `volatile` does not provide that
# ordering, so weakly ordered targets could otherwise observe stale state.
activation.append(ThreadFence('release'))
activation.append(
DummyExpr(FieldFromComposite(sdata.symbolic_flag, sdata[d]), 2)
)
# The request is published last, so the thread sees everything set above
activation.append(atomic_store(flag, 2))
name = sregistry.make_name(prefix='activate')
efunc = make_callable(name, activation)

Expand Down Expand Up @@ -134,6 +131,7 @@ def _(iet, key=None, tracker=None, sregistry=None, **kwargs):
pname=sregistry.make_name(prefix='tsdata')
)
sbase = sdata.indexed
flag = FieldFromPointer(sdata.symbolic_flag, sbase)

# PThreadArray -- that is he pthreads that will execute the AsyncCallable
# asynchronously
Expand All @@ -144,31 +142,25 @@ def _(iet, key=None, tracker=None, sregistry=None, **kwargs):
tbase = threads.indexed

# Prepend the SharedData fields available upon thread activation
preactions = [ThreadFence('acquire')]
preactions.extend(DummyExpr(i, FieldFromPointer(i.base, sbase))
for i in ncfields)
preactions = [DummyExpr(i, FieldFromPointer(i.base, sbase)) for i in ncfields]
preactions.append(BlankLine)

# Append the flag reset
postactions = [List(body=[
BlankLine,
# Whatever the task produced -- data in a buffer, a lock handed back --
# has to be visible before the flag says it is done.
ThreadFence('release'),
DummyExpr(FieldFromPointer(sdata.symbolic_flag, sbase), 1)
# has to be visible before the flag says it is done
atomic_store(flag, 1)
])]

wrap = List(body=preactions + list(iet.body.body) + postactions)

# The thread has work to do when it receives the signal that all locks have
# been set to 0 by the main thread
wrap = Conditional(
CondEq(FieldFromPointer(sdata.symbolic_flag, sbase), 2),
wrap
)
wrap = Conditional(CondEq(atomic_load(flag), 2), wrap)

# The thread keeps spinning until the alive flag is set to 0 by the main thread
wrap = While(CondNe(FieldFromPointer(sdata.symbolic_flag, sbase), 0), wrap)
wrap = While(CondNe(atomic_load(flag), 0), wrap)

# pthread functions expect exactly one argument of type void*
tparameter = Pointer(name=f'_{sdata.name}')
Expand Down Expand Up @@ -208,7 +200,7 @@ def _(iet, key=None, tracker=None, sregistry=None, **kwargs):
body.append(DummyExpr(FieldFromPointer(i.base, sbase), i.base))
body.extend([
BlankLine,
DummyExpr(FieldFromPointer(sdata.symbolic_flag, sbase), 1)
DummyExpr(flag, 1)
])
body.extend([
BlankLine,
Expand All @@ -223,8 +215,8 @@ def _(iet, key=None, tracker=None, sregistry=None, **kwargs):
# Create an efunc to shutdown the pthreads
name = sregistry.make_name(prefix='shutdown')
body = List(body=callback([
BusyWait(CondEq(FieldFromPointer(sdata.symbolic_flag, sbase), 2)),
DummyExpr(FieldFromPointer(sdata.symbolic_flag, sbase), 0),
BusyWait(CondEq(atomic_load(flag), 2)),
atomic_store(flag, 0),
Call('pthread_join', (threads[d], Null))
]))
shutdown = make_callable(name, body)
Expand Down
20 changes: 8 additions & 12 deletions devito/passes/iet/orchestration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,8 @@
from devito.exceptions import CompilationError
from devito.ir.iet import (
AsyncCall, AsyncCallable, BlankLine, Block, BusyWait, BusyWaitCall, Call, Callable,
Conditional, DummyExpr, List, SyncSpot, ThreadFence, Transformer, derive_parameters,
make_callable
Conditional, List, SyncSpot, Transformer, atomic_load, atomic_store,
derive_parameters, make_callable
)
from devito.ir.iet.visitors import Visitor
from devito.ir.support import (
Expand Down Expand Up @@ -44,7 +44,7 @@ def __init__(self, sregistry=None, options=None, **kwargs):

def _make_waitlock(self, iet, sync_ops, *args):
waitloop = List(
body=BusyWait(Or(*[CondEq(s.handle, 0) for s in sync_ops])),
body=BusyWait(Or(*[CondEq(atomic_load(s.handle), 0) for s in sync_ops])),
)

iet = List(body=(waitloop,) + iet.body)
Expand All @@ -53,12 +53,8 @@ def _make_waitlock(self, iet, sync_ops, *args):

def _make_releaselock(self, iet, sync_ops, *args):
pre = []
pre.append(BusyWait(Or(*[CondNe(s.handle, 2) for s in sync_ops])))
pre.append(ThreadFence('acquire'))
pre.extend(DummyExpr(s.handle, 0) for s in sync_ops)
# Hand the lock back before anything that follows can be observed --
# in particular before the request that asks the thread to refill it.
pre.append(ThreadFence('release'))
pre.append(BusyWait(Or(*[CondNe(atomic_load(s.handle), 2) for s in sync_ops])))
pre.extend(atomic_store(s.handle, 0) for s in sync_ops)

name = self.sregistry.make_name(prefix="release_lock")
parameters = derive_parameters(pre, ordering='canonical')
Expand Down Expand Up @@ -421,7 +417,7 @@ def _(layer, iet, sync_ops, lang, sregistry):
if lang._map_wait is not None:
body.append(lang._map_wait(qid))

body.extend([DummyExpr(s.handle, 1) for s in sync_ops])
body.extend([atomic_store(s.handle, 1) for s in sync_ops])
body.append(BlankLine)

name = f'copy_to_{layer.suffix}'
Expand All @@ -433,7 +429,7 @@ def _(layer, iet, sync_ops, lang, sregistry):
body.extend(list(iet.body))

body.append(BlankLine)
body.extend([DummyExpr(s.handle, 2) for s in sync_ops])
body.extend([atomic_store(s.handle, 2) for s in sync_ops])

return body, name

Expand All @@ -460,7 +456,7 @@ def _(layer, iet, sync_ops, lang, sregistry):
body = []
name = f'prefetch_to_{layer.suffix}'

body.extend([DummyExpr(s.handle, 2) for s in sync_ops])
body.extend([atomic_store(s.handle, 2) for s in sync_ops])

body = iet.body + (BlankLine,) + tuple(body)

Expand Down
Loading