Point-to-point transfer of serialized function batches between universe ranks. More...

#include <cloud.h>

Inheritance diagram for madness::BatchTransport:

[legend]

Collaboration diagram for madness::BatchTransport:

[legend]

Classes
struct	PendingRecv

Public Member Functions
	BatchTransport (World &universe, Cloud *cloud)

Future< batch_bytesT >	request (batch_keyT record)
	Future to the serialized bytes of `record`, fetched from its owner.

Public Member Functions inherited from madness::WorldObject< BatchTransport >
	WorldObject (const WorldObject &other)

	WorldObject (World &world)
	Constructor that associates an object (via the derived class) with a globally unique ID.

virtual	~WorldObject ()

WorldObject &	operator= (const WorldObject &)=delete

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const a6T &a6) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const a6T &a6, const a7T &a7) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const a6T &a6, const a7T &a7, const a8T &a8) const

detail::task_result_type< memfnT >::futureT	send (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const a6T &a6, const a7T &a7, const a8T &a8, const a9T &a9) const

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const a6T &a6, const a7T &a7, const a8T &a8, const a9T &a9, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1,a2,a3,a4,a5,a6,a7,a8,a9)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const a6T &a6, const a7T &a7, const a8T &a8, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1,a2,a3,a4,a5,a6,a7,a8)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const a6T &a6, const a7T &a7, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1,a2,a3,a4,a5,a6,a7)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const a6T &a6, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1,a2,a3,a4,a5,a6)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const a5T &a5, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1,a2,a3,a4,a5)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const a4T &a4, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1,a2,a3,a4)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const a3T &a3, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1,a2,a3)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const a2T &a2, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1,a2)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const a1T &a1, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)(a1)`.

detail::task_result_type< memfnT >::futureT	task (ProcessID dest, memfnT memfn, const TaskAttributes &attr=TaskAttributes()) const
	Sends task to derived class method `returnT (this->*memfn)()`.

Public Member Functions inherited from madness::WorldObjectBase
virtual	~WorldObjectBase ()

World &	get_world () const

World &	get_world_noexcept () const noexcept
	Get the world to which this object belongs, without risking a throw.

const uniqueidT &	id () const
	Returns the globally unique object ID.

bool	world_is_alive () const noexcept
	Reports whether the world to which this object belongs still exists.

Static Public Member Functions
static std::size_t	batch_chunk_bytes ()
	bytes per MPI message in a batch transfer; a larger payload is split into several

static void	set_batch_chunk_bytes (const std::size_t n)
	Set the chunk size, for tests only.

Static Public Attributes
static constexpr std::size_t	BATCH_NOT_FOUND = ~std::size_t(0)
	reply size meaning "the owner does not hold this record"; see on_trigger

static constexpr int	BATCH_TAG_BASE = 8192

static constexpr int	BATCH_TAG_CAP = 32767

Private Member Functions
int	alloc_tag ()

void	finish_recv (int tag)
	requester side, worker task: await the background-progressed Irecv and set the future

void	on_reply (int tag, std::size_t size, std::size_t chunk)
	requester side, comm thread: size the buffer, post the Irecvs, enqueue finish_recv

void	on_trigger (batch_keyT record, ProcessID requester, int tag)

void	reap_sends ()

Private Attributes
Cloud *	cloud_
	back-reference (not owned)

std::atomic< int >	next_tag_

std::map< int, std::shared_ptr< PendingRecv > >	pending_

std::mutex	pending_mtx_

std::list< SafeMPI::Request >	sends_

std::mutex	sends_mtx_

Static Private Attributes
static std::size_t	batch_chunk_bytes_ = std::size_t(1) << 30
	1 GiB

Additional Inherited Members
Public Types inherited from madness::WorldObject< BatchTransport >
typedef WorldObject< BatchTransport >	objT

Protected Member Functions inherited from madness::WorldObject< BatchTransport >
void	process_pending ()
	To be called from derived constructor to process pending messages.

Protected Member Functions inherited from madness::WorldObjectBase
	WorldObjectBase (const WorldObjectBase &other)
	Copy constructor; produces an unregistered object.

	WorldObjectBase (World &w)
	Construct a new WorldObjectBase.

World &	get_world_unchecked () const noexcept
	Unchecked access to the memoized world reference.

template<typename DerivedT >
void	register_self (DerivedT *this_ptr)
	Registers this object with its world, making it globally addressable.

Detailed Description

Point-to-point transfer of serialized function batches between universe ranks.

The bytes stream straight from the owner's local batch store to the requester by MPI point-to-point; they never ride inside an active-message payload, so there is no eager-buffer limit and no extra copy on the wire.

Both endpoints are posted from comm-thread AM handlers (WorldObject::send runs the member inline on the RMI receiver thread), never from worker tasks. That is what buys overlap under worker saturation: at a tight protocol every worker sits in the exchange kernel for a long time, so an endpoint posted as a task would queue behind it and the MPI op would not be posted until compute ended. On the comm thread the RMI loop's Testsome drives the rendezvous to completion during compute instead, leaving only the final await for the worker.

Wire protocol:

requester (worker): record a pending slot keyed by tag, send on_trigger
owner on_trigger (comm thread): Isend the local bytes, reply on_reply with the size
requester on_reply (comm thread): size the buffer, post the Irecv, enqueue finish_recv
requester finish_recv (worker): await the Irecv and set the future

The size travels in the reply of step 2 rather than a separate Isend so that step 3 can post the payload Irecv during compute; posting it at consume time would move the data transfer to post-compute and lose the overlap.

Constructor & Destructor Documentation

◆ BatchTransport()

madness::BatchTransport::BatchTransport	(	World &	universe,
		Cloud *	cloud
	)

inline

Parameters

[in]	universe	the world the cloud lives in (collective construction)
[in]	cloud	back-reference used to read owner-local batch bytes

References madness::WorldObject< BatchTransport >::process_pending().

Member Function Documentation

◆ alloc_tag()

int madness::BatchTransport::alloc_tag ( )

inlineprivate

References BATCH_TAG_BASE, BATCH_TAG_CAP, next_tag_, and madness::nonlinear_vector_solver().

Referenced by request().

◆ batch_chunk_bytes()

static std::size_t madness::BatchTransport::batch_chunk_bytes ( )

inlinestatic

bytes per MPI message in a batch transfer; a larger payload is split into several

MPI byte counts are int, and batches do exceed 2 GiB: a 161-orbital run at k=10 measured 3.03 GiB, whose count narrowed to a negative int. MPI rejects that and SafeMPI throws it on the comm thread, where nothing catches it.

References batch_chunk_bytes_.

Referenced by test_batch_fetch_chunked().

◆ finish_recv()

void madness::BatchTransport::finish_recv ( int tag )

inlineprivate

requester side, worker task: await the background-progressed Irecv and set the future

References madness::World::await(), madness::g, MADNESS_CHECK, p(), pending_, and pending_mtx_.

Referenced by on_reply().

◆ on_reply()

void madness::BatchTransport::on_reply	(	int	tag,
		std::size_t	size,
		std::size_t	chunk
	)

inlineprivate

requester side, comm thread: size the buffer, post the Irecvs, enqueue finish_recv

Parameters

chunk the chunking the owner used, echoed back so the receiver never has to infer it. See on_trigger.

References BATCH_NOT_FOUND, finish_recv(), madness::g, madness::WorldObjectBase::get_world(), MADNESS_CHECK, MPI_BYTE, madness::nonlinear_vector_solver(), p(), pending_, pending_mtx_, and u().

Referenced by on_trigger().

◆ on_trigger()

void madness::BatchTransport::on_trigger	(	batch_keyT	record,
		ProcessID	requester,
		int	tag
	)

inlineprivate

owner side, comm thread: Isend the record's bytes, then reply with the count. Must not throw: an exception here escapes every task-level handler.

References batch_chunk_bytes_, BATCH_NOT_FOUND, cloud_, madness::g, madness::WorldObjectBase::get_world(), MPI_BYTE, madness::nonlinear_vector_solver(), on_reply(), reap_sends(), madness::WorldObject< BatchTransport >::send(), sends_, sends_mtx_, madness::Cloud::try_get_local_batch_ptr(), and u().

Referenced by request().

◆ reap_sends()

void madness::BatchTransport::reap_sends ( )

inlineprivate

References madness::g, sends_, and sends_mtx_.

Referenced by on_trigger().

◆ request()

Future< batch_bytesT > madness::BatchTransport::request ( batch_keyT record )

inline

Future to the serialized bytes of record, fetched from its owner.

Resolves locally without MPI when this rank owns the record. The trigger is in flight on return, so the round trip overlaps work until the future is consumed.

References alloc_tag(), madness::Cloud::batch_owner(), cloud_, madness::g, madness::WorldObjectBase::get_world(), madness::nonlinear_vector_solver(), on_trigger(), p(), pending_, pending_mtx_, madness::WorldObject< BatchTransport >::send(), madness::Cloud::try_get_local_batch_bytes(), and u().

◆ set_batch_chunk_bytes()

static void madness::BatchTransport::set_batch_chunk_bytes ( const std::size_t n )

inlinestatic

Set the chunk size, for tests only.

Collective and not thread-safe: call it on every rank before any transfer. It exists so a unit test can reach the multi-chunk path with an affordable payload.

References batch_chunk_bytes_, MADNESS_CHECK_THROW, and madness::nonlinear_vector_solver().

Referenced by test_batch_fetch_chunked().

Member Data Documentation

◆ batch_chunk_bytes_

std::size_t madness::BatchTransport::batch_chunk_bytes_ = std::size_t(1) << 30

inlinestaticprivate

1 GiB

Referenced by batch_chunk_bytes(), on_trigger(), and set_batch_chunk_bytes().

◆ BATCH_NOT_FOUND

constexpr std::size_t madness::BatchTransport::BATCH_NOT_FOUND = ~std::size_t(0)

staticconstexpr

reply size meaning "the owner does not hold this record"; see on_trigger

Referenced by on_reply(), and on_trigger().

◆ BATCH_TAG_BASE

constexpr int madness::BatchTransport::BATCH_TAG_BASE = 8192

staticconstexpr

tags live in the range MADNESS does not manage (safempi.h); 32767 is the conservative MPI_TAG_UB floor

Referenced by alloc_tag().

◆ BATCH_TAG_CAP

constexpr int madness::BatchTransport::BATCH_TAG_CAP = 32767

staticconstexpr

Referenced by alloc_tag().

◆ cloud_

Cloud* madness::BatchTransport::cloud_

private

back-reference (not owned)

Referenced by on_trigger(), and request().

◆ next_tag_

std::atomic<int> madness::BatchTransport::next_tag_

private

Referenced by alloc_tag().

◆ pending_

std::map<int, std::shared_ptr<PendingRecv> > madness::BatchTransport::pending_

private

Referenced by finish_recv(), on_reply(), and request().

◆ pending_mtx_

std::mutex madness::BatchTransport::pending_mtx_

private

Referenced by finish_recv(), on_reply(), and request().

◆ sends_

std::list<SafeMPI::Request> madness::BatchTransport::sends_

private

Referenced by on_trigger(), and reap_sends().

◆ sends_mtx_

std::mutex madness::BatchTransport::sends_mtx_

private

owner-side in-flight Isends, reaped lazily. Their buffers live in the cloud's batch container and stay valid for the duration, so an un-reaped Isend is harmless.

Referenced by on_trigger(), and reap_sends().

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

cloud.h

Classes

Public Member Functions

Static Public Member Functions

Static Public Attributes

Private Member Functions

Private Attributes

Static Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ BatchTransport()

Member Function Documentation

◆ alloc_tag()

◆ batch_chunk_bytes()

◆ finish_recv()

◆ on_reply()

◆ on_trigger()

◆ reap_sends()

◆ request()

◆ set_batch_chunk_bytes()

Member Data Documentation

◆ batch_chunk_bytes_

◆ BATCH_NOT_FOUND

◆ BATCH_TAG_BASE

◆ BATCH_TAG_CAP

◆ cloud_

◆ next_tag_

◆ pending_

◆ pending_mtx_

◆ sends_

◆ sends_mtx_