3
\$\begingroup\$

I need split the list I get from the select into multiple list so I can use Parallel.For() to generate the next query in different threads.

  • I create the main list
  • Create the sub list
  • Keep filling sub list until reach max capacity.
  • Add sublist to main list
    • Create new sub list
  • When finish the loop save the last sublist

This works but I bet there is an easy / clean way to do this, I can change the db if needed.

List<List<long>> result = new List<List<long>>();
using (var conn = new NpgsqlConnection(strConnection))
{
    conn.Open();
    using (var cmd = new NpgsqlCommand())
    {
        cmd.Connection = conn;
        cmd.CommandText = "SELECT route_source_id " +                                      
                          "FROM route_sources " +
                          "WHERE has_route IS NULL " +          
                          "LIMIT 1000; ";

        using (var reader = cmd.ExecuteReader())
        {
            const long list_size = 100;                

            long row_id = 0;
            List<long> route_list = new List<long>();

            while (reader.Read())
            {                                
               if (row_id++ > list_size) // Add the sublist to the result
               {
                   result.Add(route_list); 
                   row_id = 0;
               }

               if (row_id == 0)  // Create new sublist
               {
                  route_list = new List<long>();
               }

               if (!reader.IsDBNull(0)) // fill the sublist
               {
                    route_list.Add(reader.GetInt64(0));                                
               }
            }

            if (route_list.Count > 0 ) // add the last sublist
            {
               result.Add(route_list);
            }
        }
   }
 }
 return result;

EDIT: aditional part of the process.

I have ticker to check the db for pending id to be process

// read 1000 rows split into 10 list, using above function
List<List<long>> route_pending = getRouteRange(); 
long total_rows = 0;

while (route_pending.Count > 0) 
{
     foreach (List<long> range in route_pending)
     {
         total_rows += range.Count;                    
     }
     routeParallelGeneration(route_pending);
     route_pending = getRouteRange(); // keep reading next 1000 rows
}

And the parallel function, here I call my db storeproc with an array parameter. So max 10 connections with array of size 100. Here was where got issues with the excesive number of connections and decide split the data into 10 lists

public void routeParallelGeneration(List<List<long>> route_pending)
{
    Action<List<long>> forEachLoop = route_source_id => //Begin definition of forLoop
    {
        using (var conn = new NpgsqlConnection(strConnection))
        {
            conn.Open();
            using (var cmd = new NpgsqlCommand())
            {
                cmd.Connection = conn;
                cmd.CommandText = "SELECT avl_db.process_route(@route_source)";
                cmd.CommandType = CommandType.Text;

                NpgsqlParameter p = new NpgsqlParameter("@route_source", NpgsqlDbType.Array | NpgsqlDbType.Bigint);
                p.Value = route_source_id;
                cmd.Parameters.Add(p);
                cmd.ExecuteNonQuery();
            }
        }
    };
    Parallel.ForEach(route_pending, forEachLoop);
}
\$\endgroup\$
1
\$\begingroup\$

Calculating ranges

I need split the list I get from the select into multiple list so I can use Parallel.For() to generate the next query in different threads.

Nope. You don't need it. Use Parallel.ForEach that automatically can distribute the load on multiple threads if necessary or use the Partitioner to calculate the ranges e.g.:

var partitions = Partitioner
    .Create(fromInclusive: 0, toExclusive: 100, rangeSize: 30)
    .GetDynamicPartitions();

the result

Item1  Item2

    0     30 
   30     60 
   60     90 
   90    100

Example

Here's an example of how it could be implemented:

  • The first change is that you don't split the list but pass all ids to this method. It'll partition it and will use these partitions for reading from the id-list.

  • The second change is that Parallel.ForEach doesn't run over the ids but over the partitions. Each partition (index range) is passed to the body.

  • The third change is that you initialize the connection and the command once per thread via the localInit. It'll be disposed/closed by localFinally.

  • The fourth change is that you don't have to recreate the command each time. All you have to do is to set the new parameter value and call ExecuteNonQuery.

Code:

public void routeParallelGeneration(List<long> route_pending)
{
    var parallelOptions = new ParallelOptions
    {
        MaxDegreeOfParallelism = Environment.ProcessorCount
    };

    var partitions =
        Partitioner
            .Create(
                fromInclusive: 0,
                toExclusive: route_pending.Count,
                rangeSize: 100
            )
            .GetDynamicPartitions();

    Parallel.ForEach(
        source: partitions,
        parallelOptions: parallelOptions,
        localInit: () => 
        {
            var conn = new NpgsqlConnection(strConnection);
            conn.Open();

            var cmd = new NpgsqlCommand();
            cmd.Connection = conn;
            cmd.CommandText = "SELECT avl_db.process_route(@route_source)";
            cmd.CommandType = CommandType.Text;

            NpgsqlParameter p = new NpgsqlParameter("@route_source", NpgsqlDbType.Array | NpgsqlDbType.Bigint);
            cmd.Parameters.Add(p);

            return new { Connection = conn, Command = cmd };
        },
        body: (source, state, local) =>
        {
            local.Command.Parameters["@route_source"].Value = route_pending.GetRange(
                index: source.Item1, 
                count: source.Item2 - source.Item1 - 1
            );
            local.Command.ExecuteNonQuery();
            return local;
        },
        localFinally: local =>
        {
            local.Connection?.Dispose();
            local.Command?.Dispose();
        }
    );
}

Improvable

Don't do everything in a single method. Write one method to get the data from the server and another one to process it and/or calculate the partitions. You need to pay more attentions to the separation of concerns. Mixing all features in a single method makes it super hard to maintain and to test.

\$\endgroup\$
  • \$\begingroup\$ Comments are not for extended discussion; this conversation has been moved to chat. \$\endgroup\$ – Mathieu Guindon Feb 17 '17 at 14:57
  • \$\begingroup\$ Im moving this code to one app in VS2013 and got an error on the line body: (source, state, local) => say Error 5 Delegate 'System.Func<System.Tuple<long,long>,System.Threading.Tasks.ParallelLoopState,long,AnonymousType#1,AnonymousType#1>' does not take 3 arguments both are using framework 4.5.2 shouldnt be any difference between 2013 and 2015 ? ` \$\endgroup\$ – Juan Carlos Oropeza Jul 2 '17 at 21:06
2
\$\begingroup\$

Pretty minor but you don't need the if (row_id == 0)

if (row_id++ > list_size) // Add the sublist to the result
{
    result.Add(route_list); 
    row_id = 0;
    route_list = new List<long>();
}

Also it seem like you should also test for

if (route_list.Count > 0 ) 

Why not just put where route_source_id is not null in the query?

Connection, Command, and Reader behave the same with many DB so this is pretty portable.

I think you can use

conn.CreateCommand()

Long is not required for row_id or list_size

Why bump row_id++ if there is no value?

In summary

    using (var reader = cmd.ExecuteReader())
    {
        const int list_size = 100;                
        List<long> route_list = new List<long>(list_size);

        while (reader.Read())
        {                                
           if (route_list.Count >= list_size) // Add the sublist to the result
           {
               result.Add(route_list); 
               route_list = new List<long>(list_size);
           }
           route_list.Add(reader.GetInt64(0));
        }

        if (route_list.Count > 0 ) // add the last sublist
        {
           result.Add(route_list);
        }
    }
\$\endgroup\$
  • \$\begingroup\$ route_source_id is PK for that table, cant be null. \$\endgroup\$ – Juan Carlos Oropeza Feb 13 '17 at 19:31
  • \$\begingroup\$ Then why are testing for !reader.IsDBNull(0) ? \$\endgroup\$ – paparazzo Feb 13 '17 at 19:32
  • \$\begingroup\$ Ups ... I just copy paste from a sample :$ \$\endgroup\$ – Juan Carlos Oropeza Feb 13 '17 at 19:34
  • \$\begingroup\$ Ohh I remember In the previous sample the query was MAX(id) and that one could get null \$\endgroup\$ – Juan Carlos Oropeza Feb 13 '17 at 19:36
  • \$\begingroup\$ Thanks for your help, this definitely is much cleaner than my first version and I learn a lot. But I found na even simpler solution doing the list in the query. Check my answer. \$\endgroup\$ – Juan Carlos Oropeza Feb 14 '17 at 14:15
0
\$\begingroup\$

I found this version. I create one query to return an array. here just hard code the size 100.

using (var cmd = new NpgsqlCommand())
{
    cmd.Connection = conn;
    cmd.CommandText = "with cte as ( " +
                      "  SELECT route_source_id::bigint, " +
                      "          (row_number() OVER() - 1) / 100 AS grp " +
                      "  FROM avl_db.route_sources " +
                      "  WHERE  has_route IS NULL " +
                      "  limit 1000 " +
                      ") " +
                      "SELECT array_agg(route_source_id)" +
                      "FROM cte " +
                      "GROUP BY grp " +
                      "ORDER BY grp; ";
    using (var reader = cmd.ExecuteReader())
    {
       while (reader.Read())
       {
           long[] sublist = (long[])reader.GetValue(0);
           result.Add(sublist.ToList());
       }
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.