txm: disallow yields after DDL operation in TX

To avoid sharing (ergo phantom reads) metadata object for different
transactions in MVCC mode, let's do following things. Firstly, after
modification of any shared objects (like space/user/function caches)
disallow transaction to yield in order to prevent other transactions see
that changes. Yield is permitted til the transaction is committed. In
case yield occurs, transaction is aborted. Secondly, on committing
transaction that provides DDL changes let's abort all other transaction
since they may refer to obsolete schema objects. The last restriction
may seem too strict, but it is OK as primitive workaround until
transactional DDL is introduced. In fact we should only abort
transactions that have read dirty (i.e. modified) objects.

Closes #5998
Closes #6140
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;
 }
 
@@ -3587,6 +3595,7 @@ on_replace_dd_func(struct trigger * /* trigger */, void *event)
 			return -1;
 		}
 	}
+	memtx_tx_acquire_ddl(txn);
 	return 0;
 }
 
@@ -3815,6 +3824,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;
 }
 
@@ -4137,6 +4147,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;
 }
 
@@ -4175,6 +4186,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;
 }
 
@@ -4307,6 +4319,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;
 }
 
@@ -4530,6 +4543,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;
 }
 
@@ -4594,6 +4608,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;
 }
 
@@ -4810,6 +4825,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;
 }
 
@@ -4985,6 +5001,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;
 }
 
@@ -5512,6 +5529,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;
 }
 
@@ -5701,6 +5719,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;
 			}
 		}
@@ -5768,6 +5787,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;
 }
 
@@ -5843,8 +5863,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)
@@ -5869,6 +5891,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,27 @@ 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=%ld) has aborted "
+			 "another TX (id=%ld)", 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