Draft: Implement a memcpy() clone that uses the threadpool

Using multiple threads to copy data can make better use of the available memory bandwidth when multiple NUMA regions are being used. This is an attempt to exploit that.

Experiments show that using it should be done only for quite large data, so a heuristic is in place to support that, which requires a number of 4k pages, at least 1 seems to be required, which makes sense, but more are required in reality (COSMA8 AMD 128 core, 8 NUMA regions). Surprisingly even then using all the threads is still not a guarantor of the best speed, so we also apply a limit to the number of threads that will be used to 25% of the total, using more than 1 gives a good improvement which tails off rapidly.

The actual gains here are modest, the largest come from particle splitting, but only when the buffers need reallocating and replication, which is only used for testing.

So the question is, is this worth it...

src/engine_redistribute.c

@@ -32,6 +32,7 @@
 /**
  * Do the exchange of one type of particles with all the other nodes.
  *
+ * @param e the current #engine
  * @param label a label for the memory allocations of this particle type.
  * @param counts 2D array with the counts of particles to exchange with
  *               each other node.
@@ -49,7 +50,8 @@
  * @result new particle data constructed from all the exchanges with the
  *         given alignment.
  */
-static void *engine_do_redistribute(const char *label, int *counts, char *parts,
+static void *engine_do_redistribute(struct engine *e, const char *label,
+                                    int *counts, char *parts,
                                     size_t new_nr_parts, size_t sizeofparts,
                                     size_t alignsize, MPI_Datatype mpi_type,
                                     int nr_nodes, int nodeID, int syncredist) {
@@ -89,9 +91,10 @@ static void *engine_do_redistribute(const char *label, int *counts, char *parts,
           /*  Just copy our own parts */
           if (counts[ind_send] > 0) {
             if (j == nodeID) {
-              memcpy(&parts_new[offset_recv * sizeofparts],
-                     &parts[offset_send * sizeofparts],
-                     sizeofparts * counts[ind_recv]);
+              threadpool_memcpy(&e->threadpool,
+                                &parts_new[offset_recv * sizeofparts],
+                                &parts[offset_send * sizeofparts],
+                                sizeofparts * counts[ind_recv]);
               offset_send += counts[ind_send];
               offset_recv += counts[ind_recv];
             } else {
@@ -177,8 +180,9 @@ static void *engine_do_redistribute(const char *label, int *counts, char *parts,
           if (k == nodeID) {
             int receiving = counts[ind_recv] - recvd;
             if (receiving > chunk) receiving = chunk;
-            memcpy(&parts_new[offset_recv * sizeofparts],
-                   &parts[offset_send * sizeofparts], sizeofparts * receiving);
+            threadpool_memcpy(
+                &e->threadpool, &parts_new[offset_recv * sizeofparts],
+                &parts[offset_send * sizeofparts], sizeofparts * receiving);
           } else {
             /* Otherwise send it. */
             int res =
@@ -1181,7 +1185,7 @@ void engine_redistribute(struct engine *e) {
 
   /* SPH particles. */
   void *new_parts = engine_do_redistribute(
-      "parts", counts, (char *)s->parts, nr_parts_new, sizeof(struct part),
+      e, "parts", counts, (char *)s->parts, nr_parts_new, sizeof(struct part),
       part_align, part_mpi_type, nr_nodes, nodeID, e->syncredist);
   swift_free("parts", s->parts);
   s->parts = (struct part *)new_parts;
@@ -1189,15 +1193,16 @@ void engine_redistribute(struct engine *e) {
   s->size_parts = engine_redistribute_alloc_margin * nr_parts_new;
 
   /* Extra SPH particle properties. */
-  new_parts = engine_do_redistribute(
-      "xparts", counts, (char *)s->xparts, nr_parts_new, sizeof(struct xpart),
-      xpart_align, xpart_mpi_type, nr_nodes, nodeID, e->syncredist);
+  new_parts =
+      engine_do_redistribute(e, "xparts", counts, (char *)s->xparts,
+                             nr_parts_new, sizeof(struct xpart), xpart_align,
+                             xpart_mpi_type, nr_nodes, nodeID, e->syncredist);
   swift_free("xparts", s->xparts);
   s->xparts = (struct xpart *)new_parts;
 
   /* Gravity particles. */
   new_parts =
-      engine_do_redistribute("gparts", g_counts, (char *)s->gparts,
+      engine_do_redistribute(e, "gparts", g_counts, (char *)s->gparts,
                              nr_gparts_new, sizeof(struct gpart), gpart_align,
                              gpart_mpi_type, nr_nodes, nodeID, e->syncredist);
   swift_free("gparts", s->gparts);
@@ -1207,7 +1212,7 @@ void engine_redistribute(struct engine *e) {
 
   /* Star particles. */
   new_parts =
-      engine_do_redistribute("sparts", s_counts, (char *)s->sparts,
+      engine_do_redistribute(e, "sparts", s_counts, (char *)s->sparts,
                              nr_sparts_new, sizeof(struct spart), spart_align,
                              spart_mpi_type, nr_nodes, nodeID, e->syncredist);
   swift_free("sparts", s->sparts);
@@ -1217,7 +1222,7 @@ void engine_redistribute(struct engine *e) {
 
   /* Black holes particles. */
   new_parts =
-      engine_do_redistribute("bparts", b_counts, (char *)s->bparts,
+      engine_do_redistribute(e, "bparts", b_counts, (char *)s->bparts,
                              nr_bparts_new, sizeof(struct bpart), bpart_align,
                              bpart_mpi_type, nr_nodes, nodeID, e->syncredist);
   swift_free("bparts", s->bparts);
@@ -1226,7 +1231,7 @@ void engine_redistribute(struct engine *e) {
   s->size_bparts = engine_redistribute_alloc_margin * nr_bparts_new;
 
   /* Sink particles. */
-  new_parts = engine_do_redistribute(
+  new_parts = engine_do_redistribute(e,
       "sinks", sink_counts, (char *)s->sinks, nr_sinks_new, sizeof(struct sink),
       sink_align, sink_mpi_type, nr_nodes, nodeID, e->syncredist);
   swift_free("sinks", s->sinks);