Khung Xử Lý Mạng Bất Đồng Bộ

Hệ thống này được xây dựng dựa trên hai mô-đun tiêu chuẩn của Python, cung cấp cơ chế xử lý I/O không chặn (non-blocking) cho các ứng dụng mạng TCP/IP.

asyncore

Mô-đun này đóng vai trò là lớp nền tảng, bao bọc trực tiếp các hàm select và socket của hệ điều hành. Nó duy trì một vòng lặp sự kiện (event loop) để giám sát trạng thái của các kênh kết nối. Khi một socket sẵn sàng đọc hoặc ghi dữ liệu, vòng lặp sẽ kích hoạt các phương thức xử lý tương ứng mà không cần tạo luồng mới cho mỗi kết nối.

• asyncore.loop(): Chạy vòng lặp giám sát cho đến khi không còn kênh nào hoạt động.
• asyncore.dispatcher: Lớp cơ sở đại diện cho một socket. Các phương thức như readable(), writable(), handle_read(), handle_write() cho phép tùy chỉnh luồng xử lý sự kiện mạng.
• handle_accept(): Được gọi khi server nhận được yêu cầu kết nối mới.

asynchat

Mở rộng từ asyncore, mô-đun này được tối ưu hóa cho các giao thức dựa trên bản tin (message-based). Nó tự động quản lý bộ đệm đầu vào/đầu ra và cho phép định nghĩa dấu hiệu kết thúc thông điệp (terminator), giúp việc phân tích chuỗi dữ liệu mạng trở nên trực quan hơn.

• collect_incoming_data(): Gom các mảnh dữ liệu nhận được vào bộ đệm.
• found_terminator(): Kích hoạt khi dữ liệu trong bộ đệm chạm đến dấu ngắt đã định.
• push(): Đẩy dữ liệu vào hàng đợi gửi đi một cách an toàn.

Triển Khai Máy Chủ (Server)

Kiến trúc server được chia thành các lớp chịu trách nhiệm riêng biệt: quản lý socket lắng nghe, duy trì phiên kết nối người dùng, định tuyến lệnh và phân phối thông điệp theo từng trạng thái phòng.

import asyncore
import asynchat

SERVER_PORT = 5555

class SessionTerminated(Exception):
    pass

class NetworkHub(asyncore.dispatcher):
    def __init__(self, port):
        asyncore.dispatcher.__init__(self)
        self.create_socket()
        self.set_reuse_addr()
        self.bind(('0.0.0.0', port))
        self.listen(10)
        self.active_registry = {}
        self.central_space = MainLobby(self)

    def handle_accept(self):
        client_sock, _ = self.accept()
        ClientSession(self, client_sock)

class ClientSession(asynchat.async_chat):
    def __init__(self, hub, sock):
        asynchat.async_chat.__init__(self, sock=sock)
        self.hub = hub
        self.set_terminator(b'\r\n')
        self.buffer = []
        self.nickname = None
        self.switch_zone(AuthZone(self.hub))

    def switch_zone(self, zone):
        if hasattr(self, 'current_zone'):
            self.current_zone.leave(self)
        self.current_zone = zone
        zone.join(self)

    def collect_incoming_data(self, chunk):
        self.buffer.append(chunk.decode('utf-8', errors='ignore'))

    def found_terminator(self):
        raw_msg = ''.join(self.buffer).strip()
        self.buffer.clear()
        if raw_msg:
            try:
                self.current_zone.dispatch_cmd(self, raw_msg.encode('utf-8'))
            except SessionTerminated:
                self.close()

    def handle_close(self):
        super().handle_close()
        self.switch_zone(ExitZone(self.hub))

class RouteEngine:
    def dispatch_cmd(self, session, payload):
        decoded = payload.decode('utf-8')
        parts = decoded.split(maxsplit=1)
        cmd = parts[0]
        arg = parts[1].strip() if len(parts) > 1 else ''
        handler = getattr(self, f'exec_{cmd}', None)
        if handler:
            handler(session, arg)
        else:
            session.push(b'Unknown command.\r\n')

class BaseRoom(RouteEngine):
    def __init__(self, hub):
        self.hub = hub
        self.participants = []

    def join(self, session): self.participants.append(session)
    def leave(self, session):
        try: self.participants.remove(session)
        except ValueError: pass

    def relay(self, data):
        for p in self.participants:
            p.push(data)

    def exec_logout(self, _, __):
        raise SessionTerminated()

class AuthZone(BaseRoom):
    def join(self, session):
        super().join(session)
        session.push(b'Identify yourself. Format: login [name]\r\n')

    def exec_login(self, session, name):
        clean_name = name.strip()
        if not clean_name:
            session.push(b'Name cannot be empty.\r\n')
        elif clean_name in self.hub.active_registry:
            session.push(b'Name already in use.\r\n')
        else:
            session.nickname = clean_name
            self.hub.active_registry[clean_name] = session
            session.switch_zone(self.hub.central_space)

class MainLobby(BaseRoom):
    def join(self, session):
        session.push(b'Entered main hall.\r\n')
        self.relay(f'[{session.nickname}] joined.\r\n'.encode('utf-8'))
        self.hub.active_registry[session.nickname] = session
        super().join(session)

    def leave(self, session):
        super().leave(session)
        self.relay(f'[{session.nickname}] departed.\r\n'.encode('utf-8'))

    def exec_say(self, session, msg):
        formatted = f'[{session.nickname}]: {msg}\r\n'.encode('utf-8')
        self.relay(formatted)

    def exec_who(self, session, _):
        names = '\n'.join(p.nickname for p in self.participants if p.nickname)
        session.push(f'Active users:\n{names}\r\n'.encode('utf-8'))

class ExitZone(BaseRoom):
    def join(self, session):
        if session.nickname:
            self.hub.active_registry.pop(session.nickname, None)

if __name__ == '__main__':
    server = NetworkHub(SERVER_PORT)
    print(f'Listening on 0.0.0.0:{SERVER_PORT}')
    asyncore.loop()

Triển Khai Giao Diện Máy Khách (Client)

Phiên bản client sử dụng tkinter kết hợp với socket chuẩn để đảm bảo tính tương thích và dễ mở rộng. Một luồng phụ được sinh ra để lắng nghe dữ liệu từ server mà không làm đóng băng giao diện chính.

import tkinter as tk
import socket
import threading
import tkinter.messagebox as msgbox

class ConnectionPanel(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("Authentication")
        self.geometry("320x200")
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        tk.Label(self, text="Endpoint (host:port):").pack(pady=8)
        self.ep_entry = tk.Entry(self)
        self.ep_entry.insert(0, "localhost:5555")
        self.ep_entry.pack(padx=20)

        tk.Label(self, text="Username:").pack(pady=8)
        self.name_entry = tk.Entry(self)
        self.name_entry.pack(padx=20)

        tk.Button(self, text="Connect", command=self.establish_link).pack(pady=15)

    def establish_link(self):
        try:
            host, port = self.ep_entry.get().split(':')
            self.sock.connect((host.strip(), int(port)))
            welcome = self.sock.recv(1024).decode('utf-8')
            if "Identify" in welcome:
                self.sock.sendall(f"login {self.name_entry.get().strip()}\r\n".encode('utf-8'))
                auth_resp = self.sock.recv(1024).decode('utf-8')
                if "hall" in auth_resp.lower() or "Identify" in auth_resp:
                    self.destroy()
                    ChatRoom(self.sock, self.name_entry.get().strip())
                else:
                    msgbox.showerror("Login Failed", auth_resp)
                    self.sock.close()
            else:
                self.sock.close()
        except Exception as err:
            msgbox.showerror("Connection Error", str(err))

class ChatRoom(tk.Toplevel):
    def __init__(self, conn, user):
        super().__init__()
        self.title(f"Chat Space - {user}")
        self.geometry("500x400")
        self.conn = conn

        self.display = tk.Text(self, state='disabled', wrap='word')
        self.display.pack(expand=True, fill='both', padx=10, pady=10)

        input_area = tk.Frame(self)
        input_area.pack(fill='x', padx=10, pady=10)

        self.msg_box = tk.Entry(input_area)
        self.msg_box.pack(side='left', fill='x', expand=True, padx=5)
        self.msg_box.bind("<Return>", self.transmit_msg)

        tk.Button(input_area, text="Send", command=lambda: self.transmit_msg(None)).pack(side='right', padx=5)
        tk.Button(input_area, text="Users", command=self.request_list).pack(side='right', padx=5)
        tk.Button(input_area, text="Exit", command=self.terminate).pack(side='right', padx=5)

        self.protocol("WM_DELETE_WINDOW", self.terminate)
        threading.Thread(target=self.poll_server, daemon=True).start()

    def transmit_msg(self, _):
        text = self.msg_box.get().strip()
        if text:
            self.conn.sendall(f"say {text}\r\n".encode('utf-8'))
            self.msg_box.delete(0, 'end')

    def request_list(self):
        self.conn.sendall(b"who\r\n")

    def terminate(self):
        try:
            self.conn.sendall(b"logout\r\n")
            self.conn.close()
        except: pass
        self.destroy()

    def poll_server(self):
        accumulator = b""
        try:
            while True:
                data = self.conn.recv(4096)
                if not data: break
                accumulator += data
                while b'\r\n' in accumulator:
                    line, accumulator = accumulator.split(b'\r\n', 1)
                    self.display.config(state='normal')
                    self.display.insert('end', line.decode('utf-8', errors='ignore') + '\n')
                    self.display.see('end')
                    self.display.config(state='disabled')
        except: pass

Phân Tích Chu Trình Vận Hành

Khi máy chủ khởi động, asyncore.dispatcher thiết lập socket lắng nghe và đưa vào vòng lặp giám sát của hệ thống. Ngay khi có kết nối TCP mới, handle_accept() được kích hoạt, tạo ra một thể hiện ClientSession kế thừa từ asynchat.async_chat. Phiên này ngay lập tức được đưa vào trạng thái xác thực (AuthZone) thông qua cơ chế chuyển vùng.

Dữ liệu từ client được asynchat gom vào bộ đệm đệm cho đến khi gặp ký tự ngắt dòng \r\n. Sự kiện found_terminator() sẽ cắt chuỗi, giải mã và chuyển tiếp đến dispatch_cmd(). Cơ chế phản xạ (reflection) getattr() sẽ ánh xạ tiền tố lệnh sang phương thức xử lý tương ứng (ví dụ: exec_say, exec_who). Nếu người dùng đang ở trạng thái xác thực, lệnh hợp lệ sẽ chuyển họ sang MainLobby, cập nhật danh sách người dùng và phát thông báo gia nhập. Ngược lại, mọi thông điệp chat hoặc truy vấn danh sách sẽ được relay() phát tán đồng loạt tới tất cả các phiên đang tồn tại trong danh sách participants. Khi client đóng kết nối hoặc gửi lệnh thoát, handle_close() dọn dẹp tài nguyên và xóa bản ghi khỏi registry toàn cục.