Thêm Chức Năng Xử Lý Đa Yêu Cầu Song Song Cho Skynet

Khi thiết kế skynet, chúng tôi đã chủ động từ chối sử dụng mô hình callback để triển khai cơ chế yêu cầu-phản hồi. Theo quan điểm của tôi, việc sử dụng callback sẽ khiến luồng xử lý phản hồi bị tách rời khỏi luồng yêu cầu gốc, điều này gây khó khăn cho việc xây dựng các logic nghiệp vụ phức tạp. Đặc biệt là việc xử lý ngoại lệ trở nên kém hiệu quả.

Trong skynet hiện tại, khi một yêu cầu được khởi phát, luồng thực thi sẽ bị chặn cho đến khi nhận được phản hồi từ xa, sau đó tiếp tục xử lý trên cùng một luồng. Cách tiếp cận này đảm bảo tính nhất quán nhưng lại tiềm ẩn vấn đề về độ trễ khi thực hiện nhiều yêu cầu tuần tự. Vì trên cùng một luồng, bạn phải chờ phản hồi của yêu cầu trước rồi mới có thể gửi yêu cầu tiếp theo, dù thực tế các yêu cầu này hoàn toàn có thể xử lý song song.

Việc đồng thời khởi phát nhiều yêu cầu không liên quan hiện đang gặp nhiều bất tiện. Nếu phải tạo riêng một luồng cho mỗi yêu cầu, việc tổng hợp kết quả về một luồng xử lý duy nhất sẽ trở thành vấn đề phức tạp. Hiện tại, chúng ta chỉ có thể dùng các hàm fork/wait/wakeup để giải quyết, nhưng cách này khá rườm rà.

Nhu cầu xử lý song song này luôn tồn tại. Tôi đã dành nhiều thời gian nghiên cứu để tìm ra phương pháp tối ưu: cùng lúc gửi đi n yêu cầu, xử lý phản hồi theo thứ tự chúng đến, và chỉ tiếp tục luồng nghiệp vụ khi tất cả đã hoàn tất. Việc triển khai chức năng này trong skynet không quá phức tạp, nhưng thách thức nằm ở việc thiết kế API sao cho người dùng dễ sử dụng.

Ban đầu, tôi nghĩ đến việc sử dụng vòng lặp for để xử lý nhiều phản hồi (ví dụ dưới đây không phải cú pháp Lua chuẩn, chỉ mang tính minh họa):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

for index, ok, a,b,c,... in skynet.request(...)(...)(...) do
  if index == 1 then
    -- Xử lý yêu cầu thứ nhất
  elseif index == 2 then
    -- Xử lý yêu cầu thứ hai
  else
    assert(index == 3)
    -- Xử lý yêu cầu thứ ba
  end
end

Đoạn mã này cùng lúc khởi phát ba yêu cầu khác nhau, dự kiến nhận về ba phản hồi. Tuy nhiên thứ tự phản hồi không thể xác định, nên tôi dùng biến index để đánh dấu thứ tự. Tham số ok cho biết yêu cầu thành công hay thất bại.

Tuy nhiên cách này khiến yêu cầu và phản hồi bị tách biệt, dẫn đến bài toán khớp nối chúng lại. Tôi đã thử nhiều phương án, trong đó có ý tưởng sử dụng hàm select():

1
2
3
4
5
6
7
8
9

for req, ok, a, b, ... in skynet.select() do
  if req(...) then
    -- Xử lý yêu cầu thứ nhất
  elseif req(...) then
    -- Xử lý yêu cầu thứ hai
  elseif req(...) then
    -- Xử lý yêu cầu thứ ba
  end
end

Nếu thiết kế skynet.select() trả về một iterator, đồng thời đặt các yêu cầu vào bên trong vòng lặp, có thể giải quyết vấn đề tách biệt. Với kỹ thuật phù hợp, các yêu cầu chỉ được gửi đi khi gọi lần đầu, còn vòng lặp sẽ liên tục so khớp phản hồi.

Tuy nhiên tôi nhận ra cách này quá “kỹ thuật hóa”, khiến ngữ nghĩa trở nên mơ hồ. Nên tôi đã từ bỏ hướng tiếp cận này.

Tôi nhận ra giải pháp đơn giản nhất là dùng một chuỗi định danh cho mỗi yêu cầu, giúp người lập trình dễ dàng phân biệt các yêu cầu song song. Mở rộng hơn, nếu yêu cầu là một đối tượng (table), chính đối tượng đó có thể tự phân biệt chính mình. Từ đó hình thành giải pháp cuối cùng:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

local reqs = skynet.request()
local req1 = reqs:add { ... }
local req2 = reqs:add { ... }
for req, resp in reqs:select() do
  if req == req1 then
    -- Xử lý yêu cầu 1
  elseif req == req2 then
    -- Xử lý yêu cầu 2
  end
end

Hoặc phiên bản rút gọn:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

for req, resp in skynet.request 
  { ... , token = "req1" }
  { ... , token = "req2" }
   :select() do
  if req.token == "req1" then
    -- Xử lý yêu cầu 1
  elseif req.token == "req2" then
    -- Xử lý yêu cầu 2
  end
end

Ở đây, iterator trả về từ select() sẽ trả về cặp yêu cầu-phản hồi mỗi khi nhận được phản hồi. Người dùng có thể tự nhận diện qua token hoặc bất kỳ thuộc tính nào thêm vào yêu cầu. Nếu resp là nil, điều đó có nghĩa yêu cầu đã xảy ra lỗi.

Trong phiên bản đầu tiên, cơ chế select() có một số hạn chế. Từ góc độ trình điều phối, điểm chặn của nghiệp vụ nằm ở select(), nơi chờ phản hồi từ nhiều yêu cầu. Nhưng code xử lý phản hồi lại nằm trong thân vòng lặp. Nếu trong quá trình xử lý phản hồi lại phát sinh yêu cầu mới (gọi hàm chặn), trình điều phối sẽ không phân biệt được phản hồi tiếp theo thuộc về yêu cầu cũ hay yêu cầu mới.

Tuy nhiên hạn chế này có thể giải quyết dễ dàng bằng cách đặt toàn bộ select() vào một coroutine độc lập. Tức là sau khi gửi các yêu cầu, một coroutine riêng sẽ chờ phản hồi, mỗi khi nhận được phản hồi sẽ chuyển tiếp về luồng gốc để xử lý tiếp.

Với cơ chế này, việc thêm xử lý timeout trở nên đơn giản. Chỉ cần đồng thời gửi một yêu cầu hẹn giờ, khi nhận được tín hiệu hẹn giờ thì bỏ qua các phản hồi chưa đến. Hoặc ngược lại, nếu tất cả yêu cầu đều đã xử lý xong thì hủy bỏ hẹn giờ.

Chúng ta cần một quy trình rõ ràng để kết thúc vòng lặp select: gộp các xử lý phản hồi về luồng nghiệp vụ chính, bỏ qua các phản hồi chưa đến. Nếu mọi thứ diễn ra bình thường, iterator sẽ tự kết thúc. Nhưng nếu xảy ra ngoại lệ (break, error…), cần có cơ chế đóng luồng chủ động.

May mắn là Lua 5.4 cung cấp cơ chế “to be closed” để giải quyết vấn đề này. Nên tính năng