1. Khái niệm Tiến trình và Trạng thái

Tiến trình là gì?

Một chương trình (ví dụ: script.py) là một thực thể tĩnh, chỉ là tập hợp các dòng lệnh. Khi chương trình được thực thi, hệ điều hành sẽ tải nó vào bộ nhớ và cấp phát tài nguyên (như CPU, RAM, I/O) để nó chạy. Lúc này, chương trình đang hoạt động được gọi là một tiến trình (process). Mỗi tiến trình là một đơn vị cơ bản để hệ điều hành phân bổ tài nguyên.

Không chỉ luồng (thread) mới thực hiện được đa nhiệm, tiến trình cũng có khả năng này.

Các trạng thái của tiến trình (Mô hình 5 trạng thái)

Trong thực tế, số lượng tác vụ thường lớn hơn số lõi CPU. Vì vậy, tại một thời điểm, chỉ một số tác vụ được thực thi, số còn lại phải chờ đợi. Điều này dẫn đến các trạng thái khác nhau của tiến trình:

• Sẵn sàng (Ready): Tiến trình đã có đủ mọi điều kiện để chạy, chỉ đang chờ CPU cấp phát thời gian thực thi.
• Đang chạy (Running): CPU đang thực thi các lệnh của tiến trình.
• Chờ đợi (Waiting/Blocked): Tiến trình đang chờ một sự kiện nào đó xảy ra, ví dụ như chờ nhập liệu từ bàn phím, chờ kết thúc thao tác đọc/ghi file, hoặc chờ một khoảng thời gian (ví dụ: lệnh sleep).

2. Tạo tiến trình với multiprocessing

Module multiprocessing trong Python cung cấp một cách tiếp cận nhất quán trên nhiều nền tảng để tạo và quản lý các tiến trình. Lớp Process đại diện cho một đối tượng tiến trình.

Ví dụ 1: Chạy đồng thời hai vòng lặp while

from multiprocessing import Process
import time

def worker_function():
    """Mã sẽ được thực thi bởi tiến trình con"""
    while True:
        print("----Tác vụ con đang chạy----")
        time.sleep(1)

if __name__ == '__main__':
    # Tạo một đối tượng Process, target trỏ đến hàm cần thực thi
    child_process = Process(target=worker_function)
    # Khởi động tiến trình con
    child_process.start()
    
    while True:
        print("----Tác vụ cha đang chạy----")
        time.sleep(1)

Giải thích: Để tạo một tiến trình con, ta chỉ cần truyền tên hàm cần thực thi vào tham số target của lớp Process và gọi phương thức start().

Ví dụ 2: Lấy Process ID (PID)

from multiprocessing import Process
import os
import time

def child_action():
    """Hàm này chạy trong tiến trình con"""
    print(f'Tiến trình con đang chạy, PID: {os.getpid()}')
    print('Tiến trình con sẽ kết thúc...')

if __name__ == '__main__':
    print(f'PID của tiến trình cha: {os.getpid()}')
    proc = Process(target=child_action)
    proc.start()

Cấu trúc của lớp Process

Cú pháp khởi tạo: Process([group [, target [, name [, args [, kwargs]]]]])

• target: Hàm hoặc phương thức sẽ được tiến trình con thực thi.
• args: Tham số dạng tuple để truyền cho hàm target.
• kwargs: Tham số dạng dictionary (từ khóa) để truyền cho hàm target.
• name: Tên của tiến trình (tùy chọn).
• group: Nhóm tiến trình (ít khi sử dụng).

Các phương thức thông dụng:

• start(): Khởi động tiến trình.
• is_alive(): Kiểm tra xem tiến trình có đang hoạt động không.
• join([timeout]): Chờ tiến trình con kết thúc. Có thể chờ tối đa timeout giây.
• terminate(): Kết thúc tiến trình ngay lập tức.

Các thuộc tính thông dụng:

• name: Tên của tiến trình.
• pid: Process ID của tiến trình.

Ví dụ 3: Truyền tham số cho tiến trình con

from multiprocessing import Process
import os
from time import sleep

def run_task(user, years, **extras):
    for i in range(10):
        print(f'Tiến trình con: user={user}, tuổi={years}, PID={os.getpid()}')
        print(extras)
        sleep(0.2)

if __name__ == '__main__':
    p = Process(target=run_task, args=('Alice', 25), kwargs={"skill": "Python"})
    p.start()
    sleep(1)  # Đợi 1 giây rồi kết thúc tiến trình con
    p.terminate()
    p.join()

Ví dụ 4: Tiến trình không chia sẻ biến toàn cục

from multiprocessing import Process
import os
import time

global_data = [10, 20]

def modify_data():
    """Tiến trình con thay đổi dữ liệu"""
    print(f"Tiến trình 1 (PID={os.getpid()}): Dữ liệu ban đầu {global_data}")
    for i in range(3):
        global_data.append(i)
        time.sleep(1)
        print(f"Tiến trình 1 (PID={os.getpid()}): Dữ liệu hiện tại {global_data}")

def read_data():
    """Tiến trình con chỉ đọc dữ liệu"""
    print(f"Tiến trình 2 (PID={os.getpid()}): Dữ liệu đọc được {global_data}")

if __name__ == '__main__':
    p1 = Process(target=modify_data)
    p1.start()
    p1.join()

    p2 = Process(target=read_data)
    p2.start()

Kết quả: Tiến trình p2 sẽ thấy global_data là [10, 20], không bị ảnh hưởng bởi các thay đổi của p1. Điều này chứng minh mỗi tiến trình có không gian bộ nhớ riêng biệt.

3. So sánh Tiến trình và Luồng (Thread)

Hãy tưởng tượng một nhà máy: Tiến trình là một dây chuyền sản xuất hoàn chỉnh, còn luồng là các công nhân trên dây chuyền đó. Mỗi dây chuyền có không gian riêng, nhưng các công nhân trên cùng một dây chuyền thì làm việc chung với nhau.

4. Giao tiếp giữa các Tiến trình với Queue

Do mỗi tiến trình có không gian bộ nhớ riêng, để trao đổi dữ liệu, chúng ta cần các cơ chế giao tiếp liên tiến trình (IPC - Inter-Process Communication). multiprocessing.Queue là một công cụ đơn giản và hiệu quả cho mục đích này.

Cách sử dụng Queue

from multiprocessing import Queue

# Tạo một hàng đợi có tối đa 3 phần tử
message_queue = Queue(3)

message_queue.put("Tin nhắn 1")
message_queue.put("Tin nhắn 2")
print(message_queue.full())  # False

message_queue.put("Tin nhắn 3")
print(message_queue.full())  # True

# Thử đặt thêm khi hàng đợi đầy
try:
    message_queue.put("Tin nhắn 4", True, 2)  # Chờ tối đa 2 giây
except:
    print(f"Hàng đợi đã đầy. Số lượng hiện tại: {message_queue.qsize()}")

# Đọc lần lượt các tin nhắn
while not message_queue.empty():
    print(message_queue.get_nowait())

Các phương thức chính của Queue:

• qsize(): Trả về số lượng phần tử hiện có.
• empty(): Kiểm tra hàng đợi rỗng hay không.
• full(): Kiểm tra hàng đợi đã đầy hay chưa.
• get([block[, timeout]]): Lấy và xóa một phần tử khỏi hàng đợi.
  • block=True, timeout=None: Chờ cho đến khi có phần tử.
  • block=False: Nếu rỗng, ném ngoại lệ Queue.Empty.
• put(item, [block[, timeout]]): Thêm một phần tử vào hàng đợi.
  • block=True, timeout=None: Chờ cho đến khi có chỗ trống.
  • block=False: Nếu đầy, ném ngoại lệ Queue.Full.

Ví dụ ứng dụng Queue để giao tiếp

from multiprocessing import Process, Queue
import os, time, random

def produce(q):
    """Tiến trình ghi dữ liệu"""
    for item in ['X', 'Y', 'Z']:
        print(f'Đặt {item} vào hàng đợi...')
        q.put(item)
        time.sleep(random.random())

def consume(q):
    """Tiến trình đọc dữ liệu"""
    while True:
        if not q.empty():
            value = q.get(True)
            print(f'Lấy {value} từ hàng đợi.')
            time.sleep(random.random())
        else:
            break

if __name__ == '__main__':
    q = Queue()
    producer = Process(target=produce, args=(q,))
    consumer = Process(target=consume, args=(q,))

    producer.start()
    producer.join()

    consumer.start()
    consumer.join()

    print('Hoàn tất trao đổi dữ liệu.')

5. Quản lý tiến trình với Pool (Bể tiến trình)

Khi số lượng tiến trình cần tạo rất lớn (ví dụ hàng trăm), việc tạo từng đối tượng Process một cách thủ công sẽ không hiệu quả. Pool cho phép quản lý một nhóm các tiến trình làm việc (worker processes).

• Giới hạn số lượng: Bạn chỉ định số lượng tối đa các tiến trình trong pool.
• Tự động quản lý: Khi có tác vụ mới, nếu trong pool còn tiến trình rảnh, nó sẽ được giao việc ngay. Nếu tất cả đều bận, tác vụ sẽ phải xếp hàng chờ.

Ví dụ minh họa

from multiprocessing import Pool
import os, time, random

def task_handler(msg):
    start = time.time()
    print(f"{msg} bắt đầu, PID: {os.getpid()}")
    time.sleep(random.random() * 2)
    end = time.time()
    print(f"{msg} hoàn thành, mất {end - start:.2f} giây")

# Tạo pool với tối đa 3 tiến trình
process_pool = Pool(3)

for i in range(10):
    # apply_async: gửi tác vụ không đồng bộ (không chặn)
    process_pool.apply_async(task_handler, args=(i,))

print("---- Bắt đầu gửi tác vụ ----")
process_pool.close()  # Không nhận thêm tác vụ mới
process_pool.join()   # Chờ tất cả tác vụ hoàn thành
print("----- Kết thúc -----")

Các phương thức quan trọng của Pool:

• apply_async(func, args, kwds): Gửi tác vụ không đồng bộ (không chặn chương trình chính).
• close(): Đóng pool, không cho phép gửi thêm tác vụ.
• terminate(): Kết thúc tất cả tiến trình trong pool ngay lập tức.
• join(): Chương trình chính đợi cho đến khi tất cả tiến trình trong pool kết thúc. Phải gọi sau close() hoặc terminate().

Sử dụng Queue với Pool

Khi dùng Pool, ta không thể dùng multiprocessing.Queue() trực tiếp mà phải dùng multiprocessing.Manager().Queue().

from multiprocessing import Manager, Pool
import os, time

def writer(q):
    print(f"Writer (PID: {os.getpid()}) bắt đầu")
    for char in "hello":
        q.put(char)

def reader(q):
    print(f"Reader (PID: {os.getpid()}) bắt đầu")
    while not q.empty():
        msg = q.get(True)
        print(f"Reader nhận: {msg}")

if __name__ == "__main__":
    print(f"Chương trình chính (PID: {os.getpid()}) bắt đầu")
    shared_queue = Manager().Queue()
    pool = Pool()

    pool.apply_async(writer, (shared_queue,))
    time.sleep(1)  # Đợi writer hoàn thành
    pool.apply_async(reader, (shared_queue,))

    pool.close()
    pool.join()
    print("Chương trình chính kết thúc")

6. Ứng dụng thực tế: Sao chép thư mục đa tiến trình

import multiprocessing
import os
import time
import random

def copy_file_process(queue, file_name, src_dir, dest_dir):
    """Sao chép một file từ thư mục nguồn sang thư mục đích"""
    with open(os.path.join(src_dir, file_name), "rb") as f_src:
        with open(os.path.join(dest_dir, file_name), "wb") as f_dest:
            while True:
                time.sleep(random.random())  # Giả lập thời gian sao chép
                data = f_src.read(1024)
                if not data:
                    break
                f_dest.write(data)
    # Báo cáo hoàn thành
    queue.put(file_name)

def main():
    src = input("Nhập tên thư mục nguồn: ").strip()
    dest = src + "_copy"

    try:
        os.mkdir(dest)
    except FileExistsError:
        pass

    files = [f for f in os.listdir(src) if os.path.isfile(os.path.join(src, f))]
    total_files = len(files)

    progress_queue = multiprocessing.Manager().Queue()
    pool = multiprocessing.Pool(4)

    for f in files:
        pool.apply_async(copy_file_process, args=(progress_queue, f, src, dest))

    pool.close()

    completed = 0
    while completed < total_files:
        # Nhận tên file vừa được sao chép xong
        done_file = progress_queue.get()
        completed += 1
        percent = (completed / total_files) * 100
        print(f"\rTiến độ: {percent:.2f}% - Đã xong: {done_file}" + " " * 30, end="", flush=True)

    pool.join()
    print("\nSao chép hoàn tất!")

if __name__ == "__main__":
    main()