class Program
        {
            private static Object ObjectLock = new Object();

            private static DataTable table;
            private static SqlBulkCopy bulkCopy;
            private static String[] Records;
            private static CultureInfo ruRuCulture;
            private static CultureInfo ci;

            private static DataRow[] rows;
            private static Semaphore semaphore = new Semaphore(2, 2);

            private class Variables
            {
                public Int32 IndexOfRead { get; set; }
                public Int32 IndexOfWrite { get; set; }
            }

            static void MakeTable(DataTable table)
            {

            }
            
            static object [][] parsePortion(object arg)
            {
                var lines = arg as string[];
                var result = new object[lines.Length][];
                int i=0;
                foreach (var line in lines)
                {
                    result[i] = importRecord(line);
                    i++;
                }
                return result;
            }
            static Task<object[][]> processParsePortion(StreamReader reader, int portionSize, TaskFactory<object [][]> taskFactory, ref bool hasMore)
            {
                var portion = new string[portionSize];
                int i = 0;
                var s = reader.ReadLine();
                while (s!=null && i<portionSize)
                {
                    portion[i] = s;
                    i++;
                    s = reader.ReadLine();
                }
                             
                hasMore =  s != null;
                return (i > 0) ?
                    taskFactory.StartNew(parsePortion, portion)
                    : null;
            }
            static DataTable makeDataTable(Task<object [][]>[] parseTasks)
            {
                DataTable result = new DataTable();
                MakeTable(result);
                foreach (var task in parseTasks)
                {
                    foreach (var row in task.Result)
                    {
                        var resultRow = result.NewRow();
                        resultRow.ItemArray = row;
                        result.Rows.Add(resultRow);
                    }
                }
                return result;
            }
            static void writeToDB(Task<DataTable> task)
            {
                lock (ObjectLock)
                {
                    bulkCopy.WriteToServer(task.Result);
                }
                semaphore.Release();
            }
            static Boolean processWritePortion(StreamReader reader, int writePortionSize, int portionSize)
            {
                bool hasMore = true; 
                semaphore.WaitOne();
                var taskFactory = new TaskFactory<object [][]>();
                var parseTasks = new Task<object [][]>[writePortionSize];
                int i=0;
                while (i < writePortionSize && hasMore)
                {
                    var currentTask = processParsePortion(reader, portionSize, taskFactory, ref hasMore);
                    if (currentTask!=null)
                    {
                        parseTasks[i] = currentTask;
                        i++;
                    }
                }
                if (i > 0)
                {
                    if (writePortionSize > i)
                    {
                        Array.Resize<Task<object[][]>>(ref parseTasks, i);
                    }
                    var dataTaskFactory = new TaskFactory<DataTable>();
                    dataTaskFactory.ContinueWhenAll<object[][]>(parseTasks, makeDataTable)
                                   .ContinueWith(writeToDB).Start();
                }
                else 
                {
                    semaphore.Release();
                }
                return hasMore;
            }
            static void Main(string[] args)
            {

                String SqlServer, SqlDatabase;

                Int32 ReadBatchSize, WriteBatchSize;
                Int32 IndexOfReadRecords, IndexOfWriteRecords;
                Int64 TotalRecords;
                Task[] tasks;

                DateTime StartTimer;
                DateTime EndTimer;

                try
                {

                    ci = CultureInfo.InvariantCulture.Clone() as CultureInfo;
                    ci.NumberFormat.NumberDecimalSeparator = ".";
                    ruRuCulture = CultureInfo.GetCultureInfo("ru-Ru");

                    SqlServer = args[0];
                    SqlDatabase = args[1];

                    table = new DataTable();
                    bulkCopy = new SqlBulkCopy(GetConnectionString(SqlServer, SqlDatabase));

                    SqlServer = args[0];
                    SqlDatabase = args[1];

                    ReadBatchSize = Convert.ToInt32(args[2]);
                    WriteBatchSize = Convert.ToInt32(args[3]);
                    Records = new String[ReadBatchSize];
                    tasks = new Task[ReadBatchSize];

                    
                    
                    ColumnMappings(bulkCopy);
                    

                    TotalRecords = 0;

                    using (bulkCopy)
                    {

                        using (var sr = new StreamReader(args[4]))
                        {

                            StartTimer = DateTime.Now;
                            IndexOfReadRecords = 0;
                            IndexOfWriteRecords = 0;
                            while (processWritePortion(sr, 2, 1000))
                            {
                                Console.Write("Portion");
                            }

                        }
                        for (var i = 0; i < 2; i++)
                        {
                            semaphore.WaitOne();
                        }
                        semaphore.Dispose();

                        bulkCopy.Close();

                        EndTimer = DateTime.Now;
                        Console.WriteLine("Records: {0}\nStartTime: {1}\nExecution Time: {2}", TotalRecords, StartTimer.ToString("yyyy-MM-dd HH:mm:ss"), new TimeSpan(EndTimer.Ticks - StartTimer.Ticks));
                        Console.ReadKey();

                    }
                }
                catch (AggregateException ae)
                {
                    foreach (var e in ae.InnerExceptions)
                    {
                        Console.WriteLine("Import error:");
                        Console.WriteLine(e.Message);
                        Console.ReadKey();
                    }
                }
                catch (Exception e)
                {
                    Console.WriteLine("Import error:");
                    Console.WriteLine(e.Message);
                    Console.ReadKey();
                }

            }

            private static void ColumnMappings(SqlBulkCopy bulkCopy)
            {
                throw new NotImplementedException();
            }

            private static string GetConnectionString(string SqlServer, string SqlDatabase)
            {
                throw new NotImplementedException();
            }


            private static object[] importRecord(string line)
            {

                Object[] rowArray = new Object[50];
                String[] Columns = new String[11];
                Int32[] Value_ID = new Int32[20];
                Decimal[] Value_Changes = new Decimal[20];

                fSplit(line, '|', Columns);

                rowArray[0] = Columns[0];
                rowArray[1] = Convert.ToDateTime(Columns[1], ruRuCulture);
                rowArray[2] = Columns[2];
                rowArray[3] = Columns[3];
                rowArray[4] = Decimal.Parse(Columns[4], ci);
                rowArray[5] = Decimal.Parse(Columns[5], ci);
                rowArray[6] = Decimal.Parse(Columns[6], ci);
                rowArray[7] = Columns[7];
                rowArray[8] = Columns[8];
                rowArray[9] = String.IsNullOrEmpty(Columns[9]) ? Convert.DBNull : Decimal.Parse(Columns[9], ci);

                String[] Values = Columns[10].Split(']');

                Int32 Id = 10;

                foreach (String Value in Values)
                {
                    if (!String.IsNullOrEmpty(Value))
                    {
                        String[] Value_structure = Value.Split('~');
                        rowArray[Id + 0] = Int32.Parse(Value_structure[0].Remove(0, 1));
                        rowArray[Id + 20] = Decimal.Parse(Value_structure[2], ci);
                        Id++;
                    }
                }

                return rowArray;
            }

            private static void fSplit(String src, char delim, String[] output)
            {
                int index = 0;
                int lindex = 0;
                int columnID = 0;
                int arrayID = 0;
                int[] array = { 0, 1, 3, 4, 5, 12, 16, 23, 25, 27, 30 };
                int i;

                while ((index = src.IndexOf(delim, lindex)) != -1)
                {
                    for (i = 0; i < 11 && array[i] != columnID; i++) ;
                    if (i < 11) output[arrayID++] = src.Substring(lindex, index - lindex);
                    if (columnID >= 30) break;
                    lindex = index + 1;
                    index++;
                    columnID++;
                }
            }


        }

using System;
using System.Threading.Tasks;
using System.IO;
using System.Threading;
using System.Data;
using System.Data.SqlClient;
using System.Collections.Generic;
using System.Collections;
using System.Linq;
using System.Globalization;


namespace ConsoleApplication4
{
         
        class VirtualTextReader: TextReader
        {
            int line;
            int count;
            public VirtualTextReader(int count)
            {
                this.count = count;
            }
            public override string ReadLine()
            {
                line++;
                return line > count ? null : "1|2222|333|4444|{A:one:B}{C:two:D}|666666";
            }
        }
        class Program
        {
            private static Object ObjectLock = new Object();

            private static DataTable table;
            private static SqlBulkCopy bulkCopy;
            private static String[] Records;
            private static CultureInfo ruRuCulture;
            private static CultureInfo ci;

            
            
            private class Variables
            {
                public Int32 IndexOfRead { get; set; }
                public Int32 IndexOfWrite { get; set; }
            }

 
            
            static object [][] parsePortion(object arg)
            {
                var lines = arg as string[];
                var result = new object[lines.Length][];
                int i=0;
                foreach (var line in lines)
                {
                    result[i] = importRecord(line);
                    i++;
                }
                return result;
            }
            static Task<object[][]> processParsePortion(TextReader reader, int portionSize, TaskFactory<object [][]> taskFactory, ref bool hasMore)
            {
                var portion = new string[portionSize];
                int i = 0;
                string s=null;
                do
                {
                    s = reader.ReadLine();
                    if (s != null)
                    {
                        portion[i] = s;
                        i++;
                    }
                } while (s != null && i < portionSize);

                if (i < portionSize)
                {
                    Array.Resize<string>(ref portion, i);
                }
                             
                hasMore =  s != null;
                return (i > 0) ?
                    taskFactory.StartNew(parsePortion, portion)
                    : null;
            }
            
            static DataTable makeDataTable(Task<object [][]>[] parseTasks)
            {
                // rethrow exceptions 
                // http://reedcopsey.com/2010/07/19/parallelism-in-net-part-18-task-continuations-with-multiple-tasks/
                Task.WaitAll(parseTasks);
                DataTable result = new DataTable();
                MakeTable(result);
                foreach (var task in parseTasks)
                {
                    foreach (var row in task.Result)
                    {
                        var resultRow = result.NewRow();
                        resultRow.ItemArray = row;
                        result.Rows.Add(resultRow);
                    }
                }
                return result;
            }
            static void writeToDB(Task<DataTable> task)
            {
                task.Wait();
                lock (ObjectLock)
                {
                    bulkCopy.WriteToServer(task.Result);
                }
            }
            static Boolean processWritePortion(TextReader reader, int writePortionSize, int portionSize, ref Task task)
            {
                bool hasMore = true; 
                var taskFactory = new TaskFactory<object [][]>();
                var parseTasks = new Task<object [][]>[writePortionSize];
                int i=0;
                do
                {
                    var currentTask = processParsePortion(reader, portionSize, taskFactory, ref hasMore);
                    if (currentTask != null)
                    {
                        parseTasks[i] = currentTask;
                        i++;
                    }
                }
                while (i < writePortionSize && hasMore);

                if (i > 0)
                {
                    if (writePortionSize > i)
                    {
                        Array.Resize<Task<object[][]>>(ref parseTasks, i);
                    }
                    var dataTaskFactory = new TaskFactory<DataTable>();
                    task = dataTaskFactory.ContinueWhenAll<object[][]>(parseTasks, makeDataTable)
                                   .ContinueWith(writeToDB);
                }
                else 
                {
                    task = null;
                }
                return hasMore;
            }
            static void Main(string[] args)
            {

                String SqlServer, SqlDatabase;

                Int32 ReadBatchSize, WriteBatchSize;
                Int64 TotalRecords;
                HashSet<Task> tasks = new HashSet<Task>();

                DateTime StartTimer;
                DateTime EndTimer;

                try
                {

                    ci = CultureInfo.InvariantCulture.Clone() as CultureInfo;
                    ci.NumberFormat.NumberDecimalSeparator = ".";
                    ruRuCulture = CultureInfo.GetCultureInfo("ru-Ru");

                    SqlServer = args[0];
                    SqlDatabase = args[1];
                    SqlServer = args[0];
                    SqlDatabase = args[1];

                    bulkCopy = new SqlBulkCopy(GetConnectionString(SqlServer, SqlDatabase));


                    ReadBatchSize = Convert.ToInt32(args[2]);
                    WriteBatchSize = Convert.ToInt32(args[3]);
                    Records = new String[ReadBatchSize];

                    
                    
                    ColumnMappings(bulkCopy);
                    

                    TotalRecords = 0;

                    using (bulkCopy)
                    {
                        
                        using (var sr = new VirtualTextReader(8000000))//new StreamReader(args[4]))
                        {

                            StartTimer = DateTime.Now;

                            Task currentTask = null;
                            while (processWritePortion(sr, 2, 5000, ref currentTask))
                            {
                                if (currentTask != null)
                                {
                                    tasks.Add(currentTask);
                                    currentTask = null;
                                }
                                if (tasks.Count > 1)
                                {
                                    Task.WaitAny(tasks.ToArray());
                                    Task.WaitAll(tasks.Where(t => t.IsFaulted).ToArray());
                                    tasks.RemoveWhere(task => task.IsCompleted);
                                }
                            }
                            
                            if (currentTask != null)
                            {
                                tasks.Add(currentTask);
                            }
                            Task.WaitAll(tasks.ToArray());
                        }

                        bulkCopy.Close();

                        EndTimer = DateTime.Now;
                        Console.WriteLine("Records: {0}\nStartTime: {1}\nExecution Time: {2}", TotalRecords, StartTimer.ToString("yyyy-MM-dd HH:mm:ss"), new TimeSpan(EndTimer.Ticks - StartTimer.Ticks));
                        Console.ReadKey();

                    }
                }
                catch (AggregateException ae)
                {
                    foreach (var e in ae.Flatten().InnerExceptions)
                    {
                        Console.WriteLine("Import error:");
                        Console.WriteLine(e.Message);                        
                        Console.ReadKey();
                    }
                }
                catch (Exception e)
                {
                    Console.WriteLine("Import error:");
                    Console.WriteLine(e.Message);
                    Console.ReadKey();
                }

            }

            private static void MakeTable(DataTable table)
            {

                table.Columns.Add("Col1", System.Type.GetType("System.String"));
                table.Columns.Add("Col2", System.Type.GetType("System.Int32"));
                table.Columns.Add("Col3", System.Type.GetType("System.Decimal"));
                table.Columns.Add("Col4", System.Type.GetType("System.String"));
                table.Columns.Add("Col5", System.Type.GetType("System.String"));
                table.Columns.Add("Col6", System.Type.GetType("System.String"));

            }

            private static string GetConnectionString(String SqlServer, String SqlDatabase)
            {
                return "Data Source=" + SqlServer + "; " +
                    " Integrated Security=true;" +
                    "Initial Catalog=" + SqlDatabase + ";";
            }


            private static void ColumnMappings(SqlBulkCopy bulkCopy)
            {
                bulkCopy.DestinationTableName = "tb_test";
                bulkCopy.ColumnMappings.Add("col1", "col1");
                bulkCopy.ColumnMappings.Add("col2", "col2");
                bulkCopy.ColumnMappings.Add("col3", "col3");
                bulkCopy.ColumnMappings.Add("col4", "col4");
                bulkCopy.ColumnMappings.Add("col5", "col5");
                bulkCopy.ColumnMappings.Add("col6", "col6");
            }

            private static String[] parseComplexColumn(string value)
            {
                var g = System.Text.RegularExpressions.Regex.Match(value, 
                    @"\{(.*):.*:(.*)}{(.*):.*:(.*)\}").Groups;
                return new String[] { g[1].Value, g[2].Value, g[3].Value, g[4].Value };
            }

            /// <summary>
            /// Создай проект
            /// пусть в цикле передается одна и та же строка "1|22|333|4444|{A:one:B}{C:two:D}|666666"
            /// в таблицу в отдельные столбцы должны быть загружены
            /// 22,4444,A,B,C,D
            /// </summary>
            /// <param name="line"></param>
            /// <returns></returns>
            private static object[] importRecord(string line)
            {
                var columns = line.Split('|');
                var result = new Object[6];
                result[0] = columns[1];
                result[1] = Int32.Parse(columns[2]);
                result[2] = Decimal.Parse(columns[3], ci);

                var complex = parseComplexColumn(columns[4]);
                result[3] = complex[0];
                result[4] = complex[1];
                result[5] = complex[2];
                //result[6] = complex[3];

                return result;
            }
            
            private static object[] importRecord_old(string line)
            {

                Object[] rowArray = new Object[50];
                String[] Columns = new String[11];
                Int32[] Value_ID = new Int32[20];
                Decimal[] Value_Changes = new Decimal[20];

                fSplit(line, '|', Columns);


                rowArray[0] = Columns[0];
                rowArray[1] = Convert.ToDateTime(Columns[1], ruRuCulture);
                rowArray[2] = Columns[2];
                rowArray[3] = Columns[3];
                rowArray[4] = Decimal.Parse(Columns[4], ci);
                rowArray[5] = Decimal.Parse(Columns[5], ci);
                rowArray[6] = Decimal.Parse(Columns[6], ci);
                rowArray[7] = Columns[7];
                rowArray[8] = Columns[8];
                rowArray[9] = String.IsNullOrEmpty(Columns[9]) ? Convert.DBNull : Decimal.Parse(Columns[9], ci);

                String[] Values = Columns[10].Split(']');

                Int32 Id = 10;

                foreach (String Value in Values)
                {
                    if (!String.IsNullOrEmpty(Value))
                    {
                        String[] Value_structure = Value.Split('~');
                        rowArray[Id + 0] = Int32.Parse(Value_structure[0].Remove(0, 1));
                        rowArray[Id + 20] = Decimal.Parse(Value_structure[2], ci);
                        Id++;
                    }
                }

                return rowArray;
            }

            private static void fSplit(String src, char delim, String[] output)
            {
                int index = 0;
                int lindex = 0;
                int columnID = 0;
                int arrayID = 0;
                int[] array = { 0, 1, 3, 4, 5, 12, 16, 23, 25, 27, 30 };
                int i;

                while ((index = src.IndexOf(delim, lindex)) != -1)
                {
                    for (i = 0; i < 11 && array[i] != columnID; i++) ;
                    
                    if (i < 11) 
                        output[arrayID++] = src.Substring(lindex, index - lindex);
                    
                    if (columnID >= 30) 
                        break;
                    
                    lindex = index + 1;
                    index++;
                    columnID++;
                }
            }
        }

}