Using new parallel ExecutionContext

Help & Support
21

From Queueing Theory the optimum queue structure has 1 queue feeding n servers.
This is faster than using n queues feeding n servers.

Thus the optimum parallel process has each thread|server fed input from 1 queue|producer as they become available. This is the classical Bank Teller|Grocery Store problem.

There are n bank teller stations (or grocery store cashiers). The sub-optimum strategy queues 1 line for each station (n queues). The optimal strategy is 1 queue feeding the n threads|servers.
(The U.S. grocery chain Trader Joe’s is famous for doing this, which I personally frequent.)

Thus the optimal parallel process has each spawn element sent to a current available thread. So while this code works correctly (no hanging, small constant memory use), it’s sub-optimal.

  ncpus = System.cpu_count
  wg = WaitGroup.new(ncpus)
  channel = Channel(Tuple(Int32,Int32)).new(ncpus)
  threads = Fiber::ExecutionContext::Parallel.new("threads", ncpus)

  threadscnt = Atomic.new(0)
  ncpus.times do
    threads.spawn do
      while inputs = channel.receive?
        r_hi, i = inputs
        lastcousins[i], cnts[i] = cousins_sieve(r_hi, ...)
        print "\r#{threadscnt.add(1)} of #{pairscnt} cousinpairs done"
      end
    ensure
      wg.done
  end end

  rescousins.each_with_index { |r_hi, i| channel.send({r_hi, i}) }
  channel.close
  wg.wait

Here the ncpus threads are spawned in parallel, then each waits to receive data.
This is the classical sub-optimal n queues for n servers strategy.

Queueing Theory, and my timing tests, establish the code below is faster.
The pairscnt jobs are fed first-come-first-serve to the n threads.
They both run in similar times for the same inputs, and are 10-15% faster as
the input values (number of spawned elements) grow.

Not only being shorter code, they’re more intuitive, and elegant.

  Fiber::ExecutionContext.default.resize(maximum: System.cpu_count)
  done = Channel(Nil).new(pairscnt)

  threadscnt = Atomic.new(0)
  restwins.each_with_index do |r_hi, i|
    spawn do
      lastwins[i], cnts[i] = twins_sieve(r_hi, ...)
      print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      done.send(nil)
  end end
  pairscnt.times { done.receive }
  print "\r#{pairscnt} of #{pairscnt} twinpairs done"

----------------------------------------------------------------------------------

  Fiber::ExecutionContext.default.resize(maximum: System.cpu_count)
  wg = WaitGroup.new(pairscnt)

  threadscnt = Atomic.new(0)
  restwins.each_with_index do |r_hi, i|
    wg.spawn do
      lastwins[i], cnts[i] = twins_sieve(r_hi, ...)
      print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      wg.done
  end end
  wg.wait

However, they also hang on certain (sometime different) input values, and use increasing memory as the inputs grow, until stabilizing on some larger constant memory amount (I have 64GB).

I think what is happening is once a spawned input|fiber is sent to an available thread it doesn’t seem to immediately terminate when the thread finishes, and free up that memory. And sometimes it
appears there is fiber-to-thread contention, which causes the hanging.

Ideally you only need about ncpu number of the wg to cycle through, to be fed to the ncpu threads. Each spawn doesn’t need a separate fiber, there only need to be ncpu fibers active at any one time.

Because I don’t understand the mechanics of fiber creation|destruction I don’t know if you can create a finite fiber pool to cycle through, versus creating a separate new fiber for each spawn.

Anyway, values up to at least 10_000_000_000_000 work with the new code with no problems, with faster times and similar small memory footprints.

I hope these observations are helpful.

22

This is wrong. In your grocery store checkout analogy, the spawned worker fibers are the cashiers. There’s one line (channel) for customers (input values) and when the cashier (worker fiber) finishes helping the customer (running the sieve with the input values), the cashier beckons the next customer from the line to help them (channel.receive?).

I’m not sure why you’re running into more scaling problems with the worker approach than your “make a fiber for every input value” approach, but it’s not because of some queueing theory issue. It doesn’t even make sense to apply queueing theory to your application in the first place. All the input values are known by the time you begin the parallelized work, which makes it a totally different kind of problem than what queueing theory was developed for.

23

For this problem, there are a fixed number of threads (for a given system) and the algorithm doesn’t know how many spawns (customers) it needs to process, like a bank or grocery store. The single line feeds the threads on a first-come-first-serve basis, to keep all the threads optimally occupied at a given instance. This is a classical Queueing Theory example.

Crystal|Go prominently use fibers for their customer lines, but you don’t have to e.g. Rust|Rayon.

The point is, the new code is faster because it parcels out the most current input to an availaable thread, instead of the threads waiting to for their inputs individually. It’s more inefficient to create nqueues to feed n threads than1 queue to feed n threads. If you’ve been to a Trader Joe’s, et al, that uses a single queue system, you’d see how amazingly fast the long lines go.

24

By the time the parallelized portion of your code runs, the program has all the input for processing in parallel that it will ever have. That’s simply not the same as the realtime queueing problems in a grocery store (where customers may show up and check out at any point throughout the day) or internet traffic routing. It doesn’t make it a queueing problem just because you didn’t write your code to take advantage of the available information.

What I’m telling you is that the worker+channel implementation is the implementation analogous to the Trader Joe’s checkout line with a single line and multiple cashiers taking customers from that one line. Your “spawn as many fibers as there are inputs” implementation is more like, I don’t know, if the grocery store had a fixed number of cash registers but knew how many total customers they would ever have and hired one cashier for each customer (only to lay them off after they checked out that one customer) and had the cashier get in line with their assigned customer. It’s an absurd analogy, but, like I’ve already said, your application isn’t actually analogous to a grocery store checkout line.

25

Let’s look at what happens for a thread under both scenarios.

  ncpus.times do
    threads.spawn do
      while inputs = channel.receive?
        r_hi, i = inputs
        lastwins[i], cnts[i] = twins_sieve(r_hi, ...)
        print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      end
    ensure
      wg.done
  end end

Under this scenario, after each thread is started, they spend some chunk of their time querying if they have data to process, using the while loop. It is sub-optimal to have each thread spend time determining if it has data. This also means each thread spends time determining when they can terminate. This creates unnecessary duplication of assessing the input queue data availability.

Its main advantage is it uses one channel|queue sent to each of the ncpus threads.
This minimizes fiber|memory use, at the expense of forcing queue processing to the threads.
This also appears to not cause any input data hanging|contention issue.

However, this design is sub-optimal for the purpose of maximum data processing speed.
To maximize speed, you only want the threads to process the data that’s delivered to them.
They either process data when given some, or are inactive otherwise.

Thus for maximum speed, each thread is provided data on a first-come-first-serve basis.
Thus, no thread has to process the input queue (or know anything about it).
All the threads do is process data, then become inactive until sent more data.
This is what this code does (as well as the other language versions.)

  restwins.each_with_index do |r_hi, i|
    wg.spawn do
      lastwins[i], cnts[i] = twins_sieve(r_hi, ...)
      print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      wg.done
  end end
  wg.wait

Here, the conceptual input queue data is spawned to whatever thread is available until all the data is processed. This DRYs (don’t repeat yourself) out the input queue processing.

This version is a great improvement using EC versus concurrency with channels, as it eventually stabilizes memory use size, which doesn’t grow unbounded.

Here wg = WaitGroup.new(pairscnt) seems to be the cause of the memory size use.

As the input ranges grow, larger modular groups size are used, which increases pairscnt.
But I have 64 GB of memory, so I can still do large input values.
Here is performance data showing the difference between the 2 versions, compared to Rust.

 Input Range  | Sub-optimal |  Optimal |   Rust   |
--------------|-------------|----------|----------|
     10^12    |     4.95    |     4.75 |     3.95 |
--------------|-------------|----------|----------|
     10^13    |    57.32    |    52.49 |    48.13 |
--------------|-------------|----------|----------|
     10^14    |   640.81    |   581.34 |   535.64 |
--------------|-------------|----------|----------|
     10^15    |  9654.49    |  9265.87 |  8638.59 |
--------------|-------------|----------|----------|

The optimal version is getting much closer to Rust now.

So ideally, if fibers are mapped to threads on a 1-to-1 basis the inputs could be cycled through and fed to the threads on a 1-to-1 basis. Therefore, the number of spawned data instances would be divorced from the input data stream. The finite fiber|threads are just cycled through until all the data is processed, however long that would take.

26

You’re talking about fibers, not threads.

I’m not sure that this is how cooperative concurrency actually works, and I think you’re making assumptions.

Have you perhaps missed the fact that channel.receive? blocks? The while block loops once for every input it processes.

Maybe. I don’t know.

Again, this is what the “sub-optimal” implementation does.

No. That’s not what this code does. This code creates as many fibers as you have inputs and gives them all to the scheduler. Each fiber runs once and then exits. I suppose you could describe a fiber as data for a thread to process, but that’s only technically true, in the same way as literally any program is data for a thread to process. I doubt that this Crystal code is equivalent to what the Rust version is doing. If you want to argue that point, feel free to analyze the LLVM IR for us.

By the way, you’re mixing usage patterns for WaitGroup. It happens to work, but it’s unnecessary.

You’re still just describing the “sub-optimal” implementation.

27

From @ralsina’s parallel processing code explanation, I did a similar technique for mine.

All results were achieved using BI (brain intelligence), not AI (artificial intelligence).

  threads = System.cpu_count                # number of system threads
  next_pair = Atomic.new(0)                 # index for next twinpair residue
  pairs_done = Atomic.new(0)                # count of finished twinpair residues
  wg = WaitGroup.new(threads)               # create as many WGs as threads
  Fiber::ExecutionContext.default.resize(threads)

  threads.times do
    spawn do
      while (i = next_pair.add(1)) < pairscnt
        lastwins[i], cnts[i] = twins_sieve(restwins[i], ...)
        print "\r#{pairs_done.add(1)} of #{pairscnt} twinpairs done"
      end
      wg.done
  end end
  wg.wait

Here each thread is first spawned in parallel, and only started once. Within each the while loop atomically increments the index value for the residue pairs array, thus they are all processed in order without contention. It now uses minimum memory, and no longer hangs for any input values. It also eliminates the need for an external data producer.

However again, it’s still suboptimal for performance because each thread still must manage its input data generation before data processing can begin. So for optimum performance, distributing data within a thread pool is faster. Conceptually I want to be able to do something like the following:

  threads = System.cpu_count               # number of system threads
  pairs_done = Atomic.new(0)               # count of finished twinpair residues
  wc = WaitCount.new(pairscnt)             # create as many WCs as residue pairs
  Fiber::ExecutionContext.default.resize(maximum: threads)

  restwins.each_with_index do |rhi, i|
    spawn do
      lastwins[i], cnts[i] = twins_sieve(rhi, ...)
      print "\r#{pairs_done.add(1)} of #{pairscnt} twinpairs done"
      wc.done
  end end
  wc.wait

Here WaitCount.new(pairscnt) replaces WaitGroups and indicates the number of times the thread pool is accessed. Then wc.done decrements the count, like wg.done decrements the fiber count, while wc.wait monitors them until it reaches 0.

Since the number of threads is small|finite, each spawned worker in the pool doesn’t need to be assigned a unique fiber, so you can have a fiber pool that cycles thru each available thread. Thus, the growing memory issue for large wg counts can be eliminated, as now only one count value is used with a small fiber|threads pool.

Well, this is one conceptualization of how it could be done using Crystal’s existing structure.