add non-implicit ghosts below the ghost_in/ghost_out tasks that do the actual work.

src/cell.c

@@ -1754,6 +1754,7 @@ int cell_unskip_tasks(struct cell *c, struct scheduler *s) {
   if (c->extra_ghost != NULL) scheduler_activate(s, c->extra_ghost);
   if (c->ghost_in != NULL) scheduler_activate(s, c->ghost_in);
   if (c->ghost_out != NULL) scheduler_activate(s, c->ghost_out);
+  if (c->ghost != NULL) scheduler_activate(s, c->ghost);
   if (c->init_grav != NULL) scheduler_activate(s, c->init_grav);
   if (c->drift != NULL) scheduler_activate(s, c->drift);
   if (c->kick1 != NULL) scheduler_activate(s, c->kick1);

src/cell.h

@@ -154,6 +154,7 @@ struct cell {
   /*! The ghost tasks */
   struct task *ghost_in;
   struct task *ghost_out;
+  struct task *ghost;
 
   /*! The extra ghost task for complex hydro schemes */
   struct task *extra_ghost;

src/engine.c

@@ -119,6 +119,24 @@ void engine_addlink(struct engine *e, struct link **l, struct task *t) {
   res->next = atomic_swap(l, res);
 }
 
+/**
+ * @brief Recursively add non-implicit ghost tasks to a cell hierarchy.
+ */
+void engine_add_ghosts(struct engine *e, struct cell *c, struct task *ghost_in,
+                       struct task *ghost_out) {
+  if (!c->split || c->count < engine_max_parts_per_ghost) {
+    struct scheduler *s = &e->sched;
+    c->ghost =
+        scheduler_addtask(s, task_type_ghost, task_subtype_none, 0, 0, c, NULL);
+    scheduler_addunlock(s, ghost_in, c->ghost);
+    scheduler_addunlock(s, c->ghost, ghost_out);
+  } else {
+    for (int k = 0; k < 8; k++)
+      if (c->progeny[k] != NULL)
+        engine_add_ghosts(e, c->progeny[k], ghost_in, ghost_out);
+  }
+}
+
 /**
  * @brief Generate the hydro hierarchical tasks for a hierarchy of cells -
  * i.e. all the O(Npart) tasks.
@@ -185,10 +203,13 @@ void engine_make_hierarchical_tasks(struct engine *e, struct cell *c) {
 
       /* Generate the ghost tasks. */
       if (is_hydro) {
-        c->ghost_in = scheduler_addtask(s, task_type_ghost, task_subtype_none, 0,
-                                     /* implicit = */ 1, c, NULL);
-        c->ghost_out = scheduler_addtask(s, task_type_ghost, task_subtype_none, 0,
-                                     /* implicit = */ 1, c, NULL);
+        c->ghost_in =
+            scheduler_addtask(s, task_type_ghost, task_subtype_none, 0,
+                              /* implicit = */ 1, c, NULL);
+        c->ghost_out =
+            scheduler_addtask(s, task_type_ghost, task_subtype_none, 0,
+                              /* implicit = */ 1, c, NULL);
+        engine_add_ghosts(e, c, c->ghost_in, c->ghost_out);
       }
 
 #ifdef EXTRA_HYDRO_LOOP

src/engine.h

@@ -82,6 +82,7 @@ extern const char *engine_policy_names[];
 #define engine_redistribute_alloc_margin 1.2
 #define engine_default_energy_file_name "energy"
 #define engine_default_timesteps_file_name "timesteps"
+#define engine_max_parts_per_ghost 1000
 
 /* The rank of the engine as a global variable (for messages). */
 extern int engine_rank;

src/scheduler.c

@@ -757,8 +757,8 @@ void scheduler_splittasks(struct scheduler *s) {
  * @param cj The second cell to interact.
  */
 struct task *scheduler_addtask(struct scheduler *s, enum task_types type,
-                               enum task_subtypes subtype, int flags, int implicit,
-                               struct cell *ci, struct cell *cj) {
+                               enum task_subtypes subtype, int flags,
+                               int implicit, struct cell *ci, struct cell *cj) {
 
 #ifdef SWIFT_DEBUG_CHECKS
   if (ci == NULL && cj != NULL)

src/scheduler.h

@@ -133,8 +133,8 @@ void scheduler_reset(struct scheduler *s, int nr_tasks);
 void scheduler_ranktasks(struct scheduler *s);
 void scheduler_reweight(struct scheduler *s, int verbose);
 struct task *scheduler_addtask(struct scheduler *s, enum task_types type,
-                               enum task_subtypes subtype, int flags, int implicit,
-                               struct cell *ci, struct cell *cj);
+                               enum task_subtypes subtype, int flags,
+                               int implicit, struct cell *ci, struct cell *cj);
 void scheduler_splittasks(struct scheduler *s);
 struct task *scheduler_done(struct scheduler *s, struct task *t);
 struct task *scheduler_unlock(struct scheduler *s, struct task *t);

src/space.c

@@ -215,6 +215,7 @@ void space_rebuild_recycle_mapper(void *map_data, int num_elements,
     c->extra_ghost = NULL;
     c->ghost_in = NULL;
     c->ghost_out = NULL;
+    c->ghost = NULL;
     c->kick1 = NULL;
     c->kick2 = NULL;
     c->timestep = NULL;