txm: disallow yields after DDL operation in TX

Closes #5998
Workaround for #6138

src/box/alter.cc

@@ -57,6 +57,7 @@
 #include "sequence.h"
 #include "sql.h"
 #include "constraint_id.h"
+#include "memtx_tx.h"
 
 /* {{{ Auxiliary functions and methods. */
 
@@ -2500,6 +2501,10 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 		}
 		alter_guard.is_active = false;
 	}
+	/*
+	 * After cache modification transaction is now allowed to yield.
+	 */
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -2829,6 +2834,7 @@ on_replace_dd_index(struct trigger * /* trigger */, void *event)
 		return -1;
 	}
 	scoped_guard.is_active = false;
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -2918,6 +2924,7 @@ on_replace_dd_truncate(struct trigger * /* trigger */, void *event)
 		return -1;
 	}
 	scoped_guard.is_active = false;
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -3210,6 +3217,7 @@ on_replace_dd_user(struct trigger * /* trigger */, void *event)
 			return -1;
 		txn_stmt_on_rollback(stmt, on_rollback);
 	}
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -3584,6 +3592,7 @@ on_replace_dd_func(struct trigger * /* trigger */, void *event)
 			return -1;
 		}
 	}
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -3812,6 +3821,7 @@ on_replace_dd_collation(struct trigger * /* trigger */, void *event)
 		diag_set(ClientError, ER_UNSUPPORTED, "collation", "alter");
 		return -1;
 	}
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -4134,6 +4144,7 @@ on_replace_dd_priv(struct trigger * /* trigger */, void *event)
 			return -1;
 		txn_stmt_on_rollback(stmt, on_rollback);
 	}
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -4172,6 +4183,7 @@ on_replace_dd_schema(struct trigger * /* trigger */, void *event)
 		REPLICASET_UUID = uu;
 		say_info("cluster uuid %s", tt_uuid_str(&uu));
 	}
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -4304,6 +4316,7 @@ on_replace_dd_cluster(struct trigger *trigger, void *event)
 			return -1;
 		txn_stmt_on_commit(stmt, on_commit);
 	}
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -4527,6 +4540,7 @@ on_replace_dd_sequence(struct trigger * /* trigger */, void *event)
 	def_guard.is_active = false;
 	if (trigger_run(&on_alter_sequence, seq) != 0)
 		return -1;
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -4591,6 +4605,7 @@ on_replace_dd_sequence_data(struct trigger * /* trigger */, void *event)
 		txn_stmt_on_rollback(stmt, on_rollback);
 		sequence_reset(seq);
 	}
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -4807,6 +4822,7 @@ on_replace_dd_space_sequence(struct trigger * /* trigger */, void *event)
 	}
 	if (trigger_run(&on_alter_space, space) != 0)
 		return -1;
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -4982,6 +4998,7 @@ on_replace_dd_trigger(struct trigger * /* trigger */, void *event)
 	txn_stmt_on_rollback(stmt, on_rollback);
 	txn_stmt_on_commit(stmt, on_commit);
 	++schema_version;
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -5509,6 +5526,7 @@ on_replace_dd_fk_constraint(struct trigger * /* trigger*/, void *event)
 		space_reset_fk_constraint_mask(parent_space);
 	}
 	++schema_version;
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -5698,6 +5716,7 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 				old_def->is_enabled = ck_def->is_enabled;
 				if (trigger_run(&on_alter_space, space) != 0)
 					return -1;
+				memtx_tx_acquire_ddl(txn);
 				return 0;
 			}
 		}
@@ -5765,6 +5784,7 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 	if (trigger_run(&on_alter_space, space) != 0)
 		return -1;
 	++schema_version;
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -5840,8 +5860,10 @@ on_replace_dd_func_index(struct trigger *trigger, void *event)
 	 * Index is already initialized for corresponding
 	 * function. Index rebuild is not required.
 	 */
-	if (index_def_get_func(index->def) == func)
+	if (index_def_get_func(index->def) == func) {
+		memtx_tx_acquire_ddl(txn);
 		return 0;
+	}
 
 	alter = alter_space_new(space);
 	if (alter == NULL)
@@ -5866,6 +5888,7 @@ on_replace_dd_func_index(struct trigger *trigger, void *event)
 	}
 
 	scoped_guard.is_active = false;
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 

src/box/memtx_engine.c

@@ -381,6 +381,8 @@ memtx_engine_prepare(struct engine *engine, struct txn *txn)
 		if (stmt->add_story != NULL || stmt->del_story != NULL)
 			memtx_tx_history_prepare_stmt(stmt);
 	}
+	if (txn->is_schema_changed)
+		memtx_tx_abort_all_for_ddl(txn);
 	return 0;
 }
 

src/box/memtx_tx.c

@@ -258,6 +258,26 @@ memtx_tx_register_tx(struct txn *tx)
 	rlist_add_tail(&txm.all_txs, &tx->in_all_txs);
 }
 
+void
+memtx_tx_acquire_ddl(struct txn *tx)
+{
+	tx->is_schema_changed = true;
+	(void) txn_can_yield(tx, false);
+}
+
+void
+memtx_tx_abort_all_for_ddl(struct txn *ddl_owner)
+{
+	struct txn *to_be_aborted;
+	rlist_foreach_entry(to_be_aborted, &txm.all_txs, in_all_txs) {
+		if (to_be_aborted == ddl_owner)
+			continue;
+		to_be_aborted->status = TXN_CONFLICTED;
+		say_warn("Transaction committing DDL (id=%d) has aborted "
+			 "another TX (id=%d)", ddl_owner->id,to_be_aborted->id);
+	}
+}
+
 int
 memtx_tx_cause_conflict(struct txn *breaker, struct txn *victim)
 {

src/box/memtx_tx.h

@@ -176,6 +176,21 @@ memtx_tx_manager_init();
 void
 memtx_tx_manager_free();
 
+/**
+ * Transaction providing DDL changes is disallowed to yield after
+ * modifications of internal caches (i.e. after ALTER operation finishes).
+ */
+void
+memtx_tx_acquire_ddl(struct txn *tx);
+
+/**
+ * Mark all transactions except for a given as aborted due to conflict:
+ * when DDL operation is about to be committed other transactions are
+ * considered to use obsolete schema so that should be aborted.
+ */
+void
+memtx_tx_abort_all_for_ddl(struct txn *ddl_owner);
+
 /**
  * Notify TX manager that if transaction @a breaker is committed then the
  * transaction @a victim must be aborted due to conflict. It is achieved

src/box/txn.c

@@ -301,6 +301,7 @@ txn_begin(void)
 	txn->engine = NULL;
 	txn->engine_tx = NULL;
 	txn->fk_deferred_count = 0;
+	txn->is_schema_changed = false;
 	rlist_create(&txn->savepoints);
 	memtx_tx_register_tx(txn);
 	txn->fiber = NULL;

src/box/txn.h

@@ -424,6 +424,8 @@ struct txn {
 	struct rlist gap_list;
 	/** Link in tx_manager::all_txs. */
 	struct rlist in_all_txs;
+	/** True in case transaction provides any DDL change. */
+	bool is_schema_changed;
 };
 
 static inline bool