ReceiveBuffer< Object, Context > Class Template Reference

#include <ReceiveBuffer.h>

Inheritance diagram for ReceiveBuffer< Object, Context >:

Public Member Functions
	ReceiveBuffer (const libMesh::Parallel::Communicator &comm, Context *const context, const ParallelStudyMethod method, const unsigned int clicks_per_receive, const Parallel::MessageTag object_buffer_tag)
	~ReceiveBuffer () override
	Destructor: ensures that all send requests have completed. More...
bool	currentlyReceiving () const
	Whether or not there are messages that are being currently received. More...
unsigned long int	objectsReceived () const
	The work received since the last reset. More...
unsigned long int	buffersReceived () const
	The number of buffers received since the last reset. More...
unsigned long int	numProbes () const
	The total number of times we've polled for messages. More...
unsigned long int	objectPoolCreated () const
	Number of buffers created in the object buffer pool. More...
unsigned long int	bufferPoolCreated () const
	Number of buffers created in the buffer pool. More...
void	receive (const bool start_receives_only=false)
	Start receives for all currently available messages. More...
void	clear ()
	Clear all existing data. More...
Real	receiveLoopTime () const
	Amount of time (in seconds) spent in the loop checking for messages and creating Receives. More...
Real	cleanupRequestsTime () const
	Amount of time (in seconds) spent finishing receives and collecting objects. More...
void	cleanupRequests ()
	Checks to see if any Requests can be finished. More...
MooseUtils::LIFOBuffer< std::shared_ptr< Object > > &	buffer ()
	Gets the buffer that the received objects are filled into after the requests are finished. More...
const Parallel::Communicator &	comm () const
processor_id_type	n_processors () const
processor_id_type	processor_id () const

Protected Attributes
const Parallel::Communicator &	_communicator

Private Member Functions
template<typename C , typename OutputIter , typename T >
void	blocking_receive_packed_range (const Parallel::Communicator &comm, const processor_id_type src_processor_id, C *context, OutputIter out, const T *, Parallel::Request &req, Parallel::Status &stat, const Parallel::MessageTag &tag) const

Private Attributes
Context *const	_context
	The context. More...
MooseUtils::LIFOBuffer< std::shared_ptr< Object > >	_buffer
	The buffer that finished requests are filled into. More...
const ParallelStudyMethod	_method
	The method. More...
const unsigned int	_clicks_per_receive
	Iterations to wait before checking for new objects. More...
std::list< std::pair< std::shared_ptr< Parallel::Request >, std::shared_ptr< std::vector< std::shared_ptr< Object > > > > >	_requests
	List of Requests and buffers for each request. More...
const Parallel::MessageTag	_object_buffer_tag
	MessageTag for sending objects. More...
std::chrono::steady_clock::duration	_receive_loop_time
	Receive loop time. More...
std::chrono::steady_clock::duration	_cleanup_requests_time
	Time cleaning up requests. More...
unsigned long int	_objects_received
	Total objects received. More...
unsigned long int	_buffers_received
	Total object buffers received. More...
unsigned long int	_num_probes
	Total number of times we've polled for messages. More...
MooseUtils::SharedPool< std::vector< std::shared_ptr< Object > > >	_object_buffer_pool
	Shared pool of object buffers for incoming messages. More...
MooseUtils::SharedPool< std::vector< typename Parallel::Packing< std::shared_ptr< Object > >::buffer_type > >	_buffer_pool
	Shared pool of buffers. More...

Detailed Description

template<typename Object, typename Context>
class ReceiveBuffer< Object, Context >

Definition at line 28 of file ReceiveBuffer.h.

Constructor & Destructor Documentation

ReceiveBuffer()

template<typename Object , typename Context >

ReceiveBuffer< Object, Context >::ReceiveBuffer	(	const libMesh::Parallel::Communicator &	comm,
		Context *const	context,
		const ParallelStudyMethod	method,
		const unsigned int	clicks_per_receive,
		const Parallel::MessageTag	object_buffer_tag
	)

Definition at line 178 of file ReceiveBuffer.h.

  : ParallelObject(comm),
    _context(context),
    _method(method),
    _clicks_per_receive(clicks_per_receive),
    _object_buffer_tag(object_buffer_tag)
{
}

~ReceiveBuffer()

template<typename Object , typename Context >

ReceiveBuffer< Object, Context >::~ReceiveBuffer ( )

override

Destructor: ensures that all send requests have completed.

Definition at line 192 of file ReceiveBuffer.h.

{
  if (!_requests.empty())
    mooseError("Some requests not serviced!");
}

Member Function Documentation

blocking_receive_packed_range()

template<typename Object , typename Context >

template<typename C , typename OutputIter , typename T >

void ReceiveBuffer< Object, Context >::blocking_receive_packed_range ( const Parallel::Communicator &  comm, const processor_id_type  src_processor_id, C *  context, OutputIter  out, const T *  , Parallel::Request &  req, Parallel::Status &  stat, const Parallel::MessageTag &  tag  ) const

inlineprivate

Definition at line 146 of file ReceiveBuffer.h.

  {
    libmesh_experimental();

    typedef typename Parallel::Packing<T>::buffer_type buffer_t;

    // Receive serialized variable size objects as a sequence of
    // buffer_t.
    // Allocate a buffer on the heap so we don't have to free it until
    // after the Request::wait()
    std::vector<buffer_t> * buffer = new std::vector<buffer_t>(stat.size());
    comm.receive(src_processor_id, *buffer, tag);

    // Make the Request::wait() handle unpacking the buffer
    req.add_post_wait_work(
        new libMesh::Parallel::PostWaitUnpackBuffer<std::vector<buffer_t>, C, OutputIter, T>(
            *buffer, context, out));

    // Make the Request::wait() handle deleting the buffer
    req.add_post_wait_work(
        new libMesh::Parallel::PostWaitDeleteBuffer<std::vector<buffer_t>>(buffer));
  }

buffer()

template<typename Object , typename Context >

MooseUtils::LIFOBuffer<std::shared_ptr<Object> >& ReceiveBuffer< Object, Context >::buffer ( )

inline

Gets the buffer that the received objects are filled into after the requests are finished.

Definition at line 105 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::blocking_receive_packed_range().

{ return _buffer; }

bufferPoolCreated()

template<typename Object , typename Context >

unsigned long int ReceiveBuffer< Object, Context >::bufferPoolCreated ( ) const

inline

Number of buffers created in the buffer pool.

Definition at line 70 of file ReceiveBuffer.h.

{ return _buffer_pool.num_created(); }

buffersReceived()

template<typename Object , typename Context >

unsigned long int ReceiveBuffer< Object, Context >::buffersReceived ( ) const

inline

The number of buffers received since the last reset.

Definition at line 55 of file ReceiveBuffer.h.

{ return _buffers_received; }

cleanupRequests()

template<typename Object , typename Context >

void ReceiveBuffer< Object, Context >::cleanupRequests

(

)

Checks to see if any Requests can be finished.

Definition at line 281 of file ReceiveBuffer.h.

{
  _requests.remove_if(
      [&](std::pair<std::shared_ptr<Parallel::Request>,
                    std::shared_ptr<std::vector<std::shared_ptr<Object>>>> & request_pair)
      {
        auto req = request_pair.first;
        auto objects = request_pair.second;

        if (req->test()) // See if the receive has completed
        {
          req->wait(); // MUST call wait() to do post_wait_work which actually fills the object
                       // buffer

          _buffers_received++;
          _objects_received += objects->size();

          if (_buffer.capacity() < _buffer.size() + objects->size())
            _buffer.setCapacity(_buffer.size() + objects->size());

          for (auto & object : *objects)
            _buffer.move(object);

          objects->clear();

          return true;
        }
        else
          return false;
      });
}

cleanupRequestsTime()

template<typename Object , typename Context >

Real ReceiveBuffer< Object, Context >::cleanupRequestsTime ( ) const

inline

Amount of time (in seconds) spent finishing receives and collecting objects.

Definition at line 92 of file ReceiveBuffer.h.

  {
    return std::chrono::duration<Real>(_cleanup_requests_time).count();
  }

clear()

template<typename Object , typename Context >

void ReceiveBuffer< Object, Context >::clear

(

)

Clear all existing data.

Definition at line 267 of file ReceiveBuffer.h.

{
  _requests.clear();

  _receive_loop_time = std::chrono::steady_clock::duration::zero();
  _cleanup_requests_time = std::chrono::steady_clock::duration::zero();

  _objects_received = 0;
  _buffers_received = 0;
  _num_probes = 0;
}

currentlyReceiving()

template<typename Object , typename Context >

bool ReceiveBuffer< Object, Context >::currentlyReceiving ( ) const

inline

Whether or not there are messages that are being currently received.

Definition at line 45 of file ReceiveBuffer.h.

{ return _requests.size(); }

numProbes()

template<typename Object , typename Context >

unsigned long int ReceiveBuffer< Object, Context >::numProbes ( ) const

inline

The total number of times we've polled for messages.

Definition at line 60 of file ReceiveBuffer.h.

{ return _num_probes; }

objectPoolCreated()

template<typename Object , typename Context >

unsigned long int ReceiveBuffer< Object, Context >::objectPoolCreated ( ) const

inline

Number of buffers created in the object buffer pool.

Definition at line 65 of file ReceiveBuffer.h.

{ return _object_buffer_pool.num_created(); }

objectsReceived()

template<typename Object , typename Context >

unsigned long int ReceiveBuffer< Object, Context >::objectsReceived ( ) const

inline

The work received since the last reset.

Definition at line 50 of file ReceiveBuffer.h.

{ return _objects_received; }

receive()

template<typename Object , typename Context >

void ReceiveBuffer< Object, Context >::receive

(

const bool

start_receives_only = false

)

Start receives for all currently available messages.

Adds the to the working buffer

Definition at line 200 of file ReceiveBuffer.h.

{
  bool flag = false;
  Parallel::Status stat;

  static unsigned int current_clicks = 0;

  if (current_clicks % _clicks_per_receive == 0)
  {
    current_clicks = 0;

    // Receive and process a bunch of objects
    do
    {
      stat = _communicator.template packed_range_probe<std::shared_ptr<Object>>(
          Parallel::any_source, _object_buffer_tag, flag);

      _num_probes++;

      if (flag)
      {
        auto req = std::make_shared<Parallel::Request>();
        std::shared_ptr<std::vector<std::shared_ptr<Object>>> objects =
            _object_buffer_pool.acquire();

        // Make sure the buffer is clear - this shouldn't resize the storage though.
        objects->clear();

        if (_method == ParallelStudyMethod::HARM || _method == ParallelStudyMethod::BS)
          blocking_receive_packed_range(comm(),
                                        stat.source(),
                                        _context,
                                        std::back_inserter(*objects),
                                        (std::shared_ptr<Object> *)(libmesh_nullptr),
                                        *req,
                                        stat,
                                        _object_buffer_tag);
        else
        {
          std::shared_ptr<
              std::vector<typename Parallel::Packing<std::shared_ptr<Object>>::buffer_type>>
              buffer = _buffer_pool.acquire();

          _communicator.nonblocking_receive_packed_range(
              stat.source(),
              _context,
              std::back_inserter(*objects),
              (std::shared_ptr<Object> *)(libmesh_nullptr),
              *req,
              stat,
              buffer,
              _object_buffer_tag);
        }

        _requests.emplace_back(req, objects);
      }
    } while (flag);
  }

  current_clicks++;

  if (!start_receives_only)
    cleanupRequests();
}

receiveLoopTime()

template<typename Object , typename Context >

Real ReceiveBuffer< Object, Context >::receiveLoopTime ( ) const

inline

Amount of time (in seconds) spent in the loop checking for messages and creating Receives.

Definition at line 87 of file ReceiveBuffer.h.

{ return std::chrono::duration<Real>(_receive_loop_time).count(); }

Member Data Documentation

_buffer

template<typename Object , typename Context >

MooseUtils::LIFOBuffer<std::shared_ptr<Object> > ReceiveBuffer< Object, Context >::_buffer

private

The buffer that finished requests are filled into.

Definition at line 111 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::buffer().

_buffer_pool

template<typename Object , typename Context >

MooseUtils::SharedPool< std::vector<typename Parallel::Packing<std::shared_ptr<Object> >::buffer_type> > ReceiveBuffer< Object, Context >::_buffer_pool

private

Shared pool of buffers.

Definition at line 143 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::bufferPoolCreated().

_buffers_received

template<typename Object , typename Context >

unsigned long int ReceiveBuffer< Object, Context >::_buffers_received

private

Total object buffers received.

Definition at line 134 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::buffersReceived().

_cleanup_requests_time

template<typename Object , typename Context >

std::chrono::steady_clock::duration ReceiveBuffer< Object, Context >::_cleanup_requests_time

private

Time cleaning up requests.

Definition at line 129 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::cleanupRequestsTime().

_clicks_per_receive

template<typename Object , typename Context >

const unsigned int ReceiveBuffer< Object, Context >::_clicks_per_receive

private

Iterations to wait before checking for new objects.

Definition at line 116 of file ReceiveBuffer.h.

_context

template<typename Object , typename Context >

Context* const ReceiveBuffer< Object, Context >::_context

private

The context.

Definition at line 109 of file ReceiveBuffer.h.

_method

template<typename Object , typename Context >

const ParallelStudyMethod ReceiveBuffer< Object, Context >::_method

private

The method.

Definition at line 114 of file ReceiveBuffer.h.

_num_probes

template<typename Object , typename Context >

unsigned long int ReceiveBuffer< Object, Context >::_num_probes

private

Total number of times we've polled for messages.

Definition at line 136 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::numProbes().

_object_buffer_pool

template<typename Object , typename Context >

MooseUtils::SharedPool<std::vector<std::shared_ptr<Object> > > ReceiveBuffer< Object, Context >::_object_buffer_pool

private

Shared pool of object buffers for incoming messages.

Definition at line 139 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::objectPoolCreated().

_object_buffer_tag

template<typename Object , typename Context >

const Parallel::MessageTag ReceiveBuffer< Object, Context >::_object_buffer_tag

private

MessageTag for sending objects.

Definition at line 124 of file ReceiveBuffer.h.

_objects_received

template<typename Object , typename Context >

unsigned long int ReceiveBuffer< Object, Context >::_objects_received

private

Total objects received.

Definition at line 132 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::objectsReceived().

_receive_loop_time

template<typename Object , typename Context >

std::chrono::steady_clock::duration ReceiveBuffer< Object, Context >::_receive_loop_time

private

Receive loop time.

Definition at line 127 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::receiveLoopTime().

_requests

template<typename Object , typename Context >

std::list<std::pair<std::shared_ptr<Parallel::Request>, std::shared_ptr<std::vector<std::shared_ptr<Object> > > > > ReceiveBuffer< Object, Context >::_requests

private

List of Requests and buffers for each request.

Definition at line 121 of file ReceiveBuffer.h.

Referenced by ReceiveBuffer< Object, Context >::currentlyReceiving().

---

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

• ray_tracing/include/utils/ReceiveBuffer.h