Allow to reuse tuple_formats for ephemeral spaces

Since under heavy load with SQL queries ephemeral
spaces might be extensively used it is possible to run out
of tuple_formats for such spaces. This occurs because
tuple_format is not immediately deleted when ephemeral space is
dropped. Its removel is postponed instead and triggered only
when tuple memory is exhausted.
As far as there's no way to alter ephemeral space's format,
let's re-use them for multiple epehemral spaces in case
they're identical.

Closes #3924

src/box/blackhole.c

@@ -157,7 +157,7 @@ blackhole_engine_create_space(struct engine *engine, struct space_def *def,
 	struct tuple_format *format;
 	format = tuple_format_new(&tuple_format_runtime->vtab, NULL, NULL, 0,
 				  def->fields, def->field_count, def->dict,
-				  false, def->exact_field_count);
+				  false, false, def->exact_field_count);
 	if (format == NULL) {
 		free(space);
 		return NULL;

src/box/box.cc

@@ -46,6 +46,7 @@
 #include <rmean.h>
 #include "main.h"
 #include "tuple.h"
+#include "tuple_format.h"
 #include "session.h"
 #include "schema.h"
 #include "engine.h"
@@ -2050,6 +2051,9 @@ box_init(void)
 	 */
 	session_init();
 
+	if (tuple_format_init() != 0)
+		diag_raise();
+
 	if (tuple_init(lua_hash) != 0)
 		diag_raise();
 

src/box/memtx_engine.c

@@ -994,6 +994,12 @@ memtx_engine_gc_f(va_list va)
 	struct memtx_engine *memtx = va_arg(va, struct memtx_engine *);
 	while (!fiber_is_cancelled()) {
 		bool stop;
+		struct errinj *delay = errinj(ERRINJ_MEMTX_DELAY_GC,
+					      ERRINJ_BOOL);
+		if (delay != NULL && delay->bparam) {
+			while (delay->bparam)
+				fiber_sleep(0.001);
+		}
 		memtx_engine_run_gc(memtx, &stop);
 		if (stop) {
 			fiber_yield_timeout(TIMEOUT_INFINITY);

src/box/memtx_space.c

@@ -992,7 +992,8 @@ memtx_space_new(struct memtx_engine *memtx,
 	struct tuple_format *format =
 		tuple_format_new(&memtx_tuple_format_vtab, memtx, keys, key_count,
 				 def->fields, def->field_count, def->dict,
-				 def->opts.is_temporary, def->exact_field_count);
+				 def->opts.is_temporary, def->opts.is_ephemeral,
+				 def->exact_field_count);
 	if (format == NULL) {
 		free(memtx_space);
 		return NULL;

src/box/space.c

@@ -194,10 +194,11 @@ space_new(struct space_def *def, struct rlist *key_list)
 struct space *
 space_new_ephemeral(struct space_def *def, struct rlist *key_list)
 {
+	assert(def->opts.is_temporary);
+	assert(def->opts.is_ephemeral);
 	struct space *space = space_new(def, key_list);
 	if (space == NULL)
 		return NULL;
-	space->def->opts.is_temporary = true;
 	space->vtab->init_ephemeral_space(space);
 	return space;
 }

src/box/space_def.c

@@ -53,6 +53,7 @@ checks_array_decode(const char **str, uint32_t len, char *opt, uint32_t errcode,
 const struct space_opts space_opts_default = {
 	/* .group_id = */ 0,
 	/* .is_temporary = */ false,
+	/* .is_ephemeral = */ false,
 	/* .view = */ false,
 	/* .sql        = */ NULL,
 	/* .checks     = */ NULL,
@@ -61,6 +62,7 @@ const struct space_opts space_opts_default = {
 const struct opt_def space_opts_reg[] = {
 	OPT_DEF("group_id", OPT_UINT32, struct space_opts, group_id),
 	OPT_DEF("temporary", OPT_BOOL, struct space_opts, is_temporary),
+	OPT_DEF("ephemeral", OPT_BOOL, struct space_opts, is_ephemeral),
 	OPT_DEF("view", OPT_BOOL, struct space_opts, is_view),
 	OPT_DEF("sql", OPT_STRPTR, struct space_opts, sql),
 	OPT_DEF_ARRAY("checks", struct space_opts, checks,
@@ -267,6 +269,7 @@ space_def_new_ephemeral(uint32_t field_count)
 						    &space_opts_default,
 						    &field_def_default, 0);
 	space_def->opts.is_temporary = true;
+	space_def->opts.is_ephemeral = true;
 	return space_def;
 }
 

src/box/space_def.h

@@ -58,6 +58,11 @@ struct space_opts {
          *   does not require manual release.
 	 */
 	bool is_temporary;
+	/**
+	 * This flag is set if space is ephemeral and hence
+	 * its format might be re-used.
+	 */
+	bool is_ephemeral;
 	/**
 	 * If the space is a view, then it can't feature any
 	 * indexes, and must have SQL statement. Moreover,

src/box/tuple.c

@@ -208,7 +208,8 @@ tuple_init(field_name_hash_f hash)
 	 * Create a format for runtime tuples
 	 */
 	tuple_format_runtime = tuple_format_new(&tuple_format_runtime_vtab, NULL,
-						NULL, 0, NULL, 0, NULL, false, 0);
+						NULL, 0, NULL, 0, NULL, false,
+						false, 0);
 	if (tuple_format_runtime == NULL)
 		return -1;
 
@@ -380,7 +381,8 @@ box_tuple_format_new(struct key_def **keys, uint16_t key_count)
 {
 	box_tuple_format_t *format =
 		tuple_format_new(&tuple_format_runtime_vtab, NULL,
-				 keys, key_count, NULL, 0, NULL, false, 0);
+				 keys, key_count, NULL, 0, NULL, false, false,
+				 0);
 	if (format != NULL)
 		tuple_format_ref(format);
 	return format;

src/box/tuple_format.c

@@ -34,6 +34,8 @@
 #include "tuple_format.h"
 #include "coll_id_cache.h"
 
+#include "third_party/PMurHash.h"
+
 /** Global table of tuple formats */
 struct tuple_format **tuple_formats;
 static intptr_t recycled_format_ids = FORMAT_ID_NIL;
@@ -276,7 +278,11 @@ tuple_format_register(struct tuple_format *format)
 			formats_capacity = new_capacity;
 			tuple_formats = formats;
 		}
-		if (formats_size == FORMAT_ID_MAX + 1) {
+		struct errinj *inj = errinj(ERRINJ_TUPLE_FORMAT_COUNT,
+					    ERRINJ_INT);
+		if ((inj != NULL && inj->iparam > 0
+		     && formats_size >= inj->iparam + 1)
+		    || formats_size == FORMAT_ID_MAX + 1) {
 			diag_set(ClientError, ER_TUPLE_FORMAT_LIMIT,
 				 (unsigned) formats_capacity);
 			return -1;
@@ -380,10 +386,77 @@ tuple_format_alloc(struct key_def * const *keys, uint16_t key_count,
 	return NULL;
 }
 
+static int
+tuple_format_cmp(const struct tuple_format *a, const struct tuple_format *b)
+{
+	if (a->exact_field_count != b->exact_field_count)
+		return a->exact_field_count - b->exact_field_count;
+	if (tuple_format_field_count(a) != tuple_format_field_count(b))
+		return tuple_format_field_count(a) - tuple_format_field_count(b);
+
+	for (uint32_t i = 0; i < tuple_format_field_count(a); ++i) {
+		struct tuple_field *field_a = tuple_format_field(
+			(struct tuple_format *)a, i);
+		struct tuple_field *field_b = tuple_format_field(
+			(struct tuple_format *)b, i);
+		if (field_a->type != field_b->type)
+			return (int)field_a->type - (int)field_b->type;
+		if (field_a->coll_id != field_b->coll_id)
+			return (int)field_a->coll_id - (int)field_b->coll_id;
+		if (field_a->nullable_action != field_b->nullable_action)
+			return (int)field_a->nullable_action -
+				(int)field_b->nullable_action;
+		if (field_a->is_key_part != field_b->is_key_part)
+			return (int)field_a->is_key_part -
+				(int)field_b->is_key_part;
+	}
+
+	return 0;
+}
+
+static uint32_t
+tuple_format_hash(struct tuple_format *format)
+{
+	uint32_t h = 0;
+	uint32_t carry = 0;
+	uint32_t size = 0;
+	for (uint32_t i = 0; i < tuple_format_field_count(format); ++i) {
+		struct tuple_field *f = tuple_format_field(format, i);
+		PMurHash32_Process(&h, &carry, &f->type,
+				   sizeof(enum field_type));
+		size += sizeof(enum field_type);
+		PMurHash32_Process(&h, &carry, &f->coll_id,
+				   sizeof(uint32_t));
+		size += sizeof(uint32_t);
+		PMurHash32_Process(&h, &carry, &f->nullable_action,
+				   sizeof(enum on_conflict_action));
+		size += sizeof(enum on_conflict_action);
+		PMurHash32_Process(&h, &carry, &f->is_key_part, sizeof(bool));
+		size += sizeof(bool);
+	}
+	return PMurHash32_Result(h, carry, size);
+}
+
+#define MH_SOURCE 1
+#define mh_name _tuple_format
+#define mh_key_t struct tuple_format *
+#define mh_node_t struct tuple_format *
+#define mh_arg_t void *
+#define mh_hash(a, arg) ((*(a))->hash)
+#define mh_hash_key(a, arg) ((a)->hash)
+#define mh_cmp(a, b, arg) (tuple_format_cmp(*(a), *(b)))
+#define mh_cmp_key(a, b, arg) (tuple_format_cmp((a), *(b)))
+#include "salad/mhash.h"
+
+struct mh_tuple_format_t *tuple_formats_hash = NULL;
+
 /** Free tuple format resources, doesn't unregister. */
 static inline void
 tuple_format_destroy(struct tuple_format *format)
 {
+	mh_int_t key = mh_tuple_format_find(tuple_formats_hash, format, NULL);
+	if (key != mh_end(tuple_formats_hash))
+		mh_tuple_format_del(tuple_formats_hash, key, NULL);
 	free(format->required_fields);
 	tuple_format_destroy_fields(format);
 	tuple_dictionary_unref(format->dict);
@@ -402,7 +475,8 @@ tuple_format_new(struct tuple_format_vtab *vtab, void *engine,
 		 struct key_def * const *keys, uint16_t key_count,
 		 const struct field_def *space_fields,
 		 uint32_t space_field_count, struct tuple_dictionary *dict,
-		 bool is_temporary, uint32_t exact_field_count)
+		 bool is_temporary, bool is_ephemeral,
+		 uint32_t exact_field_count)
 {
 	struct tuple_format *format =
 		tuple_format_alloc(keys, key_count, space_field_count, dict);
@@ -411,18 +485,46 @@ tuple_format_new(struct tuple_format_vtab *vtab, void *engine,
 	format->vtab = *vtab;
 	format->engine = engine;
 	format->is_temporary = is_temporary;
+	format->is_ephemeral = is_ephemeral;
 	format->exact_field_count = exact_field_count;
-	if (tuple_format_register(format) < 0) {
-		tuple_format_destroy(format);
-		free(format);
-		return NULL;
-	}
 	if (tuple_format_create(format, keys, key_count, space_fields,
 				space_field_count) < 0) {
 		tuple_format_delete(format);
 		return NULL;
 	}
-	return format;
+	format->hash = tuple_format_hash(format);
+	struct errinj *inj = errinj(ERRINJ_TUPLE_FORMAT_COUNT, ERRINJ_INT);
+	if (inj != NULL && inj->iparam == 200) {
+		inj = NULL;
+	}
+	if (format->is_ephemeral) {
+		mh_int_t key = mh_tuple_format_find(tuple_formats_hash, format,
+						    NULL);
+		if (key != mh_end(tuple_formats_hash)) {
+			struct tuple_format **entry = mh_tuple_format_node(tuple_formats_hash, key);
+			tuple_format_destroy(format);
+			free(format);
+			return *entry;
+		} else {
+			if (tuple_format_register(format) < 0) {
+				tuple_format_destroy(format);
+				free(format);
+				return NULL;
+			}
+		        mh_tuple_format_put(tuple_formats_hash,
+					    (const
+					     struct tuple_format **)&format,
+					     NULL, NULL);
+			return format;
+		}
+	} else {
+		if (tuple_format_register(format) < 0) {
+			tuple_format_destroy(format);
+			free(format);
+			return NULL;
+		}
+		return format;
+	}
 }
 
 bool
@@ -823,3 +925,15 @@ tuple_field_raw_by_path(struct tuple_format *format, const char *tuple,
 		 tt_sprintf("error in path on position %d", rc));
 	return -1;
 }
+
+int
+tuple_format_init()
+{
+	tuple_formats_hash = mh_tuple_format_new();
+	if (tuple_formats_hash == NULL) {
+		diag_set(OutOfMemory, sizeof(struct mh_tuple_format_t), "malloc",
+			 "tuple format hash");
+		return -1;
+	}
+	return 0;
+}

src/box/tuple_format.h

@@ -151,6 +151,11 @@ struct tuple_format {
 	 * in progress.
 	 */
 	bool is_temporary;
+	/**
+	 * This format belongs to ephemeral space and thus might
+	 * be shared with other ephemeral spaces.
+	 */
+	bool is_ephemeral;
 	/**
 	 * Size of field map of tuple in bytes.
 	 * \sa struct tuple
@@ -193,14 +198,19 @@ struct tuple_format {
 	 * tuple_field::token.
 	 */
 	struct json_tree fields;
+
+	/**
+	 * Hash key for shared formats of epeheral spaces.
+	 */
+	uint32_t hash;
 };
 
 /**
  * Return the number of top-level tuple fields defined by
  * a given format.
  */
 static inline uint32_t
-tuple_format_field_count(struct tuple_format *format)
+tuple_format_field_count(const struct tuple_format *format)
 {
 	const struct json_token *root = &format->fields.root;
 	return root->children != NULL ? root->max_child_idx + 1 : 0;
@@ -263,6 +273,7 @@ tuple_format_unref(struct tuple_format *format)
  * @param key_count The number of keys in @a keys array.
  * @param space_fields Array of fields, defined in a space format.
  * @param space_field_count Length of @a space_fields.
+ * @param is_ephemeral Set if format belongs to ephemeral space.
  * @param is_temporary Set if format is temporary.
  * @param exact_field_count Exact field count for format.
  *
@@ -274,7 +285,8 @@ tuple_format_new(struct tuple_format_vtab *vtab, void *engine,
 		 struct key_def * const *keys, uint16_t key_count,
 		 const struct field_def *space_fields,
 		 uint32_t space_field_count, struct tuple_dictionary *dict,
-		 bool is_temporary, uint32_t exact_field_count);
+		 bool is_temporary, bool is_ephemeral,
+		 uint32_t exact_field_count);
 
 /**
  * Check, if @a format1 can store any tuples of @a format2. For
@@ -459,6 +471,13 @@ tuple_field_by_part_raw(struct tuple_format *format, const char *data,
 	return tuple_field_raw(format, data, field_map, part->fieldno);
 }
 
+/**
+ * Initialize tuple formats.
+ * @retval 0 on success, -1 otherwise.
+ */
+int
+tuple_format_init();
+
 #if defined(__cplusplus)
 } /* extern "C" */
 #endif /* defined(__cplusplus) */

src/box/vinyl.c

@@ -622,7 +622,7 @@ vinyl_engine_create_space(struct engine *engine, struct space_def *def,
 		tuple_format_new(&vy_tuple_format_vtab, NULL, keys,
 				 key_count,
 				 def->fields, def->field_count, def->dict,
-				 false, def->exact_field_count);
+				 false, false, def->exact_field_count);
 	if (format == NULL) {
 		free(space);
 		return NULL;
@@ -3046,7 +3046,7 @@ vy_send_lsm(struct vy_join_ctx *ctx, struct vy_lsm_recovery_info *lsm_info)
 		goto out;
 	ctx->format = tuple_format_new(&vy_tuple_format_vtab, NULL,
 				       &ctx->key_def, 1, NULL, 0, NULL, false,
-				       0);
+				       false, 0);
 	if (ctx->format == NULL)
 		goto out_free_key_def;
 	tuple_format_ref(ctx->format);

src/box/vy_lsm.c

@@ -61,7 +61,8 @@ vy_lsm_env_create(struct vy_lsm_env *env, const char *path,
 		  void *upsert_thresh_arg)
 {
 	env->key_format = tuple_format_new(&vy_tuple_format_vtab, NULL,
-					   NULL, 0, NULL, 0, NULL, false, 0);
+					   NULL, 0, NULL, 0, NULL, false, false,
+					   0);
 	if (env->key_format == NULL)
 		return -1;
 	tuple_format_ref(env->key_format);
@@ -155,7 +156,7 @@ vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_cache_env *cache_env,
 	} else {
 		lsm->disk_format = tuple_format_new(&vy_tuple_format_vtab, NULL,
 						    &cmp_def, 1, NULL, 0, NULL,
-						    false, 0);
+						    false, false, 0);
 		if (lsm->disk_format == NULL)
 			goto fail_format;
 	}