Using new parallel ExecutionContext

Help & Support
1

I converted my residue sieves to use the new Crystal 1.21 ExecutionContext structures.

They’re a significant upgrade in convenience, flexibility, and performance to create true parallelism. :slight_smile:

Under the old regime I had to compile, then run code, as:

$ crystal build --release -Dpreview_mt --mcpu native twinprimes_ssoz.cr

CRYSTAL_WORKERS=32 ./twinprimes_ssoz val1 val2

Now I can do the way more convenient and easier:

$ crystal build --release --mcpu native twinprimes_ssoz.cr

$ ./twinprimes_ssoz val1 val2

I used the ExecutionContext documentation below to learn how to use them.

Enough was there for me to figure out, through trial-and-error, how to create working code. More thorough documentation would be useful to show and explain how all the different methods can|should be used.

I was ultimately able to convert my old code to run in parallel similar to the Rust, D, C++, etc versions. This means the code now reuses the threads to run all the sieve instances in parallel, and now uses a small constant memory footprint for the total length of the sieving process. Previously the memory footprint would grow as more fibers were used, preventing processing of input sizes that needed more system memory than available.

Below are 2 coded versions to achieve true parallel operation.

The 1st retained the old code structure using Channels|Fiber, and the 2nd Wait Groups. They perform similarly, switching in performance depending on the input sizes, but in general the times are consistently close.


Parallel implementation using Channels

  done = Channel(Nil).new(pairscnt)
  threads = Fiber::ExecutionContext::Parallel.new("threads", System.cpu_count)
  
  threadscnt = Atomic.new(0)                # count of finished threads
  restwins.each_with_index do |r_hi, i|     # sieve twinpair restracks
    threads.spawn do
      lastwins[i], cnts[i] = twins_sieve(params)
      print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      done.send(nil)
  end end
  pairscnt.times { done.receive }           # wait for all threads to finish
  
  
Parallel implementation using Wait Groups
  
  wg = WaitGroup.new(pairscnt)
  threads = Fiber::ExecutionContext::Parallel.new("threads", System.cpu_count)
  
  threadscnt = Atomic.new(0)                # count of finished threads
  restwins.each_with_index do |r_hi, i|     # sieve twinpair restracks
    threads.spawn do
      lastwins[i], cnts[i] = twins_sieve(params)
      print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
    ensure
      wg.done
  end end
  wg.wait                                   # wait for all threads to finish

Again, this is a major & beneficial upgrade for Crystal, and it’s great this is now part of the Std::lib.

The only problem I’ve seen so far, using htop, is for certain input sizes both implementations start out using all available threads, but then quickly drop to using just 4 threads, then ultimately freezes processing, requiring a hard keyboard termination to stop it.

This phenomena is not linear, as it occurs with some smaller input values, which doesn’t occur with much greater values. So it seems the parallel processing is tied to the input values somehow?

I would also like feedback to writing the code “better” if possible.

Kudos, again, to the development of these new parallel processing structures.

2

Yes I think Execution Contexts are a big improvement for Crystal. Great that it immediately helps improve your program.

We’ve been working on an extensive guide for writing parallel code at Add parallelism guide (execution contexts) by ysbaddaden · Pull Request #946 · crystal-lang/crystal-book · GitHub

Would you mind reviewing it?
I hope it provides some of the information you’d have to retrieve through trial & error.

That doesn’t seem to work as intended. Could you share a reproduction for that?
@ysbaddaden any idea what might be happening?

3

Are you sure pairscnt is strictly identical to the number of iterations in restwins.each_with_index?

You might want to try:

restwins.each_with_index do |r_hi, i|
  wg.add(1)
  threads.spawn do
    # ...
  ensure
    wg.done
  end
end

That will make sure to have an identical number of add (before spawn) vs done calls.

4

Here’s the source code for the Wait Groups version. You can create the other version by using its parallel structure that’s shown.

My laptop is a Tuxedo Gemini 3, AMD Ryzen 9 7945HX, 16C|32T, 5.4GHz.

pairscnt is number of restwins values.

# Crystal >= 1.21
# Compile as: $ crystal build --release --mcpu native twinprimes_ssoz_wg.cr
# To reduce binary size do: $ strip twinprimes_ssoz_wg
# Single val: $ ./twinprimes_ssoz_wg val1
# Range vals: $ ./twinprimes_ssoz_wg val1 val2
# val1 and val2 can be entered as either: 123456789 or 123_456_789

require "wait_group"

def modinv(a0, m0)
  return 1 if m0 == 1
  a, m = a0, m0
  x0, inv = 0, 1
  while a > 1
    inv &-= (a // m) &* x0
    a, m = m, a % m
    x0, inv = inv, x0
  end
  inv &+= m0 if inv < 0
  inv
end

def gen_pg_parameters(prime)
  puts "using Prime Generator parameters for P#{prime}"
  primes = [2, 3, 5, 7, 11, 13, 17, 19, 23]
  modpg, res_0 = 1, 0
  primes.each { |prm| res_0 = prm; break if prm > prime; modpg &*= prm }

  restwins = [] of Int32                    
  inverses = Array.new(modpg + 2, 0)        
  rc, inc, res = 5, 2, 0                    
  midmodpg = modpg >> 1                     
  while rc < midmodpg                       
    if rc.gcd(modpg) == 1                   
      mc = modpg &- rc                      
      inverses[rc] = modinv(rc, modpg)      
      inverses[mc] = modinv(mc, modpg)      
      restwins << rc << mc &+ 2 if res &+ 2 == rc
      res = rc                             
    end
    rc &+= inc; inc ^= 0b110               
  end
  restwins.sort!;          restwins << (modpg + 1)         
  inverses[modpg + 1] = 1; inverses[modpg - 1] = modpg - 1 
  {modpg, res_0, restwins.size, restwins, inverses}
end

def set_sieve_parameters(start_num, end_num)
  nrange = end_num - start_num
  bn, pg = 0, 3
  if end_num < 49
    bn = 1; pg = 3
  elsif nrange < 77_000_000
    bn = 16; pg = 5
  elsif nrange <  1_100_000_000
    bn = 32; pg = 7
  elsif nrange < 35_500_000_000
    bn = 64; pg = 11
  elsif nrange < 14_000_000_000_000
    pg = 13
    if    nrange > 7_000_000_000_000; bn = 384
    elsif nrange > 2_500_000_000_000; bn = 320
    elsif nrange >   250_000_000_000; bn = 196
    else  bn = 128
    end
  elsif nrange < 480_000_000_000_000
    bn = 448; pg = 17
  else
    bn = 640; pg = 19
  end
  modpg, res_0, pairscnt, restwins, resinvrs = gen_pg_parameters(pg)
  kmin = (start_num-2) // modpg + 1        
  kmax = (end_num - 2) // modpg + 1        
  krange = kmax - kmin + 1                 
  n = krange < 37_500_000_000_000 ? 12 : (krange < 975_000_000_000_000 ? 18 : 22)
  b = bn * n * 1024                         
  ks = krange < b ? krange : b              

  puts "segment size = #{ks.format} resgroups; seg array is [1 x #{(((ks-1) >> 6) + 1).format}] 64-bits"
  maxpairs = krange * pairscnt              
  puts "twinprime candidates = #{maxpairs.format}; resgroups = #{krange.format}"
  {modpg, res_0, ks, kmin, kmax, krange, pairscnt, restwins, resinvrs}
end

def sozp5(val, res_0, start_num, end_num)
  md, rescnt = 30u64, 8                     
  res  = [7,11,13,17,19,23,29,31]           
  range_size = end_num - start_num          

  kmax = (val &- 2) // md &+ 1              
  prms = Array(UInt8).new(kmax, 0)          
  sqrtn = Math.isqrt(val-1|1)               
  k = sqrtn//md; resk = sqrtn-md*k; r=0     
  while resk >= res[r]; r &+= 1 end         
  pcs_to_sqrtn = k &* rescnt &+ r           

  pcs_to_sqrtn.times do |i|                 
    k, r  = i.divmod rescnt                 
    next if prms[k] & (1 << r) != 0         
    prm_r = res[r]                          
    prime = md &* k &+ prm_r                
    rem   = start_num % prime               
    next unless (prime &- rem <= range_size) || rem == 0 
    res.each do |ri|                        
      kn,rn = (prm_r &* ri &- 2).divmod md  
      bit_r = 1 << res.index(rn &+ 2).not_nil! 
      kpm = k &* (prime &+ ri) &+ kn        
      while kpm < kmax; prms[kpm] |= bit_r; kpm &+= prime end 
  end end
  
  primes = [] of UInt64                     
  res.each_with_index do |r_i, i|           
    kmax.times do |k|                       
      if prms[k] & (1 << i) == 0            
        prime = md &* k &+ r_i
        rem = start_num % prime             
        primes << prime if (res_0 <= prime <= val) && (prime &- rem <= range_size || rem == 0)
  end end end
  primes                                    
end

def nextp_init(rhi, kmin, modpg, primes, resinvrs)
  nextp = Slice(UInt64).new(primes.size*2)  
  r_hi, r_lo = rhi.to_u64, rhi.to_u64 &- 2  
  primes.each_with_index do |prime, j|      
    k = (prime &- 2) // modpg.to_u64        
    r = (prime &- 2) %  modpg &+ 2          
    r_inv = resinvrs[r].to_u64              
    rl = (r_inv &* r_lo &- 2) % modpg &+ 2  
    rh = (r_inv &* r_hi &- 2) % modpg &+ 2  
    kl = (prime &+ rl) &* k &+ (rl &* r &- 2) // modpg # kl 1st mult resgroup
    kh = (prime &+ rh) &* k &+ (rh &* r &- 2) // modpg # kh 1st mult resgroup
    kl < kmin ? (kl = (kmin &- kl) % prime; kl = prime &- kl if kl > 0) : (kl &-= kmin)
    kh < kmin ? (kh = (kmin &- kh) % prime; kh = prime &- kh if kh > 0) : (kh &-= kmin)
    nextp[j << 1] = kl                      
    nextp[j << 1 | 1] = kh                  
  end
  nextp
end

def twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
  s = 6                                                
  bmask = (1 << s) &- 1                                
  sum, ki, kn  = 0_u64, kmin &- 1, ks                  
  hi_tp, k_max = 0_u64, kmax                           
  seg = Slice(UInt64).new(((ks - 1) >> s) &+ 1)        
  ki    &+= 1 if r_hi &- 2 < (start_num &- 2) % modpg &+ 2
  k_max &-= 1 if r_hi > (end_num &- 2) % modpg &+ 2    
  nextp = nextp_init(r_hi, ki, modpg, primes,resinvrs) 
  while ki < k_max                          
    kn = k_max &- ki if ks > (k_max &- ki)  
    primes.each_with_index do |prime, j|    
                                            
      k1 = nextp.to_unsafe[j << 1]          
      while k1 < kn                         
        seg.to_unsafe[k1 >> s] |= 1u64 << (k1 & bmask)
        k1 &+= prime  end                   
      nextp.to_unsafe[j << 1] = k1 &- kn    
                                            
      k2 = nextp.to_unsafe[j << 1 | 1]      
      while k2 < kn                         
        seg.to_unsafe[k2 >> s] |= 1u64 << (k2 & bmask)
        k2 &+= prime  end                   
      nextp.to_unsafe[j << 1| 1] = k2 &- kn 
    end                                     
                                            
    seg.to_unsafe[(kn - 1) >> s] |= ~1u64 << ((kn &- 1) & bmask)
    cnt = 0                                 
    seg[0..(kn - 1) >> s].each { |m| cnt &+= (~m).popcount }
    if cnt > 0                              
      sum &+= cnt                           
      upk = kn &- 1                         
      while seg.to_unsafe[upk >> s] & (1u64 << (upk & bmask)) != 0; upk &-= 1 end
      hi_tp = ki &+ upk                     
    end
    ki &+= ks                               
    seg.fill(0) if ki < k_max               
  end                                       
                                            
  hi_tp = (r_hi > end_num || sum == 0) ? 1u64 : hi_tp &* modpg &+ r_hi
  {hi_tp, sum}                              
end

def twinprimes_ssoz()
  end_num   = {(ARGV[0].to_u64 underscore: true), 3u64}.max
  start_num = ARGV.size > 1 ? {(ARGV[1].to_u64 underscore: true), 3u64}.max : 3u64
  start_num, end_num = end_num, start_num if start_num > end_num
  start_num |= 1                           
  end_num = (end_num - 1) | 1              
  start_num = end_num = 7u64 if end_num - start_num < 2

  puts "threads = #{System.cpu_count}"
  ts = Time.instant                         
                                            
  modpg, res_0, ks, kmin, kmax, krange, pairscnt, restwins, resinvrs = set_sieve_parameters(start_num, end_num)

  primes = end_num < 49 ? [5u64] : sozp5(Math.isqrt(end_num), res_0, start_num, end_num)

  puts "each of #{pairscnt.format} threads has nextp[2 x #{primes.size.format}] array"

  te = (Time.instant - ts).total_seconds.round(6)
  puts "setup time = #{te} secs"            
  puts "perform twinprimes ssoz sieve"
  t1 = Time.instant                         

  twinscnt = 0_u64                          
  twinscnt += [3, 5, 11, 17].select { |tp| start_num <= tp < res_0 }.size if end_num > 3

  cnts = Array(UInt64).new(pairscnt, 0)     
  lastwins = Array(UInt64).new(pairscnt, 0) 

  wg = WaitGroup.new(pairscnt)
  threads = Fiber::ExecutionContext::Parallel.new("threads", System.cpu_count)
  
  threadscnt = Atomic.new(0)                
  restwins.each_with_index do |r_hi, i|     
    threads.spawn do
      lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
      print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
    ensure
      wg.done
  end end
  wg.wait                                  
  print "\r#{pairscnt} of #{pairscnt} twinpairs done"

  last_twin = lastwins.max                  
  twinscnt += cnts.sum                      
  last_twin = 5 if end_num == 5 && twinscnt == 1
  kn = krange % ks                          
  kn = ks if kn == 0                        
  t2 = (Time.instant - t1).total_seconds    

  puts "\nsieve time = #{t2.round(6)} secs"      
  puts "total time = #{(t2 + te).round(6)} secs" 
  puts "last segment = #{kn.format} resgroups; segment slices = #{((krange - 1)//ks + 1).format}"
  puts "total twins = #{twinscnt.format}; last twin = #{last_twin.format}|-2"
end

twinprimes_ssoz

This freezes for input of 10^14.

➜  crystal-projects ./twinprimes_ssoz_wg 100_000_000_000_000
threads = 32
using Prime Generator parameters for P17
segment size = 5,505,024 resgroups; seg array is [1 x 86,016] 64-bits
twinprime candidates = 4,363,283,778,975; resgroups = 195,882,549
each of 22,275 threads has nextp[2 x 664,572] array
setup time = 0.023539 secs
perform twinprimes ssoz sieve
3 of 22275 twinpairs done^C

But works for 10x greater input of 10^15.
Try different values for system you run it on. Fewer threads, slower times.

➜  crystal-projects ./twinprimes_ssoz_wg 1_000_000_000_000_000
threads = 32
using Prime Generator parameters for P19
segment size = 7,864,320 resgroups; seg array is [1 x 122,880] 64-bits
twinprime candidates = 39,039,907,715,325; resgroups = 103,096,079
each of 378,675 threads has nextp[2 x 1,951,949] array
setup time = 0.186185 secs
perform twinprimes ssoz sieve
378675 of 378675 twinpairs done
sieve time = 9087.479603 secs
total time = 9087.665788 secs # 150m+|2.5h+
last segment = 859,919 resgroups; segment slices = 14
total twins = 1,177,209,242,304; last twin = 999,999,999,997,969|-2

I used htop to monitor operations.

5

Then I don’t know about the final hang. This might be a problem in the program, or maybe it’s triggering a bug. I’ll try and see if I can reproduce.

What I do know, is that the 10^15 case spawns 378,675 fibers, but those fibers are CPU bounded—only CPU calculations, no I/O, no communication, no sync. The fibers will start and die immediately, which is a huge waste of resources. For CPU bounded workloads, you should spawn as many fibers as the actual parallelism (number of logical CPU), then distribute each work item through a Channel.

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

# consumers
ncpus.times do
  threads.spawn do
    while r_hi = channel.receive?
      # ...
    end
  ensure
    wg.done
  end
end

# producer
restwins.each { |r_hi| channel.send(r_hi) }

For best performance, which is the whole point here, you’ll want to avoid cross-context communication, which is slower than in-context, and just resize the default execution context.

6

I’ve played with this a little bit on a much less powerful computer than the one you’re using, and I’m finding that the hang is nondeterministic. It hangs maybe 9 out of 10 times I run with 10^14. On my machine, this behavior starts at about 7*10^13 with closer to a 50/50 chance of hanging, and significantly smaller inputs appear unaffected.

If you add a puts "Thread done" between the twin_sieve call and the existing print, it looks like it’s printed roughly once per number of threads after the last atomic print output (which only gets to 3 or 4). And the CPU usage of the process drops to full usage of one core essentially immediately (whereas smaller numbers that don’t hang will use all 16 cores on my machine).

I unsuccessfully tried:

  • Moving the Atomic::add call to an assignment before the print
  • Removing the atomic entirely (in favor of just using i directly, despite the loss of properly-ordered output) and using puts instead of print with carriage return (messy output, but still hanging)
  • Making threadscnt an Atomic(Int64), in case that matters somehow (though it shouldn’t)

I’ve let this strangeness nerd snipe me too long already, but I hope some of the above info helps others figure out what’s going on.

7

Your code eliminates the hanging problems, as every input value I tried worked to completion.

However, the code doesn’t perform the same algorithm as my code.
It doesn’t provide the indexi needed for: lastwins[i], cnts[i] = twins_sieve(params)

I tried multiple ways to try to provide it, but none worked, so the code doesn’t correctly store all the pairs counts in cnts[i].

Ideally, I’d like to eliminate using Fibers|Channels as no intra-thread communication is needed. What just needs to happen is all the values in the array restwins be processed in parallel. So like the implementations in Rust, D, etc, I just iterate through their values, which are processed in order by the current available thread. When the last value|thread finishes the process ends.

Also, I don’t conceptually understand the purpose|need for the embedded loops using ncpus, etc. If you can explain that in some detail that would be nice.

So, if you can fix the code to generate the i | r_hi values in sync, to create correct results again, I can then compare it to my working (but non-optimal) version.

8

Just use Channel(Tuple(Int32, Int32)) and restwins.each_with_index. I’ve tried this with your code, and it works. Make sure to close the channel after all values are sent, though. I haven’t used channels much and missed that the first time.

See below. I decided to give you code.

Modifying restwins from multiple threads is inter-thread communication. Things like channels just make parallelism safer than reading and writing memory recklessly in parallel. And fibers are just an abstraction that allows for concurrency in a single-threaded context or parallelism in a multi-threaded context with the same code. Unless you want to write bindings to pthreads (perhaps if you feel you need to punish yourself for something), you pretty much need to use fibers to do parallelism in Crystal. You don’t need to use channels, but you probably should.

Hope this helps. I haven’t run this exact code, and it’s a mix of copy-paste from @ysbaddaden above and stuff from my local copy where I tried this.

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

# consumers
ncpus.times do                     # Spawn `ncpus` fibers
  threads.spawn do                 # (i.e. one fiber per available thread)
    while input = channel.receive?  # get input from channel (or break if closed)
      rhi = input[0]
      i = input[1]
      lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
      print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
    end
  ensure
    wg.done
  end
end

# producer
restwins.each_with_index { |r_hi, i| channel.send({r_hi, i}) }

# adding this because it tripped me up; I think it was meant to be implied
channel.close

wg.wait
9

Right, I forgot to close the channel :person_facepalming:

10

I can’t emphasize what @RespiteSage said enough. Use communication. Accessing, and worst, mutating shared data (:fearful:) from multiple fibers is unsafe.

I proposed to use a channel to distribute the work because that’s how it should be. Only consider something else carefully if it proves to be a performance bottleneck and the alternative is proven safe and is actually faster (spoiler: multiple threads writing to roughly the same memory is slow).

Now, you should now use another channel to report the results back and have a consumer fiber printing the results.

11

It seems that closing the channel was the problem, which channel.close fixed.

Here’s @RespiteSage version that I got to work, and mine that now closes the channel too. They both produce correct results, with comparable times, and no hanging. I’ll do true performance testing on a “quiet” system (not playing music, and browsing, in the background while code is running).

I also just do channel = Channel(Int32).new(ncpus) and not ..new(npcus * 4) because I didn’t see any performance|memory difference.

  ncpus = System.cpu_count
  wg = WaitGroup.new(ncpus)
  channel = Channel(Tuple(Int32,Int32)).new(ncpus)
  threads = Fiber::ExecutionContext::Parallel.new("threads", System.cpu_count)
  
  threadscnt = Atomic.new(0)                # count of finished thread
  ncpus.times do                            # sieve twinpair restracks
    threads.spawn do
      while inputs = channel.receive?
        r_hi, i = inputs
        lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
        print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      end
    ensure
      wg.done
  end end

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

--------------------------------------------------------
  ncpus = System.cpu_count
  wg = WaitGroup.new(ncpus)
  channel = Channel(Int32).new(ncpus)
  threads = Fiber::ExecutionContext::Parallel.new("threads", System.cpu_count)
  
  threadscnt = Atomic.new(0)                # count of finished thread
  ncpus.times do                            # sieve twinpair restracks
    threads.spawn do
      while i = channel.receive?
        r_hi = restwins[i]
        lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
        print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      end
    ensure
      wg.done
  end end

  pairscnt.times { |i| channel.send(i) }
  channel.close
  wg.wait

This is so much better now than before. I’ve been putting off doing some other projects in Crystal until it had true parallel processing ability, but now I feel I can use Crystal for them.

For me, this takes Crystal to another level of usability I had reserved for Rust, D, etc. But I have a suggestion that could make it even easier, and more intuitive, to use for parallel processing.

In Rust the Rayon crate is used most prominently for multi-threading. It provides really nice parallel iterators.

Just as WaitGroups made concurrent processing symantically easier|simpler, adding parallel iterators could do the same for Execution::Context. It could take care of some of the standard boilerplate so people don’t miss doing the simple stuff, like wg.done and channel.close and make writing code for 90% of algorithms really simple and intuitive. This could make Crystal really competitive in the ease of use department, and to Rust in particular. Crystal already beats Rust hands down for writing clean, short, beautiful code.

Here’s a top of my head example how it could look for this example.

  wg = WaitGroup.new(ncpus)
  threads = Fiber::ExecutionContext::Parallel.new("threads", System.cpu_count)
  
  threadscnt = Atomic.new(0)
  restwin.each_with_index.parallel::iterator do |r_hi, i|
     lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
     print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
     wg.done
  end end
  wg.wait

Anyway, maybe something to think about as you improve the semantics and features.

Also needed, as you already know, is better docs with conceptual explanations of the new models and coding paradigms, with coding examples. I was able to get 90% of the code working from the existing docs, but it was the details for that last 10% that got me stuck (like channel.close).

But even as it is, this now makes me really, really happy. :slightly_smiling_face:

12

The parallelism guide is now live at Parallelism - Crystal

13

I’ve taken some time to compare Crystal’s new ExecutionContext to other
languages parallel forms. I compare them based on 3 general criteria.

They are: 1) Does it work correctly as desired.; 2) How relatively performant is it.
3) Are the semantics simple, clear, and intuitive.

  1. Does it work correctly as desired.

Without a doubt, the new EC paradigm performs true parallel processing and
produces correct results. It also uses a similar small constant amount of memory
while doing so, consistent with other languages. So its a major plus for me.

  1. How relatively performant is it.

Compared to Rust, D, etc, it currently is not as fast, BUT isn’t way bad.
And the time differences seem to be linear as input values increase.
That’s to say, I see a consistent process overhead the longer processing takes.

But this means Crystal can become similarly performant as its implementation
matures, and optimizations and refinements are learned and applied. However, as
I’ll discuss later in more detail, I think it’ll be necessary to adopt a different
paradigm, at least for specific use cases, to achieve similar relative performance.

  1. Are the semantics simple, clear, and intuitive.

Here I’m speaking as a programmer (language user) and not a developer.
My observations and comments here pertain to how easy, simple, and intuitive
it is to do parallel processing in Crystal compared to other languages.

Below is the parallel code for D, C++, Nim, Rust, and Crystal.

C++ uses OpenMP and Rust the Rayon crate for parallel processing.
D, Nim, and Crystal provide it as part their standard libraries.

As you can see, the semantics for Crystal are much longer and not as intuitive.
A lot of boilerplate code has to be written compared to the others.
Hopefully, this boilerplate can be incorporated into simpler semantics to
make using Crystal just as simple and intuitive, and error free, to use.

D

  cnts = new uint[](pairscnt);           // count of tps for each thread
  lastwins = new ulong[](pairscnt);      // largest hi_tp for each thread
  shared uint threadscnt = 0;            // count of finished threads

  foreach (indx; parallel(iota(0, pairscnt))) { // do in parallel for each tp
    twinsSieve(cast(uint) indx);         // sieve selected twinpair restracks
    write("\r", threadscnt.atomicOp!"+="(1), " of ", pairscnt, " twinpairs done");
  }


C++

  vector<uint64> cnts(pairscnt);         // twins cnts for each twinpair
  vector<uint64> lastwins(pairscnt);     // last|largest twin per twinpair
  std::atomic<uint> threadscnt(0);       // count of finished threads

  #pragma omp parallel for               // process twinpairs in parrallel
  for (int i = 0; i < pairscnt; ++i) {   // process each twinpair in own thread
     uint64 l, c;
     tie(l, c) = twins_sieve(restwins[i], kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs);
     lastwins[i] = l; cnts[i] = c;
     cout << "\r" << threadscnt++ << " of " << pairscnt << " twinpairs done";
  }


Nim

  #parallel:                             # perform in parallel
  for indx, r_hi in restwins:            # for each twin pair row index
    spawn twins_sieve(r_hi.uint, Kmin, Kmax, Ks, start_num, end_num, modpg, primes, resinvrs, indx)
    stdout.write("\r", (indx + 1), " of ", pairscnt, " twinpairs done")
  sync()                                 # when all the threads finish


Rust
                                             // sieve each twinpair restracks in parallel
  let (lastwins, cnts): (Vec<_>, Vec<_>) = { // store outputs in these arrays
    let counter = RelaxedCounter::new();
    restwins.par_iter().map( |r_hi| {
      let out = twins_sieve(*r_hi, kmin, kmax, ks, start_num, end_num, modpg, &primes, &resinvrs);
      print!("\r{} of {} twinpairs done", counter.increment(), pairscnt);
      out
    }).unzip()
  };


Crystal

  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)                # count of finished thread
  ncpus.times do                            # sieve twinpair restracks
    threads.spawn do
      while inputs = channel.receive?
        r_hi, i = inputs
        lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
        print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      end
    ensure
      wg.done
  end end

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

While understanding the EC paradigm is new, and a major advancement for Crystal,
to become closer to the others (code and performance) will likely require a revisioning
of the underlying paradigm. This primarily surrounds the universal use of Fibers.

From my understanding, fibers are really good at message passing.
However when you don’t need to do it, it becomes unnecessary overhead.

There are whole classes of parallel numerical algorithms|processes they don’t pass information.
Things like FFTs (Fast Fourier Transforms, et al), video processing|rendering, audio processing
(multi-track processing), data mining, etc, parallel process data independently to compute results.

Thus, the EC reliance on fibers for these class of application is unnecessary.
For them, each parallel instance just needs to perform a 1-to-1 mapping of data-to-thread.

So optimally for these applications, I think Crystal would either have to create a unique
stripped down EC version to eliminate the fiber structure overhead, or create a new
independent direct data-to-thread paradigm, similar to other languages.

I assume you might not want to hear this, but you can go the Rust route and make this a shard.
Thus you can make this an independent project that doesn’t change the standard lib|structure.
If it turns out to be useful|beneficial, you can then migrate it into the standard language.

Since this has already been done in many languages, you don’t have to start from scratch.
It’s probably mostly a matter of deciding how to best do it for Crystal (semantics and structure).

Actually Ruby has|is facing these issues with Ractors, which allow intra-process messaging.
However, the more options you provide the more (unforseen) problems you will create.

Anyway, these observations|comments are meant to help improve Crystal and make it stronger.

14

Could you elaborate what you consider an overhead introduced by the “fiber structure”?

A fiber is just the envelope for a sequence of instructions. When executing, its code runs directly on a thread. I don’t see where there might be any overhead there.
Runtime mechanics for switching between fibers only intervene at natural stopping points when interacting with the event loop. A computation-focused algorithm wouldn’t be affected by that.

15

@jzakiya maybe it could work in a simpler way?

threads = Fiber::ExecutionContext::Parallel.new("threads", System.cpu_count)
wg = WaitGroup.new(restwins.size)

restwins.each_with_index do |r_hi, i|
  threads.spawn do
    lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
  ensure
    wg.done
  end
end

wg.wait
16

For C++, it looks like you’re using the OpenMP library. For Rust, it looks like you’re using the Rayon library (or some derivative). The “intuitive” implementations in each come from libraries, not the language or standard library. It’s absolutely reasonable to want Crystal to have convenient parallelization libraries, but that’s a different thing than suggesting that the standard library implementation(s) should be changed.

From what I can tell, the Nim implementation uses a thread pool module, so it’s just doing what the Crystal implementation is doing, but you don’t have to write it (which, again, could be addressed in libraries without changes to the standard library).

As for D, I can’t say that I find that at all intuitive. It looks like some kind of parallel iterator, but then there’s iota… :man_shrugging: I’m sure it’s a great language, but I don’t find it immediately intuitive in the way you’re suggesting.

None of that is to say that the existing standard library implementation is necessarily perfect, but changing the standard library is disruptive in a way that writing a library isn’t, and it requires the time of Crystal’s maintainers when they could be working on other worthwhile efforts, whereas anyone else with the time, expertise, and desire could write a library. So I think that the ergonomics of concurrency would be better to tackle in a shard.

Also, you could reduce the Crystal boilerplate a bit this way:

  ncpus = Fiber::ExecutionContext.default_workers_count # respects ${CRYSTAL_WORKERS}

  # instead of making a new ExecutionContext, just change the existing one
  Fiber::ExecutionContext.default.resize(num_workers)

  # note that we don't give a number of workers
  wg = WaitGroup.new 

  channel = Channel(Tuple(Int32,Int32)).new(ncpus)

  threadscnt = Atomic.new(0)                # count of finished thread
  ncpus.times do                            # sieve twinpair restracks
    wg.spawn do                             # just spawn straight from wg
      while inputs = channel.receive?
        r_hi, i = inputs
        lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
        print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      end
  end end

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

@paulocoghi your code is what I started with, and conceptually is what ideally I want to do.
It’s the fastest|shortest implementation, but hangs for various input values, so not 100% usable. Also the print statement has no affect on speed, as it’s the same w/wo it.

  wg = WaitGroup.new(pairscnt)
  threads = Fiber::ExecutionContext::Parallel.new("threads", System.cpu_count)

  threadscnt = Atomic.new(0)                # count of finished thread
  restwins.each_with_index do |r_hi, i|
    threads.spawn do
      lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
      print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
    ensure
      wg.done
    end
  end
  wg.wait

@RespiteSage this works for all values, but is no shorter|faster, and is more confusing to me.
It still requires boilerplate code to structure the fibers to operate correctly over the system threads,
and sends them data via messages instead of directly, creating the processing overhead for each thread I allude to. And I still have no conceptual understanding why|what the double embedded loops do, and why they’re necessary.

  ncpus = Fiber::ExecutionContext.default_workers_count # respects ${CRYSTAL_WORKERS}
  Fiber::ExecutionContext.default.resize(ncpus) # instead of new ExC, just change existing one
  wg = WaitGroup.new                            # note that we don't give a number of workers
  channel = Channel(Tuple(Int32,Int32)).new(ncpus)

  threadscnt = Atomic.new(0)                # count of finished thread
  ncpus.times do                            # sieve twinpair restracks
    wg.spawn do                             # just spawn straight from wg
      while inputs = channel.receive?
        r_hi, i = inputs
        lastwins[i], cnts[i] = twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
        print "\r#{threadscnt.add(1)} of #{pairscnt} twinpairs done"
      end
  end end

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

Also, I said in my comments, creating a direct threading model could be developed as a shard, independent of the standard library. And for me, the other language versions where easy to understand, and code, from their documentation, and help from their forums. And without a doubt, they’re all shorter, and more concise.

@straight-shoota, what I mean|show by “fiber structure” is the need to setup sender|receiver channels, send messages, and the whole consumer-producer pattern|paradigm. And I don’t really understand what a fiber is physically, i.e. how|what in hardware are they, and what|how in hardware are they created|destroyed.

Again as I said in my comments, I want to directly send data to an available thread on a 1-to-1 basis,
and not via proxy fibers. Go was the only other language I tried that used fibers, and it was SLOW,
and ate up memory for large inputs. But it wasn’t designed for these class of algorithms, but web stuff.

So given the current state of EC, if the hanging problem can be resolved @paulocoghi’s code would be ideal, from a conciseness, conceptual intuitiveness, and performance standpoints.

18

In LangArena benchmark, matmul test because slower: LangArena/crystal/main.cr at master · kostya/LangArena · GitHub

1.20.1 with -Dpreview_mt and CRYSTAL_WORKERS=16

Matmul::Single: OK in 5.057s
Matmul::T4: OK in 1.324s
Matmul::T8: OK in 0.769s
Matmul::T16: OK in 0.677s

in 1.21.0 with Fiber::ExecutionContext.default.resize(maximum: 16)

Matmul::Single: OK in 5.084s
Matmul::T4: OK in 2.921s
Matmul::T8: OK in 2.446s
Matmul::T16: OK in 2.303s

this is pure parallelism tests, there is maximum 16 spawns.

Golang results is (they trade off parallelism for concurrency, but still perform well):

Matmul::Single: OK in 5.111s
Matmul::T4: OK in 2.427s
Matmul::T8: OK in 1.119s
Matmul::T16: OK in 0.824s

C language with pthread:

Matmul::Single: OK in 5.055s
Matmul::T4: OK in 1.323s
Matmul::T8: OK in 0.651s
Matmul::T16: OK in 0.433s
19

The C++ and Rust comparisons aren’t fair as they depend on a C++ extension (OpenMP) and a crate (rayon). Achieving the same without extensions and libraries will blow up the code.

Nim is interesting. I suppose it’s structured concurrency at work. It feels a bit too magic, though.

Crystal can be much simpler, indeed. Extract a helper, and resize the default context which is often faster because the producer & consumers are the same context which avoids cross context synchronization:

def parallel(collection : Enumerable(F), &) forall F
  count = Fiber::ExecutionContext.current.capacity

  # channel must have more entries (avoid thread starvation)
  inputs = Channel(F).new(count * 4)

  WaitGroup.new do |wg|
    count.times do |i|
      wg.spawn("parallel::consumer:#{i}") do
        while input = inputs.receive?
          yield input
        end
      end
    end

    collection.each { |item| inputs.send(item) }
    inputs.close
  end
end

Then it becomes:

Fiber::ExecutionContext.default.resize(System.cpu_count)

parallel(restwins) do |r_hi|
  twins_sieve(r_hi, kmin, kmax, ks, start_num, end_num, modpg, primes, resinvrs)
end
20

From looking at it I don’t see where it accounts for the indexes for each rhi value. And where would the print line go?

Also, would this helper method be part of the underneath structure for ECs and the user would just use the parallel method? More explanation (w/examples) would be nice.

It would be also visually nicer to mimic Rust, so you can write it something like:

restwins.each.parallel do |rhi| ...

restwins.each_with_index.parallel do |rhi, i| ...