home

หน้าแรก
 Computer >> บทช่วยสอนคอมพิวเตอร์ >  >> การเขียนโปรแกรม >> Ruby

เชี่ยวชาญ Ruby Concurrency: เครื่องมือ เทคนิค และ ทิศทางอนาคต

Concurrency และ parallelism สำคัญกว่าที่เคยสำหรับ Ruby developers มันทำให้แอปเร็วขึ้น ใช้ hardware เต็มที่ ในบทความนี้เราจะสำรวจเครื่องมือที่มีอยู่สำหรับ Rubyist ทุกคน และสิ่งที่ Ruby สัญญาจะส่งมอบในอนาคตใกล้

ไม่ใช่ทุกคนใช้ concurrency โดยตรง แต่งเราทุกคนใช้ indirectly ผ่าน tools อย่าง Sidekiq การเข้าใจ Ruby concurrency ช่วยไม่แค่สร้าง solutions ของตัวเอง แต่ช่วยเข้าใจและ troubleshoot existing ones

แต่ก่อน เรามองภาพใหญ่ก่อน

Concurrency vs. Parallelism

คำเหล่านี้ใช้สลับกัน แต่มีความหมายแตกต่าง

  • Concurrency: ศิลปะการทำหลาย tasks ทีละอัน ด้วยการสลับกันเร็วๆ อาจดูเหมือน simultaneous
  • Parallelism: ทำหลาย tasks ในเวลาเดียวกันจริงๆ ไม่ใช่แค่ดูเหมือน คือ simultaneous จริง

Concurrency ใช้บ่อยกับแอป IO heavy เช่น web app ติดต่อ database หรือ network requests มาก ด้วย concurrency เราเก็บแอป responsive ระหว่างรอ database ตอบ query

เป็นไปได้เพราะ Ruby VM อนุญาต threads อื่นรันขณะหนึ่งรอ IO แม้โปรแกรมทำหลายสิบ requests ถ้าใช้ concurrency requests จะส่งไปพร้อมกันเกือบหมด

Parallelism ไม่ได้รองรับใน Ruby ปัจจุบัน

ทำไม Ruby ไม่มี Parallelism?

วันนี้ ไม่มีวิธีทำ parallelism ภายใน single Ruby process ด้วย default Ruby implementation (MRI/CRuby) Ruby VM enforce lock (GVL/Global VM Lock) ป้องกัน multiple threads รัน Ruby code พร้อมกัน Lock นี้มีเพื่อปกป้อง internal state ของ VM และป้องกัน scenarios ที่อาจทำให้ VM crash ไม่ได้สถานะดี แต่ยังมีหวัง: Ruby 3 จะมาเร็วๆ นี้ และสัญญาแก้ปัญหานี้ด้วย concept ชื่อ Guild (อธิบายในส่วนท้าย)

Threads

Threads คือ workhorse ของ Ruby concurrency เพื่อเข้าใจการใช้และ pitfalls เราจะให้ตัวอย่าง สร้างโปรแกรมเล็ก consume API และเก็บผลลัพธ์ใน datastore ใช้ concurrency

ก่อนสร้าง API client ต้องมี API ด้านล่างคือ tiny API implementation รับตัวเลข ตอบ plain text ว่าเลขคู่หรือคี่ ถ้า syntax ดูแปลก ไม่ต้องกังวล ไม่เกี่ยวกับ concurrency แค่เครื่องมือ

app = Proc.new do |env|
  sleep 0.05
  qs = env['QUERY_STRING']
  number = Integer(qs.match(/number=(\d+)/)[1])
  [
    '200',
    { 'Content-Type' => 'text/plain' },
    [number.even? ? 'even' : 'odd']
  ]
end
run app

รัน web app นี้ต้องมี rack gem แล้วรัน rackup config.ru

ต้องการ mock datastore นี่คือ class จำลอง key-value database:

class Datastore
  # ... accessors and initialization omitted ...
  def read(key)
    data[key]
  end

  def write(key, value)
    data[key] = value
  end
end

มา implement concurrent solution มี method run fetch 1,000 records concurrently เก็บใน datastore

class ThreadPoweredIntegration
  # ... accessors and initialization ...
  def run
    threads = []
    (1..1000).each_slice(250) do |subset|
      threads << Thread.new do
        subset.each do |number|
          uri = 'https://localhost:9292/' \
            "even_or_odd?number=#{number}"
          status, body = AdHocHTTP.new(uri).blocking_get
          handle_response(status, body)
        rescue Errno::ETIMEDOUT
          retry # Try again if the server times out.
        end
      end
    end
    threads.each(&:join)
  end
  # ... end
end

สร้าง 4 threads แต่ละอัน process 250 records ใช้กลยุทธ์นี้เพื่อไม่ท่วมท้น third-party API หรือระบบของตัวเอง

ด้วย requests concurrently ด้วย multiple threads total execution ใช้เวลาส่วนเล็กของ sequential implementation ขณะแต่ละ thread มีช่วง inactive ระหว่างขั้นตอน HTTP request Ruby VM ให้ thread อื่นรัน นี่คือเหตุผลที่ implementation นี้เร็วกว่า sequential มาก

Class AdHocHTTP เป็น HTTP client straightforward สร้างสำหรับบทความนี้เพื่อโฟกัสที่ความแตกต่างระหว่าง threads และ fibers implementation อยู่นอกขอบเขต แต่ดูได้ที่นี่ถ้าสนใจ

สุดท้าย handle server response ท้าย inner loop นี่คือ handle_response:

# ... inside the ThreadPoweredIntegration class ...
attr_reader :ds

def initialize
  @ds = Datastore.new(even: 0, odd: 0)
end

# ...

def handle_response(status, body)
  return if status != '200'
  key = body.to_sym
  curr_count = ds.read(key)
  ds.write(key, curr_count + 1)
end

method นี้ดูโอเคไหม? รันดูสิว่าสิ่งไหนอยู่ใน datastore:

{ even: 497, odd: 489 }

แปลกมาก เพราะระหว่าง 1 ถึง 1000 มีเลขคู่ 500 และคี่ 500 ในส่วนต่อไป มาเข้าใจว่เกิดอะไรและสำรวจวิธีแก้ bug นี้

Threads และ Data Races: เทพเจ้าอยู่ในรายละเอียด

Threads ทำให้ IO heavy programs เร็วขึ้น แต่ยากที่จะทำถูก Error ในผลลัพธ์ข้างต้นมาจาก race condition ใน handle_response Race condition เกิดเมื่อ 2 threads จัดการข้อมูลเดียวกัน

เรา operate บน shared resource (object ds datastore) ต้องระมัดระวัง non-atomic operations มองว่าเรา read จาก datastore ก่อน และ — ใน statement ที่สอง — write count incremented by 1 นี่ problematic เพราะ thread อาจหยุดหลัง read แต่ก่อน write ถ้า thread อื่นรันและ increment ค่าของ key ที่เราสนใจ เราจะ write out-of-date count เมื่อ thread เดิม resume

หนึ่งในวิธีลดอันตรายของ threads คือใช้ higher-level abstractions โครงสร้าง concurrent implementation ดู concurrent-ruby gem สำหรับ patterns ต่างๆ และ thread-powered program ที่ปลอดภัยกว่า

มีหลายวิธีแก้ data race วิธีง่ายคือใช้ mutex synchronization mechanism บังคับ one-at-a-time access โค้ด segment นี่คือ implementation ก่อนหน้าแก้ด้วย mutex:

# ... inside ThreadPoweredIntegration class ...
def initialize
  # ...
  @semaphore = Mutex.new
end

# ...

def handle_response(status, body)
  return if status != '200'
  key = body.to_sym
  semaphore.synchronize do
    curr_count = ds.read(key)
    ds.write(key, curr_count + 1)
  end
end

ถ้าวางแผนใช้ threads ใน Rails app คู่มือทางการ Threading and Code Execution in Rails ต้องอ่าน ไม่ปฏิบัติตาม guidelines อาจทำให้ leak database connections

รัน implementation ที่แก้แล้ว ได้ผลลัพธ์ที่คาดหวัง:

{ even: 500, odd: 500 }

แทน mutex เราสามารถกำจัด data races โดย drop threads เลย และใช้ concurrency tool อีกอย่างใน Ruby ในส่วนต่อไป มาดู Fiber กลไกสำหรับปรับปรุงประสิทธิภาพ IO-heavy apps

Fiber: เครื่องมือบางสำหรับ Concurrency

Ruby Fibers ให้ cooperative concurrency ภายใน single thread หมายถึง fibers ไม่ preempted โปรแกรมตองทำ scheduling เอง เพราะโปรแกรมเมอร์ควบคุมเมื่อ fibers เริ่มและหยุด จึงหลีกเลี่ยง race conditions ได้ง่ายกว่า

ไม่เหมือน threads fibers ไม่ให้ better performance ตอน IO เกิดขึ้น โชคดีที่ Ruby มี asynchronous reads/writes ผ่าน IO class ด้วย async methods เราสามารถป้องกัน IO operations block fiber-based code ได้

สถานการณ์เดียวกัน ตอนนี้กับ Fibers

มาทำตัวอย่างเดียวกัน แต่กับ fibers รวม async capabilities ของ Ruby IO class ไปไกลกว่าขอบเขตบทความอธิบาย async IO ใน Ruby ทั้งหมด แต่จะแตะส่วนสำคัญ และดู implementation ของ relevant methods ของ AdHocHTTP (client เดียวกับ threaded solution) ถ้าสนใจ

เริ่มจาก run method ของ fiber-powered implementation:

class FiberPoweredIntegration
  # ... accessors and initialization ...
  def run
    (1..1000).each_slice(250) do |subset|
      Fiber.new do
        subset.each do |number|
          uri = 'https://127.0.0.1:9292/' \
            "even_or_odd?number=#{number}"
          client = AdHocHTTP.new(uri)
          socket = client.init_non_blocking_get
          yield_if_waiting(client, socket, :connect_non_blocking_get)
          yield_if_waiting(client, socket, :write_non_blocking_get)
          status, body = yield_if_waiting(client, socket, :read_non_blocking_get)
          handle_response(status, body)
        ensure
          client&.close_non_blocking_get
        end
      end.resume
    end
    wait_all_requests
  end
  # ... end
end

สร้าง fiber สำหรับแต่ละ subset ของตัวเลขที่ต้องการเช็คคู่คี่

loop ตัวเลข เรียก yield_if_waiting method นี้หยุด fiber ปัจจุบัน ให้อีก fiber resume

สังเกตหลังสร้าง fiber เรียก resume ทำให้ fiber เริ่มรัน เรียก resume ทันทีหลังสร้าง เริ่มทำ HTTP requests ก่อน main loop 1 ถึง 1000 จบ

ท้าย run method มี wait_all_requests method เลือก fibers พร้อมรัน และ guarantee เราทำ requests ตามเป้า จะดูในส่วนท้าย

มาดู yield_if_waiting รายละเอียด:

# ... inside FiberPoweredIntegration ...
def initialize
  @ds = Datastore.new(even: 0, odd: 0)
  @waiting = { wait_readable: {}, wait_writable: {} }
end

# ...

def yield_if_waiting(client, socket, operation)
  res_or_status = client.send(operation)
  is_waiting = [:wait_readable, :wait_writable].include?(res_or_status)
  return res_or_status unless is_waiting

  waiting[res_or_status][socket] = Fiber.current
  Fiber.yield
  waiting[res_or_status].delete(socket)
  yield_if_waiting(client, socket, operation)
rescue Errno::ETIMEDOUT
  retry # Try again if the server times out.
end

ลองทำ operation (connect, read, write) ด้วย client 2 ผลลัพธ์หลัก:

  • Success: return
  • Receive symbol: ต้องรอ

ยังไงถึง "รอ"?

  1. สร้าง checkpoint โดยเพิ่ม socket รวม current fiber ไป instance variable waiting (Hash)
  2. เก็บคู่นี้ใน collection ที่ hold IO รอ read หรือ write (จะเห็นสำคัญในหน่อย) ขึ้นกับผลลัพธ์จาก client
  3. หยุด execution ของ current fiber ให้อีก fiber รัน fiber ที่หยุดจะได้ resume ทีหลังเมื่อ network socket พร้อม IO operation จะ retry (ครั้งนี้จะสำเร็จ)

ทุก Ruby program รันใน fiber ที่อยู่ใน thread (ทุกอย่างใน process) ดังนั้นตอนสร้าง fiber แรก รัน แล้ว yield เรากำลัง resume execution ของส่วนกลางของโปรแกรม

เข้าใจ mechanism yield execution เมื่อ fiber รอ IO มาแล้ว มาสำรวจส่วนสุดท้ายของ fiber-powered implementation

def wait_all_requests
  while(waiting[:wait_readable].any? || waiting[:wait_writable].any?)
    ready_to_read, ready_to_write = IO.select(waiting[:wait_readable].keys, waiting[:wait_writable].keys)
    ready_to_read.each do |socket|
      waiting[:wait_readable][socket].resume
    end
    ready_to_write.each do |socket|
      waiting[:wait_writable][socket].resume
    end
  end
end

ความคิดหลักคือรอ (loop) จนกว่า pending IO operations จบหมด

ใช้ IO.select รับ 2 collections pending IO objects: read และ write return IO objects ที่ทำเสร็จ เพราะเรา associate IO objects กับ fibers ที่รัน มันง่ายที่จะ resume fibers นั้นๆ

ทำซ้ำจน requests ทั้งหมด fired และ completed

The Grand Finale: ประสิทธิภาพเทียบได้ ไม่ต้อง locks

handle_response method เหมือนใน threaded code (version ไม่มี mutex) แต่เนื่องจาก fibers ทั้งหมดรันใน thread เดียวกัน ไม่มี data races รันโค้ด ได้ผลลัพธ์ที่คาดหวัง:

{ even: 500, odd: 500 }

คุณไม่อยากจัดการ fiber switching ทุกครั้งที่ใช้ async IO โชคดีที่มี gems abstract ทำให้ fibers ไม่ต้องคิดถึง ดู async project เป็นจุดเริ่มที่ดี

Fibers ส่องแสงเมื่อ High Scalability จำเป็น

แม้เราได้ประโยชน์ virtually eliminating data race risks แม้ใน small scale fibers เป็น tool ยอดเยี่ยมเมื่อ high scalability จำเป็น Fibers lightweight กว่า threads มาก ด้วย resources เท่ากัน สร้าง threads จะ overwhelm ระบบเร็วกว่า fibers สำหรับการสำรวจลึก แนะนำ presentation The Journey to One Million โดย Samuel Williams ของ Ruby Core Team

Guild - Parallel Programming ใน Ruby

จนนี้เราเห็น 2 tools concurrency ใน Ruby ไม่มีตัวไหนปรับปรุงประสิทธิภาพ pure computations สำหรับนั้นต้องการ true parallelism ซึ่งไม่มีใน Ruby (พิจารณา MRI)

อาจเปลี่ยนใน Ruby 3 ด้วย feature ใหม่ชื่อ "Guilds" รายละเอียดยัง unclear ในส่วนต่อไป เราจะดูว่า work-in-progress feature นี้สัญญา allow parallelism ใน Ruby อย่างไร

Guilds อาจทำงานอย่างไร

แหล่งปัญหาใหญ่ใน concurrent/parallel solutions คือ shared memory ในส่วน threads เราเห็นว่าง่ายมากที่จะสะดุดและเขียนโค้ดที่ดูโศกนุ่ยแตกต่างจริงๆ มี subtle bugs

Koichi Sasada — Ruby Core Team member หัวหน้าพัฒนา Guild feature — กำลัง.design solution ที่ tackle อันตรายของ shared memory ระหว่าง multiple threads ตรงๆ ในการนำเสนอที่ RubyConf 2018 เขาอธิบายว่าใช้ guilds จะไม่สามารถ share mutable objects ได้แค่ immutable objects เท่านั้นที่ share ระหว่าง guilds ต่างๆ

Specialized data structures จะเพิ่มใน Ruby เพื่อ shared memory ระหว่าง guilds บ้าง แต่รายละเอียดยังไม่ fleshed out สมบูรณ์ จะมี API ให้ copy/move objects ระหว่าง guilds และ safeguard ป้องกัน object ถูก reference หลัง move ไป guild อื่น

ใช้ Guilds สำรวจสถานการณ์ทั่วไป

มีหลายสถานการณ์ที่อยาก speed up computations ด้วย parallel จินตนาการต้องคำนวณ average และ mean ของ dataset เดียวกัน

ตัวอย่างด้านล่างแสดงวิธีทำกับ guilds จะจำว่า code นี้ยังไม่ทำงาน และอาจไม่ทำงานเลยแม้หลัง guilds released

# A frozen array of numeric values is an immutable object.
dataset = [88, 43, 37, 85, 84, 38, 13, 84, 17, 87].freeze

# The overhead of using guilds will probably be
# considerable, so it will only make sense to
# parallelize work when a dataset is large / when
# performing lots of operations.
g1 = Guild.new do
  mean = dataset.reduce(:+).fdiv(dataset.length)
  Guild.send_to(:mean, Guild.parent)
end

g2 = Guild.new do
  median = Median.calculate(dataset.sort)
  Guild.send_to(:median, Guild.parent)
end

results = {}

# Every Ruby program will be run inside a main guild;
# therefore, we can also receive messages in the main
# section of our program.
Guild.receive(:mean, :median) do |tag, result|
  results[tag] = result
end

สรุป

Concurrency และ parallelism ไใช่จุดแข็งของ Ruby แต่มันก็มี tools ที่พอดีกับ most use cases Ruby 3 จะมา และดูเหมือนจะดีขึ้นมากด้วย Guild primitive ตามความคิดผม Ruby ยังคงเป็นตัวเลือกเหมาะสมในหลายสถานการณ์ และ community กำลังทำงานให้ภาษาดีขึ้น มาติดตามสิ่งที่จะมาถึง!