diff --git a/src/core/MpiCallbacks.hpp b/src/core/MpiCallbacks.hpp
index 03f3a014b7..5ea39c0e55 100644
--- a/src/core/MpiCallbacks.hpp
+++ b/src/core/MpiCallbacks.hpp
@@ -31,28 +31,20 @@
  * e.g. to broadcast global variables or run an algorithm in parallel.
  *
  * Callbacks are registered on the head node as function pointers via
- * the @ref callback_macros "callback macros". The visitor pattern
- * allows using arbitrary function signatures. For non-void returning
- * callbacks, several return value policies are available: ignore return
- * value, return only one value (this is achieved using a boost optional
- * that is empty on all but one node), return the value of the head node,
- * or return a reduced value (by specifying the reduction operation).
+ * the @ref REGISTER_CALLBACK. The visitor pattern allows using arbitrary
+ * function signatures.
  */
 
 #include <utils/NumeratedContainer.hpp>
 #include <utils/tuple.hpp>
 #include <utils/type_traits.hpp>
 
-#include <boost/mpi/collectives/all_reduce.hpp>
 #include <boost/mpi/collectives/broadcast.hpp>
-#include <boost/mpi/collectives/reduce.hpp>
 #include <boost/mpi/communicator.hpp>
-#include <boost/optional.hpp>
+#include <boost/mpi/packed_iarchive.hpp>
 #include <boost/range/algorithm/remove_if.hpp>
 
 #include <cassert>
-#include <functional>
-#include <initializer_list>
 #include <memory>
 #include <tuple>
 #include <type_traits>
@@ -63,26 +55,6 @@ namespace Communication {
 
 class MpiCallbacks;
 
-/**
- * This namespace contains tag types
- * to indicate what to do with the return
- * values of callbacks.
- */
-namespace Result {
-/** %Ignore result */
-struct Ignore {};
-constexpr auto ignore = Ignore{};
-/** Return value from one rank */
-struct OneRank {};
-constexpr auto one_rank = OneRank{};
-/** Return value from the head node */
-struct MainRank {};
-constexpr auto main_rank = MainRank{};
-/** Reduce return value over all ranks */
-struct Reduction {};
-constexpr auto reduction = Reduction{};
-} // namespace Result
-
 namespace detail {
 /**
  * @brief Check if a type can be used as a callback argument.
@@ -153,8 +125,7 @@ struct callback_concept_t {
  * @brief Callback without a return value.
  *
  * This is an implementation of a callback for a specific callable
- * @p F and a set of arguments to call it with, where the there
- * is no return value.
+ * @p F and a set of arguments to call it with.
  */
 template <class F, class... Args>
 struct callback_void_t final : public callback_concept_t {
@@ -171,112 +142,6 @@ struct callback_void_t final : public callback_concept_t {
   }
 };
 
-/**
- * @brief Callback where the return value is ignored.
- *
- * This is an implementation of a callback for a specific callable
- * @p F and a set of arguments to call it with, where the valued from
- * all ranks are ignored.
- */
-template <class F, class... Args>
-struct callback_ignore_t final : public callback_concept_t {
-  F m_f;
-
-  callback_ignore_t(callback_ignore_t const &) = delete;
-  callback_ignore_t(callback_ignore_t &&) = delete;
-
-  template <class FRef>
-  explicit callback_ignore_t(FRef &&f) : m_f(std::forward<FRef>(f)) {}
-  void operator()(boost::mpi::communicator const &,
-                  boost::mpi::packed_iarchive &ia) const override {
-    detail::invoke<F, Args...>(m_f, ia);
-  }
-};
-
-/**
- * @brief Callback with a return value from one rank.
- *
- * This is an implementation of a callback for a specific callable
- * @p F and a set of arguments to call it with, where the value from
- * one rank is returned. Only one node is allowed to return a value.
- */
-template <class F, class... Args>
-struct callback_one_rank_t final : public callback_concept_t {
-  F m_f;
-
-  callback_one_rank_t(callback_one_rank_t const &) = delete;
-  callback_one_rank_t(callback_one_rank_t &&) = delete;
-
-  template <class FRef>
-  explicit callback_one_rank_t(FRef &&f) : m_f(std::forward<FRef>(f)) {}
-  void operator()(boost::mpi::communicator const &comm,
-                  boost::mpi::packed_iarchive &ia) const override {
-    auto const result = detail::invoke<F, Args...>(m_f, ia);
-
-    assert(1 == boost::mpi::all_reduce(comm, static_cast<int>(!!result),
-                                       std::plus<>()) &&
-           "Incorrect number of return values");
-
-    /* If this rank returned a result, send it to the head node. */
-    if (!!result) {
-      comm.send(0, 42, *result);
-    }
-  }
-};
-
-/**
- * @brief Callback with a return value from the head node.
- *
- * This is an implementation of a callback for a specific callable
- * @p F and a set of arguments to call it with, where the value from
- * the head node is returned.
- */
-template <class F, class... Args>
-struct callback_main_rank_t final : public callback_concept_t {
-  F m_f;
-
-  callback_main_rank_t(callback_main_rank_t const &) = delete;
-  callback_main_rank_t(callback_main_rank_t &&) = delete;
-
-  template <class FRef>
-  explicit callback_main_rank_t(FRef &&f) : m_f(std::forward<FRef>(f)) {}
-  void operator()(boost::mpi::communicator const &,
-                  boost::mpi::packed_iarchive &ia) const override {
-    std::ignore = detail::invoke<F, Args...>(m_f, ia);
-  }
-};
-
-/**
- * @brief Callback with return value reduction.
- *
- * This is an implementation of a callback for a specific callable
- * @p F and a set of arguments to call it with, where the return
- * value is reduced over the communicator.
- */
-template <class Op, class F, class... Args>
-struct callback_reduce_t final : public callback_concept_t {
-  Op m_op;
-  F m_f;
-
-  template <class OpRef, class FRef>
-  explicit callback_reduce_t(OpRef &&op, FRef &&f)
-      : m_op(std::forward<OpRef>(op)), m_f(std::forward<FRef>(f)) {}
-
-  /**
-   * @brief Execute the callback.
-   *
-   * Receive parameters for this callback, and then call it.
-   *
-   * @param comm The communicator to receive the parameters on.
-   */
-  void operator()(boost::mpi::communicator const &comm,
-                  boost::mpi::packed_iarchive &ia) const override {
-    /* Call the callback function, and reduce over the results with
-     * the stored reduction operation. */
-    boost::mpi::reduce(comm, detail::invoke<F, Args...>(m_f, ia), m_op, 0);
-  }
-};
-
 template <class F, class R, class... Args> struct FunctorTypes {
   using functor_type = F;
   using return_type = R;
@@ -293,7 +158,7 @@ using functor_types =
     decltype(functor_types_impl(&std::remove_reference_t<F>::operator()));
 
 template <class CRef, class C, class R, class... Args>
-auto make_model_impl(Result::Ignore, CRef &&c, FunctorTypes<C, R, Args...>) {
+auto make_model_impl(CRef &&c, FunctorTypes<C, R, Args...>) {
   return std::make_unique<callback_void_t<C, Args...>>(std::forward<CRef>(c));
 }
 
@@ -304,43 +169,15 @@ auto make_model_impl(Result::Ignore, CRef &&c, FunctorTypes<C, R, Args...>) {
  * to exist and can not be overloaded.
  */
 template <typename F> auto make_model(F &&f) {
-  return make_model_impl(Result::Ignore{}, std::forward<F>(f),
-                         functor_types<F>{});
+  return make_model_impl(std::forward<F>(f), functor_types<F>{});
 }
 
 /**
  * @brief Make a @ref callback_model_t for a function pointer.
- *
- * This instantiates an implementation of a callback for a function
- * pointer. The main task here is to transfer the signature from
- * the pointer to the callback_model_t by template argument type
- * deduction.
  */
 template <class... Args> auto make_model(void (*f_ptr)(Args...)) {
   return std::make_unique<callback_void_t<void (*)(Args...), Args...>>(f_ptr);
 }
-
-template <class Op, class R, class... Args>
-auto make_model(Result::Reduction, R (*f_ptr)(Args...), Op &&op) {
-  return std::make_unique<
-      callback_reduce_t<std::remove_reference_t<Op>, R (*)(Args...), Args...>>(
-      std::forward<Op>(op), f_ptr);
-}
-
-template <class R, class... Args>
-auto make_model(Result::Ignore, R (*f_ptr)(Args...)) {
-  return std::make_unique<callback_ignore_t<R (*)(Args...), Args...>>(f_ptr);
-}
-
-template <class R, class... Args>
-auto make_model(Result::OneRank, R (*f_ptr)(Args...)) {
-  return std::make_unique<callback_one_rank_t<R (*)(Args...), Args...>>(f_ptr);
-}
-
-template <class R, class... Args>
-auto make_model(Result::MainRank, R (*f_ptr)(Args...)) {
-  return std::make_unique<callback_main_rank_t<R (*)(Args...), Args...>>(f_ptr);
-}
 } // namespace detail
 
 /**
@@ -484,25 +321,6 @@ class MpiCallbacks {
                                     detail::make_model(fp));
   }
 
-  /**
-   * @brief Add a new callback with a return value.
-   *
-   * Add a new callback to the system. This is a collective
-   * function that must be run on all nodes.
-   * @p tag is one of the tag types from @ref Communication::Result,
-   * which indicates what to do with the return values.
-   *
-   * @param tag Tag type indicating return operation
-   * @param tag_args Argument for the return operation, if any.
-   * @param fp Pointer to the static callback function to add.
-   */
-  template <class Tag, class R, class... Args, class... TagArgs>
-  static void add_static(Tag tag, R (*fp)(Args...), TagArgs &&...tag_args) {
-    static_callbacks().emplace_back(
-        reinterpret_cast<void (*)()>(fp),
-        detail::make_model(tag, fp, std::forward<TagArgs>(tag_args)...));
-  }
-
 private:
   /**
    * @brief Remove callback.
@@ -566,8 +384,7 @@ class MpiCallbacks {
    */
   template <class... Args, class... ArgRef>
   auto call(void (*fp)(Args...), ArgRef &&...args) const ->
-      /* Enable only if fp can be called with the provided arguments,
-       * e.g. if fp(args...) is well-formed. */
+      /* enable only if fp can be called with the provided arguments */
       std::enable_if_t<std::is_void_v<decltype(fp(args...))>> {
     const int id = m_func_ptr_to_id.at(reinterpret_cast<void (*)()>(fp));
 
@@ -585,104 +402,13 @@ class MpiCallbacks {
    * @param args Arguments for the callback.
    */
   template <class... Args, class... ArgRef>
-  void call_all(void (*fp)(Args...), ArgRef &&...args) const {
+  auto call_all(void (*fp)(Args...), ArgRef &&...args) const ->
+      /* enable only if fp can be called with the provided arguments */
+      std::enable_if_t<std::is_void_v<decltype(fp(args...))>> {
     call(fp, args...);
     fp(args...);
   }
 
-  /**
-   * @brief Call a callback and reduce the result over all nodes.
-   *
-   * This calls a callback on all nodes, including the head node,
-   * and does a mpi reduction with the registered operation.
-   *
-   * This method can only be called on the head node.
-   */
-  template <class Op, class R, class... Args>
-  auto call(Result::Reduction, Op op, R (*fp)(Args...), Args... args) const
-      -> std::remove_reference_t<decltype(op(std::declval<R>(),
-                                             std::declval<R>()))> {
-    const int id = m_func_ptr_to_id.at(reinterpret_cast<void (*)()>(fp));
-
-    call(id, args...);
-
-    std::remove_cv_t<std::remove_reference_t<decltype(op(std::declval<R>(),
-                                                         std::declval<R>()))>>
-        result{};
-    boost::mpi::reduce(m_comm, fp(args...), result, op, 0);
-
-    return result;
-  }
-
-  /**
-   * @brief Call a callback and ignore the result over all nodes.
-   *
-   * This calls a callback on all nodes, including the head node,
-   * and ignore all return values.
-   *
-   * This method can only be called on the head node.
-   */
-  template <class R, class... Args, class... ArgRef>
-  auto call(Result::Ignore, R (*fp)(Args...), ArgRef... args) const
-      -> std::remove_reference_t<R> {
-
-    const int id = m_func_ptr_to_id.at(reinterpret_cast<void (*)()>(fp));
-    call(id, args...);
-
-    fp(std::forward<Args>(args)...);
-
-    return {};
-  }
-
-  /**
-   * @brief Call a callback and reduce the result over all nodes.
-   *
-   * This calls a callback on all nodes, including the head node,
-   * and returns the value from the node which returned an engaged
-   * optional. Only one node is allowed to return a value.
-   *
-   * This method can only be called on the head node.
-   */
-  template <class R, class... Args, class... ArgRef>
-  auto call(Result::OneRank, boost::optional<R> (*fp)(Args...),
-            ArgRef... args) const -> std::remove_reference_t<R> {
-
-    const int id = m_func_ptr_to_id.at(reinterpret_cast<void (*)()>(fp));
-    call(id, args...);
-
-    auto const local_result = fp(std::forward<Args>(args)...);
-
-    assert(1 == boost::mpi::all_reduce(m_comm, static_cast<int>(!!local_result),
-                                       std::plus<>()) &&
-           "Incorrect number of return values");
-
-    if (!!local_result) {
-      return *local_result;
-    }
-
-    std::remove_cv_t<std::remove_reference_t<R>> result;
-    m_comm.recv(boost::mpi::any_source, boost::mpi::any_tag, result);
-    return result;
-  }
-
-  /**
-   * @brief Call a callback and return the result of the head node.
-   *
-   * This calls a callback on all nodes, including the head node,
-   * and returns the value from the head node.
-   *
-   * This method can only be called on the head node.
-   */
-  template <class R, class... Args, class... ArgRef>
-  auto call(Result::MainRank, R (*fp)(Args...), ArgRef... args) const
-      -> std::remove_reference_t<R> {
-
-    const int id = m_func_ptr_to_id.at(reinterpret_cast<void (*)()>(fp));
-    call(id, args...);
-
-    return fp(std::forward<Args>(args)...);
-  }
-
   /**
    * @brief Start the MPI loop.
    *
@@ -771,20 +497,9 @@ class RegisterCallback {
   template <class... Args> explicit RegisterCallback(void (*cb)(Args...)) {
     MpiCallbacks::add_static(cb);
   }
-
-  template <class Tag, class R, class... Args, class... TagArgs>
-  explicit RegisterCallback(Tag tag, R (*cb)(Args...), TagArgs &&...tag_args) {
-    MpiCallbacks::add_static(tag, cb, std::forward<TagArgs>(tag_args)...);
-  }
 };
 } /* namespace Communication */
 
-/**
- * @name Callback macros
- * @anchor callback_macros
- */
-/**@{*/
-
 /**
  * @brief Register a static callback without return value.
  *
@@ -798,70 +513,4 @@ class RegisterCallback {
   static ::Communication::RegisterCallback register_##cb(&(cb));               \
   }
 
-/**
- * @brief Register a static callback whose return value is reduced.
- *
- * This registers a function as an mpi callback with
- * reduction of the return values from the nodes.
- * The macro should be used at global scope.
- *
- * @param cb A function
- * @param op Reduction operation
- */
-#define REGISTER_CALLBACK_REDUCTION(cb, op)                                    \
-  namespace Communication {                                                    \
-  static ::Communication::RegisterCallback                                     \
-      register_reduction_##cb(::Communication::Result::Reduction{}, &(cb),     \
-                              (op));                                           \
-  }
-
-/**
- * @brief Register a static callback whose return value is to be ignored.
- *
- * This registers a function as an mpi callback with
- * ignored return values.
- * The macro should be used at global scope.
- *
- * @param cb A function
- */
-#define REGISTER_CALLBACK_IGNORE(cb)                                           \
-  namespace Communication {                                                    \
-  static ::Communication::RegisterCallback                                     \
-      register_ignore_##cb(::Communication::Result::Ignore{}, &(cb));          \
-  }
-
-/**
- * @brief Register a static callback which returns a value on only one node.
- *
- * This registers a function as an mpi callback with
- * reduction of the return values from one node
- * where the value of the optional is set.
- * The macro should be used at global scope.
- *
- * @param cb A function
- */
-#define REGISTER_CALLBACK_ONE_RANK(cb)                                         \
-  namespace Communication {                                                    \
-  static ::Communication::RegisterCallback                                     \
-      register_one_rank_##cb(::Communication::Result::OneRank{}, &(cb));       \
-  }
-
-/**
- * @brief Register a static callback whose return value is ignored except on
- * the head node.
- *
- * This registers a function as an mpi callback with
- * reduction of the return values from the head node.
- * The macro should be used at global scope.
- *
- * @param cb A function
- */
-#define REGISTER_CALLBACK_MAIN_RANK(cb)                                        \
-  namespace Communication {                                                    \
-  static ::Communication::RegisterCallback                                     \
-      register_main_rank_##cb(::Communication::Result::MainRank{}, &(cb));     \
-  }
-
-/**@}*/
-
 #endif
diff --git a/src/core/communication.hpp b/src/core/communication.hpp
index 2f7d149259..a7dbcf4280 100644
--- a/src/core/communication.hpp
+++ b/src/core/communication.hpp
@@ -47,6 +47,7 @@
 #include "MpiCallbacks.hpp"
 
 #include <boost/mpi/communicator.hpp>
+#include <boost/mpi/environment.hpp>
 
 #include <memory>
 #include <utility>
@@ -97,38 +98,6 @@ void mpi_call_all(void (*fp)(Args...), ArgRef &&...args) {
   Communication::mpiCallbacks().call_all(fp, std::forward<ArgRef>(args)...);
 }
 
-/** @brief Call a local function.
- *  @tparam Tag    Any tag type defined in @ref Communication::Result
- *  @tparam R      Return type of the local function
- *  @tparam Args   Local function argument types
- *  @tparam ArgRef Local function argument types
- *  @param tag     Reduction strategy
- *  @param fp      Local function
- *  @param args    Local function arguments
- */
-template <class Tag, class R, class... Args, class... ArgRef>
-auto mpi_call(Tag tag, R (*fp)(Args...), ArgRef &&...args) {
-  return Communication::mpiCallbacks().call(tag, fp,
-                                            std::forward<ArgRef>(args)...);
-}
-
-/** @brief Call a local function.
- *  @tparam Tag    Any tag type defined in @ref Communication::Result
- *  @tparam TagArg Types of arguments to @p Tag
- *  @tparam R      Return type of the local function
- *  @tparam Args   Local function argument types
- *  @tparam ArgRef Local function argument types
- *  @param tag     Reduction strategy
- *  @param tag_arg Arguments to the reduction strategy
- *  @param fp      Local function
- *  @param args    Local function arguments
- */
-template <class Tag, class TagArg, class R, class... Args, class... ArgRef>
-auto mpi_call(Tag tag, TagArg &&tag_arg, R (*fp)(Args...), ArgRef &&...args) {
-  return Communication::mpiCallbacks().call(tag, std::forward<TagArg>(tag_arg),
-                                            fp, std::forward<ArgRef>(args)...);
-}
-
 /** Process requests from head node. Worker nodes main loop. */
 void mpi_loop();
 
diff --git a/src/core/observables/CylindricalDensityProfile.hpp b/src/core/observables/CylindricalDensityProfile.hpp
index 01e387e6ba..4b0c18f787 100644
--- a/src/core/observables/CylindricalDensityProfile.hpp
+++ b/src/core/observables/CylindricalDensityProfile.hpp
@@ -23,6 +23,7 @@
 
 #include "BoxGeometry.hpp"
 #include "grid.hpp"
+#include "utils_histogram.hpp"
 
 #include <utils/Histogram.hpp>
 #include <utils/math/coordinate_transformation.hpp>
@@ -53,9 +54,8 @@ class CylindricalDensityProfile : public CylindricalPidProfileObservable {
               transform_params->orientation()));
     }
 
-    std::vector<decltype(local_folded_positions)> global_folded_positions{};
-    boost::mpi::gather(comm, local_folded_positions, global_folded_positions,
-                       0);
+    auto const global_folded_positions =
+        detail::gather(comm, local_folded_positions);
 
     if (comm.rank() != 0) {
       return {};
diff --git a/src/core/observables/CylindricalFluxDensityProfile.hpp b/src/core/observables/CylindricalFluxDensityProfile.hpp
index 5796705a60..9e27a6c339 100644
--- a/src/core/observables/CylindricalFluxDensityProfile.hpp
+++ b/src/core/observables/CylindricalFluxDensityProfile.hpp
@@ -60,21 +60,15 @@ class CylindricalFluxDensityProfile : public CylindricalPidProfileObservable {
               traits.velocity(p), transform_params->axis(), pos));
     }
 
-    auto const world_size = comm.size();
-    std::vector<decltype(local_folded_positions)> global_folded_positions{};
-    std::vector<decltype(local_velocities)> global_velocities{};
-    global_folded_positions.reserve(world_size);
-    global_velocities.reserve(world_size);
-    boost::mpi::gather(comm, local_folded_positions, global_folded_positions,
-                       0);
-    boost::mpi::gather(comm, local_velocities, global_velocities, 0);
+    auto const [global_folded_positions, global_velocities] =
+        detail::gather(comm, local_folded_positions, local_velocities);
 
     if (comm.rank() != 0) {
       return {};
     }
 
     Utils::CylindricalHistogram<double, 3> histogram(n_bins(), limits());
-    accumulate(histogram, global_folded_positions, global_velocities);
+    detail::accumulate(histogram, global_folded_positions, global_velocities);
     histogram.normalize();
     return histogram.get_histogram();
   }
diff --git a/src/core/observables/CylindricalLBFluxDensityProfileAtParticlePositions.cpp b/src/core/observables/CylindricalLBFluxDensityProfileAtParticlePositions.cpp
index d881e82dcb..a836063099 100644
--- a/src/core/observables/CylindricalLBFluxDensityProfileAtParticlePositions.cpp
+++ b/src/core/observables/CylindricalLBFluxDensityProfileAtParticlePositions.cpp
@@ -61,17 +61,15 @@ CylindricalLBFluxDensityProfileAtParticlePositions::evaluate(
     local_flux_densities.emplace_back(flux_cyl);
   }
 
-  std::vector<decltype(local_folded_positions)> global_folded_positions{};
-  std::vector<decltype(local_flux_densities)> global_flux_densities{};
-  boost::mpi::gather(comm, local_folded_positions, global_folded_positions, 0);
-  boost::mpi::gather(comm, local_flux_densities, global_flux_densities, 0);
+  auto const [global_folded_positions, global_flux_densities] =
+      detail::gather(comm, local_folded_positions, local_flux_densities);
 
   if (comm.rank() != 0) {
     return {};
   }
 
   Utils::CylindricalHistogram<double, 3> histogram(n_bins(), limits());
-  accumulate(histogram, global_folded_positions, global_flux_densities);
-  return normalize_by_bin_size(histogram);
+  detail::accumulate(histogram, global_folded_positions, global_flux_densities);
+  return detail::normalize_by_bin_size(histogram);
 }
 } // namespace Observables
diff --git a/src/core/observables/CylindricalLBVelocityProfile.cpp b/src/core/observables/CylindricalLBVelocityProfile.cpp
index 2f0ae68ce5..6a12d4285e 100644
--- a/src/core/observables/CylindricalLBVelocityProfile.cpp
+++ b/src/core/observables/CylindricalLBVelocityProfile.cpp
@@ -25,9 +25,6 @@
 #include <utils/Histogram.hpp>
 #include <utils/math/coordinate_transformation.hpp>
 
-#include <boost/mpi/collectives/gather.hpp>
-#include <boost/serialization/vector.hpp>
-
 #include <vector>
 
 namespace Observables {
@@ -55,21 +52,16 @@ std::vector<double> CylindricalLBVelocityProfile::operator()(
     }
   }
 
-  auto const world_size = comm.size();
-  std::vector<decltype(sampling_positions)> global_positions{};
-  std::vector<decltype(local_velocities)> global_velocities{};
-  global_positions.reserve(world_size);
-  global_velocities.reserve(world_size);
-  boost::mpi::gather(comm, local_positions, global_positions, 0);
-  boost::mpi::gather(comm, local_velocities, global_velocities, 0);
+  auto const [global_positions, global_velocities] =
+      detail::gather(comm, local_positions, local_velocities);
 
   if (comm.rank() != 0) {
     return {};
   }
 
   Utils::CylindricalHistogram<double, 3> histogram(n_bins(), limits());
-  accumulate(histogram, global_positions, global_velocities);
-  return normalize_by_bin_size(histogram);
+  detail::accumulate(histogram, global_positions, global_velocities);
+  return detail::normalize_by_bin_size(histogram);
 }
 
 } // namespace Observables
diff --git a/src/core/observables/CylindricalLBVelocityProfileAtParticlePositions.cpp b/src/core/observables/CylindricalLBVelocityProfileAtParticlePositions.cpp
index a139561a16..f498b168f0 100644
--- a/src/core/observables/CylindricalLBVelocityProfileAtParticlePositions.cpp
+++ b/src/core/observables/CylindricalLBVelocityProfileAtParticlePositions.cpp
@@ -26,9 +26,6 @@
 #include <utils/Histogram.hpp>
 #include <utils/math/coordinate_transformation.hpp>
 
-#include <boost/mpi/collectives/gather.hpp>
-#include <boost/serialization/vector.hpp>
-
 #include <utility>
 #include <vector>
 
@@ -59,21 +56,16 @@ std::vector<double> CylindricalLBVelocityProfileAtParticlePositions::evaluate(
     local_velocities.emplace_back(vel_cyl);
   }
 
-  auto const world_size = comm.size();
-  std::vector<decltype(local_folded_positions)> global_folded_positions{};
-  std::vector<decltype(local_velocities)> global_velocities{};
-  global_folded_positions.reserve(world_size);
-  global_velocities.reserve(world_size);
-  boost::mpi::gather(comm, local_folded_positions, global_folded_positions, 0);
-  boost::mpi::gather(comm, local_velocities, global_velocities, 0);
+  auto const [global_folded_positions, global_velocities] =
+      detail::gather(comm, local_folded_positions, local_velocities);
 
   if (comm.rank() != 0) {
     return {};
   }
 
   Utils::CylindricalHistogram<double, 3> histogram(n_bins(), limits());
-  accumulate(histogram, global_folded_positions, global_velocities);
-  return normalize_by_bin_size(histogram);
+  detail::accumulate(histogram, global_folded_positions, global_velocities);
+  return detail::normalize_by_bin_size(histogram);
 }
 
 } // namespace Observables
diff --git a/src/core/observables/CylindricalVelocityProfile.hpp b/src/core/observables/CylindricalVelocityProfile.hpp
index ab54d4610d..394df7cb03 100644
--- a/src/core/observables/CylindricalVelocityProfile.hpp
+++ b/src/core/observables/CylindricalVelocityProfile.hpp
@@ -28,7 +28,6 @@
 #include <utils/math/coordinate_transformation.hpp>
 
 #include <cstddef>
-#include <utility>
 #include <vector>
 
 namespace Observables {
@@ -59,22 +58,16 @@ class CylindricalVelocityProfile : public CylindricalPidProfileObservable {
               traits.velocity(p), transform_params->axis(), pos));
     }
 
-    auto const world_size = comm.size();
-    std::vector<decltype(local_folded_positions)> global_folded_positions{};
-    std::vector<decltype(local_velocities)> global_velocities{};
-    global_folded_positions.reserve(world_size);
-    global_velocities.reserve(world_size);
-    boost::mpi::gather(comm, local_folded_positions, global_folded_positions,
-                       0);
-    boost::mpi::gather(comm, local_velocities, global_velocities, 0);
+    auto const [global_folded_positions, global_velocities] =
+        detail::gather(comm, local_folded_positions, local_velocities);
 
     if (comm.rank() != 0) {
       return {};
     }
 
     Utils::CylindricalHistogram<double, 3> histogram(n_bins(), limits());
-    accumulate(histogram, global_folded_positions, global_velocities);
-    return normalize_by_bin_size(histogram);
+    detail::accumulate(histogram, global_folded_positions, global_velocities);
+    return detail::normalize_by_bin_size(histogram);
   }
 
   std::vector<std::size_t> shape() const override {
diff --git a/src/core/observables/DensityProfile.hpp b/src/core/observables/DensityProfile.hpp
index 325d38908c..4ebad035dc 100644
--- a/src/core/observables/DensityProfile.hpp
+++ b/src/core/observables/DensityProfile.hpp
@@ -23,6 +23,7 @@
 #include "Particle.hpp"
 #include "PidProfileObservable.hpp"
 #include "grid.hpp"
+#include "utils_histogram.hpp"
 
 #include <utils/Histogram.hpp>
 
@@ -49,11 +50,8 @@ class DensityProfile : public PidProfileObservable {
           folded_position(traits.position(p), box_geo));
     }
 
-    auto const world_size = comm.size();
-    std::vector<decltype(local_folded_positions)> global_folded_positions{};
-    global_folded_positions.reserve(world_size);
-    boost::mpi::gather(comm, local_folded_positions, global_folded_positions,
-                       0);
+    auto const global_folded_positions =
+        detail::gather(comm, local_folded_positions);
 
     if (comm.rank() != 0) {
       return {};
diff --git a/src/core/observables/FluxDensityProfile.hpp b/src/core/observables/FluxDensityProfile.hpp
index 1304ce263e..1887c8d8c6 100644
--- a/src/core/observables/FluxDensityProfile.hpp
+++ b/src/core/observables/FluxDensityProfile.hpp
@@ -27,8 +27,6 @@
 
 #include <utils/Histogram.hpp>
 
-#include <boost/range/combine.hpp>
-
 #include <cstddef>
 #include <utility>
 #include <vector>
@@ -60,21 +58,15 @@ class FluxDensityProfile : public PidProfileObservable {
       local_velocities.emplace_back(traits.velocity(p));
     }
 
-    auto const world_size = comm.size();
-    std::vector<decltype(local_folded_positions)> global_folded_positions{};
-    std::vector<decltype(local_velocities)> global_velocities{};
-    global_folded_positions.reserve(world_size);
-    global_velocities.reserve(world_size);
-    boost::mpi::gather(comm, local_folded_positions, global_folded_positions,
-                       0);
-    boost::mpi::gather(comm, local_velocities, global_velocities, 0);
+    auto const [global_folded_positions, global_velocities] =
+        detail::gather(comm, local_folded_positions, local_velocities);
 
     if (comm.rank() != 0) {
       return {};
     }
 
     Utils::Histogram<double, 3> histogram(n_bins(), limits());
-    accumulate(histogram, global_folded_positions, global_velocities);
+    detail::accumulate(histogram, global_folded_positions, global_velocities);
     histogram.normalize();
     return histogram.get_histogram();
   }
diff --git a/src/core/observables/ForceDensityProfile.hpp b/src/core/observables/ForceDensityProfile.hpp
index 3bdda403ca..bd3ded0f03 100644
--- a/src/core/observables/ForceDensityProfile.hpp
+++ b/src/core/observables/ForceDensityProfile.hpp
@@ -62,21 +62,15 @@ class ForceDensityProfile : public PidProfileObservable {
       local_forces.emplace_back(traits.force(p));
     }
 
-    auto const world_size = comm.size();
-    std::vector<decltype(local_folded_positions)> global_folded_positions{};
-    std::vector<decltype(local_forces)> global_forces{};
-    global_folded_positions.reserve(world_size);
-    global_forces.reserve(world_size);
-    boost::mpi::gather(comm, local_folded_positions, global_folded_positions,
-                       0);
-    boost::mpi::gather(comm, local_forces, global_forces, 0);
+    auto const [global_folded_positions, global_forces] =
+        detail::gather(comm, local_folded_positions, local_forces);
 
     if (comm.rank() != 0) {
       return {};
     }
 
     Utils::Histogram<double, 3> histogram(n_bins(), limits());
-    accumulate(histogram, global_folded_positions, global_forces);
+    detail::accumulate(histogram, global_folded_positions, global_forces);
     histogram.normalize();
     return histogram.get_histogram();
   }
diff --git a/src/core/observables/LBVelocityProfile.cpp b/src/core/observables/LBVelocityProfile.cpp
index c15c6c6310..706d824afe 100644
--- a/src/core/observables/LBVelocityProfile.cpp
+++ b/src/core/observables/LBVelocityProfile.cpp
@@ -24,9 +24,6 @@
 #include <utils/Histogram.hpp>
 #include <utils/Vector.hpp>
 
-#include <boost/mpi/collectives/gather.hpp>
-#include <boost/serialization/vector.hpp>
-
 #include <stdexcept>
 #include <vector>
 
@@ -48,23 +45,18 @@ LBVelocityProfile::operator()(boost::mpi::communicator const &comm) const {
     }
   }
 
-  auto const world_size = comm.size();
-  std::vector<decltype(sampling_positions)> global_positions{};
-  std::vector<decltype(local_velocities)> global_velocities{};
-  global_positions.reserve(world_size);
-  global_velocities.reserve(world_size);
-  boost::mpi::gather(comm, local_positions, global_positions, 0);
-  boost::mpi::gather(comm, local_velocities, global_velocities, 0);
+  auto const [global_positions, global_velocities] =
+      detail::gather(comm, local_positions, local_velocities);
 
   if (comm.rank() != 0) {
     return {};
   }
 
   Utils::Histogram<double, 3> histogram(n_bins(), limits());
-  accumulate(histogram, global_positions, global_velocities);
+  detail::accumulate(histogram, global_positions, global_velocities);
   try {
-    return normalize_by_bin_size(histogram, allow_empty_bins);
-  } catch (empty_bin_exception const &) {
+    return detail::normalize_by_bin_size(histogram, allow_empty_bins);
+  } catch (detail::empty_bin_exception const &) {
     throw std::runtime_error(
         "Decrease sampling delta(s), some bins have no hit");
   }
diff --git a/src/core/observables/PidObservable.hpp b/src/core/observables/PidObservable.hpp
index ee2e0ba3f0..1f61f9eb97 100644
--- a/src/core/observables/PidObservable.hpp
+++ b/src/core/observables/PidObservable.hpp
@@ -31,9 +31,9 @@
 
 #include <boost/mpi/collectives/gather.hpp>
 #include <boost/mpi/collectives/reduce.hpp>
-
 #include <boost/range/algorithm/copy.hpp>
 #include <boost/serialization/utility.hpp>
+#include <boost/serialization/vector.hpp>
 
 #include <cassert>
 #include <cstddef>
diff --git a/src/core/observables/utils_histogram.hpp b/src/core/observables/utils_histogram.hpp
index 7f4bbc003e..591423e2c3 100644
--- a/src/core/observables/utils_histogram.hpp
+++ b/src/core/observables/utils_histogram.hpp
@@ -21,10 +21,40 @@
 
 #include <utils/Histogram.hpp>
 
+#include <boost/mpi/collectives/gather.hpp>
+#include <boost/mpi/communicator.hpp>
+#include <boost/serialization/vector.hpp>
+
 #include <cstddef>
+#include <utility>
 #include <vector>
 
-namespace Observables {
+namespace Observables::detail {
+
+/** @brief Gather data from all MPI ranks. */
+template <class Pos>
+auto gather(boost::mpi::communicator const &comm,
+            std::vector<Pos> const &local_pos) {
+  std::vector<std::vector<Pos>> global_pos{};
+  global_pos.reserve(comm.size());
+  boost::mpi::gather(comm, local_pos, global_pos, 0);
+  return global_pos;
+}
+
+/** @brief Gather data from all MPI ranks. */
+template <class Pos, class Val>
+auto gather(boost::mpi::communicator const &comm,
+            std::vector<Pos> const &local_pos,
+            std::vector<Val> const &local_val) {
+  auto const world_size = comm.size();
+  std::vector<std::vector<Pos>> global_pos{};
+  global_pos.reserve(world_size);
+  boost::mpi::gather(comm, local_pos, global_pos, 0);
+  std::vector<std::vector<Val>> global_val{};
+  global_val.reserve(world_size);
+  boost::mpi::gather(comm, local_val, global_val, 0);
+  return std::make_pair(global_pos, global_val);
+}
 
 /** @brief Accumulate histogram data gathered from multiple MPI ranks. */
 template <class T, std::size_t N, std::size_t M, class U, class Pos, class Val>
@@ -58,4 +88,4 @@ auto normalize_by_bin_size(Utils::Histogram<T, N, M, U> &histogram,
   return hist_data;
 }
 
-} // Namespace Observables
+} // Namespace Observables::detail
diff --git a/src/core/particle_node.cpp b/src/core/particle_node.cpp
index 8c8d46d37a..ef66de66b8 100644
--- a/src/core/particle_node.cpp
+++ b/src/core/particle_node.cpp
@@ -36,6 +36,7 @@
 #include <utils/mpi/gatherv.hpp>
 
 #include <boost/mpi/collectives/all_gather.hpp>
+#include <boost/mpi/collectives/all_reduce.hpp>
 #include <boost/mpi/collectives/gather.hpp>
 #include <boost/mpi/collectives/reduce.hpp>
 #include <boost/mpi/collectives/scatter.hpp>
@@ -155,17 +156,18 @@ Utils::Cache<int, Particle> particle_fetch_cache(max_cache_size);
 void invalidate_fetch_cache() { particle_fetch_cache.invalidate(); }
 std::size_t fetch_cache_max_size() { return particle_fetch_cache.max_size(); }
 
-static boost::optional<const Particle &> get_particle_data_local(int p_id) {
-  auto p = cell_structure.get_local_particle(p_id);
-
-  if (p and (not p->is_ghost())) {
-    return *p;
+static void mpi_send_particle_data_local(int p_id) {
+  auto const p = cell_structure.get_local_particle(p_id);
+  auto const found = p and not p->is_ghost();
+  assert(1 == boost::mpi::all_reduce(::comm_cart, static_cast<int>(found),
+                                     std::plus<>()) &&
+         "Particle not found");
+  if (found) {
+    ::comm_cart.send(0, 42, *p);
   }
-
-  return {};
 }
 
-REGISTER_CALLBACK_ONE_RANK(get_particle_data_local)
+REGISTER_CALLBACK(mpi_send_particle_data_local)
 
 const Particle &get_particle_data(int p_id) {
   auto const pnode = get_particle_node(p_id);
@@ -183,10 +185,10 @@ const Particle &get_particle_data(int p_id) {
 
   /* Cache miss, fetch the particle,
    * put it into the cache and return a pointer into the cache. */
-  auto const cache_ptr = particle_fetch_cache.put(
-      p_id, Communication::mpiCallbacks().call(Communication::Result::one_rank,
-                                               get_particle_data_local, p_id));
-  return *cache_ptr;
+  Communication::mpiCallbacks().call_all(mpi_send_particle_data_local, p_id);
+  Particle result{};
+  ::comm_cart.recv(boost::mpi::any_source, boost::mpi::any_tag, result);
+  return *(particle_fetch_cache.put(p_id, std::move(result)));
 }
 
 static void mpi_get_particles_local() {
diff --git a/src/core/unit_tests/MpiCallbacks_test.cpp b/src/core/unit_tests/MpiCallbacks_test.cpp
index 3afbe3f2a6..27884057eb 100644
--- a/src/core/unit_tests/MpiCallbacks_test.cpp
+++ b/src/core/unit_tests/MpiCallbacks_test.cpp
@@ -175,38 +175,20 @@ BOOST_AUTO_TEST_CASE(CallbackHandle) {
   }
 }
 
-BOOST_AUTO_TEST_CASE(reduce_callback) {
-  auto cb = []() -> int { return boost::mpi::communicator().rank(); };
-  Communication::MpiCallbacks::add_static(Communication::Result::Reduction{},
-                                          static_cast<int (*)()>(cb),
-                                          std::plus<int>());
-
-  boost::mpi::communicator world;
-  Communication::MpiCallbacks cbs(world);
-
-  if (0 == world.rank()) {
-    auto const ret = cbs.call(Communication::Result::reduction,
-                              std::plus<int>(), static_cast<int (*)()>(cb));
-    auto const n = world.size();
-    BOOST_CHECK_EQUAL(ret, (n * (n - 1)) / 2);
-  } else {
-    cbs.loop();
-  }
-}
-
-BOOST_AUTO_TEST_CASE(ignore_callback) {
+BOOST_AUTO_TEST_CASE(call) {
   called = false;
-  auto cb = []() -> int { return called = true, -1; };
+  auto cb = []() { called = true; };
 
-  auto const fp = static_cast<int (*)()>(cb);
+  auto const fp = static_cast<void (*)()>(cb);
 
-  Communication::MpiCallbacks::add_static(Communication::Result::ignore, fp);
+  Communication::MpiCallbacks::add_static(fp);
 
   boost::mpi::communicator world;
   Communication::MpiCallbacks cbs(world);
 
   if (0 == world.rank()) {
-    cbs.call(Communication::Result::ignore, fp);
+    cbs.call(fp);
+    fp();
   } else {
     cbs.loop();
   }
@@ -214,56 +196,6 @@ BOOST_AUTO_TEST_CASE(ignore_callback) {
   BOOST_CHECK(called);
 }
 
-BOOST_AUTO_TEST_CASE(one_rank_callback) {
-  auto cb = []() -> boost::optional<int> {
-    boost::mpi::communicator world;
-    if (world.rank() == (world.size() - 1)) {
-      return world.rank();
-    }
-
-    return {};
-  };
-
-  auto const fp = static_cast<boost::optional<int> (*)()>(cb);
-
-  Communication::MpiCallbacks::add_static(Communication::Result::one_rank, fp);
-
-  boost::mpi::communicator world;
-  Communication::MpiCallbacks cbs(world);
-
-  if (0 == world.rank()) {
-    BOOST_CHECK_EQUAL(cbs.call(Communication::Result::one_rank, fp),
-                      world.size() - 1);
-  } else {
-    cbs.loop();
-  }
-}
-
-BOOST_AUTO_TEST_CASE(main_rank_callback) {
-  auto cb = []() -> int {
-    boost::mpi::communicator world;
-    if (world.rank() == 0) {
-      return world.size();
-    }
-
-    return -1;
-  };
-
-  auto const fp = static_cast<int (*)()>(cb);
-
-  Communication::MpiCallbacks::add_static(Communication::Result::main_rank, fp);
-
-  boost::mpi::communicator world;
-  Communication::MpiCallbacks cbs(world);
-
-  if (0 == world.rank()) {
-    BOOST_CHECK_EQUAL(cbs.call(Communication::Result::main_rank, fp),
-                      world.size());
-  } else {
-    cbs.loop();
-  }
-}
-
 BOOST_AUTO_TEST_CASE(call_all) {
   called = false;
   auto cb = []() { called = true; };