44. 多进程 vs 多线程：vLLM中的并发选择

# vllm/multiprocessing/process_manager.py
import multiprocessing
from multiprocessing.managers import BaseManager
from typing import Dict, Any, Optional

class ProcessManager:
    """多进程管理器"""
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.processes = []
        self.manager = None
        self.shared_state = None
    
    def start(self):
        """启动多进程"""
        # 初始化共享管理器
        BaseManager.register('SharedState', SharedState)
        self.manager = BaseManager()
        self.manager.start()
        self.shared_state = self.manager.SharedState()
        
        # 创建工作进程
        num_processes = self.config.get('num_processes', multiprocessing.cpu_count())
        for i in range(num_processes):
            process = multiprocessing.Process(
                target=self._worker_process, 
                args=(i, self.shared_state)
            )
            self.processes.append(process)
            process.start()
    
    def _worker_process(self, process_id: int, shared_state: Any):
        """工作进程"""
        from vllm.worker import Worker
        
        worker = Worker(process_id, self.config, shared_state)
        worker.run()
    
    def stop(self):
        """停止所有进程"""
        for process in self.processes:
            if process.is_alive():
                process.terminate()
                process.join()
        
        if self.manager:
            self.manager.shutdown()

# vllm/threading/thread_manager.py
import threading
from typing import Dict, Any, List

class ThreadManager:
    """多线程管理器"""
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.threads: List[threading.Thread] = []
        self.lock = threading.Lock()
        self.condition = threading.Condition(self.lock)
    
    def start(self):
        """启动多线程"""
        num_threads = self.config.get('num_threads', threading.cpu_count())
        for i in range(num_threads):
            thread = threading.Thread(
                target=self._worker_thread, 
                args=(i,)
            )
            self.threads.append(thread)
            thread.start()
    
    def _worker_thread(self, thread_id: int):
        """工作线程"""
        from vllm.worker import ThreadWorker
        
        worker = ThreadWorker(thread_id, self.config)
        worker.run()
    
    def stop(self):
        """停止所有线程"""
        with self.lock:
            self.running = False
        for thread in self.threads:
            thread.join()

# vllm/hybrid/hybrid_manager.py
import multiprocessing
import threading
from typing import Dict, Any

class HybridManager:
    """混合并发管理器"""
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.processes = []
    
    def start(self):
        """启动混合并发"""
        # 创建多个进程
        num_processes = self.config.get('num_processes', multiprocessing.cpu_count())
        
        for i in range(num_processes):
            # 每个进程内部包含多个线程
            process = multiprocessing.Process(
                target=self._process_worker, 
                args=(i,)
            )
            self.processes.append(process)
            process.start()
    
    def _process_worker(self, process_id: int):
        """进程工作函数"""
        # 每个进程内部创建多个线程
        num_threads = self.config.get('num_threads_per_process', 4)
        threads = []
        
        for i in range(num_threads):
            thread = threading.Thread(
                target=self._thread_worker, 
                args=(process_id, i)
            )
            threads.append(thread)
            thread.start()
        
        # 等待所有线程完成
        for thread in threads:
            thread.join()
    
    def _thread_worker(self, process_id: int, thread_id: int):
        """线程工作函数"""
        from vllm.worker import HybridWorker
        worker = HybridWorker(process_id, thread_id, self.config)
        worker.run()

# vllm/async_manager.py
import asyncio
from typing import Dict, Any

class AsyncManager:
    """异步并发管理器"""
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.loop = asyncio.get_event_loop()
    
    async def start(self):
        """启动异步服务"""
        # 创建异步服务器
        server = await asyncio.start_server(
            self._handle_connection, 
            self.config.get('host', '0.0.0.0'),
            self.config.get('port', 8000)
        )
        
        print(f"Async server started on {self.config.get('host', '0.0.0.0')}:{self.config.get('port', 8000)}")
        await server.serve_forever()
    
    async def _handle_connection(self, reader, writer):
        """处理客户端连接"""
        # 异步处理请求
        request = await self._parse_request(reader)
        response = await self._process_request(request)
        await self._write_response(writer, response)
    
    async def _process_request(self, request: Any) -> Any:
        """异步处理请求"""
        from vllm.async_worker import AsyncWorker
        worker = AsyncWorker(self.config)
        return await worker.process(request)

开始
  ├── 任务类型？
  │   ├── CPU密集型 → 多进程或混合模型
  │   └── I/O密集型 → 异步模型或多线程
  ├── 部署规模？
  │   ├── 单节点 → 多线程或混合模型
  │   └── 分布式 → 多进程或Ray分布式
  ├── 实时性要求？
  │   ├── 高（<100ms）→ 混合模型或异步
  │   └── 中低 → 多线程或多进程
  └── 开发复杂度容忍度？
      ├── 高 → 混合模型或异步
      └── 低 → 多线程或简单多进程

# 多进程配置
multiprocess_config = {
    'num_processes': 4,
    'shared_memory': True,
    'communication_method': 'queue',  # queue, pipe, shared_memory
}

# 多线程配置
threading_config = {
    'num_threads': 8,
    'lock_strategy': 'fine-grained',  # coarse-grained, fine-grained
}

# 混合模型配置
hybrid_config = {
    'num_processes': 2,
    'num_threads_per_process': 4,
    'communication_method': 'shared_memory',
    'load_balancing': 'dynamic',  # static, dynamic, adaptive
}

# 异步模型配置
async_config = {
    'event_loop': 'uvloop',  # asyncio, uvloop
    'max_concurrent_connections': 10000,
    'timeout': 30.0,
}

import time
import concurrent.futures
from vllm import LLM, SamplingParams

def test_multiprocess():
    """测试多进程性能"""
    llm = LLM(model="meta-llama/Llama-2-70b-hf", tensor_parallel_size=4)
    sampling_params = SamplingParams(max_tokens=100)
    
    prompts = [
        "Hello, how are you?" * 5
        for _ in range(100)
    ]
    
    start_time = time.time()
    outputs = llm.generate(prompts, sampling_params)
    end_time = time.time()
    
    total_tokens = sum(len(output.outputs[0].text.split()) for output in outputs)
    throughput = total_tokens / (end_time - start_time)
    print(f"多进程吞吐量: {throughput:.2f} tokens/s")
    print(f"总耗时: {end_time - start_time:.2f}s")

def test_multithreading():
    """测试多线程性能"""
    llm = LLM(model="meta-llama/Llama-2-70b-hf", tensor_parallel_size=1)
    sampling_params = SamplingParams(max_tokens=100)
    
    prompts = [
        "Hello, how are you?" * 5
        for _ in range(100)
    ]
    
    def generate(prompt):
        return llm.generate([prompt], sampling_params)
    
    start_time = time.time()
    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
        futures = [executor.submit(generate, prompt) for prompt in prompts]
        outputs = [future.result()[0] for future in concurrent.futures.as_completed(futures)]
    end_time = time.time()
    
    total_tokens = sum(len(output.outputs[0].text.split()) for output in outputs)
    throughput = total_tokens / (end_time - start_time)
    print(f"多线程吞吐量: {throughput:.2f} tokens/s")
    print(f"总耗时: {end_time - start_time:.2f}s")

if __name__ == "__main__":
    print("=== 测试多进程 ===")
    test_multiprocess()
    
    print("\n=== 测试多线程 ===")
    test_multithreading()