Lập trình Đà Nẵng: Sử dụng coroutine trong python để cài đặt thuật toán điều phối các request

Một trong các vấn đề của một hệ thống backend là bài toán điều phối request tới các nguồn dữ liệu.

Xét bài toán với một hệ thống blog như Kipalog:
Mỗi một User có nhiều bài Post
Mỗi một Post có nhiều Comment
Mỗi Comment lại thuộc về một User khác nhau.

Một trong những tác vụ chính của Kipalog là render một bài Post. Khi render Post cũng cần hiển thị
luôn danh sách các comment, và tác giả của comment đó. Nếu mọi việc chỉ dừng lại ở dây, thì không có gì để bàn. Bạn đơn giản chỉ cần thực hiện một câu query join 3 bảng Post, Comment, và User
là có thể trả về dữ liệu để render Post.

Nhưng do admin và cộng đồng của Kipalog rất tích cực tham gia vào phong trào #hardcore, nên chất lượng các bài viết của Kipalog ngày càng tốt. Chính vì thế càng ngày càng nhiều user tham gia vào Kipalog hơn. Các bài viết càng ngày càng nhiều hơn. Việc lưu trữ tất cả dữ liệu trên một Database là không còn đủ nữa. Các kỹ sư của Kipalog phải thực hiện nhiều việc thay đổi kiến trúc hệ thống bằng cách cài đặt việc sharding database. Giờ đây một một bảng User, Post, Comment được tổ chức nằm trên một database riêng biệt. Đến lúc này việc render một Post đã không còn đơn giản như xưa, đoạn code để trả về dữ liệu có thể mô tả bằng Python như sau:

def get_post_info(pid):
   post = db.get_post(pid)                               
   cids = db.get_post_comment_ids(pid)
   comments = [get_comment(cid) for cid in cids]
   return post, comments

def get_comment(cid):
   comment = db.get_comment(cid)
   user = db.get_user(comment.user_id)
   comment.user = user
   return comment

Trong đoạn code trên, giả sử rằng các hàm db.get_<something> là những hàm sẽ gọi và Database để query dữ liệu. Với cách implement như trên để render được một post, sẽ cần tới 2 + 2N query vào database (1 cho việc lấy get_post, 1 cho get_post_comment_ids, N cho lấy nội dung comment, và N cho lấy thông tin của người comment). Rõ ràng đây là cách không thể nào chấp nhận được!

Với đoạn code trên, có thể thấy mấy điểm để cải thiện như sau: (1) việc get_post và get_post_comment_ids là có thể thực hiện được song song với nhau. (2) thay vì chạy tuần tự việc lấy ra các comment và user, chúng ta có thể chạy chúng song song với nhau. Hoặc tốt hơn, nếu như database hỗ trợ việc batching (ví dụ data được cached vào memcached), thì có thể gộp các cid và user_id lại với nhau để chỉ cần thực hiện 2 requests (việc gộp các request này lại với nhau được gọi là batching)

Để cài đặt giải pháp nói trên cần phải refactor lại code, đồng thời nó cũng phá vỡ tính module hoá của từng hàm, khiến cho các hàm khó dùng lại hơn.

Việc điều phối các request để lấy dữ liệu từ những nguồn khác nhau như nói ở trên là bài toán rất phổ biến đối với các công ty có dữ liệu lớn. Một giải pháp tốt của bài toán này cần implement một framework để giúp cho các lập trình viên ứng dụng: không cần phải suy nghĩ nhiều về việc dữ liệu được fetch như nào, thứ tự ra sao, đồng thời giúp cho việc viết code trở nên đơn giản và hiệu quả.

Có một vài kết quả nghiên cứu đã được đưa ra và cài đặt thành framework cụ thể. Ví dụ: Facebook sử dụng Applicative Fuctor để cài đặt thư viện Haxl, Twitter sử dụng Free monad để cài đặt framework Stitch (rất tiếc rằng Stitch chưa được opensource), và mới đây nhất là Quora đưa ra thư viện Asynq sử dụng Python Coroutine để batching request (

Bài viết này sẽ mô tả thuật toán được sử dụng trong thư viện Asynq cũng như đề xuất một vài cải tiến cho thư viện này.

Chú ý: Tại thời điểm bài viết được thực hiện, Repo Asynq mới chỉ có một commit (https://github.com/quora/asynq/tree/c25fe0834ca64b35b2cd4a4e64666d0b5423c44f). Bài viết chỉ mô tả thuật toán được sử dụng tại thời điểm này.

Coroutine là một hàm có thể có nhiều entry point. Nói cách khác, hàm này có thể được pause và resume lại trong khi chạy.

Trong Python, Coroutine được implement bằng Generator. Chính vì thế, từ đây về sau, thay vì dùng Coroutine, bài viết sẽ dùng Generator

Sau đây là một ví dụ về Generator như sau:

def gen():
   yield 1
   yield 2
   yield 3

g = gen()
g.next() # 1
g.next() # 2
g.next() # 3
g.next() # StopIterator

Các entry point của gen đó là các dòng có lời gọi lệnh yield. Khi lệnh yield được gọi, con trỏ lệnh đang từ hàm gen sẽ được trả về hàm caller của gen. Lúc này caller có thể thực hiện chạy gen tiếp hoặc là chạy hàm khác. Bất cứ hàm nào mà trong thân hàm có sử dụng yield đều được gọi là Generator trong Python.

Một điểm quan trọng của Generator trong Python đó là caller có thể truyền giá trị vào trong gen, bằng cách gọi hàm gen.send.

Ví dụ

def gen():
   a = yield 1
   yield a + 10


g = gen()
value = g.send(None)   # 1
g.send(value)          # 11

Luôn luôn cần send giá trị None như giá trị khởi đầu vào một Generator. Bằng cách lưu lại kết quả trước đó của hàm g.send, chúng ta có thể truyền lại giá trị đó vào lần chạy tiếp theo của Generator.

Tới đây bạn có thể thắc mắc vậy yield thì có tác dụng gì.

Tôi sẽ giới thiệu cho các bạn một cách tư duy mà tôi gọi là tư duy hardcore – Đây là cách tư duy mà chúng tôi (những thành viên của channel #hardcore trên RubyVietnam) rất hay sử dụng. Tư tưởng của cách tư duy này là: Chúng tôi cho rằng tất cả các bài toán ở tầng ứng dụng phần lớn đều đã có lời giải nhưng ở mức độ low level hơn, nên nếu bạn tìm hiểu đủ về cách CPU điều phối giữa các core, cách compiler traslate code của bạn sang assembly, hay cách Kernel handle resource ra sao, bạn có thể hiểu và giải quyết hầu hết các bài toán ở tầng ứng dụng khác. Nếu bạn thấy hứng thú, hãy tham gia vào channel ngày nhé

Quay trở yield, nếu sử dụng bài toán tương tự từ việc thực hiện các chỉ thị của CPU, chúng ta sẽ thấy bằng cách sử dụng yield, chúng ta đã chia việc Execute một Generator thành rất nhiều steps. Do đó khi cần Execute nhiều Generators cùng lúc, chúng ta có thể chạy các Generators này theo cơ chế PipeLine. Giả sử chúng ta có 2 Generators G1, G2. Mỗi generator có 5 Steps. Thay vì phải chạy G1 xong rồi mới chạy đến G2, chúng ta có thể pipeline bằng cách chạy (G1 Step1) (G2 Step1) (G1 Step 2) (G2 Step2), …Việc phân chia các step là dựa vào việc gọi hàm yield trong Generator.
Hay tốt hơn nữa, với các step của G1 và G2 mà có thể gộp lại, chúng ta có thể sắp xếp chúng theo thứ tự sao cho có thể gộp chung lại được, hoặc với những step liên quan tới IO, chúng ta có thể để chúng chạy song song. Đây chính là nền tảng của thuật toán trong thư viện Asynq

Quay trở lại ví dụ ở phần đầu, bằng Asynq với yield chúng ta có thể viết lại code như sau:

from asynq import async

@async()
def get_post_info(pid):
   post = yield db.get_post.async(pid)                               
   cids = yield db.get_post_comment_ids.async(pid)
   futures = [get_comment(cid) for cid in cids]
   comments = yield futures
   raise StopIterator((user, comments))

@async()
def get_comment(cid):
   comment = yield db.get_comment(cid)
   user = yield db.get_user(comment.user_id)
   comment.user = user
   raise StopIterator(comment)


result = get_post_info.async(pid)()

Đoạn code trên chỉ thay đổi đôi chút so với đoạn code ở phần đầu, nhưng khi chạy với Asynq, việc gộp chung các request đê lấy Comments và lấy User của các comments đó là được thực hiện tự động.

Để làm việc này, Asynq tạo bao lấy các Generator bằng các Future object. Future object là một object mà việc tính toán ra dữ liệu của nó là chưa được thực hiện ngay.

Future object được tạo ra bởi việc gọi là Future object được tạo ra khi gọi decorator asynccủa function, hoặc gọi lệnh yield từ trong Generator. Mỗi khi Future object được tạo ra từ lênh yield, Future object đó sẽ tạo ra một mối quan hệ phụ thuộc với caller của nó. Một Future object chỉ có kết quả khi mà các Future con của nó đều đã chạy và ra kết quả

Trong Asynq có 2 loại Future objects cần chú ý:

AsyncTask: Là các task mà bao lấy các generator. Mỗi một task sẽ chứa một generator.
BatchItemBase: là Future object mà việc thực hiện nó có thể đươc gộp lại với nhau. Mỗi một BatchItem sẽ thuộc về một Batch object. Việc thực hiện batching các item sẽ được thực hiện bởi Batch object.

Việc chạy các Future object sẽ được đẩy cho một scheduler. Scheduler sẽ được kích hoạt khi chúng ta gọi một async task như result = get_post_info.async(pid)(). Ở đây get_post_info.async(pid) là một AsyncTask. Việc gọi một async task sẽ tương đương với việc gọi hàm scheduler để schedule task đó.

Scheduler sẽ chứa 2 queues:

queue _tasks chứa các task cần schedule
queue _batches chứa các batch cần schedule.

Ở trạng thái khởi tạo, cả 2 queue này đều rỗng, không chưa bất cứ một phần tử nào.

Khi Scheduler được yêu cần schedule một list các task, scheduler sẽ đẩy các task vào trong queue _tasks. Sau đó nó sẽ lấy các task ra lần lượt. Mỗi khi một task được lấy khỏi queue _task, task đó sẽ được execute bằng cách gọi hàm generator.send để execute từng step của Generator (như mô tả ở phần 2). Khi AsyncTask nhận được kết quả từ generator, kết quả này có thể là một trong 3 loại:

Một AsyncTask khác. Khi đó AsyncTask mới sẽ là AsyncTask con của AsyncTask kể trên. AsyncTask con này sẽ được đẩy tiếp vào queue _tasks của scheduler
Một BatchItemBase, khi đó batch object của BatchItemBase sẽ được đẩy vào trong queue _batches của scheduler. Chú ý rằng chúng
Một giá trị cụ thể, tức là scheduler đã hoàn thành nhiệm vụ của nó.

Scheduler sau khi schedule các task từ queue _tasks, nó sẽ schedule tiếp các object từ trong queue _batches. Các batch này sẽ được thực hiện tuần tự bằng cách gọi hàm flush của từng batch.

Với thuật toán như trên, và đoạn code như chúng ta mô tả ở phần đầu của mục 3. Nếu các tác vụ db.get_post.async được cài đặt như các BatchItemBase, chúng ta sẽ có những batch sau:

batch 1: db.get_post.async với tham số pid
batch 2: db.get_post_comment_ids.async với tham số pid
batch 3: db.get_comment với N items có các tham số từ mảng cids
batch 4: db.get_user với N items có các tham số từ mảng user_ids

Như thế, chúng tả chỉ mất 4 requests thay vì 2 + 2N như ban đầu.

Một điểm Asynq làm chưa tốt đó là khi schedule các batchs chúng ta có thể schedules chúng song song với nhau. Ví dụ batch 1 và batch 2 có thể thực hiện song song được.

Tất nhiên để làm được việc đó, đoạn code ở trên (phần đầu của 3) cũng phải thay đổi một chút (yield 2 futures get_post và get_post_comment_ids cùng lúc) và đồng thời cần phải implement một cơ chế EventLoop trong Asynq để việc thực hiện các batch sẽ đẩy về EventLoop

Việc sử dụng Generator để scheduler các request không phải là idea mới trong Python. Thư viện NDB của Google App Engine đã implement cơ chế giống như Asynq với EventLoop từ những năm 2012. Tuy nhiên source code của NDB phức tạp và khó đọc hơn rất nhiều so với Asynq.

Bằng mô hình coroutine, Asynq sử dụng một thuật toán khá đơn giản để điều phối các request: cố gắng batching càng nhiều request càng tốt.
Áp dụng phương pháp điều phối này cùng với GraphQL tôi nghĩ chúng ta có thể build được những backend vừa rất mềm dẻo trong việc trả về dữ liệu cho client, lại vừa có performance tốt mà chi phí cho việc viết code là không lớn (không thay đổi nhiều so với cách viết truyền thống).

Bài viết này xin dừng lại ở đây, bài viết sau tôi sẽ trình bày thêm về cách tiếp cận sử dụng Free Monad để implement một thuật toán điều phối khác.

Bạn muốn học lập trình hãy liên hệ ngay với Mỹ Vân để được nhận ưu đãi từ học viện nhé