SQL Server Elastic Demo 1

gistfile1.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Azure.SqlDatabase.ElasticScale.ShardManagement;
using System.Data.SqlClient;
using Microsoft.Azure.SqlDatabase.ElasticScale.ShardManagement.Schema;

/**
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
 * IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
 * OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
 * ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 * OTHER DEALINGS IN THE SOFTWARE.
*/

namespace elasticdemo
{
    class Program
    {
        const string SHARDMAPMANAGER_DB = "ShardMapManager";
        const string SHARD_DB_PREFIX = "Shard_";

        const string SHARDMAP = "Main";

        const string SCHEMA_NAME = "log";
        const string TABLE_NAME = "LogEntry";
        const string TABLE_ID = "ID";

        static ListShardMap<int> listShardMapMain = null;
        static List<System.Data.SqlClient.SqlConnectionStringBuilder> lServersShards = null;
        static System.Data.SqlClient.SqlConnectionStringBuilder ssShardMapManager = null;
        static int iThreads = -1;
        static int iInsertsPerThread = -1;

        static void Main(string[] args)
        {
            Main_Code(args);
        }

        static void Main_Code(string[] args)
        {
            ssShardMapManager = new System.Data.SqlClient.SqlConnectionStringBuilder();

            ShardMapManager shardMapManager;

            lServersShards = new List<System.Data.SqlClient.SqlConnectionStringBuilder>();

            #region Parameter reading
            {
                if (args.Length < 4)
                {
                    Console.WriteLine("Syntax error.\nelasticdemo <shard map instance> #thread_count #inserts_total <shard0> | <shard1> ... <shardN>");
                    return;
                }

                ssShardMapManager.DataSource = args[0];
                ssShardMapManager.InitialCatalog = "master";
                ssShardMapManager.IntegratedSecurity = true;

                iThreads = int.Parse(args[1]);

                int iInsertsTotal = int.Parse(args[2]);
                iInsertsPerThread = iInsertsTotal / iThreads;

                for (int i = 3; i < args.Length; i++)
                {
                    lServersShards.Add(new SqlConnectionStringBuilder());
                    lServersShards[lServersShards.Count - 1].IntegratedSecurity = true;
                    lServersShards[lServersShards.Count - 1].DataSource = args[i];
                }
            }
            #endregion

            #region Initialize shard map manager

            #region Create shard map manager db
            CreateDatabaseIfNotExists(ssShardMapManager.ConnectionString, SHARDMAPMANAGER_DB);
            #endregion

            ssShardMapManager.InitialCatalog = SHARDMAPMANAGER_DB;
            if (!ShardMapManagerFactory.TryGetSqlShardMapManager(
                ssShardMapManager.ConnectionString,
                ShardMapManagerLoadPolicy.Lazy,
                out shardMapManager))
            {
                Console.WriteLine("Shard map manager to be created...");
                shardMapManager = ShardMapManagerFactory.CreateSqlShardMapManager(ssShardMapManager.ConnectionString);
            }
            else
            {
                Console.WriteLine("Shard map manager already in place...");
            }

            #endregion

            #region Initialize shard map

            if (!shardMapManager.TryGetListShardMap<int>(SHARDMAP, out listShardMapMain))
            {
                listShardMapMain = shardMapManager.CreateListShardMap<int>(SHARDMAP);
            }
            #endregion

            #region Initialize schema info
            {
                SchemaInfo si = new SchemaInfo();

                // create it only if it doesn't exists
                if (shardMapManager.GetSchemaInfoCollection().FirstOrDefault(item =>
                {
                    return item.Key == SHARDMAP;
                }).Key == null)
                {
                    si.Add(new ShardedTableInfo(SCHEMA_NAME, TABLE_NAME, TABLE_ID));
                    shardMapManager.GetSchemaInfoCollection().Add(SHARDMAP, si);
                }
            }
            #endregion

            #region Shard creation
            // create SHARD_COUNT shards...
            for (int i = 0; i < lServersShards.Count; i++)
            {
                #region Database and schema creation
                ssShardMapManager.InitialCatalog = "master";
                if (CreateDatabaseIfNotExists(lServersShards[i].ConnectionString, SHARD_DB_PREFIX + i))
                {
                    lServersShards[i].InitialCatalog = SHARD_DB_PREFIX + i;

                    Console.ForegroundColor = ConsoleColor.Yellow;
                    Console.WriteLine("Created shard {0:N0} at {1:S}.", i, lServersShards[i].ConnectionString);

                    #region Create log table
                    using (SqlConnection conn = new SqlConnection(lServersShards[i].ConnectionString))
                    {
                        conn.Open();

                        SqlCommand cmd = new SqlCommand(string.Format("CREATE SCHEMA [{0:S}];", SCHEMA_NAME), conn);
                        cmd.ExecuteNonQuery();

                        cmd = new SqlCommand(string.Format("CREATE TABLE [{0:S}].[{1:S}]({2:S} INT PRIMARY KEY, Entry NVARCHAR(MAX), EventTime DATETIME DEFAULT GETDATE());",
                            SCHEMA_NAME, TABLE_NAME, TABLE_ID), conn);

                        cmd.ExecuteNonQuery();
                    }
                    #endregion
                }
                #endregion

                ssShardMapManager.InitialCatalog = SHARD_DB_PREFIX + i;

                Shard shard;

                #region Store this new db as shard location
                ShardLocation sl = new ShardLocation(lServersShards[i].DataSource, SHARD_DB_PREFIX + i);
                if (!listShardMapMain.TryGetShard(sl, out shard))
                {
                    shard = listShardMapMain.CreateShard(sl);

                    // create shard mapping
                    listShardMapMain.CreatePointMapping(i, shard);
                }
                #endregion
            }

            Console.ForegroundColor = ConsoleColor.Cyan;
            Console.WriteLine("This test will use {0:N0} shards.", lServersShards.Count);
            #endregion

            #region Data insert (multithreaded)
            System.Threading.Thread[] threads = new System.Threading.Thread[iThreads];

            ssShardMapManager.DataSource = "";
            ssShardMapManager.InitialCatalog = "";

            DateTime dtStart = DateTime.Now;
            Console.ForegroundColor = ConsoleColor.Cyan;
            Console.WriteLine("{0:N0} threads starting at {1:S}. Each one will insert {2:N0} rows. Total rows to insert {3:N0}.",
                iThreads, dtStart.ToLongTimeString(), iInsertsPerThread, iInsertsPerThread * iThreads);

            for (int i = 0; i < iThreads; i++)
            {
                System.Threading.ParameterizedThreadStart pts = new System.Threading.ParameterizedThreadStart(ThreadInsert);
                threads[i] = new System.Threading.Thread(pts);
                threads[i].Start(i * iInsertsPerThread);
            }

            for (int i = 0; i < iThreads; i++)
            {
                threads[i].Join();
            }
            #endregion

            DateTime dtEnd = DateTime.Now;
            Console.ForegroundColor = ConsoleColor.Cyan;
            Console.WriteLine("Threads finished at {0:S}.", dtEnd.ToLongTimeString());

            TimeSpan tsTotal = dtEnd - dtStart;
            Console.ForegroundColor = ConsoleColor.White;
            Console.WriteLine("Time taken {0:S}.", tsTotal.ToString());
        }

        private static void ThreadInsert(object par)
        {
            int iSegment = (int)par;

            for (int i = iSegment; i < iSegment + iInsertsPerThread; i++)
            {
                int iServerIndex = i % lServersShards.Count;
                using (SqlConnection conn = listShardMapMain.OpenConnectionForKey(iServerIndex, ssShardMapManager.ConnectionString))
                {
                    SqlCommand cmd = new SqlCommand(
                        @"INSERT INTO [log].[LogEntry]
                                       ([ID]
                                       ,[Entry])
                            SELECT @ID, COUNT(*) FROM [log].[LogEntry] WITH(TABLOCKX)",
                        conn);

                    SqlParameter param = new SqlParameter("ID", System.Data.SqlDbType.Int);
                    param.Value = i;
                    cmd.Parameters.Add(param);

                    cmd.ExecuteNonQuery();
                }
            }
        }

        protected static bool CreateDatabaseIfNotExists(string connectionString, string databaseName)
        {
            using (SqlConnection conn = new SqlConnection(connectionString))
            {
                conn.Open();

                SqlCommand cmd = new SqlCommand(
                    string.Format("SELECT * FROM sys.databases WHERE [name]=\'{0:S}\'", databaseName),
                    conn);

                if (cmd.ExecuteScalar() == null)
                {
                    SqlCommand cmd2 = new SqlCommand(
                        string.Format("CREATE DATABASE [{0:S}];", databaseName),
                        conn);

                    cmd2.ExecuteNonQuery();

                    return true;
                }
                else
                    return false;
            }
        }
    }
}