Added locking and unlocking functions for the sparts in cells.

src/cell.c

@@ -81,6 +81,47 @@ int cell_getsize(struct cell *c) {
   return count;
 }
 
+/**
+ * @brief Pack the data of the given cell and all it's sub-cells.
+ *
+ * @param c The #cell.
+ * @param pc Pointer to an array of packed cells in which the
+ *      cells will be packed.
+ *
+ * @return The number of packed cells.
+ */
+int cell_pack(struct cell *c, struct pcell *pc) {
+
+#ifdef WITH_MPI
+
+  /* Start by packing the data of the current cell. */
+  pc->h_max = c->h_max;
+  pc->ti_end_min = c->ti_end_min;
+  pc->ti_end_max = c->ti_end_max;
+  pc->count = c->count;
+  pc->gcount = c->gcount;
+  pc->scount = c->scount;
+  c->tag = pc->tag = atomic_inc(&cell_next_tag) % cell_max_tag;
+
+  /* Fill in the progeny, depth-first recursion. */
+  int count = 1;
+  for (int k = 0; k < 8; k++)
+    if (c->progeny[k] != NULL) {
+      pc->progeny[k] = count;
+      count += cell_pack(c->progeny[k], &pc[count]);
+    } else
+      pc->progeny[k] = -1;
+
+  /* Return the number of packed cells used. */
+  c->pcell_size = count;
+  return count;
+
+#else
+  error("SWIFT was not compiled with MPI support.");
+  return 0;
+#endif
+}
+
 /**
  * @brief Unpack the data of a given cell and its sub-cells.
  *
@@ -100,6 +141,7 @@ int cell_unpack(struct pcell *pc, struct cell *c, struct space *s) {
   c->ti_end_max = pc->ti_end_max;
   c->count = pc->count;
   c->gcount = pc->gcount;
+  c->scount = pc->scount;
   c->tag = pc->tag;
 
   /* Number of new cells created. */
@@ -111,6 +153,7 @@ int cell_unpack(struct pcell *pc, struct cell *c, struct space *s) {
       struct cell *temp = space_getcell(s);
       temp->count = 0;
       temp->gcount = 0;
+      temp->scount = 0;
       temp->loc[0] = c->loc[0];
       temp->loc[1] = c->loc[1];
       temp->loc[2] = c->loc[2];
@@ -191,46 +234,6 @@ int cell_link_gparts(struct cell *c, struct gpart *gparts) {
   return c->gcount;
 }
 
-/**
- * @brief Pack the data of the given cell and all it's sub-cells.
- *
- * @param c The #cell.
- * @param pc Pointer to an array of packed cells in which the
- *      cells will be packed.
- *
- * @return The number of packed cells.
- */
-int cell_pack(struct cell *c, struct pcell *pc) {
-
-#ifdef WITH_MPI
-
-  /* Start by packing the data of the current cell. */
-  pc->h_max = c->h_max;
-  pc->ti_end_min = c->ti_end_min;
-  pc->ti_end_max = c->ti_end_max;
-  pc->count = c->count;
-  pc->gcount = c->gcount;
-  c->tag = pc->tag = atomic_inc(&cell_next_tag) % cell_max_tag;
-
-  /* Fill in the progeny, depth-first recursion. */
-  int count = 1;
-  for (int k = 0; k < 8; k++)
-    if (c->progeny[k] != NULL) {
-      pc->progeny[k] = count;
-      count += cell_pack(c->progeny[k], &pc[count]);
-    } else
-      pc->progeny[k] = -1;
-
-  /* Return the number of packed cells used. */
-  c->pcell_size = count;
-  return count;
-
-#else
-  error("SWIFT was not compiled with MPI support.");
-  return 0;
-#endif
-}
-
 /**
  * @brief Pack the time information of the given cell and all it's sub-cells.
  *
@@ -421,6 +424,70 @@ int cell_glocktree(struct cell *c) {
   }
 }
 
+/**
+ * @brief Lock a cell for access to its array of #spart and hold its parents.
+ *
+ * @param c The #cell.
+ * @return 0 on success, 1 on failure
+ */
+int cell_slocktree(struct cell *c) {
+
+  TIMER_TIC
+
+  /* First of all, try to lock this cell. */
+  if (c->shold || lock_trylock(&c->slock) != 0) {
+    TIMER_TOC(timer_locktree);
+    return 1;
+  }
+
+  /* Did somebody hold this cell in the meantime? */
+  if (c->shold) {
+
+    /* Unlock this cell. */
+    if (lock_unlock(&c->slock) != 0) error("Failed to unlock cell.");
+
+    /* Admit defeat. */
+    TIMER_TOC(timer_locktree);
+    return 1;
+  }
+
+  /* Climb up the tree and lock/hold/unlock. */
+  struct cell *finger;
+  for (finger = c->parent; finger != NULL; finger = finger->parent) {
+
+    /* Lock this cell. */
+    if (lock_trylock(&finger->slock) != 0) break;
+
+    /* Increment the hold. */
+    atomic_inc(&finger->shold);
+
+    /* Unlock the cell. */
+    if (lock_unlock(&finger->slock) != 0) error("Failed to unlock cell.");
+  }
+
+  /* If we reached the top of the tree, we're done. */
+  if (finger == NULL) {
+    TIMER_TOC(timer_locktree);
+    return 0;
+  }
+
+  /* Otherwise, we hit a snag. */
+  else {
+
+    /* Undo the holds up to finger. */
+    for (struct cell *finger2 = c->parent; finger2 != finger;
+         finger2 = finger2->parent)
+      atomic_dec(&finger2->shold);
+
+    /* Unlock this cell. */
+    if (lock_unlock(&c->slock) != 0) error("Failed to unlock cell.");
+
+    /* Admit defeat. */
+    TIMER_TOC(timer_locktree);
+    return 1;
+  }
+}
+
 /**
  * @brief Unlock a cell's parents for access to #part array.
  *
@@ -459,6 +526,25 @@ void cell_gunlocktree(struct cell *c) {
   TIMER_TOC(timer_locktree);
 }
 
+/**
+ * @brief Unlock a cell's parents for access to #spart array.
+ *
+ * @param c The #cell.
+ */
+void cell_sunlocktree(struct cell *c) {
+
+  TIMER_TIC
+
+  /* First of all, try to unlock this cell. */
+  if (lock_unlock(&c->slock) != 0) error("Failed to unlock cell.");
+
+  /* Climb up the tree and unhold the parents. */
+  for (struct cell *finger = c->parent; finger != NULL; finger = finger->parent)
+    atomic_dec(&finger->shold);
+
+  TIMER_TOC(timer_locktree);
+}
+
 /**
  * @brief Sort the parts into eight bins along the given pivots.
  *

src/cell.h

@@ -76,7 +76,7 @@ struct pcell {
   int ti_end_min, ti_end_max;
 
   /* Number of particles in this cell. */
-  int count, gcount;
+  int count, gcount, scount;
 
   /* tag used for MPI communication. */
   int tag;
@@ -117,6 +117,9 @@ struct cell {
   /*! Pointer to the #gpart data. */
   struct gpart *gparts;
 
+  /*! Pointer to the #spart data. */
+  struct gpart *sparts;
+
   /*! Pointer for the sorted indices. */
   struct entry *sort;
 
@@ -229,6 +232,9 @@ struct cell {
   /*! Nr of #gpart in this cell. */
   int gcount;
 
+  /*! Nr of #spart in this cell. */
+  int scount;
+
   /*! The size of the sort array */
   int sortsize;
 
@@ -241,6 +247,9 @@ struct cell {
   /*! Spin lock for various uses (#gpart case). */
   swift_lock_type glock;
 
+  /*! Spin lock for various uses (#spart case). */
+  swift_lock_type slock;
+
   /*! ID of the previous owner, e.g. runner. */
   int owner;
 
@@ -250,6 +259,9 @@ struct cell {
   /*! Number of #gpart updated in this cell. */
   int g_updated;
 
+  /*! Number of #spart updated in this cell. */
+  int s_updated;
+
   /*! ID of the node this cell lives on. */
   int nodeID;
 
@@ -259,6 +271,9 @@ struct cell {
   /*! Is the #gpart data of this cell being used in a sub-cell? */
   int ghold;
 
+  /*! Is the #spart data of this cell being used in a sub-cell? */
+  int shold;
+
   /*! Number of tasks that are associated with this cell. */
   short int nr_tasks;
 
@@ -285,6 +300,8 @@ int cell_locktree(struct cell *c);
 void cell_unlocktree(struct cell *c);
 int cell_glocktree(struct cell *c);
 void cell_gunlocktree(struct cell *c);
+int cell_slocktree(struct cell *c);
+void cell_sunlocktree(struct cell *c);
 int cell_pack(struct cell *c, struct pcell *pc);
 int cell_unpack(struct pcell *pc, struct cell *c, struct space *s);
 int cell_pack_ti_ends(struct cell *c, int *ti_ends);