Document multicore scaling of transactions

[mmnosek]

Type of issue:

We miss documentation of using transactions in parallel threads. It might be very important in high-performance systems but might be tricky. Probably it would be great to write a separate page about it. The original question I started to investigate this topic with was: "Is it possible to use C# Parallel API for UPDATE/DELETE instead of simple foreach loop?"

Links to related github issues:

• https://github.com/Starcounter/level1/issues/4020
• https://github.com/Starcounter/level1/issues/3006#issuecomment-157060380
• https://github.com/Starcounter/level1/issues/3210

Requires some serious input from the core team.

[mmnosek]

@k-rus @bigwad - could you help us with this?

[bigwad]

To make parallel processing coherent - one has to use transaction duplication so that several transactions share the same snapshot. Then the problem is how to partition input set and how to schedule processing on a starcounter thread.
Scheduling on starcounter thread can be done with Scheduling.RunTask or wrapped into task scheduler with simple code: https://github.com/Starcounter/TPCC/blob/36ea18c7847797664383b75f4b26ec435721a6de/TPCC/DbTask.cs#L11

Partitioning is harder because built-in TPL and PLINQ partitioning don't work well with starcounter iterators. The problem is that TPL and PLINQ call Enumerator.MoveNext() on an arbitrary thread (though ensure thread-safe access to iterator), while starcounter disallow using of iterator from several threads.
If we manage to fix this issue with iterator than TPL would work quite good. Employing PLINQ also would be possible but requires attention to manually schedule all delegates on a starcounter thread. This is because PLINQ doesn't support supplying custom task scheduler, unlike TPL.

Until starcounter iterators are fixed I can suggest following partitioning schemas:

• the easiest way of partitioning is if you can statically design effective queries that return disjoint sets. then just run those queries in parallel using Scheduling.RunTask:

Db.Transact( ()=>
{
  var token = Db.CurrentTransactionToken;
  string[] queries = { "<query 1>", "<query 2>", ..., "<query n>" }; //makes sense to generate up to number_of_cores queries

  var tasks = queries.Select( q=>Scheduling.RunTask(
     ()=>{
         Db.Transact( 
               ()=>{ //run query q and process result set },
               new Db.Advanced.TransactOptions { source_token = token });
    });
  
  Task.WaitAll(tasks.ToArray());
});

• another easy solution would be to enumerate all objects in one thread and launch a task on every object. This approach has a problem of the potentially unbound growth of the number of tasks if processing is slow. Obviously, you can batch objects to run fewer tasks.

• to cap the number of tasks, one can use a queue:

    Db.Transact( () =>{
        var token = Db.CurrentTransactionToken;

        CancellationTokenSource cts = new CancellationTokenSource();
        BlockingCollection<B> buf = new BlockingCollection<B>(10000);

        var t = Task.Run(
            () => Parallel.ForEach(
                  buf.GetConsumingEnumerable(),
                  new ParallelOptions { TaskScheduler = new DbTaskScheduler()},
                  (b) => Db.Transact(
                                 () =>{ /*process  object b*/ },       
                                 new Db.Advanced.TransactOptions { source_token = token })));

        t.ContinueWith((_) => cts.Cancel(), TaskContinuationOptions.OnlyOnFaulted);

        try
        {
            foreach (var b in Db.SQL<B>("select b from B b"))
            {
                    buf.Add(b, cts.Token);
            }
        }
        catch (OperationCanceledException)
        {
        }
        finally
        {
            buf.CompleteAdding();
        }

        t.GetAwaiter().GetResult();

    });

• an alternative to manual queue would be using of TPL dataflow. What is nice about it is that it has built-in batching capabilities:

            Db.Transact(() => {
                var token = Db.CurrentTransactionToken;

                ActionBlock<B[]> ab = new ActionBlock<B[]>((B[] l) => 
                    Db.Transact(() =>{/*process array of B's here*/},
                                       new Db.Advanced.TransactOptions { source_token = token }),
                    new ExecutionDataflowBlockOptions { 
                         MaxDegreeOfParallelism = ExecutionDataflowBlockOptions.Unbounded, 
                         BoundedCapacity = 10000, TaskScheduler= new DbTaskScheduler()}
                 );

                BatchBlock<B> ab2 = new BatchBlock<B>(100, new GroupingDataflowBlockOptions { BoundedCapacity = 10000 });

                ClassLibrary1.Class1.CompletionForBlock(ab).ContinueWith(
                            (t) =>(ab2 as IDataflowBlock).Fault(t.Exception), TaskContinuationOptions.OnlyOnFaulted
                            );

                ab2.LinkTo(ab);

                foreach(var b in Db.SQL<B>("select b from B b"))
                {
                    ab2.SendAsync(b).Wait();
                    if (ClassLibrary1.Class1.CompletionForBlock(ab2).IsCompleted)
                        break;
                }

                ab2.Complete();
                ClassLibrary1.Class1.CompletionForBlock(ab2).Wait();
                ab.Complete();
                ClassLibrary1.Class1.CompletionForBlock(ab).Wait();
            });

The final note about write transactions. One certainly can do updates/deletes as described, but has to note two caveats:

1. Writes are serialized in the kernel, so update of the database wouldn't be parallelized regardless. The only thing that will run in parallel is some preparatory work if you have one.
2. Each duplicated transaction is a separate transaction. Indeed. Hence writes won't be atomic if made in several parallel transactions.

[mmnosek]

@bigwad, many thanks for the very extensive explanation! I would sum it up like this (in simple words):

• writes don't make sense because at the end everything serialized in kernel anyway and you lose atomicity as in fact transactions will be executed separately
• For reads it makes sense, but you cannot relay either on TPL nor PLINQ. You need to do partitioning manually and schedule all tasks on Starcounter scheduler and use common transaction token to use the same snapshot for all threads.
• Here are examples:

Is this correct?

If we manage to fix this issue with iterator than TPL would work quite right.

Do you know if we have the issue for it somewhere, is it planned to be fixed? Do you think it's big work?

@miyconst, do you have anything to add based on your experience, e.g. in the Optym POC project?

[bigwad]

For reads it makes sense, but you cannot relay either on TPL nor PLINQ. You need to do partitioning manually and schedule all tasks on Starcounter scheduler and use common transaction token to use the same snapshot for all threads.

Not exactly. You can use TPL and PLINQ if you prepare data so that TPL's (or PLINQ) partitioning doesn't call starcounter iterator directly. I did so in the example with the queue implemented via BlockingCollection: I filled the queue in starcounter thread and passed it to TPL for partitioning.

If we manage to fix this issue with iterator than TPL would work quite right.

Do you know if we have the issue for it somewhere, is it planned to be fixed? Do you think it's big work?

Haven't seen such issue. Regarding implementation - can't say right away.