活动介绍

torch.multiprocessing.set_start_method('spawn')这行代码是什么意思

时间: 2024-05-20 14:12:13 浏览: 392
这行代码是用来设置PyTorch使用的进程启动方法的，它指定了在使用torch.multiprocessing时，创建新进程的方式。具体来说，它设置了PyTorch使用"spawn"方法来启动新进程。"spawn"方法是Python中的一种多进程启动方式，它会在父进程中创建一个新的Python解释器子进程，然后在子进程中执行指定的Python代码，这样可以避免一些常见的多进程问题，例如共享内存和文件描述符的问题。在PyTorch中，使用"spawn"方法可以确保在多GPU环境下，每个子进程都会被分配到一个独立的GPU资源。
相关问题

torch.multiprocessing.set_start_method('spawn') with tempfile.TemporaryDirectory() as temp_dir: if c.num_gpus == 1: subprocess_fn(rank=0, c=c, temp_dir=temp_dir) else: torch.multiprocessing.spawn(fn=subprocess_fn, args=(c, temp_dir), nprocs=c.num_gpus)代码什么意思

这段代码是用于启动分布式训练的。`torch.multiprocessing.set_start_method('spawn')` 表示使用 spawn 方式启动多进程，这种方式可以避免一些多进程启动的问题，例如在 Windows 上使用 multiprocessing 模块时需要使用 spawn 方式。 `with tempfile.TemporaryDirectory() as temp_dir:` 表示创建一个临时目录，该目录在 with 语句块结束时会被自动删除。`temp_dir` 变量保存了该临时目录的路径。 `if c.num_gpus == 1:` 表示如果只有一个 GPU，则只启动一个进程，执行 `subprocess_fn` 函数。 `else: torch.multiprocessing.spawn(fn=subprocess_fn, args=(c, temp_dir), nprocs=c.num_gpus)` 表示如果有多个 GPU，则使用 `torch.multiprocessing.spawn()` 函数启动多个进程。`fn` 参数表示要运行的函数，`args` 参数是传递给该函数的参数，`nprocs` 参数表示启动的进程数。在这里，`subprocess_fn` 函数会被启动 `c.num_gpus` 次，每个进程会被分配一个 GPU 来运行。

给下列代码添加注释：def vision(): imgQueue = mp.Queue() # 先进先出队列，实现不同进程数据交互 Mps = []#存储进程对象的列表 Mps.append(mp.Process(target=imgRead, args=(imgQueue,))) [Mp.start() for Mp in Mps]#启动所有进程 # Mps[0].join() while imgQueue.empty(): pass while True: Key = input('Press Q or q to quit:') if Key == 'Q' or Key == 'q': break [Mp.terminate() for Mp in Mps]#终止所有进程 #torch.multiprocessing.set_start_method('spawn') if __name__ == '__main__': torch.multiprocessing.set_start_method('spawn') vision()

```python def vision(): imgQueue = mp.Queue() # 创建一个先进先出队列，用于实现不同进程之间的数据交互 Mps = [] # 存储进程对象的列表 Mps.append(mp.Process(target=imgRead, args=(imgQueue,))) # 创建一个进程对象，目标函数为imgRead，传入参数为imgQueue [Mp.start() for Mp in Mps] # 启动所有进程 while imgQueue.empty(): # 当队列为空时，循环等待 pass while True: Key = input('Press Q or q to quit:') # 获取用户输入的按键信息 if Key == 'Q' or Key == 'q': # 如果按键为Q或q，则跳出循环 break [Mp.terminate() for Mp in Mps] # 终止所有进程 # torch.multiprocessing.set_start_method('spawn') if __name__ == '__main__': torch.multiprocessing.set_start_method('spawn') vision() ```
阅读全文

相关推荐

pdf

torch.cuda.is_available（）返回False解决方案

torch.cuda.is_available() 是一个重要的函数，用于检查当前系统是否能够使用CUDA，即NVIDIA的GPU加速库。当这个函数返回 False 时，意味着你的程序将无法在GPU上运行，而会退化到使用CPU。以下是一些可能导致 ...
zip

lbcnn.torch-master.zip_..累lbcnn;x3_LBCNN_lbp_lbp 神经网络_torch

这表明压缩包中的代码不仅涉及LBCNN模型，还可能包括LBP的实现，以及如何在Torch环境中训练和应用这些模型。 由于压缩包的文件名称列表只给出了“lbcnn.torch-master”，我们可以推断这是项目的主要源码目录或者...
zip

DataLoader.py_torch数据_.DataLoader数据加载器_

torch.utils.data.DataLoader 数据加载器，结合了数据集和取样器，并且可以提供多个线程处理数据集在训练模型时使用到此函数，用来把训练数据分成多个小组，此函数每次抛出一组数据，直到把所有的数据都抛出。...

mp.set_start_method("spawn") torch.backends.cuda.matmul.allow_tf32 = True torch.set_grad_enabled(False)

为了防止因多进程初始化而导致的潜在死锁问题，在脚本的入口处应调用 mp.set_start_method('spawn') 来显式指定子进程的启动方式[^1]。此操作应在主模块保护下完成，以确保仅在主进程中执行： python if __...

--------------------------------------------------------------------------- ProcessExitedException Traceback (most recent call last) Cell In[11], line 37 34 if __name__ == '__main__': 35 # 启动多进程训练 36 world_size = torch.cuda.device_count() # 使用所有可用GPU ---> 37 mp.spawn( 38 train, # 目标函数 39 args=(world_size,), # 传递给train的参数 40 nprocs=world_size, # 进程数=GPU数量 41 join=True # 等待所有进程结束 42 ) File ~/miniconda3/envs/dev0.4/lib/python3.8/site-packages/torch/multiprocessing/spawn.py:240, in spawn(fn, args, nprocs, join, daemon, start_method) 236 msg = ('This method only supports start_method=spawn (got: %s).\n' 237 'To use a different start_method use:\n\t\t' 238 ' torch.multiprocessing.start_processes(...)' % start_method) 239 warnings.warn(msg) --> 240 return start_processes(fn, args, nprocs, join, daemon, start_method='spawn') File ~/miniconda3/envs/dev0.4/lib/python3.8/site-packages/torch/multiprocessing/spawn.py:198, in start_processes(fn, args, nprocs, join, daemon, start_method) 195 return context 197 # Loop on join until it returns True or raises an exception. --> 198 while not context.join(): 199 pass File ~/miniconda3/envs/dev0.4/lib/python3.8/site-packages/torch/multiprocessing/spawn.py:149, in ProcessContext.join(self, timeout) 140 raise ProcessExitedException( 141 "process %d terminated with signal %s" % 142 (error_index, name), (...) 146 signal_name=name 147 ) 148 else: --> 149 raise ProcessExitedException( 150 "process %d terminated with exit code %d" % 151 (error_index, exitcode), 152 error_index=error_index, 153 error_pid=failed_process.pid, 154 exit_code=exitcode 155 ) 157 original_trace = self.error_queues[error_index].get() 158 msg = "\n\n-- Process %d terminated with the following error:\n" % error_index ProcessExitedException: process 2 terminated with exit code 1

mp.set_start_method('spawn', force=True) # 获取实际可用GPU数量 world_size = torch.cuda.device_count() print(f"启动 {world_size} 个进程") try: mp.spawn(train, args=(world_size,), nprocs=world...

if __name__ == "__main__": mp.set_start_method("spawn") torch.backends.cuda.matmul.allow_tf32 = True torch.set_grad_enabled(False) device = "cuda:0" save_frames = False datetime_now = str(datetime.datetime.now()).replace(" ", "_") parser = argparse.ArgumentParser() parser.add_argument("--dataset", default="datasets/tum/rgbd_dataset_freiburg1_desk") parser.add_argument("--config", default="config/base.yaml") parser.add_argument("--save-as", default="default") parser.add_argument("--no-viz", action="store_true") parser.add_argument("--calib", default="") args = parser.parse_args() load_config(args.config) print(args.dataset) print(config) manager = mp.Manager() main2viz = new_queue(manager, args.no_viz) viz2main = new_queue(manager, args.no_viz) dataset = load_dataset(args.dataset) dataset.subsample(config["dataset"]["subsample"]) h, w = dataset.get_img_shape()[0] 分析代码

mp.set_start_method("spawn") # 设置多进程启动方式为'spawn' torch.backends.cuda.matmul.allow_tf32 = True # 允许使用TensorFloat-32进行矩阵乘法以提升性能 torch.set_grad_enabled(False) # 关闭自动求导功能...

import torch.multiprocessing as multiprocessing

可以通过调用 mp.set_start_method('spawn') 来完成这一操作。注意，在某些情况下可能还需要调整 CUDA 变量来确保 GPU 资源分配正常。 python import torch import torch.multiprocessing as mp def worker...

如何在 PyTorch 中正确地使用 torch.multiprocessing.spawn 来创建和管理多个进程？

3. **初始化CUDA环境**: 如果在使用CUDA并且需要在多个进程间共享模型或数据，需要先调用 torch.multiprocessing.set_start_method('spawn') 来初始化环境，确保CUDA环境在多进程下正常工作。 下面是一个简单的...

from concurrent.futures import ThreadPoolExecutor import multiprocessing import multiprocessing.queues import threading import time import asyncio import numpy as np import grpc from grpc import aio import argparse import os import video_service_pb2 import video_service_pb2_grpc class FPSCounter: def __init__(self, window_size=0.5): self.window_size = window_size # in seconds self.frame_timestamps = [] self.last_print_time = time.time() def add_frame(self): now = time.time() self.frame_timestamps.append(now) # Remove frames older than window_size while self.frame_timestamps and now - self.frame_timestamps[0] > self.window_size: self.frame_timestamps.pop(0) def get_fps(self): if not self.frame_timestamps: return 0.0 time_span = self.frame_timestamps[-1] - self.frame_timestamps[0] if time_span <= 0: return len(self.frame_timestamps) return len(self.frame_timestamps) / time_span def should_print(self): now = time.time() if now - self.last_print_time >= self.window_size: self.last_print_time = now return True return False class InferencerProcess(multiprocessing.Process): def __init__(self, inference_queue, result_queue): super().__init__() self.inference_queue = inference_queue # 从推理队列获取数据 self.result_queue = result_queue # 发送结果到结果队列 self.running = True self.lpd = None self.lnr = None self.tensor = None def init_model(self): """初始化推理模型""" import torch pid = os.getpid() print(f"[推理进程-{pid}] 初始化模型...") if torch.cuda.is_available(): torch.cuda.init() torch.cuda.set_device(0) print(f"[推理进程-{pid}] CUDA初始化成功，设备: {torch.cuda.get_device_name(0)}") else: raise RuntimeError("CUDA not available in worker process") from src.alpr.core.model_loader import Predictor, get_model from src.alpr.core.LicensePlateProcessor import project_root lpd_classes = ("LicensePlate",) lnr_classes = ("0","1","2","3","4","5","6","7","8","9", "A","B","C","D","E","F","G","H","J","K", "L","M","N","P","Q","R","S","T","U","V", "W","X","Y","Z") lpd_model = get_model(project_root/"models/yolox_lpd_s_20240201.pth", num_classes=len(lpd_classes)) lnr_model = get_model(project_root/"models/yolox_lnr_s_20240201.pth", num_classes=len(lnr_classes)) self.lpd = Predictor(lpd_model, obj_labels=lpd_classes, confthre=0.01, img_size=(544, 960)) self.lnr = Predictor(lnr_model, lnr_classes, confthre=0.7, img_size=(320, 640)) print(f"[推理进程-{pid}] 模型初始化完成") # self.tensor = torch.as_tensor(np.ones((3, 1080, 1920), dtype=np.uint8), device='cuda') def run(self): pid = os.getpid() print(f"[推理进程-{pid}] 启动成功") try: self.init_model() while self.running: try: # 从推理队列获取数据 data = self.inference_queue.get(block=False) if not data: continue tensor, rtsp_id, request_id = data #tensor = self.tensor # 执行推理 try: out, _, _ = self.lpd.inference(tensor) out, T3, T4 = self.lnr.inference(tensor) #out, T3, T4 = [], 0, 0 # 将检测结果转换为仅包含基本类型的字典 detections = [] for det in out: # 确保所有值都是Python基本类型 detection = { 'label': str(det['label']), 'confidence': float(det['confidence']), 'x1': float(det['bbox'][0]), 'y1': float(det['bbox'][1]), 'x2': float(det['bbox'][2]), 'y2': float(det['bbox'][3]) } detections.append(detection) # 推理成功，将结果放入结果队列 (仅包含可序列化对象) res = { 'status': 'success', 'data': { 'detections': detections, 'rtsp_id': rtsp_id, 'request_id': request_id, 'timing': { 't3': float(T3) if T3 is not None else 0, 't4': float(T4) if T4 is not None else 0 } } } self.result_queue.put_nowait(res) except Exception as e: error_msg = f"推理失败: {str(e)}" print(f"[推理进程-{pid}] {error_msg}") self.result_queue.put_nowait({ 'status': 'error', 'data': { 'message': error_msg, 'rtsp_id': rtsp_id, 'request_id': request_id } }) except multiprocessing.queues.Empty: time.sleep(0.001) except Exception as e: print(f"[推理进程-{pid}] 异常: {str(e)}") except Exception as e: print(f"[推理进程-{pid}] 初始化失败: {str(e)}") print(f"[推理进程-{pid}] 已停止") class DecodeProcess(multiprocessing.Process): def __init__(self, decode_queue, inference_queue, result_queue): super().__init__() self.decode_queue = decode_queue # 接收h264数据 self.inference_queue = inference_queue # 发送解码结果 self.result_queue = result_queue # 发送错误信息 self.running = True self.fps_counter = FPSCounter() from av_decoder import AVDecoder self.decoder = AVDecoder() print(f"[Decode进程-{os.getpid()}] 初始化完成") def run(self): pid = os.getpid() print(f"[Decode进程-{pid}] 启动成功") while self.running: try: # 从队列获取h264数据 data = self.decode_queue.get(block=False) if not data: continue h264_data, rtsp_id, request_id = data tensor, t2, t5 = self.decoder.decode(h264_data, rtsp_id) if tensor is None: error_msg = "Failed to decode H264 to tensor" print(f"[Decode进程-{pid}] {error_msg}, request_id={request_id}") self.result_queue.put_nowait({ 'status': 'error', 'data': { 'message': error_msg, 'rtsp_id': rtsp_id, 'request_id': request_id } }) continue if self.fps_counter.should_print(): print(f"Decoder FPS: {self.fps_counter.get_fps():.2f}") # 解码成功，发送到推理队列 try: self.inference_queue.put_nowait((tensor, rtsp_id, request_id)) self.fps_counter.add_frame() except multiprocessing.queues.Full: self.result_queue.put_nowait({ 'status': 'error', 'data': { 'message': '推理队列已满', 'rtsp_id': rtsp_id, 'request_id': request_id } }) print(f'Inference 隊列已滿') self.fps_counter.add_frame() except multiprocessing.queues.Empty: time.sleep(0.001) # 避免忙等待 except Exception as e: print(f"[Decode进程-{pid}] 异常: {str(e)}") break print(f"[Decode进程-{pid}] 已停止") class FinalizerProcess(multiprocessing.Process): def __init__(self, result_queue, response_queue): super().__init__() self.result_queue = result_queue # 从推理队列获取结果 self.response_queue = response_queue # 将处理好的结果发送回主进程 self.running = True def run(self): pid = os.getpid() print(f"[Finalizer进程-{pid}] 启动成功") while self.running: try: # 从结果队列获取数据 result_dict = self.result_queue.get(block=False) #print("RES:", result_dict) if result_dict: # 将结果发送回主进程 self.response_queue.put(result_dict) except multiprocessing.queues.Empty: continue except Exception as e: print(f"[Finalizer进程-{pid}] 异常: {str(e)}") break print(f"[Finalizer进程-{pid}] 已停止") class ResponseHandler: """在主进程中处理响应的类""" def __init__(self): self.pending_requests = {} # request_id -> (context, response_future) self.lock = threading.Lock() def register_request(self, request_id, context, response_future): """注册新请求""" with self.lock: self.pending_requests[request_id] = (context, response_future) def process_response(self, response_dict): """处理从Finalizer进程返回的响应""" try: data = response_dict['data'] request_id = data['request_id'] with self.lock: if request_id not in self.pending_requests: print(f"请求ID {request_id} 未找到上下文") return context, response_future = self.pending_requests.pop(request_id) # 当前全部返回 Empty Response response_future.set_result(video_service_pb2.DecodeAndInferResponse()) return if response_dict['status'] == 'success': # 构造gRPC响应 detections = [] for det in data['detections']: proto_det = video_service_pb2.Detection( label=det['label'], confidence=det['confidence'], x1=det['x1'], y1=det['y1'], x2=det['x2'], y2=det['y2'] ) detections.append(proto_det) response = video_service_pb2.DecodeAndInferResponse(detections=detections) response_future.set_result(response) else: # 设置错误状态 context.set_code(grpc.StatusCode.INTERNAL) context.set_details(data['message']) response_future.set_result(video_service_pb2.DecodeAndInferResponse()) except Exception as e: print(f"响应处理异常: {str(e)}") class DecodeAndInferService(video_service_pb2_grpc.DecodeAndInferServiceServicer): decode_frame = 0 start_time = None def __init__(self, decode_queue: multiprocessing.Queue, result_queue, response_handler): self.decode_queue = decode_queue self.result_queue = result_queue self.response_handler = response_handler self.request_counter = 0 self.counter_lock = threading.Lock() self.fps_counter = FPSCounter() # 在主进程中初始化解码器 from h264_reassembler import H264Reassembler self.reassembler = H264Reassembler() print("[主进程] 解码器初始化完成") async def DecodeAndInfer(self, request, context): # 生成唯一请求ID with self.counter_lock: self.request_counter += 1 request_id = self.request_counter if DecodeAndInferService.start_time == None: DecodeAndInferService.start_time = time.time() DecodeAndInferService.decode_frame += 1 # 创建Future对象 response_future = asyncio.Future() # 注册请求 self.response_handler.register_request(request_id, context, response_future) try: # 提取需要的数据，创建可序列化的字典 h264_data = self.reassembler.reassemble(request.packets, request.rtsp_id) if h264_data is not None: self.decode_queue.put_nowait((h264_data, request.rtsp_id, request_id)) else: self.result_queue.put_nowait({ 'status': 'error', 'data': { 'message': 'cant convert rtp packets to h264', 'rtsp_id': request.rtsp_id, 'request_id': request_id } }) except multiprocessing.queues.Full: self.result_queue.put_nowait({ 'status': 'error', 'data': { 'message': 'decode队列已满', 'rtsp_id': request.rtsp_id, 'request_id': request_id } }) print(f'Decode 隊列已滿') except Exception as e: context.set_code(grpc.StatusCode.INTERNAL) context.set_details(f'处理请求失败: {str(e)}') return video_service_pb2.DecodeAndInferResponse() finally: # 等待结果 result = await response_future print(f"D&F Service FPS: {DecodeAndInferService.decode_frame / (time.time() - DecodeAndInferService.start_time)}") return result async def response_worker(response_queue, response_handler): """响应处理工作线程，在主进程中运行""" while True: try: response_dict = response_queue.get(block=False) response_handler.process_response(response_dict) except multiprocessing.queues.Empty: await asyncio.sleep(0.001) except Exception as e: print(f"响应处理线程异常: {str(e)}") async def serve(port, decode_queue, result_queue, response_queue): """启动gRPC服务""" # 创建响应处理器 response_handler = ResponseHandler() # 启动响应处理工作线程 asyncio.create_task(response_worker(response_queue, response_handler)) server = aio.server() service = DecodeAndInferService(decode_queue, result_queue, response_handler) video_service_pb2_grpc.add_DecodeAndInferServiceServicer_to_server(service, server) server.add_insecure_port(f'[::]:{port}') print(f"[gRPC服务] 启动成功，监听端口 {port}") await server.start() await server.wait_for_termination() def main(): parser = argparse.ArgumentParser(description='视频解码推理服务') parser.add_argument('--port', type=int, default=50151, help='gRPC服务端口') args = parser.parse_args() # 创建进程间通信队列 decode_queue = multiprocessing.Queue(maxsize=50) # 主进程→Decode进程 inference_queue = multiprocessing.Queue(maxsize=50) # Decode进程→Inference进程 result_queue = multiprocessing.Queue(maxsize=50) # Decode进程/Inference进程→Finalizer response_queue = multiprocessing.Queue(maxsize=50) # Finalizer→主进程 # 启动工作进程 decoder = DecodeProcess(decode_queue, inference_queue, result_queue) inferencer = InferencerProcess(inference_queue, result_queue) finalizer = FinalizerProcess(result_queue, response_queue) decoder.start() inferencer.start() finalizer.start() print(f"[主进程] 启动gRPC服务，端口 {args.port}") try: # 启动gRPC服务 (主进程同时负责解码) asyncio.run(serve(args.port, decode_queue, result_queue, response_queue)) except KeyboardInterrupt: print("\n[主进程] 收到停止信号") # 停止工作进程 inferencer.running = False finalizer.running = False # 等待进程结束 inferencer.join() finalizer.join() print("[主进程] 所有进程已停止") if __name__ == '__main__': # 确保使用spawn启动方法 multiprocessing.set_start_method('spawn', force=True) main() 修改代码，让各自组件都可以以其全速进行处理

我们需要修改代码，使得各个组件（解码、推理、结果处理）能够以全速运行，避免因为队列满而导致阻塞或丢弃数据。主要思路包括： 1. 增加队列容量：适当增大队列的maxsize，但注意不要过大以免消耗过多内存。 2. ...

# data/dataset.py import os import numpy as np import torch from torch.utils.data import Dataset from plyfile import PlyData import open3d as o3d def read_ply_points(file_path): """ 使用 plyfile 解析 .ply 文件，提取 x, y, z 字段 返回一个 numpy 数组 (N, 3) """ ply = PlyData.read(file_path) # 提取 x, y, z 字段 x = ply['vertex']['x'] y = ply['vertex']['y'] z = ply['vertex']['z'] # 拼接为点云数组 points = np.vstack([x, y, z]).T points = points.astype(np.float32) return points class UnifiedPointCloudDataset(Dataset): def __init__(self, root_dirs, file_exts=['.ply', '.stl', '.obj'], num_points=1024): """ :param root_dirs: 包含所有数据文件夹的列表 :param file_exts: 支持的点云格式 :param num_points: 每个点云采样点数 """ self.file_list = [] self.num_points = num_points # 收集所有点云文件路径 for root_dir in root_dirs: if not os.path.exists(root_dir): raise FileNotFoundError(f"❌ 数据目录不存在: {root_dir}") for root, _, files in os.walk(root_dir): for file in files: if any(file.lower().endswith(ext) for ext in file_exts): full_path = os.path.join(root, file) if os.path.exists(full_path): # 确保文件真实存在 self.file_list.append(full_path) print(f"✅ 共发现 {len(self.file_list)} 个点云文件，用于训练") if len(self.file_list) == 0: raise ValueError("⚠️ 没有发现任何点云文件，请检查路径和文件格式") def __len__(self): return len(self.file_list) def __getitem__(self, idx): path = self.file_list[idx] ext = os.path.splitext(path)[1].lower() try: if ext == '.ply': # 使用 plyfile 读取 .ply 文件 points = read_ply_points(path) elif ext == '.stl': # 使用 open3d 读取 STL 文件并采样成点云 mesh = o3d.io.read_triangle_mesh(path) pcd = mesh.sample_points_uniformly(number_of_points=100000) points = np.asarray(pcd.points) elif ext == '.obj': # 使用 open3d 读取 OBJ 文件 pcd = o3d.io.read_point_cloud(path) if not pcd.has_points(): raise ValueError(f"点云为空或损坏: {path}") points = np.asarray(pcd.points) else: raise ValueError(f"不支持的格式: {ext}") # 检查点云是否为空 if len(points) < 10: raise ValueError(f"点云为空或损坏: {path}") # 固定点数采样 if len(points) < self.num_points: indices = np.random.choice(len(points), self.num_points, replace=True) else: indices = np.random.choice(len(points), self.num_points, replace=False) points = points[indices] return torch.tensor(points, dtype=torch.float32) except Exception as e: print(f"❌ 读取失败: {path}，错误: {str(e)}") return self.__getitem__((idx + 1) % len(self.file_list)) # models/ballmamba.py import torch import torch.nn as nn from torch_geometric.nn import knn_graph, radius_graph from models.mamba_block import MambaBlock from utils.pointcloud_utils import farthest_point_sample class FPS_Causal(nn.Module): def __init__(self, in_channels, hidden_channels, k=512): super(FPS_Causal, self).__init__() self.k = k self.downsample = nn.Linear(in_channels, hidden_channels) self.mamba = MambaBlock(hidden_channels) def forward(self, x, pos): batch_size, num_points, _ = pos.shape idxs = [] for i in range(batch_size): idx = farthest_point_sample(pos[i], self.k) # 现在支持 (N, 3) idxs.append(idx) idx = torch.stack(idxs, dim=0) # (B, k) # 使用 gather 正确采样 x_sampled = torch.gather(pos, 1, idx.unsqueeze(-1).expand(-1, -1, 3)) # (B, k, 3) x_sampled = self.downsample(x_sampled) # (B, k, hidden_channels) x_sampled = self.mamba(x_sampled) # (B, k, hidden_channels) return x_sampled class BallQuery_Sort(nn.Module): def __init__(self, radius=0.3, k=64): super(BallQuery_Sort, self).__init__() self.radius = radius self.k = k def forward(self, pos, x): edge_index = radius_graph(pos, r=self.radius) row, col = edge_index neighbor_pos = pos[col] neighbor_x = x[col] dist = torch.norm(neighbor_pos - pos[row], dim=1) sorted_indices = torch.argsort(dist.view(-1, self.k), dim=1) neighbors = neighbor_x.view(-1, self.k, neighbor_x.shape[1]) return neighbors.gather(1, sorted_indices.unsqueeze(-1).expand(-1, -1, neighbor_x.shape[1])) class BallMambaModel(nn.Module): def __init__(self, in_channels=3, num_keypoints=1024): super(BallMambaModel, self).__init__() self.fps = FPS_Causal(in_channels, 64, k=num_keypoints) self.ball_query = BallQuery_Sort(radius=0.1, k=64) self.mamba = MambaBlock(64) self.decoder = nn.Linear(64, 3) def forward(self, pos): print("pos shape:", pos.shape) # 添加此行，查看 pos 的形状 x = self.fps(None, pos) x = self.ball_query(pos, x) x = x.mean(dim=1) x = self.mamba(x.unsqueeze(0)).squeeze(0) out = self.decoder(x) return out import torch import torch.nn as nn import torch.nn.functional as F class MambaBlock(nn.Module): def __init__(self, d_model: int, d_state: int = 16, d_conv: int = 4, expand: int = 2, dt_rank: int or str = "auto"): """ 完整的 MambaBlock 实现（无 LSTM，基于状态空间模型 SSM） :param d_model: 输入特征维度（通道数） :param d_state: 状态维度（SSM 中的 N） :param d_conv: 卷积核大小（用于局部依赖建模） :param expand: 扩展因子，中间维度 = d_model * expand :param dt_rank: 离散时间步秩，控制参数量，若为 "auto" 则自动计算 """ super(MambaBlock, self).__init__() self.d_model = d_model self.d_state = d_state self.d_inner = int(expand * d_model) self.dt_rank = dt_rank if dt_rank != "auto" else max(1, self.d_model // 16) # 输入投影：将输入特征升维 self.in_proj = nn.Linear(d_model, self.d_inner * 2, bias=False) # 卷积分支（局部建模） self.conv = nn.Conv1d( in_channels=self.d_inner, out_channels=self.d_inner, kernel_size=d_conv, bias=True, groups=self.d_inner, padding=d_conv - 1, # 保证 causal ) # x_proj 将 x 映射到 dt、B、C self.x_proj = nn.Linear(self.d_inner, self.dt_rank + 2 * d_state, bias=False) # dt_proj 将 dt 映射到 d_inner self.dt_proj = nn.Linear(self.dt_rank, self.d_inner, bias=True) # A 和 D 参数（状态矩阵和跳跃连接） A = torch.arange(1, d_state + 1, dtype=torch.float32)[None, :].repeat(self.d_inner, 1) self.A_log = nn.Parameter(torch.log(A)) self.D = nn.Parameter(torch.ones(self.d_inner)) # (d_inner, ) # 输出投影 self.out_proj = nn.Linear(self.d_inner, d_model, bias=False) def forward(self, x: torch.Tensor) -> torch.Tensor: """ 前向传播 :param x: 输入张量，形状 (B, L, d_model) :return: 输出张量，形状 (B, L, d_model) """ batch, seqlen, dim = x.shape # 1. 输入投影 + 拆分 xz = self.in_proj(x) # (B, L, 2 * d_inner) x, z = torch.split(xz, [self.d_inner, self.d_inner], dim=-1) # (B, L, d_inner) # 2. 卷积处理 x = x.transpose(1, 2) # (B, d_inner, L) x = self.conv(x)[:, :, :seqlen] # (B, d_inner, L) x = x.transpose(1, 2) # (B, L, d_inner) # 3. x_proj 分割出 dt、B、C x_dbl = self.x_proj(x) # (B, L, dt_rank + 2 * d_state) dt, B, C = torch.split( x_dbl, [self.dt_rank, self.d_state, self.d_state], dim=-1 ) # 4. dt_proj 映射 dt dt = F.softplus(self.dt_proj(dt)) # (B, L, d_inner) # 5. 获取 A A = -torch.exp(self.A_log) # (d_inner, d_state) A = A.unsqueeze(0).unsqueeze(0) # (1, 1, d_inner, d_state) B = B.unsqueeze(dim=2) # (B, L, 1, d_state) C = C.unsqueeze(dim=2) # (B, L, 1, d_state) # 调试信息 print(f"[MambaBlock] A.shape = {A.shape}") print(f"[MambaBlock] B.shape = {B.shape}") print(f"[MambaBlock] C.shape = {C.shape}") print(f"[MambaBlock] dt.shape = {dt.shape}") states = torch.zeros(batch, self.d_inner, self.d_state, device=x.device) outputs = [] for t in range(seqlen): # 更新状态 states = states + x[:, t:t+1, :, None] * B[:, t:t+1, :, :] states = states * torch.exp(A * dt[:, t:t+1, :, None]) # 添加 dt 到状态更新 # 获取当前时间步的 C 并进行 einsum current_C = C[:, t] # (B, 1, d_state) current_C = current_C.squeeze(1) # (B, d_state) # 使用广播机制 y = torch.einsum("binc,bc->bin", states, current_C) # bc 会广播为 binc outputs.append(y) y = torch.stack(outputs, dim=1) # (B, L, d_inner) # ✅ 修复：self.D 扩展为 (1, 1, d_inner) 以便广播 y = y + x * self.D.view(1, 1, -1) # 加上跳跃连接 # 激活 + 输出 y = y * F.silu(z) out = self.out_proj(y) # 调试信息 print(f"[MambaBlock] y.shape = {y.shape}") print(f"[MambaBlock] out.shape = {out.shape}") return out # train.py import os import torch import torch.nn as nn from torch.utils.data import DataLoader from data.dataset import UnifiedPointCloudDataset from models.ballmamba import BallMambaModel from utils.loss import ChamferLoss # 👇 Windows 多进程训练必须放在 if __name__ == '__main__' 里面 if __name__ == '__main__': # 设置多进程启动方式（Windows 下推荐 spawn） torch.multiprocessing.set_start_method('spawn') # ✅ 修改为你自己的路径 ROOT_DIRS = [ r"D:\桌面\point\data1\part1", r"D:\桌面\point\data1\part2", r"D:\桌面\point\data1\part3", r"D:\桌面\point\data1\part4", r"D:\桌面\point\data1\part5", r"D:\桌面\point\data1\part6", r"D:\桌面\point\data1\part7", r"D:\桌面\point\data1\part8", r"D:\桌面\point\data1\part9", r"D:\桌面\point\data1\part10", r"D:\桌面\point\data1\part11", r"D:\桌面\point\data1\part12", r"D:\桌面\point\data1\part13", r"D:\桌面\point\data1\part14", r"D:\桌面\point\data1\part15", r"D:\桌面\point\data1\part16", r"D:\桌面\point\data1\part17", r"D:\桌面\point\data1\part18", r"D:\桌面\point\data1\part19", r"D:\桌面\point\data1\part20", ] # ✅ 创建 Dataset dataset = UnifiedPointCloudDataset( root_dirs=ROOT_DIRS, file_exts=['.ply', '.stl'], num_points=1024 ) print(f"✅ 共发现 {len(dataset)} 个点云文件，用于训练") if len(dataset) == 0: raise ValueError("⚠️ 没有发现任何点云文件，请检查路径和文件格式") # ✅ 创建 DataLoader（num_workers=0 可临时绕过问题） loader = DataLoader( dataset, batch_size=16, shuffle=True, num_workers=0, # 👈 Windows 下训练先设置为 0，后续再尝试 4 pin_memory=True ) # ✅ 模型初始化 device = torch.device("cpu") model = BallMambaModel(in_channels=3, num_keypoints=512).to(device) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) criterion = ChamferLoss().to(device) # ✅ 训练循环 for epoch in range(50): model.train() total_loss = 0 for i, points in enumerate(loader): points = points.to(device) # 输入输出一致（重构任务） recon_points = model(points) loss = criterion(recon_points, points) optimizer.zero_grad() loss.backward() optimizer.step() total_loss += loss.item() if i % 10 == 0: print(f"Epoch [{epoch+1}/50], Batch [{i+1}/{len(loader)}], Loss: {loss.item():.4f}") print(f"Epoch [{epoch+1}/50] 完成，平均 Loss: {total_loss / len(loader):.4f}") torch.save(model.state_dict(), f"models/ballmamba_epoch_{epoch+1}.pth") 上述是我的项目模型训练的代码，现在运行后出现问题C:\ProgramData\miniconda3\envs\torch\python.exe D:\桌面\point\scripts\train_model.py ✅ 共发现 907 个点云文件，用于训练 ✅ 共发现 907 个点云文件，用于训练 pos shape: torch.Size([16, 1024, 3]) [MambaBlock] A.shape = torch.Size([1, 1, 128, 16]) [MambaBlock] B.shape = torch.Size([16, 512, 1, 16]) [MambaBlock] C.shape = torch.Size([16, 512, 1, 16]) [MambaBlock] dt.shape = torch.Size([16, 512, 128]) Traceback (most recent call last): File "D:\桌面\point\scripts\train_model.py", line 76, in <module> recon_points = model(points) File "C:\ProgramData\miniconda3\envs\torch\lib\site-packages\torch\nn\modules\module.py", line 727, in _call_impl result = self.forward(*input, **kwargs) File "D:\桌面\point\models\ballmamba.py", line 63, in forward x = self.fps(None, pos) File "C:\ProgramData\miniconda3\envs\torch\lib\site-packages\torch\nn\modules\module.py", line 727, in _call_impl result = self.forward(*input, **kwargs) File "D:\桌面\point\models\ballmamba.py", line 30, in forward x_sampled = self.mamba(x_sampled) # (B, k, hidden_channels) File "C:\ProgramData\miniconda3\envs\torch\lib\site-packages\torch\nn\modules\module.py", line 727, in _call_impl result = self.forward(*input, **kwargs) File "D:\桌面\point\models\mamba_block.py", line 111, in forward y = y + x * self.D.view(1, 1, -1) # 加上跳跃连接 RuntimeError: The size of tensor a (16) must match the size of tensor b (512) at non-singleton dimension 2 进程已结束，退出代码为 1 应该如何修改代码？给我修改后的完整代码

- 检查错误堆栈跟踪，确定具体哪行代码触发错误（如张量运算、损失函数或模型层） - 在关键位置插入形状打印语句： python print("Tensor A shape:", tensor_a.shape) # 预期输出类似 torch.Size([B, C, 16]) ...

提供一个简单的torch.multiprocessing使用示例

mp.set_start_method('spawn') # 设置多进程启动方式 processes = [] for i in range(4): process = mp.Process(target=func, args=(i,)) processes.append(process) # 启动多个进程 for process in ...

import numpy as np import cv2 from ultralytics import YOLO import torch import time import queue from threading import Thread import psutil import multiprocessing import multiprocessing from plate_recognition.plate_rec import init_model, get_plate_result import GPUtil import pynvml import argparse import statistics from collections import defaultdict import os import threading import platform import subprocess import signal from concurrent.futures import ThreadPoolExecutor # 全局配置 CLASSES = ['danger', 'car_danger', 'headstock', 'light', 'number', '1number', 'double_number'] DETECTOR_MODEL_PATH = './weights/best.engine' # DETECTOR_MODEL_PATH = './weights/best_fp32.engine' TEXT_MODEL_PATH = './weights/plate_rec.pth' def show_frame(frame_data, stream_id, latency_queue): frame, capture_time = frame_data window_name = f"Stream {stream_id}" if frame is not None and frame.size > 0: cv2.imshow(window_name, frame) if cv2.waitKey(1) & 0xFF == ord('q'): return latency = time.time() - capture_time latency_queue.put((stream_id, latency)) def display_process(display_queue, stream_id, latency_queue): window_name = f"Stream {stream_id}" cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) cv2.resizeWindow(window_name, 800, 600) try: while True: frame_data = display_queue.get() if frame_data is None: break frame, capture_time = frame_data if frame is not None and frame.size > 0: cv2.imshow(window_name, frame) latency = time.time() - capture_time latency_queue.put((stream_id, latency)) if cv2.waitKey(1) & 0xFF == ord('q'): break finally: cv2.destroyWindow(window_name) print(f"显示进程{stream_id}退出") class EnhancedResourceMonitor: def __init__(self, gpu_id, process_mgr, interval=0.5): self.gpu_id = gpu_id self.interval = interval self.running = False self.lock = threading.Lock() self.data = defaultdict(list) self.process_mgr = process_mgr # 进程管理器引用 # GPU硬件信息 self.gpu_arch = "Ada Lovelace" self.sm_count = 56 self.peak_tflops = 35.6 self.cores_per_sm = 128 def start(self): pynvml.nvmlInit() self.handle = pynvml.nvmlDeviceGetHandleByIndex(self.gpu_id) self.running = True self.thread = Thread(target=self._monitor, daemon=True) self.thread.start() def _monitor(self): while self.running: try: # 监控所有工作进程 worker_stats = [] for p in self.process_mgr.processes: try: proc = psutil.Process(p.pid) with proc.oneshot(): worker_stats.append({ 'cpu': proc.cpu_percent(), 'mem': proc.memory_info().rss / (1024 ** 2), 'threads': proc.num_threads() }) except (psutil.NoSuchProcess, psutil.AccessDenied): continue # GPU监控 util = pynvml.nvmlDeviceGetUtilizationRates(self.handle) mem_info = pynvml.nvmlDeviceGetMemoryInfo(self.handle) clock_mhz = pynvml.nvmlDeviceGetClockInfo(self.handle, pynvml.NVML_CLOCK_SM) # 计算实际算力 current_tflops = (self.sm_count * (clock_mhz / 1000) * self.cores_per_sm * 2) / 1000 util_percent = (current_tflops / self.peak_tflops) * 100 # 记录数据 with self.lock: if worker_stats: self.data['worker_cpu'].append(sum(s['cpu'] for s in worker_stats)) self.data['worker_mem'].append(sum(s['mem'] for s in worker_stats)) self.data['worker_threads'].append(sum(s['threads'] for s in worker_stats)) self.data['gpu_util'].append(util.gpu) self.data['gpu_mem'].append(mem_info.used / (1024 ** 2)) self.data['gpu_tflops'].append(current_tflops) except Exception as e: print(f"监控错误: {str(e)}") time.sleep(self.interval) def stop(self): self.running = False if hasattr(self, 'thread'): self.thread.join() pynvml.nvmlShutdown() return self._generate_report() def _generate_report(self): report = "\n[程序资源报告]\n" # 进程统计 if self.data.get('worker_threads'): report += f"- 工作进程数: {len(self.process_mgr.processes)}\n" report += f"- 总线程数: {max(self.data['worker_threads'])}\n" report += f"- 峰值CPU使用: {max(self.data['worker_cpu']):.1f}%\n" report += f"- 峰值内存占用: {max(self.data['worker_mem']):.1f}MB\n" # GPU统计 if self.data.get('gpu_tflops'): avg_tflops = statistics.mean(self.data['gpu_tflops']) report += "\n[GPU资源]\n" report += f"- 平均利用率: {statistics.mean(self.data['gpu_util']):.1f}%\n" report += f"- 峰值显存: {max(self.data['gpu_mem']):.1f}MB\n" report += f"- 平均算力: {avg_tflops:.1f} TFLOPS\n" report += f"- 算力利用率: {avg_tflops/self.peak_tflops*100:.1f}%\n" return report class ResourceMonitor: def __init__(self, gpu_id, interval=0.5): self.gpu_id = gpu_id self.interval = interval self.running = False self.data = defaultdict(list) self.gpu_arch = "Ada Lovelace" self.sm_count = 56 # RTX 4070 SUPER有56个SM self.peak_tflops = 35.6 # 理论算力35.6 TFLOPS self.cores_per_sm = 128 # Ada架构每个SM有128个CUDA核心 self.lock = threading.Lock() # 添加锁 self.main_pid = os.getpid() # 记录主进程PID def start(self): pynvml.nvmlInit() self.handle = pynvml.nvmlDeviceGetHandleByIndex(self.gpu_id) self.running = True self.thread = Thread(target=self._monitor, daemon=True) self.thread.start() def _monitor(self): while self.running: try: # 改进的进程监控 main_process = psutil.Process(self.main_pid) with main_process.oneshot(): # 原子化读取 process_cpu = main_process.cpu_percent(interval=0.1) # 更短间隔 process_mem = main_process.memory_info().rss / (1024 ** 2) process_threads = main_process.num_threads() # 确保不会记录到0值 if process_cpu == 0 and len(self.data['process_cpu']) > 0: process_cpu = self.data['process_cpu'][-1] * 0.9 # 使用上次值的90% # 记录数据 with self.lock: self.data['process_cpu'].append(process_cpu) self.data['process_memory'].append(process_mem) self.data['process_threads'].append(process_threads) # 系统进程统计 process_count = len(list(psutil.process_iter())) cpu_percent = psutil.cpu_percent() mem = psutil.virtual_memory() # 获取当前Python进程的资源使用 current_process = psutil.Process() process_cpu = current_process.cpu_percent() process_mem = current_process.memory_info().rss / (1024 ** 2) # MB process_threads = current_process.num_threads() # GPU监控 util = pynvml.nvmlDeviceGetUtilizationRates(self.handle) mem_info = pynvml.nvmlDeviceGetMemoryInfo(self.handle) graphics_clock = pynvml.nvmlDeviceGetClockInfo(self.handle, pynvml.NVML_CLOCK_GRAPHICS) sm_clock = pynvml.nvmlDeviceGetClockInfo(self.handle, pynvml.NVML_CLOCK_SM) power_usage = pynvml.nvmlDeviceGetPowerUsage(self.handle) / 1000 # 瓦特 total_mem = sum(p.memory_info().rss for p in psutil.process_iter(['pid', 'name']) if 'python' in p.info['name'].lower()) / (1024**2) # MB # 获取当前GPU时钟频率 clock_mhz = pynvml.nvmlDeviceGetClockInfo( self.handle, pynvml.NVML_CLOCK_SM ) # 收集数据 self.data['system_processes'].append(process_count) self.data['system_cpu'].append(cpu_percent) self.data['system_memory'].append(mem.used / (1024**3)) # GB self.data['process_cpu'].append(process_cpu) self.data['process_memory'].append(process_mem) self.data['process_threads'].append(process_threads) self.data['gpu_util'].append(util.gpu) self.data['gpu_mem'].append(mem_info.used / (1024**2)) # MB self.data['gpu_power'].append(power_usage) self.data['gpu_clock_graphics'].append(graphics_clock) self.data['gpu_clock_sm'].append(sm_clock) # 实时算力计算 (TFLOPS = SM数 * 时钟频率(GHz) * 每SM核心数 * 2 / 1000) current_tflops = (self.sm_count * (clock_mhz / 1000) * self.cores_per_sm * 2) / 1000 util_percent = (current_tflops / self.peak_tflops) * 100 self.data['gpu_tflops'].append(current_tflops) self.data['gpu_sm_clock'].append(clock_mhz) self.data['gpu_util_actual'].append(util_percent) except Exception as e: print(f"算力监控错误: {e}") except (psutil.NoSuchProcess, pynvml.NVMLError) as e: print(f"监控错误(忽略): {str(e)}") except Exception as e: print(f"意外的监控错误: {str(e)}") time.sleep(self.interval) def stop(self): self.running = False self.thread.join() pynvml.nvmlShutdown() return self._generate_report() def _generate_report(self): if not any(len(v) > 0 for v in self.data.values()): return "无监控数据" report = "\n[资源使用报告]\n" report += "\n[算力分析 - RTX 4070 SUPER]\n" report += f"- GPU架构: {self.gpu_arch}\n" report += f"- 流式多处理器(SM): {self.sm_count}\n" report += f"- CUDA核心: {self.sm_count * self.cores_per_sm}\n" report += f"- 理论峰值算力: {self.peak_tflops} TFLOPS\n" if self.data.get('gpu_tflops'): avg_tflops = statistics.mean(self.data['gpu_tflops']) max_tflops = max(self.data['gpu_tflops']) avg_clock = statistics.mean(self.data['gpu_sm_clock']) report += "\n[实际运行数据]\n" report += f"- 平均SM时钟: {avg_clock} MHz\n" report += f"- 平均算力: {avg_tflops:.1f} TFLOPS\n" report += f"- 峰值算力: {max_tflops:.1f} TFLOPS\n" report += f"- 算力利用率: {avg_tflops/self.peak_tflops*100:.1f}%\n" # 瓶颈分析 avg_util = statistics.mean(self.data['gpu_util']) if avg_util > 90 and util_percent < 70: report += "\n[警告] 高GPU利用率但低算力利用率，可能存在内存带宽瓶颈\n" elif avg_tflops < 0.7 * self.peak_tflops: report += "\n[提示] 算力未充分利用，建议检查:\n" report += " • 批次大小(batch size)是否过小\n" report += " • 模型是否存在大量分支操作\n" # 系统级统计 report += "[系统资源]\n" system_metrics = { 'system_processes': ('系统进程数', '{:.0f}'), 'system_cpu': ('系统CPU使用率(%)', '{:.1f}'), 'system_memory': ('系统内存使用(GB)', '{:.2f}') } for key, (name, fmt) in system_metrics.items(): values = self.data.get(key, []) if values: report += ( f"{name}:\n" f" 平均值: {fmt.format(statistics.mean(values))}\n" f" 最大值: {fmt.format(max(values))}\n" f" 最小值: {fmt.format(min(values))}\n" f" 采样数: {len(values)}\n\n" ) # 进程级统计 report += "[主进程资源]\n" process_metrics = { 'process_cpu': ('进程CPU使用率(%)', '{:.1f}'), 'process_memory': ('进程内存使用(MB)', '{:.1f}'), 'process_threads': ('程内部的线程数量', '{:.0f}') } for key, (name, fmt) in process_metrics.items(): values = self.data.get(key, []) if values: report += ( f"{name}:\n" f" 平均值: {fmt.format(statistics.mean(values))}\n" f" 最大值: {fmt.format(max(values))}\n" f" 最小值: {fmt.format(min(values))}\n" f" 采样数: {len(values)}\n\n" ) # GPU统计 report += "[GPU资源]\n" gpu_metrics = { 'gpu_util': ('GPU利用率(%)', '{:.1f}'), 'gpu_mem': ('显存使用(MB)', '{:.1f}'), 'gpu_power': ('GPU功耗(W)', '{:.1f}'), 'gpu_clock_graphics': ('图形时钟(MHz)', '{:.0f}'), 'gpu_clock_sm': ('SM时钟(MHz)', '{:.0f}') } for key, (name, fmt) in gpu_metrics.items(): values = self.data.get(key, []) if values: report += ( f"{name}:\n" f" 平均值: {fmt.format(statistics.mean(values))}\n" f" 最大值: {fmt.format(max(values))}\n" f" 最小值: {fmt.format(min(values))}\n" f" 采样数: {len(values)}\n\n" ) return report class VideoProcessor: def __init__(self, device): # 添加CUDA初始化 torch.cuda.empty_cache() # 加载模型前设置优化选项 torch.backends.cudnn.benchmark = True torch.backends.cuda.matmul.allow_tf32 = True torch.backends.cudnn.allow_tf32 = True # 加载模型 self.model = YOLO(DETECTOR_MODEL_PATH, task='detect') self.plate_rec_model = init_model(device, TEXT_MODEL_PATH) self.rtsp_url = "rtsp://admin:[email protected]:554/cam/realmonitor?channel=1&subtype=0" self.max_retries = 3 # 预热GPU # with torch.no_grad(): # dummy_input = torch.randn(1, 3, 640, 640).to(device) # _ = self.model(dummy_input) self.device = device self.frame_count = 0 self.plate_text_cache = {} def _reconnect(self): cap = cv2.VideoCapture(self.rtsp_url, cv2.CAP_FFMPEG) cap.set(cv2.CAP_PROP_RTSP_TRANSPORT, cv2.CAP_RTSP_TRANSPORT_TCP) return cap # 在VideoProcessor类中添加中文显示支持 def process_frame(self, frame): # 增强版中文显示函数（带错误处理和字体回退） def put_chinese_text(img, text, position, font_scale, color, thickness): """支持中文显示的增强函数""" try: from PIL import Image, ImageDraw, ImageFont img_pil = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB)) draw = ImageDraw.Draw(img_pil) # 尝试加载字体（优先使用自定义字体，失败则回退系统字体） try: font_path = os.path.join("fonts", "platech.ttf") font = ImageFont.truetype(font_path, int(font_scale * 30)) except: font = ImageFont.load_default() print("警告：使用默认字体，中文显示可能不正常") draw.text(position, text, font=font, fill=color) return cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR) except Exception as e: print(f"文本渲染失败，使用OpenCV默认显示: {str(e)}") cv2.putText(img, text, position, cv2.FONT_HERSHEY_SIMPLEX, font_scale, color, thickness) return img self.frame_count += 1 # 1. 输入帧验证 if frame is None or frame.size == 0: print("错误：接收到空帧") return frame # 2. 模型推理（添加详细日志） try: results = self.model.track( frame, persist=True, imgsz=640, verbose=False # 关闭YOLO内置输出 ) # 调试输出 print(f"帧 {self.frame_count}: 检测到 {len(results)} 个结果") except Exception as e: print(f"模型推理错误: {str(e)}") return frame # 3. 结果解析与渲染 class_colors = { 'danger': (0, 0, 255), 'car_danger': (0, 165, 255), 'headstock': (255, 0, 0), 'light': (255, 255, 0), 'number': (0, 255, 0), '1number': (0, 255, 255), 'double_number': (128, 0, 128) } for result in results: # 验证检测结果有效性 if not hasattr(result, 'boxes') or result.boxes is None: print("警告：结果中未包含有效检测框") continue for box in result.boxes: try: # 解析检测框数据 cls_id = int(box.cls[0].item()) class_name = CLASSES[cls_id] x1, y1, x2, y2 = map(int, box.xyxy[0].tolist()) conf = box.conf[0].item() track_id = int(box.id[0].item()) if box.id is not None else None # 车牌特殊处理 if class_name == 'number' and (track_id not in self.plate_text_cache or self.frame_count % 5 == 0): plate_img = frame[y1:y2, x1:x2] if plate_img.size > 0: plate_text = get_plate_result(plate_img, self.device, self.plate_rec_model) or "识别失败" self.plate_text_cache[track_id] = plate_text try: if track_id not in self.plate_text_cache or self.frame_count % 5 == 0: plate_img = frame[y1:y2, x1:x2] if plate_img.size > 0: plate_text = get_plate_result(plate_img, self.device, self.plate_rec_model) or "识别失败" self.plate_text_cache[track_id] = plate_text else: plate_text = "无效区域" display_text = f"{self.plate_text_cache.get(track_id, '加载中...')} ID:{track_id} {conf:.2f}" except Exception as e: print(f"车牌处理异常: {str(e)}") display_text = f"车牌识别错误 ID:{track_id}" else: display_text = f"{class_name} {conf:.2f}" + (f" ID:{track_id}" if track_id else "") # 渲染检测框和文本 color = class_colors.get(class_name, (255, 255, 255)) cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2) # 文本位置修正（确保不超出画面） y_text = max(y1 - 10, 10) frame = put_chinese_text(frame, display_text, (x1, y_text), 0.7, color, 2) except Exception as e: print(f"单检测框处理错误: {str(e)}") continue return frame def display_thread(display_queue, stream_id, latency_queue): window_name = f"Stream {stream_id}" cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) cv2.resizeWindow(window_name, 800, 600) try: while True: frame_data = display_queue.get() if frame_data is None: break frame, capture_time = frame_data if frame is not None and frame.size > 0: cv2.imshow(window_name, frame) latency = time.time() - capture_time latency_queue.put((stream_id, latency)) if cv2.waitKey(1) & 0xFF == ord('q'): break finally: cv2.destroyWindow(window_name) print(f"显示线程{stream_id}退出") class StreamSimulator: def __init__(self, source_url, num_streams, shared_frame_queue): self.source_url = source_url self.num_streams = num_streams self.shared_frame_queue = shared_frame_queue self.display_queues = [multiprocessing.Queue(maxsize=2000) for _ in range(num_streams)] # 使用 multiprocessing.Queue self.stop_flag = multiprocessing.Event() self.capture_process = None def start(self): self.capture_process = multiprocessing.Process(target=self._capture_and_distribute) self.capture_process.start() def stop(self): self.stop_flag.set() if self.capture_process: self.capture_process.join(timeout=5) if self.capture_process.is_alive(): self.capture_process.terminate() print("强制终止捕获进程") def _capture_and_distribute(self): rtsp_url = self.source_url cap = cv2.VideoCapture(rtsp_url, cv2.CAP_FFMPEG) cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) cap.set(cv2.CAP_PROP_FPS, 15) skip_frames = 2 # 每 2 帧处理 1 帧 frame_count = 0 try: while not self.stop_flag.is_set(): ret, frame = cap.read() if not ret: print("帧读取失败，重连中...") cap.release() time.sleep(2) cap = cv2.VideoCapture(rtsp_url, cv2.CAP_FFMPEG) continue frame_count += 1 if frame_count % skip_frames == 0: for i in range(self.num_streams): try: if not self.shared_frame_queue.full(): self.shared_frame_queue.put((frame.copy(), i, time.time()), block=False) else: print(f"共享帧队列已满，丢弃旧帧") self.shared_frame_queue.get_nowait() self.shared_frame_queue.put((frame.copy(), i, time.time()), block=False) except Exception as e: print(f"帧队列操作警告: {type(e).__name__}") finally: cap.release() self.shared_frame_queue.put(None) for q in self.display_queues: q.put(None) def dispatch_process(result_queue, display_queues): frame_buffer = [] while True: data = result_queue.get() if data is None: break # 检查是否为 'stats' 二元组 if isinstance(data, tuple) and len(data) == 2 and data[0] == 'stats': continue # 跳过中间统计数据 # 检查是否为帧数据（三元组） elif isinstance(data, tuple) and len(data) == 3: processed_frame, stream_id, capture_time = data frame_buffer.append((processed_frame, stream_id, capture_time)) frame_buffer.sort(key=lambda x: x[2]) # 按时间戳排序 for frame, sid, _ in frame_buffer: if not display_queues[sid].full(): display_queues[sid].put((frame, capture_time), block=False) else: print(f"显示队列 {sid} 已满，丢帧") frame_buffer = [] # 清空缓冲区 # 检查是否为字典（最终统计数据） elif isinstance(data, dict): continue # 跳过最终统计数据 else: print(f"警告：未知数据类型: {type(data)}") def display_process(display_queue, stream_id, latency_queue): window_name = f"Stream {stream_id}" cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) cv2.resizeWindow(window_name, 800, 600) frame_count = 0 try: while True: frame_data = display_queue.get() if frame_data is None: break frame, capture_time = frame_data if frame is not None and frame.size > 0: cv2.imshow(window_name, frame) frame_count += 1 if frame_count % 10 == 0: latency = time.time() - capture_time latency_queue.put((stream_id, latency)) if cv2.getWindowProperty(window_name, cv2.WND_PROP_VISIBLE) < 1: break if cv2.waitKey(1) & 0xFF == ord('q'): break finally: cv2.destroyWindow(window_name) print(f"显示进程{stream_id}退出") def worker_process(input_queue, gpu_id, result_queue, stats_queue, monitor_interval=5): import numpy as np # 显式导入 numpy print(f"In worker process {os.getpid()}, np is {np}, type(np.empty((1,))) = {type(np.empty((1,)))}") from collections import defaultdict torch.set_num_threads(1) cv2.setNumThreads(1) torch.cuda.set_device(gpu_id) device = torch.device(f"cuda:{gpu_id}" if torch.cuda.is_available() else "cpu") processor = VideoProcessor(device) start_time = time.time() stats = { 'frame_count': 0, 'avg_fps': 0, 'max_gpu_mem': 0, 'process_time': 0, 'stream_id': None } frame_counts_per_stream = defaultdict(int) try: while True: frame_data = input_queue.get() if frame_data is None: break frame, stream_id, capture_time = frame_data stats['stream_id'] = stream_id # 单帧处理 start_process = time.time() results = processor.model.track(frame, imgsz=640, verbose=False) # 处理单帧 stats['process_time'] += time.time() - start_process processed_frame = processor.process_frame(frame) # 移除 cProfile if processed_frame is not None and processed_frame.size > 0: stats['frame_count'] += 1 frame_counts_per_stream[stream_id] += 1 result_queue.put((processed_frame, stream_id, capture_time)) # 定期更新统计信息 if stats['frame_count'] % monitor_interval == 0: duration = time.time() - start_time stats['avg_fps'] = stats['frame_count'] / duration if torch.cuda.is_available(): mem = torch.cuda.max_memory_allocated() / (1024 ** 2) stats['max_gpu_mem'] = max(stats['max_gpu_mem'], mem) stats['worker_pid'] = os.getpid() stats['frame_counts_per_stream'] = dict(frame_counts_per_stream) stats_queue.put(('stats', stats.copy())) except Exception as e: print(f"工作进程错误: {e}") finally: stats['worker_pid'] = os.getpid() stats['frame_counts_per_stream'] = dict(frame_counts_per_stream) stats_queue.put(stats) def get_gpu_info(): """获取GPU信息""" pynvml.nvmlInit() gpu_count = pynvml.nvmlDeviceGetCount() gpus = [] for i in range(gpu_count): handle = pynvml.nvmlDeviceGetHandleByIndex(i) name = pynvml.nvmlDeviceGetName(handle) mem = pynvml.nvmlDeviceGetMemoryInfo(handle) gpus.append({ 'id': i, 'name': name.decode('utf-8') if isinstance(name, bytes) else name, 'total_mem': mem.total / (1024 ** 2) }) pynvml.nvmlShutdown() return gpus import os import argparse def monitor_resources(gpu_id, interval=5): """资源监控线程""" pynvml.nvmlInit() handle = pynvml.nvmlDeviceGetHandleByIndex(gpu_id) while True: # GPU监控 util = pynvml.nvmlDeviceGetUtilizationRates(handle) mem_info = pynvml.nvmlDeviceGetMemoryInfo(handle) # CPU监控 cpu_percent = psutil.cpu_percent() mem = psutil.virtual_memory() print(f"\n[资源监控] GPU: {util.gpu}% 显存: {mem_info.used/1024**2:.1f}/{mem_info.total/1024**2:.1f}MB | " f"CPU: {cpu_percent}% 内存: {mem.used/1024**3:.1f}/{mem.total/1024**3:.1f}GB") time.sleep(interval) class DynamicProcessManager: def __init__(self, num_workers): self.num_workers = num_workers self.processes = [] self.result_queues = [] def start_workers(self, input_queue, gpu_id, result_queue, stats_queue): for i in range(self.num_workers): p = multiprocessing.Process( target=worker_process, args=(input_queue, gpu_id, result_queue, stats_queue) ) self.processes.append(p) p.start() def stop_workers(self): for p in self.processes: if p.is_alive(): p.terminate() try: p.join(timeout=1) except: pass if p.is_alive(): if platform.system() == "Windows": subprocess.run(['taskkill', '/F', '/PID', str(p.pid)], check=False) else: os.kill(p.pid, signal.SIGKILL) print(f"强制终止进程 {p.pid}") self.processes = [] def get_gpu_info(): pynvml.nvmlInit() gpu_count = pynvml.nvmlDeviceGetCount() gpus = [] for i in range(gpu_count): handle = pynvml.nvmlDeviceGetHandleByIndex(i) name = pynvml.nvmlDeviceGetName(handle) mem = pynvml.nvmlDeviceGetMemoryInfo(handle) gpus.append({ 'id': i, 'name': name.decode('utf-8') if isinstance(name, bytes) else name, 'total_mem': mem.total / (1024 ** 2) }) pynvml.nvmlShutdown() return gpus class ProgramMonitor: def __init__(self, gpu_id, process_manager, result_queue, stats_queue, args): self.gpu_id = gpu_id self.result_queue = result_queue self.stats_queue = stats_queue self.process_manager = process_manager self.args = args self.running = False self.stop_flag = threading.Event() self.data = { 'process': defaultdict(list), 'workers': defaultdict(list), 'gpu': defaultdict(list), 'fps_per_stream': defaultdict(list), 'total_fps': [], 'worker_stats': [], 'cpu_per_core': [], 'mem_bandwidth': [] } self.lock = threading.Lock() self.gpu_info = { 'arch': "Ada Lovelace", 'sm_count': 56, 'cores_per_sm': 128, 'peak_tflops': 35.6 } self.total_frame_counts = defaultdict(int) self.last_frame_counts = defaultdict(lambda: defaultdict(int)) self.start_time = None self.stop_time = None self.last_mem_time = time.time() self.last_mem_bytes = psutil.virtual_memory().used def start(self): pynvml.nvmlInit() self.handle = pynvml.nvmlDeviceGetHandleByIndex(self.gpu_id) self.running = True self.start_time = time.time() self.thread = Thread(target=self._monitor, daemon=True) self.thread.start() def _monitor(self): last_cpu_times = {} while not self.stop_flag.is_set(): try: # Process stats from stats_queue try: data = self.stats_queue.get_nowait() if isinstance(data, tuple) and data[0] == 'stats': stats = data[1] worker_pid = stats['worker_pid'] frame_counts_per_stream = stats['frame_counts_per_stream'] with self.lock: for stream_id, count in frame_counts_per_stream.items(): delta = count - self.last_frame_counts[worker_pid][stream_id] self.total_frame_counts[stream_id] += delta self.last_frame_counts[worker_pid][stream_id] = count except queue.Empty: pass # Main process monitoring main_process = psutil.Process(os.getpid()) with main_process.oneshot(): current_cpu_time = main_process.cpu_times() pid = main_process.pid if pid in last_cpu_times: cpu_usage = self._calculate_cpu_usage(last_cpu_times[pid], current_cpu_time) self.data['process']['cpu'].append(cpu_usage) last_cpu_times[pid] = current_cpu_time self.data['process']['mem'].append(main_process.memory_info().rss / (1024 ** 2)) self.data['process']['threads'].append(main_process.num_threads()) # Worker processes monitoring for p in self.process_manager.processes: try: proc = psutil.Process(p.pid) with proc.oneshot(): current_cpu_time = proc.cpu_times() pid = p.pid if pid in last_cpu_times: cpu_usage = self._calculate_cpu_usage(last_cpu_times[pid], current_cpu_time) self.data['workers']['cpu'].append(cpu_usage) last_cpu_times[pid] = current_cpu_time self.data['workers']['mem'].append(proc.memory_info().rss / (1024 ** 2)) self.data['workers']['threads'].append(proc.num_threads()) except (psutil.NoSuchProcess, psutil.AccessDenied): continue # Memory bandwidth monitoring current_time = time.time() current_mem_bytes = psutil.virtual_memory().used time_delta = current_time - self.last_mem_time if time_delta > 0: mem_bandwidth = (current_mem_bytes - self.last_mem_bytes) / time_delta / (1024 ** 2) with self.lock: self.data['mem_bandwidth'].append(mem_bandwidth) self.last_mem_time = current_time self.last_mem_bytes = current_mem_bytes # CPU per core monitoring cpu_per_core = psutil.cpu_percent(percpu=True) with self.lock: self.data['cpu_per_core'].append(cpu_per_core) self._monitor_gpu() except Exception as e: print(f"监控错误: {str(e)}") time.sleep(0.5) def _monitor_gpu(self): try: util = pynvml.nvmlDeviceGetUtilizationRates(self.handle) mem_info = pynvml.nvmlDeviceGetMemoryInfo(self.handle) clock_mhz = pynvml.nvmlDeviceGetClockInfo(self.handle, pynvml.NVML_CLOCK_SM) current_tflops = (self.gpu_info['sm_count'] * (clock_mhz / 1000) * self.gpu_info['cores_per_sm'] * 2) / 1000 with self.lock: self.data['gpu']['util'].append(util.gpu) self.data['gpu']['mem'].append(mem_info.used / (1024 ** 2)) self.data['gpu']['tflops'].append(current_tflops) except pynvml.NVMLError as e: print(f"GPU监控错误: {str(e)}") def stop(self): self.stop_time = time.time() self.running = False self.stop_flag.set() if self.thread.is_alive(): self.thread.join(timeout=2) report = self.generate_report() pynvml.nvmlShutdown() return report def generate_report(self): report = "\n=== 程序资源使用报告 ===\n" # System information (unchanged) report += "\n[系统信息]\n" report += f"- CPU核心数: {psutil.cpu_count(logical=False)}物理/{psutil.cpu_count()}逻辑\n" report += f"- 系统内存: {psutil.virtual_memory().total / (1024**3):.1f}GB\n" report += f"- 系统CPU使用率: {psutil.cpu_percent(interval=1):.1f}%\n" report += f"- 系统内存使用: {psutil.virtual_memory().used / (1024**3):.1f}GB / {psutil.virtual_memory().total / (1024**3):.1f}GB\n" gpu_name_raw = pynvml.nvmlDeviceGetName(self.handle) gpu_name = gpu_name_raw.decode('utf-8') if isinstance(gpu_name_raw, bytes) else gpu_name_raw total_gpu_mem = pynvml.nvmlDeviceGetMemoryInfo(self.handle).total / (1024 ** 2) report += f"- GPU型号: {gpu_name}\n" report += f"- GPU总显存: {total_gpu_mem:.1f}MB\n" # Main process stats (unchanged) if self.data['process']['cpu']: report += "\n[主进程资源]\n" report += f"- 平均CPU使用率: {statistics.mean(self.data['process']['cpu']):.1f}%\n" report += f"- 峰值CPU使用率: {max(self.data['process']['cpu']):.1f}%\n" report += f"- 平均内存占用: {statistics.mean(self.data['process']['mem']):.1f}MB\n" report += f"- 峰值内存占用: {max(self.data['process']['mem']):.1f}MB\n" report += f"- 线程数: {max(self.data['process']['threads'])}\n" # Worker processes stats (unchanged except for FPS section) if self.data['workers']['cpu']: num_workers = min(self.args.streams * 4, psutil.cpu_count(logical=True) * 2) num_samples = len(self.data['workers']['cpu']) // num_workers if num_samples > 0: worker_cpu_per_sample = [self.data['workers']['cpu'][i*num_workers:(i+1)*num_workers] for i in range(num_samples)] worker_mem_per_sample = [self.data['workers']['mem'][i*num_workers:(i+1)*num_workers] for i in range(num_samples)] worker_threads_per_sample = [self.data['workers']['threads'][i*num_workers:(i+1)*num_workers] for i in range(num_samples)] avg_worker_cpu = statistics.mean([statistics.mean(sample) for sample in worker_cpu_per_sample]) total_worker_cpu = statistics.mean([sum(sample) for sample in worker_cpu_per_sample]) avg_worker_mem = statistics.mean([statistics.mean(sample) for sample in worker_mem_per_sample]) total_worker_mem = statistics.mean([sum(sample) for sample in worker_mem_per_sample]) max_total_worker_threads = max([sum(sample) for sample in worker_threads_per_sample]) report += f"\n[工作进程资源 ({num_workers}个)]\n" report += f"- 平均CPU使用率（每个进程）: {avg_worker_cpu:.1f}%\n" report += f"- 总CPU使用率: {total_worker_cpu:.1f}%\n" report += f"- 平均内存占用（每个进程）: {avg_worker_mem:.1f}MB\n" report += f"- 总内存占用: {total_worker_mem:.1f}MB\n" report += f"- 总线程数（峰值）: {max_total_worker_threads}\n" # Video stream performance with accurate FPS if self.total_frame_counts: elapsed_time = self.stop_time - self.start_time report += "\n[视频流性能]\n" for stream_id in range(self.args.streams): if stream_id in self.total_frame_counts: avg_fps = self.total_frame_counts[stream_id] / elapsed_time report += f"- 视频流 {stream_id}: 平均 FPS {avg_fps:.1f}\n" total_frames = sum(self.total_frame_counts.values()) total_fps = total_frames / elapsed_time report += f"- 总吞吐量: {total_fps:.1f} FPS\n" # CPU per core (unchanged) if self.data.get('cpu_per_core'): avg_cpu_per_core = [statistics.mean([sample[i] for sample in self.data['cpu_per_core']]) for i in range(len(self.data['cpu_per_core'][0]))] overall_avg_cpu = statistics.mean(avg_cpu_per_core) report += "\n[CPU 硬件线程利用率]\n" for i, avg in enumerate(avg_cpu_per_core): report += f"- 逻辑处理器 {i}: {avg:.1f}%\n" report += f"- 16 个硬件线程平均利用率: {overall_avg_cpu:.1f}%\n" # Total process stats (unchanged) if self.data['process']['cpu'] and self.data['workers']['cpu']: num_display_processes = self.args.streams total_cpu = statistics.mean(self.data['process']['cpu']) + total_worker_cpu total_mem = statistics.mean(self.data['process']['mem']) + total_worker_mem total_threads = max(self.data['process']['threads']) + max_total_worker_threads total_processes = 1 + num_workers + num_display_processes + 1 report += "\n[所有进程总计]\n" report += f"- 总CPU使用率: {total_cpu:.1f}%\n" report += f"- 总内存占用: {total_mem:.1f}MB\n" report += f"- 总线程数: {total_threads}\n" report += f"- 总进程数: {total_processes}（1个主进程 + {num_workers}个工作进程 + {num_display_processes}个显示进程 + 1个分发进程）\n" # GPU stats (unchanged) if self.data['gpu']['tflops']: avg_tflops = statistics.mean(self.data['gpu']['tflops']) util_percent = min((avg_tflops / self.gpu_info['peak_tflops']) * 100, 100.0) report += "\n[GPU资源]\n" report += f"- 平均利用率: {statistics.mean(self.data['gpu']['util']):.1f}%\n" report += f"- 峰值显存: {max(self.data['gpu']['mem']):.1f}MB\n" report += f"- 平均算力: {avg_tflops:.1f}/{self.gpu_info['peak_tflops']} TFLOPS\n" report += f"- 算力利用率: {util_percent:.1f}%\n" # Memory bandwidth (unchanged) if self.data.get('mem_bandwidth'): avg_mem_bandwidth = statistics.mean(self.data['mem_bandwidth']) max_mem_bandwidth = max(self.data['mem_bandwidth']) report += "\n[存储器带宽]\n" report += f"- 平均内存带宽: {avg_mem_bandwidth:.1f} MB/s\n" report += f"- 峰值内存带宽: {max_mem_bandwidth:.1f} MB/s\n" return report def _calculate_cpu_usage(self, prev_times, curr_times): """ 计算基于前后的 CPU 时间的使用率百分比。 参数: prev_times: 上一次的 CPU 时间（psutil.cpu_times 对象） curr_times: 当前的 CPU 时间（psutil.cpu_times 对象） 返回: CPU 使用率（百分比） """ delta_user = curr_times.user - prev_times.user delta_system = curr_times.system - prev_times.system delta_total = (curr_times.user + curr_times.system) - (prev_times.user + prev_times.system) if delta_total > 0: cpu_usage = ((delta_user + delta_system) / delta_total) * 100 else: cpu_usage = 0.0 return cpu_usage # _monitor_gpu and _calculate_cpu_usage remain unchanged def main(): parser = argparse.ArgumentParser() parser.add_argument('--streams', type=int, default=1) parser.add_argument('--source', type=str, default="") parser.add_argument('--gpu_id', type=int, default=0) args = parser.parse_args() camera_config = { 'username': 'admin', 'password': 'guoxinzhike901' } source_url = args.source if args.source else \ f"rtsp://{camera_config['username']}:{camera_config['password']}@192.168.1.108/" gpus = get_gpu_info() print("\n[硬件配置]") print(f"- CPU核心: {psutil.cpu_count(logical=False)}物理/{psutil.cpu_count()}逻辑") print(f"- 内存: {psutil.virtual_memory().total / (1024**3):.1f}GB") print(f"- 使用GPU {args.gpu_id}: {gpus[args.gpu_id]['name']}") print(f" 显存: {gpus[args.gpu_id]['total_mem']:.1f}MB") os.environ['OMP_NUM_THREADS'] = '1' os.environ['MKL_NUM_THREADS'] = '1' print(f"\n[测试配置]") print(f"- 模拟视频流数: {args.streams}") print(f"- 视频源: {source_url}") # 创建共享队列 frame_queue_size = max(2000, 200 * args.streams) shared_frame_queue = multiprocessing.Queue(maxsize=frame_queue_size) display_queue_size = max(50, 20 * args.streams) shared_result_queue = multiprocessing.Queue(maxsize=2000) stats_queue = multiprocessing.Queue() # New queue for stats # 固定工作进程数为 16 num_workers = 1 #min(args.streams * 8, psutil.cpu_count(logical=True) * 2) process_mgr = DynamicProcessManager(num_workers) simulator = StreamSimulator(source_url, args.streams, shared_frame_queue) monitor = ProgramMonitor(args.gpu_id, process_mgr, shared_result_queue, stats_queue, args) monitor.args = args # 传递 args latency_queue = multiprocessing.Queue() # 启动工作进程 process_mgr.start_workers(shared_frame_queue, args.gpu_id, shared_result_queue, stats_queue) # 启动分发进程 dispatch_p = multiprocessing.Process( target=dispatch_process, args=(shared_result_queue, simulator.display_queues), daemon=True ) dispatch_p.start() simulator.start() monitor.start() # 启动显示进程 display_threads = [] for i in range(args.streams): t = Thread(target=display_thread, args=(simulator.display_queues[i], i+1, latency_queue)) display_threads.append(t) t.start() time.sleep(0.5) print("\n[测试开始] 程序将运行30秒...") start_time = time.time() end_time = start_time + 60*5 try: while time.time() < end_time: time.sleep(1) remaining = int(end_time - time.time()) if remaining % 10 == 0 or remaining <= 5: print(f"剩余时间: {remaining}秒") finally: runtime = time.time() - start_time print(f"\n[测试完成] 实际运行时间: {runtime:.1f}秒") print("停止模拟器...") simulator.stop() print("生成报告并停止监控...") report = monitor.stop() print("停止工作进程...") process_mgr.stop_workers() # 停止显示线程 for q in simulator.display_queues: q.put(None) for t in display_threads: t.join() # 停止分发进程 shared_result_queue.put(None) dispatch_p.join(timeout=5) if dispatch_p.is_alive(): dispatch_p.terminate() # 收集延迟测量 latencies = [] while not latency_queue.empty(): try: stream_id, latency = latency_queue.get_nowait() latencies.append(latency) except queue.Empty: break if latencies: min_latency = min(latencies) max_latency = max(latencies) avg_latency = sum(latencies) / len(latencies) report += f"\n[延迟统计]\n" report += f"- 测量次数: {len(latencies)}\n" report += f"- 最低延迟: {min_latency:.3f}秒\n" report += f"- 最高延迟: {max_latency:.3f}秒\n" report += f"- 平均延迟: {avg_latency:.3f}秒\n" else: report += "\n[延迟统计]\n- 无延迟数据\n" if torch.cuda.is_available(): torch.cuda.empty_cache() print(report) if __name__ == '__main__': multiprocessing.set_start_method('spawn') # multiprocessing.set_start_method('fork') # Linux 默认方法 main() # 测试4路视频流 # python det_ocr_shipinliu_pre.py --streams 1 --gpu_id 0 """ === 程序资源使用报告 === === 程序资源使用报告 === [系统信息] - CPU核心数: 10物理/16逻辑 - 系统内存: 63.8GB - 系统CPU使用率: 14.1% - 系统内存使用: 26.3GB / 63.8GB - GPU型号: NVIDIA GeForce RTX 4070 SUPER - GPU总显存: 12282.0MB [主进程资源] - 平均CPU使用率: 16.3% - 峰值CPU使用率: 28.1% - 平均内存占用: 385.1MB - 峰值内存占用: 385.7MB - 线程数: 9 [工作进程资源 (16个)] - 平均CPU使用率（每个进程）: 22.1% - 总CPU使用率: 354.2% - 平均内存占用（每个进程）: 801.5MB - 总内存占用: 12823.3MB - 总线程数（峰值）: 304 [所有进程总计] - 总CPU使用率: 370.5% - 总内存占用: 13208.4MB - 总线程数: 313 - 总进程数: 19（1个主进程 + 16个工作进程 + 1个显示进程 + 1个分发进程） [GPU资源] - 平均利用率: 31.3% - 峰值显存: 8226.7MB - 平均算力: 22.7/35.6 TFLOPS - 算力利用率: 63.8% [延迟统计] - 测量次数: 67 - 最低延迟: 0.024秒 - 最高延迟: 2.499秒 - 平均延迟: 0.287秒 """ # python det_ocr_shipinliu_pre.py --streams 2 --gpu_id 0 """ === 程序资源使用报告 === [系统信息] - CPU核心数: 10物理/16逻辑 - 系统内存: 63.8GB - 系统CPU使用率: 9.8% - 系统内存使用: 26.3GB / 63.8GB - GPU型号: NVIDIA GeForce RTX 4070 SUPER - GPU总显存: 12282.0MB [主进程资源] - 平均CPU使用率: 15.3% - 峰值CPU使用率: 40.6% - 平均内存占用: 386.4MB - 峰值内存占用: 387.1MB - 线程数: 9 [工作进程资源 (16个)] - 平均CPU使用率（每个进程）: 20.8% - 总CPU使用率: 333.1% - 平均内存占用（每个进程）: 960.3MB - 总内存占用: 15364.2MB - 总线程数（峰值）: 328 [所有进程总计] - 总CPU使用率: 348.4% - 总内存占用: 15750.6MB - 总线程数: 337 - 总进程数: 20（1个主进程 + 16个工作进程 + 2个显示进程 + 1个分发进程） [GPU资源] - 平均利用率: 50.5% - 峰值显存: 8328.6MB - 平均算力: 12.6/35.6 TFLOPS - 算力利用率: 35.4% [延迟统计] - 测量次数: 327 - 最低延迟: 0.027秒 - 最高延迟: 0.757秒 - 平均延迟: 0.080秒 """ # python det_ocr_shipinliu_pre.py --streams 3 --gpu_id 0 """ [系统信息] - CPU核心数: 10物理/16逻辑 - 系统内存: 63.8GB - 系统CPU使用率: 9.5% - 系统内存使用: 26.2GB / 63.8GB - GPU型号: NVIDIA GeForce RTX 4070 SUPER - GPU总显存: 12282.0MB [主进程资源] - 平均CPU使用率: 26.2% - 峰值CPU使用率: 53.1% - 平均内存占用: 386.1MB - 峰值内存占用: 386.6MB - 线程数: 9 [工作进程资源 (16个)] - 平均CPU使用率（每个进程）: 43.9% - 总CPU使用率: 702.5% - 平均内存占用（每个进程）: 1018.8MB - 总内存占用: 16301.3MB - 总线程数（峰值）: 322 [所有进程总计] - 总CPU使用率: 728.7% - 总内存占用: 16687.5MB - 总线程数: 331 - 总进程数: 21（1个主进程 + 16个工作进程 + 3个显示进程 + 1个分发进程） [GPU资源] - 平均利用率: 52.2% - 峰值显存: 7861.9MB - 平均算力: 18.9/35.6 TFLOPS - 算力利用率: 53.1% [延迟统计] - 测量次数: 327 - 最低延迟: 0.030秒 - 最高延迟: 3.756秒 - 平均延迟: 1.077秒 """ # python det_ocr_shipinliu_pre.py --streams 4 --gpu_id 0 cpu100 """ === 程序资源使用报告 === [系统信息] - CPU核心数: 10物理/16逻辑 - 系统内存: 63.8GB - 系统CPU使用率: 58.6% - 系统内存使用: 36.3GB / 63.8GB - GPU型号: NVIDIA GeForce RTX 4070 SUPER - GPU总显存: 12282.0MB [主进程资源] - 平均CPU使用率: 28.0% - 峰值CPU使用率: 53.1% - 平均内存占用: 386.4MB - 峰值内存占用: 386.8MB - 线程数: 9 [工作进程资源 (16个)] - 平均CPU使用率（每个进程）: 48.0% - 总CPU使用率: 768.7% - 平均内存占用（每个进程）: 1585.2MB - 总内存占用: 25363.6MB - 总线程数（峰值）: 320 [所有进程总计] - 总CPU使用率: 796.7% - 总内存占用: 25750.1MB - 总线程数: 329 - 总进程数: 22（1个主进程 + 16个工作进程 + 4个显示进程 + 1个分发进程） [GPU资源] - 平均利用率: 52.9% - 峰值显存: 7991.3MB - 平均算力: 20.2/35.6 TFLOPS - 算力利用率: 56.8% [延迟统计] - 测量次数: 327 - 最低延迟: 1.480秒 - 最高延迟: 14.222秒 - 平均延迟: 8.113秒 """ # python det_ocr_shipinliu_pre.py --streams 5 --gpu_id 0 """ """ # python det_ocr_shipinliu_pre.py --streams 16 --gpu_id 0 """ """ # python det_ocr_shipinliu_pre.py --streams 20 --gpu_id 0 """ """ (yolov8_bt) (base) zhang@zhang:~/danger/yolov7_crnn_ocr_detection$ python det_ocr_shipinliu_pre.py --streams 1 --gpu_id 0 [硬件配置] - CPU核心: 10物理/16逻辑 - 内存: 62.6GB - 使用GPU 0: NVIDIA GeForce RTX 4070 SUPER 显存: 12282.0MB [测试配置] - 模拟视频流数: 1 - 视频源: rtsp://admin:[email protected]/ [测试开始] 程序将运行30秒... In worker process 35804, np is <module 'numpy' from '/home/zhang/miniconda3/envs/yolov8_bt/lib/python3.9/site-packages/numpy/__init__.py'>, type(np.empty((1,))) = <class 'numpy.ndarray'> Loading weights/best.engine for TensorRT inference... [06/07/2025-18:54:11] [TRT] [I] Loaded engine size: 39 MiB [06/07/2025-18:54:11] [TRT] [W] Using an engine plan file across different models of devices is not recommended and is likely to affect performance or even cause errors. [06/07/2025-18:54:13] [TRT] [I] [MemUsageChange] TensorRT-managed allocation in engine deserialization: CPU +0, GPU +33, now: CPU 0, GPU 33 (MiB) [06/07/2025-18:54:13] [TRT] [I] [MemUsageChange] TensorRT-managed allocation in IExecutionContext creation: CPU +0, GPU +39, now: CPU 0, GPU 72 (MiB) 剩余时间: 290秒 剩余时间: 280秒 剩余时间: 270秒 (yolov8_bt) (base) zhang@zhang:~/danger/yolov7_crnn_ocr_detection$ python -c "import numpy as np; print(np.__version__)" 1.23.0 代码运行后报错

memory_pool_limit(trt.MemoryPoolType.WORKSPACE,1)#1GBengine=builder.build_engine(network,config)returnengineif__name__=="__main__":#使用spawn方法启动进程mp.set_start_method('spawn')#初始化主进程CUDA...

(VScode) zhaohy@workspace2:~/VSCode-main$ python train_test_eval.py --Training True --Testing True --Evaluation True /home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/timm/models/layers/__init__.py:48: FutureWarning: Importing from timm.models.layers is deprecated, please import via timm.layers warnings.warn(f"Importing from {__name__} is deprecated, please import via timm.layers", FutureWarning) /home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/timm/models/layers/__init__.py:48: FutureWarning: Importing from timm.models.layers is deprecated, please import via timm.layers warnings.warn(f"Importing from {__name__} is deprecated, please import via timm.layers", FutureWarning) /home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/timm/models/layers/__init__.py:48: FutureWarning: Importing from timm.models.layers is deprecated, please import via timm.layers warnings.warn(f"Importing from {__name__} is deprecated, please import via timm.layers", FutureWarning) W0311 15:54:21.788752 136232691705664 torch/multiprocessing/spawn.py:145] Terminating process 2342884 via signal SIGTERM Traceback (most recent call last): File "train_test_eval.py", line 67, in <module> Training.train_net(num_gpus=num_gpus, args=args) File "/home/zhaohy/VSCode-main/Training.py", line 64, in train_net mp.spawn(main, nprocs=num_gpus, args=(num_gpus, args)) File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/multiprocessing/spawn.py", line 281, in spawn return start_processes(fn, args, nprocs, join, daemon, start_method="spawn") File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/multiprocessing/spawn.py", line 237, in start_processes while not context.join(): File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/multiprocessing/spawn.py", line 188, in join raise ProcessRaisedException(msg, error_index, failed_process.pid) torch.multiprocessing.spawn.ProcessRaisedException: -- Process 1 terminated with the following error: Traceback (most recent call last): File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/multiprocessing/spawn.py", line 75, in _wrap fn(i, *args) File "/home/zhaohy/VSCode-main/Training.py", line 71, in main dist.init_process_group(backend='nccl', init_method='env://') # 自动从环境变量读取 RANK/WORLD_SIZE File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/distributed/c10d_logger.py", line 75, in wrapper return func(*args, **kwargs) File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/distributed/c10d_logger.py", line 89, in wrapper func_return = func(*args, **kwargs) File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/distributed/distributed_c10d.py", line 1305, in init_process_group store, rank, world_size = next(rendezvous_iterator) File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/distributed/rendezvous.py", line 234, in _env_rendezvous_handler rank = int(_get_env_or_raise("RANK")) File "/home/zhaohy/anaconda3/envs/VScode/lib/python3.8/site-packages/torch/distributed/rendezvous.py", line 219, in _get_env_or_raise raise _env_error(env_var) ValueError: Error initializing torch.distributed using env:// rendezvous: environment variable RANK expected, but not set

这通常发生在使用torch.multiprocessing.spawn启动多进程时，没有正确配置环境变量，或者在单机多卡训练中没有使用正确的启动命令。 可能的解决方法是检查代码中是否在启动多进程时正确传递了参数，或者是否应该...

Traceback (most recent call last): File "DT_001_X01_P01.py", line 150, in DT_001_X01_P01.Module.load_model File "/home/kejia/Server/tf/Bin_x64/DeepLearning/DL_Lib_02/mmdet/apis/inference.py", line 42, in init_detector checkpoint = load_checkpoint(model, checkpoint, map_location=map_loc) File "/home/kejia/Server/tf/Bin_x64/DeepLearning/DL_Lib_02/mmcv/runner/checkpoint.py", line 529, in load_checkpoint checkpoint = _load_checkpoint(filename, map_location, logger) File "/home/kejia/Server/tf/Bin_x64/DeepLearning/DL_Lib_02/mmcv/runner/checkpoint.py", line 467, in _load_checkpoint return CheckpointLoader.load_checkpoint(filename, map_location, logger) File "/home/kejia/Server/tf/Bin_x64/DeepLearning/DL_Lib_02/mmcv/runner/checkpoint.py", line 244, in load_checkpoint return checkpoint_loader(filename, map_location) File "/home/kejia/Server/tf/Bin_x64/DeepLearning/DL_Lib_02/mmcv/runner/checkpoint.py", line 261, in load_from_local checkpoint = torch.load(filename, map_location=map_location) File "torch/serialization.py", line 594, in load return _load(opened_zipfile, map_location, pickle_module, **pickle_load_args) File "torch/serialization.py", line 853, in _load result = unpickler.load() File "torch/serialization.py", line 845, in persistent_load load_tensor(data_type, size, key, _maybe_decode_ascii(location)) File "torch/serialization.py", line 834, in load_tensor loaded_storages[key] = restore_location(storage, location) File "torch/serialization.py", line 175, in default_restore_location result = fn(storage, location) File "torch/serialization.py", line 157, in _cuda_deserialize return obj.cuda(device) File "torch/_utils.py", line 71, in _cuda with torch.cuda.device(device): File "torch/cuda/__init__.py", line 225, in __enter__ self.prev_idx = torch._C._cuda_getDevice() File "torch/cuda/__init__.py", line 164, in _lazy_init "Cannot re-initialize CUDA in forked subprocess. " + msg) RuntimeError: Cannot re-initialize CUDA in forked subprocess. To use CUDA with multiprocessing, you must use the 'spawn' start method ('异常抛出', None) DT_001_X01_P01 load_model ret=1, version=V1.0.0.0

在上述代码中，mp.set_start_method('spawn') 将设置 'spawn' 启动方法为当前进程的默认启动方法。这样，在创建子进程时就会使用 'spawn' 启动方法，避免 CUDA 初始化错误。 希望这个解决方案对你有所帮助。如果...

ImportError: cannot import name 'reduction' from partially initialized module 'multiprocessing' (most likely due to a circular import) (D:\software\anoconda\envs\efficientteacher-main\Lib\site-packages\torch\multiprocessing\__init__.py)

multiprocessing.set_start_method('spawn', force=True) # Windows必须设置 import torch **注意**：Windows 必须设置 spawn 启动方法[^4]。 #### 3. **检查 Python 路径** 在导入前添加当前工作目录到...

torch.no_grad():出现RuntimeError: An attempt has been made to start a new process before the current process has finished its bootstrapping phase. This probably means that you are not using fork to start your child processes and you have forgotten to use the proper idiom in the main module: if name == 'main': freeze_support() ... The "freeze_support()" line can be omitted if the program is not going to be frozen to produce an executable.

这个错误通常是由于在 Windows 操作系统下使用 torch....torch.multiprocessing.set_start_method('spawn', force=True) 这样可以确保在 Windows 操作系统下使用 torch.multiprocessing 时不会出现上述错误。

Traceback (most recent call last): File "<string>", line 1, in <module> File "C:\Users\14754\anaconda3\Lib\multiprocessing\spawn.py", line 122, in spawn_main exitcode = _main(fd, parent_sentinel) ^^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\14754\anaconda3\Lib\multiprocessing\spawn.py", line 131, in _main prepare(preparation_data) File "C:\Users\14754\anaconda3\Lib\multiprocessing\spawn.py", line 246, in prepare _fixup_main_from_path(data['init_main_from_path']) File "C:\Users\14754\anaconda3\Lib\multiprocessing\spawn.py", line 297, in _fixup_main_from_path main_content = runpy.run_path(main_path, ^^^^^^^^^^^^^^^^^^^^^^^^^ File "<frozen runpy>", line 287, in run_path File "<frozen runpy>", line 98, in _run_module_code File "<frozen runpy>", line 88, in _run_code File "C:\Users\14754\PycharmProjects\PythonProject2\预训练01.py", line 101, in <module> images, labels = next(iter(train_loader)) ^^^^^^^^^^^^^^^^^^ File "C:\Users\14754\anaconda3\Lib\site-packages\torch\utils\data\dataloader.py", line 439, in __iter__ return self._get_iterator() ^^^^^^^^^^^^^^^^^^^^ File "C:\Users\14754\anaconda3\Lib\site-packages\torch\utils\data\dataloader.py", line 387, in _get_iterator return _MultiProcessingDataLoaderIter(self) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\14754\anaconda3\Lib\site-packages\torch\utils\data\dataloader.py", line 1040, in __init__ w.start() File "C:\Users\14754\anaconda3\Lib\multiprocessing\process.py", line 121, in start self._popen = self._Popen(self) ^^^^^^^^^^^^^^^^^ File "C:\Users\14754\anaconda3\Lib\multiprocessing\context.py", line 224, in _Popen return _default_context.get_context().Process._Popen(process_obj) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\14754\anaconda3\Lib\multiprocessing\context.py", line 337, in _Popen return Popen(process_obj) ^^^^^^^^^^^^^^^^^^ File "C:\Users\14754\anaconda3\Lib\multiprocessing\popen_spawn_win32.py", line 46, in __init__ prep_data = spawn.get_preparation_data(process_obj._name) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\14754\anaconda3\Lib\multiprocessing\spawn.py", line 164, in get_preparation_data _check_not_importing_main() File "C:\Users\14754\anaconda3\Lib\multiprocessing\spawn.py", line 140, in _check_not_importing_main raise RuntimeError(''' RuntimeError: An attempt has been made to start a new process before the current process has finished its bootstrapping phase. This probably means that you are not using fork to start your child processes and you have forgotten to use the proper idiom in the main module: if __name__ == '__main__': freeze_support() ... The "freeze_support()" line can be omitted if the program is not going to be frozen to produce an executable. To fix this issue, refer to the "Safe importing of main module" section in https://docs.python.org/3/library/multiprocessing.html

multiprocessing.set_start_method('spawn') # 强制使用 spawn 方法[^5] p = multiprocessing.Process(target=worker) p.start() p.join() 4. **检查代码逻辑** 确保没有在主模块之外直接调用可能触发新...

File "<string>", line 1, in <module> from multiprocessing.spawn import spawn_main; spawn_main(parent_pid=26856, pipe_handle=1552) ~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\spawn.py", line 122, in spawn_main exitcode = _main(fd, parent_sentinel) File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\spawn.py", line 131, in _main prepare(preparation_data) ~~~~~~~^^^^^^^^^^^^^^^^^^ File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\spawn.py", line 246, in prepare _fixup_main_from_path(data['init_main_from_path']) ~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\spawn.py", line 297, in _fixup_main_from_path main_content = runpy.run_path(main_path, run_name="__mp_main__") File "<frozen runpy>", line 287, in run_path File "<frozen runpy>", line 98, in _run_module_code File "<frozen runpy>", line 88, in _run_code File "f:\yolov8\ultralytics-main\test.py", line 8, in <module> model.train(data='yolo_bvn.yaml',workers=1,epochs=1, batch=16) ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "f:\yolov8\ultralytics-main\ultralytics\engine\model.py", line 799, in train self.trainer.train() ~~~~~~~~~~~~~~~~~~^^ File "f:\yolov8\ultralytics-main\ultralytics\engine\trainer.py", line 227, in train self._do_train(world_size) ~~~~~~~~~~~~~~^^^^^^^^^^^^ File "f:\yolov8\ultralytics-main\ultralytics\engine\trainer.py", line 348, in _do_train self._setup_train(world_size) ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^ File "f:\yolov8\ultralytics-main\ultralytics\engine\trainer.py", line 307, in _setup_train self.train_loader = self.get_dataloader( ~~~~~~~~~~~~~~~~~~~^ self.data["train"], batch_size=batch_size, rank=LOCAL_RANK, mode="train" ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ) ^ File "f:\yolov8\ultralytics-main\ultralytics\models\yolo\detect\train.py", line 90, in get_dataloader return build_dataloader(dataset, batch_size, workers, shuffle, rank) # return dataloader File "f:\yolov8\ultralytics-main\ultralytics\data\build.py", line 183, in build_dataloader return InfiniteDataLoader( dataset=dataset, ...<8 lines>... drop_last=drop_last, ) File "f:\yolov8\ultralytics-main\ultralytics\data\build.py", line 58, in __init__ self.iterator = super().__iter__() ~~~~~~~~~~~~~~~~^^ File "F:\Minicond3\envs\yolov8\Lib\site-packages\torch\utils\data\dataloader.py", line 493, in __iter__ return self._get_iterator() ~~~~~~~~~~~~~~~~~~^^ File "F:\Minicond3\envs\yolov8\Lib\site-packages\torch\utils\data\dataloader.py", line 424, in _get_iterator return _MultiProcessingDataLoaderIter(self) File "F:\Minicond3\envs\yolov8\Lib\site-packages\torch\utils\data\dataloader.py", line 1171, in __init__ w.start() ~~~~~~~^^ File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\process.py", line 121, in start self._popen = self._Popen(self) ~~~~~~~~~~~^^^^^^ File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\context.py", line 224, in _Popen return _default_context.get_context().Process._Popen(process_obj) ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^ File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\context.py", line 337, in _Popen return Popen(process_obj) File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\popen_spawn_win32.py", line 47, in __init__ prep_data = spawn.get_preparation_data(process_obj._name) File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\spawn.py", line 164, in get_preparation_data _check_not_importing_main() ~~~~~~~~~~~~~~~~~~~~~~~~~^^ File "F:\Minicond3\envs\yolov8\Lib\multiprocessing\spawn.py", line 140, in _check_not_importing_main raise RuntimeError(''' ...<16 lines>... ''') RuntimeError: An attempt has been made to start a new process before the current process has finished its bootstrapping phase. This probably means that you are not using fork to start your child processes and you have forgotten to use the proper idiom in the main module: if __name__ == '__main__': freeze_support() ... The "freeze_support()" line can be omitted if the program is not going to be frozen to produce an executable. To fix this issue, refer to the "Safe importing of main module" section in https://docs.python.org/3/library/multiprocessing.html

- 使用 torch.multiprocessing.set_start_method('spawn') 显式声明 3. 跨平台兼容： python import platform if platform.system() == 'Windows': import multiprocessing multiprocessing.freeze_...

你好，你好。

交互系统公司（Interactive Systems Corporation，或者INITERACTIVE Systems Corporation，简称ISC）是AT&T以外的第一家Unix操作系统销售商，一家来自加利福尼亚州圣摩尼卡的美国软件公司。 ISC在1977年提供IS/1，PDP-11的一个办公自动化增强的Version 6 Unix变体。IS/3和IS5是PDP-11和VAX的Uni
pptx

精选ppt蓝色科技风区块链物联网信息科技.pptx

精选ppt蓝色科技风区块链物联网信息科技.pptx

大家在看

recommend-type

Visual+Basic.NET程序设计教程》作者李兰友

绝对经典的vb教程，强力推荐初学者看。这个压缩包里包含pdf版和word版两个版本，制作vb笔记和课件都很方便。
recommend-type

北大青鸟net培训ppt

讲得比较基础，很适合刚接触C#.net的人使用
recommend-type

Kvaser CANLIB API.pdf

Kvaser CANLIB是Kvaser 的CAN测试设备提供的API
recommend-type

CHM转HTML及汉化工具.rar

看CHM英文文档，有时候很累，这时候可以使用chmdecoder和google将其快速转化成中文，然后使用CHM汉化工具复制
recommend-type

STM8 LIN2.x 协议栈

这个LIN 的协议栈是ST 官方的, 早期是在官网可以直接搜索下载的. 最近一段时间去ST 官网看的时候, 发现直接搜索LIN 协议栈搜索不到了(没有积分的同志可以直接在官网搜索 "en.stsw-stm8a-lin" 还是可以找到的.). 所以在这里上传分享一下!

最新推荐

recommend-type

Pansophica开源项目：智能Web搜索代理的探索

Pansophica开源项目是一个相对较新且具有创新性的智能Web搜索代理，它突破了传统搜索引擎的界限，提供了一种全新的交互方式。首先，我们来探讨“智能Web搜索代理”这一概念。智能Web搜索代理是一个软件程序或服务，它可以根据用户的查询自动执行Web搜索，并尝试根据用户的兴趣、历史搜索记录或其他输入来提供个性化的搜索结果。 Pansophica所代表的不仅仅是搜索结果的展示，它还强调了一个交互式的体验，在动态和交互式虚拟现实中呈现搜索结果。这种呈现方式与现有的搜索体验有着根本的不同。目前的搜索引擎，如Google、Bing和Baidu等，多以静态文本和链接列表的形式展示结果。而Pansophica通过提供一个虚拟现实环境，使得搜索者可以“扭转”视角，进行“飞行”探索，以及“弹网”来浏览不同的内容。这种多维度的交互方式使得信息的浏览变得更加快速和直观，有望改变用户与网络信息互动的方式。 接着，我们关注Pansophica的“开源”属性。所谓开源，指的是软件的源代码可以被公众获取，任何个人或组织都可以自由地使用、学习、修改和分发这些代码。开源软件通常由社区进行开发和维护，这样的模式鼓励了协作创新并减少了重复性劳动，因为全世界的开发者都可以贡献自己的力量。Pansophica项目作为开源软件，意味着其他开发者可以访问和使用其源代码，进一步改进和扩展其功能，甚至可以为Pansophica构建新的应用或服务。 最后，文件名称“Pansophica-src-1.3”表明了我们讨论的特定版本的Pansophica开源代码。数字“1.3”很可能指的是该版本号，表明这是Pansophica项目的第1.3个公开版本。这个版本号对于开发者来说很重要，因为它可以帮助他们追踪更新的进度、比较版本间的差异以及选择最适合其项目的版本。 综合以上信息，Pansophica开源项目不仅在搜索代理领域引入了创新的交互方式，而且通过开源的方式，提供了技术共享和协作的可能性。对于IT行业来说，开源项目是推动技术发展的重要力量，它激发了社区的活力，促进了知识的传播和技术的进步。通过构建一个开源社区，Pansophica可以吸引来自世界各地的开发者和研究人员，共同探索如何将搜索技术与交互式虚拟现实技术结合起来，为用户带来更加丰富和有效的信息检索体验。同时，Pansophica的开源性质也使其更容易适应未来技术的发展，随着社区的不断贡献和反馈，这一平台有望在功能和性能上得到持续的优化和增强。
recommend-type

跨平台内容提取无忧：coze工作流应对社交媒体挑战

# 1. 社交媒体内容提取的必要性与挑战 ## 1.1 内容提取的必要性 在数字时代，社交媒体内容是信息与数据流动的主干道，公司、个人和研究者都在寻求从这些信息中提取价值。无论是为了市场趋势分析、品牌监控、消费者洞察还是公共舆论监测，有效的社交媒体内容提取变得至关重要。这不仅有助于企业做出数据驱动的决策，也能使个人更好地理
recommend-type

vrrp主设备发送的免费arp

<think>好的，我现在需要回答用户关于VRRP主设备发送免费ARP的目的和工作原理的问题。首先，我得回忆一下VRRP的基本概念和工作机制。VRRP，虚拟路由冗余协议，主要用于在局域网中提供默认网关的冗余备份，确保网络的高可用性。主设备（Master）负责转发流量，而备用设备（Backup）则在主设备故障时接管工作。 用户的问题集中在主设备发送免费ARP的目的和机制上。根据引用[2]，免费ARP用于通知下游设备虚拟MAC地址的变更。当主设备被选举出来后，它需要让局域网内的其他设备知道虚拟IP对应的MAC地址已经指向自己，这样流量才会被正确路由到主设备。免费ARP的作用应该就是更新这些设备的
recommend-type

为Ghost博客平台打造的Meteor流星包装使用指南

从给定文件信息中，我们可以提炼出以下IT知识点： ### 标题知识点：流星Ghost软件包 1. **流星Ghost软件包的用途**：流星Ghost软件包是专为Ghost博客平台设计的流星（Meteor）应用程序。流星是一个开源的全栈JavaScript平台，用于开发高性能和易于编写的Web应用程序。Ghost是一个开源博客平台，它提供了一个简单且专业的写作环境。 2. **软件包的作用**：流星Ghost软件包允许用户在流星平台上轻松集成Ghost博客。这样做的好处是可以利用流星的实时特性以及易于开发和部署的应用程序框架，同时还能享受到Ghost博客系统的便利和美观。 ### 描述知识点：流星Ghost软件包的使用方法 1. **软件包安装方式**：用户可以通过流星的命令行工具添加名为`mrt:ghost`的软件包。`mrt`是流星的一个命令行工具，用于添加、管理以及配置软件包。 2. **初始化Ghost服务器**：描述中提供了如何在服务器启动时运行Ghost的基本代码示例。这段代码使用了JavaScript的Promise异步操作，`ghost().then(function (ghostServer) {...})`这行代码表示当Ghost服务器初始化完成后，会在Promise的回调函数中提供一个Ghost服务器实例。 3. **配置Ghost博客**：在`then`方法中，首先会获取到Ghost服务器的配置对象`config`，用户可以在此处进行自定义设置，例如修改主题、配置等。 4. **启动Ghost服务器**：在配置完成之后，通过调用`ghostServer.start()`来启动Ghost服务，使其能够处理博客相关的请求。 5. **Web浏览器导航**：一旦流星服务器启动并运行，用户便可以通过Web浏览器访问Ghost博客平台。 ### 标签知识点：JavaScript 1. **JavaScript作为流星Ghost软件包的开发语言**：标签指出流星Ghost软件包是使用JavaScript语言开发的。JavaScript是一种在浏览器端广泛使用的脚本语言，它也是流星平台的基础编程语言。 2. **流星和Ghost共同使用的语言**：JavaScript同样也是Ghost博客平台的开发语言。这表明流星Ghost软件包可以无缝集成，因为底层技术栈相同。 ### 压缩包子文件的文件名称列表知识点：meteor-ghost-master 1. **版本控制和软件包结构**：文件名称`meteor-ghost-master`暗示了该软件包可能托管在像GitHub这样的版本控制系统上。文件名中的`master`通常指的是主分支或主版本。 2. **软件包的目录结构**：通过文件名称可以推断出该软件包可能拥有一个标准的流星软件包结构，包含了初始化、配置、运行等必要的模块和文件。 3. **软件包的维护状态**：由于文件名没有包含特定的版本号，我们无法直接得知软件包的最新更新情况。通常，软件包维护者会将最新的版本代码放在`master`分支上。 ### 总结 流星Ghost软件包提供了一个有效的解决方案，使得流星平台的开发者能够在他们的应用中添加Ghost博客功能。软件包的使用简便，通过流星的命令行工具安装，并通过JavaScript代码配置和启动Ghost服务。通过流星Ghost软件包，开发者能够享受流星的实时特性以及Ghost博客系统的便利性。此外，软件包的命名和结构也暗示了其维护和版本控制的模式，有助于开发者更好地理解如何使用和维护这一软件包。
recommend-type

抖音标题生成自动化：用coze工作流释放创意

# 1. 抖音标题生成自动化的重要性 随着社交媒体平台的崛起，内容的吸引力很大程度上取决于标题的创意与精准性。抖音作为一个日活亿级的短视频平台，高质量的标题能够有效提高视频的点击率，增加内容的传播。但是，人工撰写标题不仅耗时耗力，而且很难做到快速响应热点，自动化标题生成工具应运而生。coze工作流，作为一种实现自动化生成抖音标题的工具，其重要性不言而喻。它能够利用大数据分析和机器学习技术，提高标题的吸引
recommend-type

spss消费结构因子分析

<think>好的，我现在需要帮助用户在SPSS中进行消费结构的因子分析。首先，我要回忆一下因子分析的基本步骤和SPSS的操作流程。用户可能对SPSS不太熟悉，所以步骤要详细，同时需要结合他们提供的引用内容，特别是引用[2]中的适用条件和检验方法。 首先，用户提到了消费结构的数据，这可能包括多个变量，如食品、住房、交通等支出。因子分析适用于这种情况，可以降维并找出潜在因子。根据引用[2]，需要检查样本量是否足够，变量间是否有相关性，以及KMO和Bartlett检验的结果。 接下来，我需要按照步骤组织回答：数据准备、适用性检验、因子提取、因子旋转、命名解释、计算得分。每个步骤都要简明扼要，说
recommend-type

OpenMediaVault的Docker映像：快速部署与管理指南

根据提供的文件信息，我们将详细讨论与标题和描述中提及的Docker、OpenMediaVault以及如何部署OpenMediaVault的Docker镜像相关的一系列知识点。 首先，Docker是一个开源的应用容器引擎，允许开发者打包应用及其依赖包到一个可移植的容器中，然后发布到任何流行的Linux机器上，也可以实现虚拟化。容器是完全使用沙箱机制，相互之间不会有任何接口（类似 iPhone 的 app）。 OpenMediaVault是一个基于Debian的NAS（网络附加存储）解决方案。它专为家庭或小型办公室提供文件共享、网络附加存储以及打印服务。它提供了一个易用的Web界面，通过这个界面用户可以管理服务器配置、网络设置、用户权限、文件服务等。 在描述中提到了一些Docker命令行操作： 1. `git clone`：用于克隆仓库到本地，这里的仓库指的是“docker-images-openmedivault”。 2. `docker build -t omv`：这是一个构建Docker镜像的命令，其中`-t`参数用于标记镜像名称和标签，这里是标记为“omv”。 3. `docker run`：运行一个容器实例，`-t`参数用于分配一个伪终端，`-i`参数用于交互式操作，`-p 80:80`则是将容器的80端口映射到宿主机的80端口。 启动服务的部分涉及OpenMediaVault的配置和初始化： - ssh服务：用于远程登录到服务器的协议。 - php5-fpm：是PHP的一个FastCGI实现，用于加速PHP的运行。 - nginx：是一个高性能的HTTP和反向代理服务器，常用于优化静态内容的分发。 - openmediavault引擎：指的是OpenMediaVault的核心服务。 - rrdcached：用于收集和缓存性能数据，这些数据可以被rrdtool图形化工具读取。 - collectd：是一个守护进程，用于收集系统性能和提供各种存储方式和传输方式来存储所收集的数据。 为了访问服务，需要在浏览器中输入"http:// IP_OF_DOCKER"，其中`IP_OF_DOCKER`指的是运行Docker容器的主机IP地址。 描述中还提到了一个步骤：“在System-> Network-> Interfaces中添加带有dhcp的eth0”，这指的是需要在OpenMediaVault的Web管理界面中配置网络接口。`eth0`是网络接口的名称，通常代表第一个以太网接口。DHCP（动态主机配置协议）是一种自动为网络中的设备分配IP地址的协议，这样设备就可以连接网络并开始通信，无需手动配置IP地址。 【压缩包子文件的文件名称列表】中的“docker-images-openmediavault-master”暗示了这是一个包含Docker镜像文件的代码仓库。通常，“master”分支是代码的主分支，包含了代码库中最新且通常是最稳定的版本。用户可以通过克隆该仓库到本地来获取所有相关的Dockerfile、配置脚本及依赖文件，以便能够自行构建和运行OpenMediaVault的Docker镜像。 综上所述，这些知识点涵盖了从基本的Docker概念、Docker命令行操作、OpenMediaVault服务启动和管理，到具体的网络配置及Docker仓库操作，都是进行Docker化OpenMediaVault部署的关键步骤。
recommend-type

小红书文案提取一步到位：coze工作流操作全攻略

# 1. coze工作流概述 工作流系统是企业信息化和数字化转型的核心组件之一，它通过自动化流程管理提升效率，确保业务流程的顺畅执行。coze工作流作为当前市场上较为先进的工作流解决方案，它不仅仅是一套软件工具，更是一个集成化的平台，旨在通过流程自动化和智能化提升企业运营效率。 coze工作流的引入不仅有助于标准化和优化企业的业务流程，还可以通过可配置的流程设计，满足不同部门的特定需求。在组织的业务流程中
recommend-type

戴尔R630设置来电自动开机

<think>嗯，用户需要设置Dell R630服务器在断电恢复后自动开机。这个问题很实用，特别是对于托管在机房或需要远程管理的服务器来说，电源恢复后自动启动能减少运维压力。 从用户提到的R630型号来看，这是Dell PowerEdge系列的主流机型，设置方法应该和其他同代服务器类似。用户可能遇到意外断电后需要手动开机的情况，这会影响服务可用性。 我回忆Dell服务器的电源管理逻辑：主要有两个关键设置点。首先是BIOS里的电源恢复策略，选项包括"保持关机"、"自动开机"和"恢复断电前状态"。其次是iDRAC远程管理卡里的设置，这个更灵活但需要配置网络。 用户可能不熟悉服务器管理，所以需
recommend-type

React-Glide：掌握React轻量级多用途轮播模块

React-glide是一个专为React框架设计的内置多用途轮播模块，它允许开发者在网页中轻松地实现图片或内容的轮播效果。轮播图是一种常见的网页元素，常用于展示一系列的图片或内容，以轮动的方式切换显示，以达到吸引用户注意和提供信息的功能。 首先，需要了解React框架。React是由Facebook开发的一个用于构建用户界面的JavaScript库，它遵循组件化思想，能够将复杂的应用分解成小型的、独立的、可复用的组件。React-glide正是建立在React组件化的基础上，提供了一个轻量级且功能丰富的轮播组件。 安装React-glide非常简单，可以通过npm（Node Package Manager）这个包管理器进行安装。npm是目前流行的JavaScript包管理工具，它能够帮助开发者快速找到合适的库，管理项目依赖，并且能够方便地进行版本控制。安装命令为： ```bash $ npm install react-glide ``` 安装完成后，开发者可以将react-glide作为一个依赖模块引入到React项目中。在组件页面顶部导入模块时，需要同时导入Glide组件以及对应的样式文件。在React中，导入组件和样式的语句如下： ```javascript import { Glide } from 'react-glide'; import 'react-glide/lib/reactGlide.css'; ``` 在使用时，Glide组件充当一个包装器，它能够包裹任何类型的元素。通常，轮播组件中会包含多个图片元素，Glide会自动管理这些图片的显示逻辑。例如，以下是一个基本的使用示例： ```jsx < Glide > < img src = 'http://path/to/image/url' /> < img src = 'http://path/to/image/url2' /> < img src = 'http://path/to/image/url3' /> < /Glide > ``` 在上述代码中，`<Glide>`标签内可以添加任意数量的子元素（通常是`<img>`标签），而Glide组件则会负责这些子元素的轮播展示。 react-glide的轮播组件也支持自定义配置选项，以满足不同场景下的需求。这些选项可以作为属性传递给Glide组件。例如，你可以设置轮播的自动播放间隔、切换动画效果、轮播方向等。具体的属性配置方法需要查阅react-glide的文档。 另外，提到的“种类”和“interface GlideProps”表明react-glide支持TypeScript。TypeScript是JavaScript的一个超集，它在JavaScript的基础上添加了类型系统和对ES6+的新特性的支持。TypeScript最终会被编译成JavaScript代码。使用TypeScript的优势在于能够在开发阶段就发现类型相关的错误，提升代码的健壮性和可维护性。而`GlideProps`可能是指Glide组件可以接收的属性接口，具体如何使用，需要结合TypeScript的类型定义和接口特性来确定。 最后，提到的“压缩包子文件的文件名称列表”中，`react-glide-master`很可能是压缩包的名称。在项目构建和发布过程中，通常会将项目文件打包成一个或多个压缩文件，以便于传输和部署。这种命名方式表明这个压缩文件中包含的是react-glide项目的主分支（master branch）的源代码。 综上所述，react-glide作为一个专为React框架设计的轮播组件，极大地简化了轮播功能的实现过程。开发者只需要通过简单的配置和少量的代码，就可以在React应用中实现美观且功能丰富的轮播效果。无论是用于图片展示、内容更新、引导教程还是广告轮播，react-glide都是一个不错的选择。