MEDIUM: stats-file/counters: store and preload stats counters as shm file objects

This is the last patch of the shm stats file series, in this patch we
implement the logic to store and fetch shm stats objects and associate
them to existing shared counters on the current process.

Shm objects are stored in the same memory location as the shm stats file
header. In fact they are stored right after it. All objects (struct
shm_stats_file_object) have the same size (no matter their type), which
allows for easy object traversal without having to check the object's
type, and could permit the use of external tools to scan the SHM in the
future. Each object stores a guid (of GUID_MAX_LEN+1 size) and tgid
which allows to match corresponding shared counters indexes. Also,
as stated before, each object stores the list of users making use of
it. Objects are never released (the map can only grow), but unused
objects (when no more users or active users are found in objects->users),
the object is automatically recycled. Also, each object stores its
type which defines how the object generic data member should be handled.

Upon startup (or reload), haproxy first tries to scan existing shm to
find objects that could be associated to frontends, backends, listeners
or servers in the current config based on GUID. For associations that
couldn't be made, haproxy will automatically create missing objects in
the SHM during late startup. When haproxy matches with an existing object,
it means the counter from an older process is preserved in the new
process, so multiple processes temporarily share the same counter for as
long as required for older processes to eventually exit.

include/haproxy/stats-file-t.h

@@ -43,11 +43,28 @@ struct shm_stats_file_hdr {
 		pid_t pid;
 		int heartbeat; // last activity of this process + heartbeat timeout, in ticks
 	} slots[64];
+	int objects; /* actual number of objects stored in the shm */
+	int objects_slots; /* total available objects slots unless map is resized */
 };
 
+#define SHM_STATS_FILE_OBJECT_TYPE_FE 0x0
+#define SHM_STATS_FILE_OBJECT_TYPE_BE 0x1
+
 struct shm_stats_file_object {
+	char guid[GUID_MAX_LEN + 1];
+	uint8_t tgid; // thread group ID from 1 to 64
+	uint8_t type; // SHM_STATS_FILE_OBJECT_TYPE_* to know how to handle object.data
+	uint64_t users; // bitfield that corresponds to users of the object (see shm_stats_file_hdr slots)
+	/* as the struct may hold any of the types described here, let's make it
+	 * so it may store up to the heaviest one using an union
+	 */
+	union {
+		struct fe_counters_shared_tg fe;
+		struct be_counters_shared_tg be;
+	} data;
 };
 
 #define SHM_STATS_FILE_MAPPING_SIZE(obj) (sizeof(struct shm_stats_file_hdr) + (obj) * sizeof(struct shm_stats_file_object))
+#define SHM_STATS_FILE_OBJECT(mem, it) (struct shm_stats_file_object *)((char *)mem + sizeof(struct shm_stats_file_hdr) + (it) * sizeof(struct shm_stats_file_object))
 
 #endif /* _HAPROXY_STATS_FILE_T_H */

include/haproxy/stats-file.h

@@ -10,6 +10,7 @@
 
 extern struct shm_stats_file_hdr *shm_stats_file_hdr;
 extern int shm_stats_file_max_objects;
+extern int shm_stats_file_slot;
 
 int stats_dump_fields_file(struct buffer *out,
                            const struct field *stats, size_t stats_count,
@@ -25,5 +26,6 @@ void stats_dump_file_header(int type, struct buffer *out);
 
 void apply_stats_file(void);
 int shm_stats_file_prepare(void);
+struct shm_stats_file_object *shm_stats_file_add_object(char **errmsg);
 
 #endif /* _HAPROXY_STATS_FILE_H */

src/counters.c

@@ -19,10 +19,13 @@
  */
 
 #include <stdlib.h>
+#include <stdio.h>
 #include <haproxy/atomic.h>
 #include <haproxy/clock.h>
 #include <haproxy/counters.h>
 #include <haproxy/global.h>
+#include <haproxy/guid.h>
+#include <haproxy/stats-file.h>
 #include <haproxy/time.h>
 #include <haproxy/tools.h>
 
@@ -34,9 +37,20 @@ static void _counters_shared_drop(void *counters)
 	if (!shared)
 		return;
 
-	/* memory was allocated using calloc(), simply free it */
-	while (it < global.nbtgroups) {
-		free(shared->tg[it]);
+	while (it < global.nbtgroups && shared->tg[it]) {
+		if (shared->flags & COUNTERS_SHARED_F_LOCAL) {
+			/* memory was allocated using calloc(), simply free it */
+			free(shared->tg[it]);
+		}
+		else {
+			struct shm_stats_file_object *obj;
+
+			/* inside shared memory, retrieve associated object and remove
+			 * ourselves from its users
+			 */
+			obj = container_of(shared->tg[it], struct shm_stats_file_object, data);
+			HA_ATOMIC_OR(&obj->users, (1 << shm_stats_file_slot));
+		}
 		it += 1;
 	}
 }
@@ -53,26 +67,55 @@ void counters_be_shared_drop(struct be_counters_shared *counters)
 	_counters_shared_drop(counters);
 }
 
-/* prepare shared counters pointer for a given <guid> object
- * <size> hint is expected to reflect the actual tg member size (fe/be)
+/* prepare shared counters pointers for a given <shared> parent
+ * pointer and for <guid> object
+ * <is_be> hint is expected to be set to 1 when the guid refers to be_shared
+ * struct, else fe_shared stuct is assumed.
+ *
  * if <guid> is not set, then sharing is disabled
  * Returns the pointer on success or NULL on failure, in which case
- * <errmsg> will contain additional hints about the error and must be freed accordingly
+ * <errmsg> may contain additional hints about the error and must be freed accordingly
  */
 static int _counters_shared_prepare(struct counters_shared *shared,
-                                    const struct guid_node *guid, size_t size, char **errmsg)
+                                    const struct guid_node *guid, int is_be, char **errmsg)
 {
+	struct fe_counters_shared *fe_shared;
+	struct be_counters_shared *be_shared;
 	int it = 0;
 
-	/* no shared memory for now, simply allocate a memory block
-	 * for the counters (zero-initialized), ignore guid
-	 */
-	if (!guid->node.key)
+	if (!guid->node.key || !shm_stats_file_hdr)
 		shared->flags |= COUNTERS_SHARED_F_LOCAL;
+
 	while (it < global.nbtgroups) {
-		shared->tg[it] = calloc(1, size);
+		if (shared->flags & COUNTERS_SHARED_F_LOCAL) {
+			size_t tg_size;
+
+			tg_size = (is_be) ? sizeof(*be_shared->tg[0]) : sizeof(*fe_shared->tg[0]);
+			shared->tg[it] = calloc(1, tg_size);
+			if (!shared->tg[it])
+				memprintf(errmsg, "memory error, calloc failed");
+		}
+		else if (!shared->tg[it]) {
+			struct shm_stats_file_object *shm_obj;
+
+			shm_obj = shm_stats_file_add_object(errmsg);
+			if (shm_obj) {
+				snprintf(shm_obj->guid, sizeof(shm_obj->guid)- 1, "%s", guid_get(guid));
+				if (is_be) {
+					shm_obj->type = SHM_STATS_FILE_OBJECT_TYPE_BE;
+					be_shared = (struct be_counters_shared *)shared;
+					be_shared->tg[it] = &shm_obj->data.be;
+				}
+				else {
+					shm_obj->type = SHM_STATS_FILE_OBJECT_TYPE_FE;
+					fe_shared = (struct fe_counters_shared *)shared;
+					fe_shared->tg[it] = &shm_obj->data.fe;
+				}
+				/* we use atomic op to make the object visible by setting valid tgid value */
+				HA_ATOMIC_STORE(&shm_obj->tgid, it + 1);
+			}
+		}
 		if (!shared->tg[it]) {
-			memprintf(errmsg, "memory error, calloc failed");
 			_counters_shared_drop(shared);
 			return 0;
 		}
@@ -89,11 +132,11 @@ static int _counters_shared_prepare(struct counters_shared *shared,
 /* prepare shared fe counters pointer for a given <guid> object */
 int counters_fe_shared_prepare(struct fe_counters_shared *shared, const struct guid_node *guid, char **errmsg)
 {
-	return _counters_shared_prepare((struct counters_shared *)shared, guid, sizeof(struct fe_counters_shared_tg), errmsg);
+	return _counters_shared_prepare((struct counters_shared *)shared, guid, 0, errmsg);
 }
 
 /* prepare shared be counters pointer for a given <guid> object */
 int counters_be_shared_prepare(struct be_counters_shared *shared, const struct guid_node *guid, char **errmsg)
 {
-	return _counters_shared_prepare((struct counters_shared *)shared, guid, sizeof(struct be_counters_shared_tg), errmsg);
+	return _counters_shared_prepare((struct counters_shared *)shared, guid, 1, errmsg);
 }

src/stats-file.c

@@ -473,6 +473,16 @@ static inline int shm_hb_is_stale(int hb)
 	return (hb == TICK_ETERNITY || tick_is_expired(hb, now_ms));
 }
 
+/* returns 1 if the slot <id> is free in <hdr>, else 0
+ */
+static int shm_stats_file_slot_isfree(struct shm_stats_file_hdr *hdr, int id)
+{
+	int hb;
+
+	hb = HA_ATOMIC_LOAD(&hdr->slots[id].heartbeat);
+	return shm_hb_is_stale(hb);
+}
+
 /* returns free slot id on success or -1 if no more slots are available
  * on success, the free slot is already reserved for the process pid
  */
@@ -524,6 +534,147 @@ static int shm_file_grow(unsigned int bytes)
 	return 1;
 }
 
+/* returns NULL if no free object or pointer to existing object if
+ * object can be reused
+ */
+static struct shm_stats_file_object *shm_stats_file_reuse_object(void)
+{
+	int it = 0;
+	int objects;
+	struct shm_stats_file_object *free_obj;
+
+	BUG_ON(!shm_stats_file_hdr);
+	objects = HA_ATOMIC_LOAD(&shm_stats_file_hdr->objects);
+	if (!objects)
+		return NULL;
+	while (it < objects) {
+		uint64_t users;
+		int free = 0;
+
+		free_obj = SHM_STATS_FILE_OBJECT(shm_stats_file_hdr, it);
+		users = HA_ATOMIC_LOAD(&free_obj->users);
+		if (!users)
+			free = 1; // no doubt, no user using this object
+		else {
+			int slot = 0;
+
+			/* if one or multiple users crashed or forgot to remove their bit
+			 * from obj->users but aren't making use of it anymore, we can detect
+			 * it by checking if the process related to "used" users slot are still
+			 * effectively active
+			 */
+			free = 1; // consider all users are inactive for now
+
+			while (slot < sizeof(shm_stats_file_hdr->slots) / sizeof(shm_stats_file_hdr->slots[0])) {
+				if ((users & (1 << slot)) &&
+				    !shm_stats_file_slot_isfree(shm_stats_file_hdr, slot)) {
+					/* user still alive, so supposedly making use of it */
+					free = 0;
+					break;
+				}
+				slot++;
+			}
+		}
+		if (free) {
+			uint64_t nusers = (1ULL << shm_stats_file_slot);
+
+			/* we use CAS here because we want to make sure that we are the only
+			 * process who exclusively owns the object as we are about to reset it.
+			 * In case of failure, we also don't expect our bit to be set, so
+			 * CAS is the best fit here. First we set the obj's users bits to 0
+			 * to make sure no other process will try to preload it (it may hold
+			 * garbage content) as we are about to reset it with our data, then
+			 * we do another CAS to confirm we are the owner of the object
+			 */
+			if (HA_ATOMIC_CAS(&free_obj->users, &users, 0)) {
+				/* we set obj tgid to 0 so it can't be looked up in
+				 * shm_stats_file_preload (tgid 0 is invalid)
+				 */
+				HA_ATOMIC_STORE(&free_obj->tgid, 0);
+
+				/* now we finally try to acquire the object */
+				users = 0;
+				if (HA_ATOMIC_CAS(&free_obj->users, &users, nusers))
+					return free_obj;
+			}
+			/* failed to CAS because of concurrent access, give up on this one */
+		}
+		it += 1;
+	}
+	return NULL;
+}
+
+/* returns pointer to new object in case of success and NULL in case
+ * of failure (if adding the maximum number of objects is already
+ * reached)
+ *
+ * <errmsg> will be set in case of failure to give more hints about the
+ * error, it must be freed accordingly
+ */
+struct shm_stats_file_object *shm_stats_file_add_object(char **errmsg)
+{
+	struct shm_stats_file_object *new_obj;
+	uint64_t expected_users;
+	int objects, objects_slots;
+
+	new_obj = shm_stats_file_reuse_object();
+	if (new_obj) {
+		return new_obj;
+	}
+
+ add:
+	objects = HA_ATOMIC_LOAD(&shm_stats_file_hdr->objects);
+
+	if (objects >= shm_stats_file_max_objects) {
+		memprintf(errmsg, "Cannot add additionnal object to '%s' file, maximum number already reached (%d). "
+		           "Adjust \"shm-stats-file-max-objects\" directive if needed.",
+		           global.shm_stats_file, shm_stats_file_max_objects / global.nbtgroups);
+		return NULL;
+	}
+
+	objects_slots = HA_ATOMIC_LOAD(&shm_stats_file_hdr->objects_slots);
+	/* we increase objects slots by following half power of two curve to
+	 * reduce waste while ensuring we don't grow the shm file (costly)
+	 * too often
+	 */
+	if (objects + 1 > objects_slots) {
+		int nobjects_slots;
+
+		if (objects_slots < 2)
+			nobjects_slots = objects_slots + 1;
+		else if ((objects_slots & (objects_slots - 1)) == 0)
+			nobjects_slots = objects_slots + objects_slots / 2;
+		else
+			nobjects_slots = (objects_slots & (objects_slots - 1)) * 2;
+
+		if (shm_file_grow((nobjects_slots - objects_slots) * sizeof(struct shm_stats_file_object)) == 0) {
+			memprintf(errmsg, "Error when trying to increase shm stats file size for '%s': %s",
+			          global.shm_stats_file, strerror(errno));
+			return NULL;
+		}
+		HA_ATOMIC_ADD(&shm_stats_file_hdr->objects_slots, nobjects_slots - objects_slots);
+	}
+
+	/* try to use this new slot */
+	new_obj = SHM_STATS_FILE_OBJECT(shm_stats_file_hdr, objects);
+	memset(new_obj, 0, sizeof(*new_obj)); // ensure object is reset before using it
+
+	if (HA_ATOMIC_FETCH_ADD(&shm_stats_file_hdr->objects, 1) != objects) {
+		/* a concurrent shm_stats_file_add_object stole our slot, retry */
+		__ha_cpu_relax();
+		goto add;
+	}
+
+	expected_users = 0;
+	if (!HA_ATOMIC_CAS(&new_obj->users, &expected_users, (1ULL << shm_stats_file_slot))) {
+		/* a parallel reuse stole us the object, retry */
+		__ha_cpu_relax();
+		goto add;
+	}
+
+	return new_obj;
+};
+
 static struct task *shm_stats_file_hb(struct task *task, void *context, unsigned int state)
 {
 	if (stopping)
@@ -545,12 +696,98 @@ static struct task *shm_stats_file_hb(struct task *task, void *context, unsigned
 	return task;
 }
 
+/* loads shm_stats_file content and tries to associate existing objects from
+ * the shared memory (if any) to objects defined in current haproxy config
+ * based on GUIDs
+ */
+static void shm_stats_file_preload(void)
+{
+	int it = 0;
+	int objects;
+	struct shm_stats_file_object *curr_obj;
+
+	BUG_ON(!shm_stats_file_hdr);
+	objects = HA_ATOMIC_LOAD(&shm_stats_file_hdr->objects);
+	if (!objects)
+		return; // nothing to do
+
+	while (it < objects) {
+		struct guid_node *node;
+		uint64_t users;
+		uint8_t obj_tgid;
+
+		curr_obj = SHM_STATS_FILE_OBJECT(shm_stats_file_hdr, it);
+
+		users = HA_ATOMIC_FETCH_OR(&curr_obj->users, (1ULL << shm_stats_file_slot));
+
+		/* ignore object if not used by anyone: when a process properly deinits,
+		 * it removes its user bit from the object, thus an object without any
+		 * bit should be considered as empty object
+		*/
+		if (!users)
+			goto release;
+
+		obj_tgid = HA_ATOMIC_LOAD(&curr_obj->tgid);
+
+		/* ignore object if greater than our max tgid */
+		if (obj_tgid <= global.nbtgroups &&
+		    (node = guid_lookup(curr_obj->guid))) {
+			switch (*node->obj_type) {
+				case OBJ_TYPE_LISTENER:
+				{
+					struct listener *li;
+
+					BUG_ON(curr_obj->type != SHM_STATS_FILE_OBJECT_TYPE_FE);
+					li = __objt_listener(node->obj_type);
+					if (li->counters) // counters are optional for listeners
+						li->counters->shared.tg[obj_tgid - 1] = &curr_obj->data.fe;
+					break;
+				}
+				case OBJ_TYPE_SERVER:
+				{
+					struct server *sv;
+
+					BUG_ON(curr_obj->type != SHM_STATS_FILE_OBJECT_TYPE_BE);
+					sv = __objt_server(node->obj_type);
+					sv->counters.shared.tg[obj_tgid - 1] = &curr_obj->data.be;
+					break;
+				}
+				case OBJ_TYPE_PROXY:
+				{
+					struct proxy *px;
+
+					px = __objt_proxy(node->obj_type);
+					if (curr_obj->type == SHM_STATS_FILE_OBJECT_TYPE_FE)
+						px->fe_counters.shared.tg[obj_tgid - 1] = &curr_obj->data.fe;
+					else if (curr_obj->type == SHM_STATS_FILE_OBJECT_TYPE_BE)
+						px->be_counters.shared.tg[obj_tgid - 1] = &curr_obj->data.be;
+					else
+						goto release; // not supported
+					break;
+				}
+				default:
+					/* not supported */
+					goto release;
+			}
+			/* success */
+			goto next;
+		}
+
+release:
+		/* we don't use this object, remove ourselves from object's users */
+		HA_ATOMIC_AND(&curr_obj->users, ~(1ULL << shm_stats_file_slot));
+next:
+		it += 1;
+	}
+}
+
 /* prepare and and initialize shm stats memory file as needed */
 int shm_stats_file_prepare(void)
 {
 	struct task *heartbeat_task;
 	int first = 0; // process responsible for initializing the shm memory
 	int slot;
+	int objects;
 
 	/* do nothing if master process or shm_stats_file not configured */
 	if (master || !global.shm_stats_file)
@@ -637,6 +874,34 @@ int shm_stats_file_prepare(void)
 	/* sync local and global clocks, so all clocks are consistent */
 	clock_update_date(0, 1);
 
+	/* check if the map is outdated and must be reset:
+	 * let's consider the map is outdated unless we find an occupied slot
+	 */
+ check_outdated:
+	if (first)
+		goto skip_check_outdated; // not needed
+	first = 1;
+	slot = 0;
+	objects = HA_ATOMIC_LOAD(&shm_stats_file_hdr->objects);
+	while (slot < sizeof(shm_stats_file_hdr->slots) / sizeof(shm_stats_file_hdr->slots[0])) {
+		if (!shm_stats_file_slot_isfree(shm_stats_file_hdr, slot)) {
+			first = 0;
+			break;
+		}
+		slot += 1;
+	}
+	if (first) {
+		/* no more slots occupied, let's reset the map but take some precautions
+		 * to ensure another reset doesn't occur in parallel
+		 */
+		if (!HA_ATOMIC_CAS(&shm_stats_file_hdr->objects, &objects, 0)) {
+			__ha_cpu_relax();
+			goto check_outdated;
+		}
+	}
+
+ skip_check_outdated:
+
 	/* reserve our slot */
 	slot = shm_stats_file_get_free_slot(shm_stats_file_hdr);
 	if (slot == -1) {
@@ -656,6 +921,9 @@ int shm_stats_file_prepare(void)
 	heartbeat_task->process = shm_stats_file_hb;
 	task_schedule(heartbeat_task, tick_add(now_ms, 1000));
 
+	/* try to preload existing objects in the shm (if any) */
+	shm_stats_file_preload();
+
  end:
 	return ERR_NONE;