ReceiveBuffer.h

Go to the documentation of this file.

//* This file is part of the MOOSE framework
//* https://mooseframework.inl.gov
//*
//* All rights reserved, see COPYRIGHT for full restrictions
//* https://github.com/idaholab/moose/blob/master/COPYRIGHT
//*
//* Licensed under LGPL 2.1, please see LICENSE for details
//* https://www.gnu.org/licenses/lgpl-2.1.html
 
#pragma once
 
// Moose Includes
#include "LIFOBuffer.h"
#include "SharedPool.h"
 
// Local includes
#include "ParallelStudyMethod.h"
 
// libMesh Includes
#include "libmesh/parallel.h"
#include "libmesh/parallel_object.h"
 
// System Includes
#include <list>
#include <chrono>
 
template <typename Object, typename Context>
class ReceiveBuffer : public libMesh::ParallelObject
{
public:
  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;
 
  bool currentlyReceiving() const { return _requests.size(); }
 
  unsigned long int objectsReceived() const { return _objects_received; }
 
  unsigned long int buffersReceived() const { return _buffers_received; }
 
  unsigned long int numProbes() const { return _num_probes; }
 
  unsigned long int objectPoolCreated() const { return _object_buffer_pool.num_created(); }
 
  unsigned long int bufferPoolCreated() const { return _buffer_pool.num_created(); }
 
  void receive(const bool start_receives_only = false);
 
  void clear();
 
  Real receiveLoopTime() const { return std::chrono::duration<Real>(_receive_loop_time).count(); }
 
  Real cleanupRequestsTime() const
  {
    return std::chrono::duration<Real>(_cleanup_requests_time).count();
  }
 
  void cleanupRequests();
 
  MooseUtils::LIFOBuffer<std::shared_ptr<Object>> & buffer() { return _buffer; }
 
private:
  Context * const _context;
  MooseUtils::LIFOBuffer<std::shared_ptr<Object>> _buffer;
 
  const ParallelStudyMethod _method;
  const unsigned int _clicks_per_receive;
 
  std::list<std::pair<std::shared_ptr<Parallel::Request>,
                      std::shared_ptr<std::vector<std::shared_ptr<Object>>>>>
      _requests;
 
  const Parallel::MessageTag _object_buffer_tag;
 
  std::chrono::steady_clock::duration _receive_loop_time;
  std::chrono::steady_clock::duration _cleanup_requests_time;
 
  unsigned long int _objects_received;
  unsigned long int _buffers_received;
  unsigned long int _num_probes;
 
  MooseUtils::SharedPool<std::vector<std::shared_ptr<Object>>> _object_buffer_pool;
  MooseUtils::SharedPool<
      std::vector<typename Parallel::Packing<std::shared_ptr<Object>>::buffer_type>>
      _buffer_pool;
 
  template <typename C, typename OutputIter, typename T>
  inline void blocking_receive_packed_range(const Parallel::Communicator & comm,
                                            const processor_id_type src_processor_id,
                                            C * context,
                                            OutputIter out,
                                            const T * /* output_type */,
                                            Parallel::Request & req,
                                            Parallel::Status & stat,
                                            const Parallel::MessageTag & tag) const
  {
    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));
  }
};
 
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)
  : ParallelObject(comm),
    _context(context),
    _method(method),
    _clicks_per_receive(clicks_per_receive),
    _object_buffer_tag(object_buffer_tag)
{
}
 
template <typename Object, typename Context>
ReceiveBuffer<Object, Context>::~ReceiveBuffer()
{
  if (!_requests.empty())
    mooseError("Some requests not serviced!");
}
 
template <typename Object, typename Context>
void
ReceiveBuffer<Object, Context>::receive(const bool start_receives_only /* = false */)
{
  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();
}
 
template <typename Object, typename Context>
void
ReceiveBuffer<Object, Context>::clear()
{
  _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;
}
 
template <typename Object, typename Context>
void
ReceiveBuffer<Object, Context>::cleanupRequests()
{
  _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;
      });
}