Giới Thiệu Về Celluloid - Part 2

Đây là phần 2 trong series về Celluloid, nếu các bạn chưa xem phần 1, hãy xem ở đây nhé:

Giới thiệu về Celluloid - Part 1

Celluloid có rất nhiều những công cụ hữu ích giúp việc lập trình đồng thời (concurrent programming) trở nên dễ dàng hơn bao giờ hết (honho)

Futures

Rất nhiều trường hợp chúng ta không muốn bỏ đi giá trị trả lại của một method vừa call ở một actor, mà muốn giữ lại để sử dụng tại một nơi khác. Để đáp ứng cho yêu cầu này, Celluloid cung cấp futures.

Thử viết một đoạn script để tính toán giá trị SHA1 checksum của một mảng các files rồi output ra console.

require 'celluloid/current'
require 'digest/sha1'

class SHAPutter
  include Celluloid

  def initialize(filename)
    @filename = filename
  end

  def output(checksum_future)
    puts "#{@filename} - #{checksum_future.value}"
  end

  def checksum
    @file_contents = File.read(@filename)
    Digest::SHA1.hexdigest @file_contents
  end

end

files = ["test1.txt", "test2.txt", "test3.txt", "test4.txt", "test5.txt"]

files.each do |file|
  sha = SHAPutter.new file
  checksum_future = sha.future :checksum
  sha.output checksum_future
end

Đầu tiên, hãy nhìn qua method checksum. Rất đơn giản, chỉ là sử dụng Digest::SHA1 để tính toán checksum cho nội dung file mà actor nhận.

Điều thú vị nằm ở trong vòng lặp (honho)

Sau khi chỉ định file cho actor, thay vì chỉ gọi checksum một cách đơn thuần, chúng ta sử dụng future. Bằng cách này, ngay lập tức một object Celluloid::Future được return. Tiếp đó, future object được chuyển cho method output bên trong actor. output cần giá trị checksum, nên nó sẽ lấy từ giá trị của method trong future object.

Có thể các bạn sẽ đặt câu hỏi "ơ ví dụ này chả khác j ví dụ cuối của Part 1 cả?" tuy nhiên nếu nhìn kỹ, bạn sẽ thấy ở ví dụ trước, để không đồng bộ các xử lý, chúng ta đã phải gộp hết tất cả vào một method duy nhất. Còn với futures, chúng ta có thể tách các đoạn code ra thoải mái (dance2)

Ngoài ra, có những trường hợp bắt buộc phải dùng futures. Ví dụ nếu bạn muốn viết một thư viện, thì giá trị của hàm checksum phải là future kể từ khi người sử dụng add vào source code của họ.

Tạo các block chạy đồng thời

Với futures, chúng ta có thể đẩy các code block sang thread khác ngon lành (honho)

require 'celluloid'

def some_method(future)
  #do something crazy
  val = future.value
  #do something with val
end

future = Celluloid::Future.new do
  #incredibly complex computation
end

some_method(future)

Sử dụng Celluloid::Future để đẩy block đó sang thread riêng , Celluloid sẽ quản lý mọi thứ liên quan đến thread đó, để chúng ta có thể sử dụng giá trị return vào sau này. Chức năng này của Celluloid có thể đưa vào bất cứ một chương trình nào, và nếu bạn có thể thành thạo được thì sẽ vô cùng hữu ích đấy (yeah)

Catching errors với Supervisors

Trong một ngày đẹp trời thì các threads chạy mượt mà không lỗi, thế nhưng hôm mưa nào đó tự dưng có lỗi xảy ra thì sao? Chúng ta phải làm gì để có thể kiểm soát được tình huống đó? (huhuhu)

Celluloid cung cấp cho chúng ta một giải pháp được gọi là supervisor.

Thử áp dụng với ví dụ ở đầu bài viết:

class SHAPutter
  include Celluloid

  def initialize(filename)
    @filename = filename
  end

  def output(checksum_future)
    puts "#{@filename} - #{checksum_future.value}"
  end

  def checksum
    @file_contents = File.read(@filename)
    Digest::SHA1.hexdigest @file_contents
  end

end

files = ["test1.txt", "test2.txt", "test3.txt", "test4.txt", "test5.txt"]

files.each do |file|
  supervisor = SHAPutter.supervise_as :sp, file
  sha = Celluloid::Actor[:sp]
  checksum_future = sha.future :checksum
  sha.output checksum_future
end

Class SHAPutter không thay đổi gì cả, như vậy là những đoạn code về business logic sẽ không phải chỉnh sửa gì (thatlatuyetvoi)

Giờ đến method supervise_as được gọi trong vòng lặp. Method này sẽ thực hiện 3 điều. Thứ nhất, nó tạo một actor là instance của SHAPutter. Thứ hai, nó trả về một supervisor object. Thứ ba, nó lấy tham số đầu tiên (ở đây là :sp) và đặt tham số đó vào registry.

Celluloid registry giống như cuốn danh bạ điện thoại vậy - bạn có thể sử dụng actors trong đó dựa theo tên. Do đó, ở dòng code tiếp theo, chúng ta sử dụng :sp để tìm actor đó trong registry.

Vậy là chỉ với 2 dòng code thêm vào, Celluloid đã tự động giúp chúng ta trong việc restart và track theo các actors khi chúng crash. Khi có một actors bị exception, actor đó sẽ ngay lập tức đc restart bởi Celluloid core.

Communication giữa các Actors

Trong hầu hết các chương trình, actors không làm việc trong một môi trường cách ly mà sẽ phải giao tiếp với các actors khác.

Ví dụ một cách đơn giản, chúng ta sẽ thử dùng 3 actors để in ra dòng chữ I am superman:

require 'celluloid/current'

class FirstActor
  include Celluloid
  def say_msg
    print "I "
    Celluloid::Actor[:second].say_msg
  end
end

class SecondActor
  include Celluloid
  def say_msg
    print "am "
    Celluloid::Actor[:third].say_msg
  end
end

class ThirdActor
  include Celluloid
  def say_msg
    print "superman.
"
  end
end

Celluloid::Actor[:second] = SecondActor.new
Celluloid::Actor[:third] = ThirdActor.new
FirstActor.new.say_msg

FirstActor sử dụng registry để tìm instance của SecondActor và gọi say_msg của đối tượng vừa tìm được, sau đó SecondActor lại làm tương tự với ThirdActor.

Tóm lại, giao tiếp giữa các actors được thực hiện qua actor Registry, nơi chúng ta có thể đặt tên cho các actors.

Ngoài ra, còn có một phương pháp nữa giúp các actors có thể làm việc cùng nhau. Đó là sử dụng futures để truyền giá trị trả về giữa các actors.

Blocking call trong actors

Với Celluloid, các actors được đặt trong các threads của mình, do đó việc thực hiện gọi các methods gây block sẽ không gặp vấn đề gì, nó chỉ block một actor đó thôi.

Tuy nhiên phải cẩn thận với việc gọi những blocks vô hạn trong actors, điều đó sẽ khiến cho các actors đó không thể tiếp tục nhận được các messages khác.

Pooling

Pools của Celluloid thực sự là tuyệt vời (tanghoa)

Chúng ta sẽ xem thử một ví dụ về sử dụng Pools của Celluloid:

require 'celluloid/current'
require 'mathn'

class PrimeWorker
  include Celluloid

  def prime(number)
    if number.prime?
      puts number
    end
  end
end

pool = PrimeWorker.pool

(2..1000).to_a.map do |i|
  pool.async.prime i
end

sleep 100

Method prime trong PrimeWorker sẽ in ra nếu số đó là số nguyên tố.

Điều thú vị nằm ở dòng code sử dụng pool của PrimeWorker. Object pool mang đủ các methods của PrimeWorker, số lượng instances của PrimeWorker được tạo ra ngang số nhân của CPU. Nếu bạn sử dụng một quad core CPU, sẽ có 4 actors được tạo ra. Celluloid sẽ quyết định actor nào rời khỏi pool để xử lý.

Wow, vậy là chỉ cần 4-5 dòng code, bạn đã có thể tạo ra các xử lý đồng thời (concurrent) để chia sẻ khối lượng công việc cho processor của bạn. (thatlatuyetvoi)

Cuối cùng là việc call hàm sleep ở cuối chương trình. Do hàm prime được gọi không đồng bộ (asynchronously) nên nếu thread chính kết thúc trước khi các actors kịp xử lý thì số nguyên tố của actors đó sẽ không thể được in ra màn hình. Câu lệnh sleep chính là để thread chính có thể tồn tại đủ lâu cho đến khi tất cả các actors thực hiện xong xử lý của mình. (yeah)

Mọi thứ sẽ còn tiếp diễn ở phần 3 (hihi)

Source: An Introduction to Celluloid, Part II