MySQL connection pool and how it can be used for parallelization

I want to share with all readers an interesting topic that I recently encountered, and I liked it. The development of this topic has pleased me and added a little experience to the piggy bank. Most likely, or maybe not, they encountered a pool of database connections. After reading this interesting option, I wanted to write an article and share it with you. Perhaps the article will turn out a little long, but I think that someone will still be interested in reading this post, and this topic will interest him. Maybe someone uses this article from this article in their project, in any case, it will be interesting for me to write it and tell it to you.

Engaged in freelance. Somehow, a friend of mine with the owners of a driving school brought me together and there they asked me to create a program to form my own collection of tickets. Something like his SDA ticket book. All questions are stored in the MySQL DBMS in three tables, where in one collection of questions, in another subject of these questions and in the last answers to questions. As a result, 800 questions with or without a picture, including answers. The initial version of the formation was a conventional method with the sequential formation of questions within. I’m quite interested in the topic of multi-threaded program execution, so after creating a completely working method, I decided to make everything more convenient, and at the same time I added the sampling speed.

To begin with, in order to form your own version of the SDA tickets, it is necessary to take into account the topics of issues. However, only 12 topics. The first 8 topics consist of 40 questions, the remaining of 120. Each topic has a specific number on the ticket, while the first 8 topics correspond to 1 question from the ticket, the other 3 questions from the ticket. To store the generated version of the questions, the Dictionary dictionary is used, where each key stores a list of questions on a specific topic. Moreover, the order of questions should always be different, i.e. you need a method that will generate a sequence of numbers from 0 to 40 without repetitions, so it will turn out to choose any ticket and, accordingly, a question from it. With all this in mind, we get the following algorithm for generating all the questions:

• we get all the topics that we have;
• from the topics we form a list of question numbers for each of the topics;
• generate random numbers;
• based on the list of questions and the sequence of random numbers, we make a selection and form a list of questions on the subject;
• enter each list in the dictionary.

The implementation and operation of the first two steps depends on the logic and, in my opinion, can be performed once upon request for sampling. The third step depends on how much we want to make the questions different, to do a new generation before each topic, or once for all topics. I chose to generate only once, in my opinion this is enough. The fourth step will work most often and in this place should be the most time-consuming when sampling. The last step is simple and just skip it.

DB connection

Consider connecting to a database. To do this, we use an abstract class that will store the connection string, Connection and DataReader:

public abstract class SqlBase 
	{
		protected static readonly String Connect;
		protected readonly MySqlConnection SqlConnection;
		protected MySqlDataReader SqlDataReader;
		static SqlBase()
		{
			Connect = String.Format("Database={0};Data Source={1};User ID={2};Password={3};CharSet=utf8",
				Settings.Default.Database, Settings.Default.DataSource, Settings.Default.UserId, Settings.Default.Password);
		}
		protected SqlBase()
		{
			try
			{
				this.SqlConnection = new MySqlConnection(Connect);
				this.SqlConnection.Open();
			}
			catch (Exception ex)
			{
				throw new Exception(ex.Message, ex);
			}
		}
	}

I like abstract classes in that it allows them to create and hide in themselves a convenient logic of work, allowing the derivative to completely concentrate on other specifics and not to worry about other details, assigning responsibilities to the base class. In this case, creating an instance of a derived class, we automatically get the necessary conditions for working with the database and it remains for us to implement the logic of the class itself to form a sample of questions:

public sealed class SqlPerceptionQuestions : SqlBase
	{
        public Dictionary> GetQuestion()
        {
           Generation();
           GetTheme();
           return Request();
        }
        private Dictionary> Request()
        {
           var _collectionDictionary = new Dictionary>();
           for(int ctr = 0; ctr < 12; ctr++)
           {
              using (var _questions = new SqlQuestions())
              {
				if (ctr < 8)
				{
					_collectionDictionary[ctr] = _questions.GetQuestionSmall((Int16)ctr);
				}
				else
				{
					_collectionDictionary[ctr] = _questions.GetQuestionGreat((Int16)ctr);
				}          
              }
           return _collectionDictionary;
          }
        private async void GetTheme()
		{
			//логика метода
		}
            //метод для получения данных для темы с 40 вопросами
            private List GetQuestionSmall(Int16 numTheme)
			{
				var _listQuestions = new List();
				for (Int16 numCard = 0; numCard < 40; numCard++)
				{
					_listQuestions.Add(GetQuestion(numCard, numTheme));
				}
				return _listQuestions;
			}
           //метод для получения данных для темы со 120 вопросами
			private List GetQuestionGreat(Int16 numTheme)
			{
				var _listQuestions = new List();
				for (Int16 numQuestion = 0; numQuestion < 3; numQuestion++)
					for (int numCard = 0; numCard < 40; numCard++)
					{
						_listQuestions.Add(GetQuestion(numQuestion, numTheme, numQuestion));
					}
				return _listQuestions;
			}
            // метод для получения одного вопроса из БД по параметрам
			private Question GetQuestion(Int16 numCard, Int16 numTheme, Int16 numQuestion = 0)
			{
				//логика метода
			}
            //получение ответа на вопрос
			private List GetResponse(Int32 questions_id)
			{
				//логика метода
			}
    }

This is the simplest and most synchronous version, which works for about 2 seconds and 200-400 milliseconds, which will accordingly block the user interface for all this time. This is a good working version, since the very first implementation worked for quite some time, about 6 seconds. After the improvement, only about 2 seconds came out.

Creating an Asynchronous Sampling Version

Everything is fine and everything is already working, but is it as it should be? After all, we have a synchronous method (blocking) and not a console application. A correct and fully working program is needed, which will not block even for half a second, but will work properly under any load. To do this, first of all, we rewrite the GetQuestion () method. Let's make it asynchronous in accordance with the TAP (Task-based Asynchronous Pattern) pattern. Who cares, read on the Internet or have a pretty good book that I really like - Asynchronous Programming in C # 5.0 by Alex Davis, where this topic is described very well, or take a look here . We rewrite it and it will look like this:

public async Task>> GetQuestionAsync()
		{
			return await Task.Factory.StartNew(() =>
			{
				Generation();
				GetTheme();
				return Request();
			}, TaskCreationOptions.LongRunning);
		}

Consider the most interesting in this method: Task.Factory.StartNew (). Starting with .NET 4.5, you can use the version of Task.Run (), which differs from the previous one by a simpler declaration with fewer parameters when creating. In essence, Task.Run () it is just a more convenient shell over Task.Factory.StartNew () and is very suitable for simple creation of asynchronous tasks, but at the same time it has a little less control flexibility. If you need more precise control over which thread performs calculations or how it is planned, use Task.Factory.StartNew (). If interested, check here. In this case, I used this option because I also specified a parameter such as TaskCreationOptions.LongRunning, which marks this task as long and means that this work item will run for a long period of time and may block other work items . This flag also provides information for TaskScheduler that you should expect an excessive subscription and this will allow you to create more threads than the number of available hardware threads. Using this parameter can completely avoid ThreadPool, including global and local queues. Read more about the "Task Scheduler . "

Thus, we get the asynchronous execution of the formation of the data collection and do not block the main stream.

Parallelization of the sample

After that, we proceed to parallelization, which will give us an increase in the speed of data sampling. But for this you will have to redo the main class a bit, change the base class and add another one to ensure parallel operation of the loop, to avoid data races, critical sections and variable problems for different threads.
First, create a new nested class in the SqlPerceptionQuestions class - SqlQuestions . We transfer methods to get questions and answers to them, and also make it derived from SqlBase, while in the outer class we leave the method to get those, we will call it only once, as well as the formation of a sequence of numbers. SqlQuestions

class code :

		internal sealed class SqlQuestions : SqlBase
		{
			internal List GetQuestionSmall(Int16 numTheme)
			{
				var _listQuestions = new List();
				for (Int16 numCard = 0; numCard < 40; numCard++)
				{
					_listQuestions.Add(GetQuestion(numCard, numTheme));
				}
				return _listQuestions;
			}
			internal List GetQuestionGreat(Int16 numTheme)
			{
				var _listQuestions = new List();
				for (Int16 numQuestion = 0; numQuestion < 3; numQuestion++)
					for (int numCard = 0; numCard < 40; numCard++)
					{
						_listQuestions.Add(GetQuestion(numQuestion, numTheme, numQuestion));
					}
				return _listQuestions;
			}
			private Question GetQuestion(Int16 numCard, Int16 numTheme, Int16 numQuestion = 0)
			{
				//логика метода
			}
			private List GetResponse(Int32 questions_id)
			{
                //логика метода
			}
		}

To parallelize the loop, we will use Parallel.For (). This is a fairly convenient way to organize the loading of data into multiple streams. But this is also fraught with the fact that we will need to create at least several database connections, since one Connection is capable of processing one DataReader. We rewrite the .Request () method:

private Dictionary> Request()
		{
			var _collectionDictionary = new Dictionary>();
			var _po = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount };
			Parallel.For(0, 12, _po, ctr =>
				{
					using (var _questions = new SqlQuestions())
					{
						if (ctr < 8)
						{
							_collectionDictionary[ctr] = _questions.GetQuestionSmall((Int16)ctr);
						}
						else
						{
							_collectionDictionary[ctr] = _questions.GetQuestionGreat((Int16)ctr);
						}
					}
				});
			return _collectionDictionary;
		}

And after opening the connection, it must be closed. Moreover, all this will happen in a cycle. To implement all this, I decided to create a separate derived class SqlQuestions . To close the connection, we will call .Dispose (), in which we will write down what we need to do when closing. To do this, first declare the .Dispose () method in the base class:

	public void Dispose()
		{
			Dispose(true);
			GC.SuppressFinalize(this);
		}
		protected abstract void Dispose(Boolean disposing);

And we will implement it differently in derived classes. Why is that? To begin with, what happens when you create a database connection and close it? If you open a connection and then close it, then it is placed for some time (about 3 minutes) in the MySQL connection pool, and if you open a new one, the connection is taken from the pool, which means that time and resources for reopening are not used. Let's launch our new methods and see how much time it takes to open the database connections. To do this, insert the Stopwatch in the base class in the code where the connection is opened and see what we have at the output. The code:

protected SqlBase()
		{
			Stopwatch st = new Stopwatch();
			st.Start();
			try
			{
				this.SqlConnection = new MySqlConnection(Connect);
				this.SqlConnection.Open();
			}
			catch (MySqlException ex)
			{
				throw new Exception(ex.Message, ex);
			}
			st.Stop();
			Debug.WriteLine("Время открытия соединения БД : " + st.Elapsed.Seconds.ToString() + " секунд " + st.Elapsed.Milliseconds.ToString() + " миллисекунд");
		}

The first connection is the longest, it opens a connection for the SqlPerceptionQuestions class , which will be open for the entire duration of the method. Subsequent connections are those that were opened when in a loop when instantiating the SqlQuestions class. Taking into account the number of processors on my computer, of which 4 we get that in the cycle maximum 4 connections will be open. We get that in total 5 connections will be initially opened, while in the cycle they will both open and close. Therefore, the first 5 connections require time to open, and after that, when the old connections are closed in the cycle and new ones are opened, they will not take time and resources, since the connections are in the pool and they are simply issued every time they are needed. Because of this, class cleanup is implemented in slightly different ways. In the SqlPerceptionQuestions class , the method will look like this:

protected override void Dispose(bool disposing)
		{
			if (!this.disposed)
			{
				if (SqlDataReader != null)
				{
					SqlDataReader.Close();
					SqlDataReader.Dispose();
				}
				SqlConnection.Close();
				SqlConnection.Dispose();
				MySqlConnection.ClearAllPools();
			}
			disposed = true;
		}

And in the SqlQuestions class , just as well, with the exception of the line MySqlConnection.ClearAllPools (); After all, if we leave it, we will get this situation:


As we see, the constant cleaning of the thread pool leads to the constant opening of the connection with the ensuing consequences.

MySql Connection Pool

Let's consider this moment in more detail. Connector / Net MySql supports connection pooling to increase the performance and scalability of high-usage application databases. This feature is enabled by default. You can disable it or change the characteristics using the connection string options. The connection string supports options regarding the connection pool:

Option Name	Default value	Details
Cache Server Properties, CacheServerProperties	Off	Determines whether to update the parameters of some system variables (SHOW VARIABLES) every time a connection from the pool is returned. Enabling this option speeds up the connection to the connection pool context. Your application is not informed of configuration changes made by other connections. This option has been added since version Connector / Net 6.3.
Connection Lifetime, ConnectionLifeTime	0	When a connection is returned to the pool, its creation time is compared with the current time and, if it exceeds the value of ConnectionLifeTime, then this connection is destroyed. This is useful in clustered configurations for balancing the load between a production server and an online server. A value of zero (0) makes the connection in the pool standby for the maximum possible time.
Connection Reset, ConnectionReset	false	If true, the connection state when retrieving from the pool is reset. The default value allows you to avoid an additional processing cycle by the server to obtain a connection, but the connection status is not reset.
Maximum Pool Size, Max Pool Size, MaximumPoolsize, maxpoolsize	100	The maximum number of connections that are allowed to be in the pool.
Minimum Pool Size, Min Pool Size, MinimumPoolSize, minpoolsize	0	The minimum number of compounds that are allowed to be in the pool.
Pooling	true	If true, then the MySqlConnection object is taken from the pool, if necessary, it is created and added to the corresponding pool. The defined values ​​are true, false, yes, and no.

Using these parameters, you can manage the pool as needed. For example:

• Connection Reset resets the connection context, and if it is not used by default, the time to receive a connection from the pool increases slightly.
• The Minimum Pool Size allows you to set, if necessary, the number of connections in the pool that will exist indefinitely, if the number of connections is less than or equal to the value of the Minimum Pool Size.
• Disabling Pooling will lead to the same results if you constantly clear the pool of threads (as in the example of this program)

By default, starting with MySQL Connector / Net 6.2, there is a background job that runs every three minutes and removes connections from the pool that are in standby (not in use) for more than three minutes. Clearing the pool frees up resources both on the client and on the server side. This is because on the client side, each connection uses a socket, and on the server side, each connection uses a socket and stream.

Parallel call stack

For the sake of interest, if you put a breakpoint, for example, in the .GetQuestion () method and see a parallel call stack, we will see:


As you can see from the screenshot, we are in one of the threads, which is suspended, and we determine from the call stack that this method was called from a method that loads a small collection of questions (40). To the left of it are 3 more threads, two of which are stopped at this moment on the line for adding a question to the collection, which also process a small collection of questions. And the last of them, thread 4 is engaged in processing and receives the answer at this moment to the question, only for a question from a large collection (120). All these 4 threads were created in a parallel loop and work almost simultaneously with the rest of the loop iteration threads. These threads are included in the total number of threads in the program, of which 8, where the remaining 4 solve other tasks of the program.

The final touch - exception handling

And finally, for the program to work, we need exception handling. Suddenly, fields or some other parameters changed from the database or an unexpected error occurred in the program itself. Rewrite the GetQuestionAsync () method:

public async Task>> GetQuestionAsync()
		{
			return await Task.Factory.StartNew(() =>
			{
				try
				{
					Generation();
					GetTheme();
					return Request();
				}
				catch (AggregateException ex)
				{
					throw new AggregateException(ex.Message, ex.InnerExceptions);
				}
				catch (Exception ex)
				{
					throw new AggregateException(ex.Message, ex.InnerException);
				}
			}, TaskCreationOptions.LongRunning);
		}

AggregateException exception handling is related to the fact that the Parallel.For loop, if an exception occurs, will raise an error of this type and, therefore, it must be processed and passed to the caller. It is logical that a parallel loop generates this kind of error. Let's consider this moment in more detail: for this, I changed the Sql query in .GetQuestion (), deliberately incorrectly indicating one of the parameters that does not exist in the database table. We get:


Moreover, if you continue debugging, then in total this exception will occur 4 times, which is quite logical. In order to handle all 4, albeit related to one reason, you need to somehow arrange them, for which an AggregateException is suitable.

Exception handling is connected with the fact that if an exception occurs in the .GetTheme () method, then there will be one exception and it must also be caught.

The calling code is as follows:

private async void Button_Click(object sender, RoutedEventArgs e)
		{
			Stopwatch st = new Stopwatch();
			st.Start();
			try
			{
				SqlQuest = new SqlPerceptionQuestions();
				collectionQuest = await SqlQuest.GetQuestionAsync();
			}
			catch (AggregateException ex)
			{
				ex.ShowError();
			}
            catch(Exception ex)
			{
				ex.ShowError();
			}
			finally
			{
				if (SqlQuest != null)
					SqlQuest.Dispose();
			}
			st.Stop();
			Debug.WriteLine("Время выполнения метода: " + st.Elapsed.Seconds.ToString() + " секунд " + st.Elapsed.Milliseconds.ToString() + " миллисекунд");
			Debugger.Break();
		}

Finally ...

In general, it started with the fact that when I wrote the parallel version of the cycle, I thought about how much the frequent opening and closing of the database connection is very much affected. Visiting forums and asking smart people, I learned about the connection pool. Then he asked himself, is it necessary to create it or is it created implicitly? I experimented a bit and read the MySQL documentation, as a result I came to the conclusion that this thing is very successful, the connection pool!