【Bug已解决】PyTorch on M1 Mac: RuntimeError: Placeholder storage has not been allocated on MPS devi 解决方案

【Bug已解决】PyTorch on M1 Mac: RuntimeError: Placeholder storage has not been allocated on MPS device 解决方案

问题描述

在 Apple Silicon（M1/M2/M3）Mac 上使用 PyTorch 的 MPS（Metal Performance Shaders）后端进行 GPU 加速时，开发者可能会遇到以下错误：

RuntimeError: Placeholder storage has not been allocated on MPS device!

这个错误通常发生在以下场景中：

• 在 MPS 设备上创建或操作 tensor
• 将模型或数据从 CPU 转移到 MPS 设备
• 在 MPS 设备上进行前向传播或反向传播
• 使用 DataLoader 加载数据并传输到 MPS 设备
• 在 MPS 设备上执行某些不支持的算子时自动回退到 CPU

MPS 是 Apple 为其自研芯片提供的 GPU 计算后端，PyTorch 从 1.12 版本开始支持 MPS 后端。但由于 MPS 仍在不断发展和完善中，某些操作可能会触发此错误。

错误复现

场景一：基本 tensor 操作

import torch # 检查MPS是否可用 print(f"MPS可用: {torch.backends.mps.is_available()}") # 在MPS设备上创建tensor device = torch.device('mps') x = torch.randn(100, 100, device=device) y = torch.randn(100, 100, device=device) # 尝试矩阵乘法 z = torch.mm(x, y) print(f"结果形状: {z.shape}") print(f"结果设备: {z.device}")

在某些情况下，上述代码可能正常运行，但在更复杂的操作中会触发错误：

# 更复杂的操作可能触发错误 model = torch.nn.Sequential( torch.nn.Linear(100, 256), torch.nn.ReLU(), torch.nn.Linear(256, 10) ).to(device) # 前向传播 x = torch.randn(32, 100, device=device) output = model(x) # 可能在此处报错 loss = torch.nn.functional.cross_entropy(output, torch.randint(0, 10, (32,), device=device)) loss.backward() # 或者在此处报错 print(f"Loss: {loss.item()}")

场景二：完整训练循环中的错误

import torch import torch.nn as nn device = torch.device('mps') # 创建模型 model = nn.Sequential( nn.Conv2d(3, 64, kernel_size=3, padding=1), nn.ReLU(), nn.MaxPool2d(2), nn.Conv2d(64, 128, kernel_size=3, padding=1), nn.ReLU(), nn.MaxPool2d(2), nn.Flatten(), nn.Linear(128 * 8 * 8, 10) ).to(device) optimizer = torch.optim.Adam(model.parameters(), lr=0.001) criterion = nn.CrossEntropyLoss() # 训练循环 for epoch in range(10): x = torch.randn(64, 3, 32, 32, device=device) target = torch.randint(0, 10, (64,), device=device) optimizer.zero_grad() output = model(x) loss = criterion(output, target) # 可能在此处触发错误 loss.backward() optimizer.step() print(f"Epoch {epoch}, Loss: {loss.item():.4f}")

错误信息：

RuntimeError: Placeholder storage has not been allocated on MPS device!

场景三：混合设备操作

import torch device = torch.device('mps') # CPU上创建tensor x_cpu = torch.randn(100, 100) # 部分转移到MPS x_mps = x_cpu.to(device) # 尝试与CPU tensor进行操作 y_cpu = torch.randn(100, 100) result = x_mps + y_cpu # 设备不匹配，可能触发错误

根因分析

1. MPS 内存管理机制

MPS 后端使用 Metal 框架管理 GPU 内存，其内存分配策略与 CUDA 有所不同。MPS 使用"占位符存储"（Placeholder Storage）机制来延迟分配 GPU 内存。当 PyTorch 需要在 MPS 设备上分配内存时，它首先创建一个占位符，然后在实际需要时才分配真实的 GPU 内存。

如果占位符存储没有被正确转换为实际分配的存储（例如，由于某些操作路径没有被正确处理），就会触发 "Placeholder storage has not been allocated on MPS device" 错误。

2. 算子不支持导致的回退问题

并非所有 PyTorch 算子都在 MPS 后端上实现了 GPU 版本。当遇到不支持的算子时，PyTorch 会尝试将 tensor 从 MPS 设备转移到 CPU 上执行，然后再转移回来。这个转移过程中可能出现占位符存储未正确分配的问题。

# 某些操作在MPS上不支持，会触发CPU回退 x = torch.randn(10, 10, device='mps') # 以下操作可能在早期PyTorch版本中不支持MPS result = torch.linalg.svd(x) # 可能触发CPU回退和错误

3. PyTorch 版本问题

MPS 后端在 PyTorch 的不同版本中成熟度不同：

• PyTorch 1.12：初步支持 MPS，很多操作有 bug
• PyTorch 1.13：改进了 MPS 支持，但仍有不少问题
• PyTorch 2.0+：大幅改进了 MPS 支持，大部分操作正常工作
• PyTorch 2.1+：进一步优化，支持更多算子

使用较旧版本的 PyTorch 更容易遇到此错误。

4. 内存压力

当 MPS 设备的统一内存（Unified Memory）压力较大时，内存分配可能失败，导致占位符存储无法被正确分配。Apple Silicon 使用统一内存架构，CPU 和 GPU 共享同一块物理内存，因此内存管理更加复杂。

5. 异步操作与时序问题

MPS 后端的某些操作是异步执行的，如果在操作完成之前就尝试访问结果，可能导致占位符存储尚未被分配。

6. DataLoader 与多进程

使用num_workers > 0的 DataLoader 时，工作进程在 CPU 上运行，将数据从工作进程传输到 MPS 设备可能触发此错误。

解决方案

方案一：升级 PyTorch 到最新版本

这是最优先的解决方案，因为新版本修复了大量 MPS 相关的 bug：

# 升级PyTorch到最新版本 pip install --upgrade torch torchvision torchaudio # 或者指定版本 pip install torch==2.1.0 torchvision==0.16.0 # 验证版本 python -c "import torch; print(torch.__version__)"

升级后验证 MPS 是否正常工作：

import torch print(f"PyTorch版本: {torch.__version__}") print(f"MPS可用: {torch.backends.mps.is_available()}") print(f"MPS已构建: {torch.backends.mps.is_built()}") # 基本功能测试 device = torch.device('mps') x = torch.randn(100, 100, device=device) y = torch.randn(100, 100, device=device) z = torch.mm(x, y) print(f"矩阵乘法测试: {z.shape}, {z.device}") # 反向传播测试 x = torch.randn(50, 50, device=device, requires_grad=True) y = x.sum() y.backward() print(f"反向传播测试: {x.grad is not None}")

方案二：使用torch.mps.synchronize()同步操作

在关键操作后添加同步调用，确保 MPS 操作完成：

import torch device = torch.device('mps') # 在关键操作后同步 x = torch.randn(100, 100, device=device) y = torch.randn(100, 100, device=device) z = torch.mm(x, y) # 同步MPS操作，确保完成 torch.mps.synchronize() print(f"结果: {z.shape}, {z.device}")

在训练循环中使用同步：

import torch import torch.nn as nn device = torch.device('mps') model = nn.Linear(100, 10).to(device) optimizer = torch.optim.Adam(model.parameters()) criterion = nn.CrossEntropyLoss() for epoch in range(10): x = torch.randn(32, 100, device=device) target = torch.randint(0, 10, (32,), device=device) optimizer.zero_grad() output = model(x) loss = criterion(output, target) loss.backward() optimizer.step() # 每个epoch后同步 torch.mps.synchronize() print(f"Epoch {epoch}, Loss: {loss.item():.4f}")

方案三：使用 CPU 回退机制

对于不支持 MPS 的操作，手动将 tensor 转移到 CPU 执行后再转回：

import torch device = torch.device('mps') def safe_mps_operation(tensor, op, *args, **kwargs): """安全执行MPS操作，不支持时回退到CPU""" original_device = tensor.device try: result = op(tensor, *args, **kwargs) torch.mps.synchronize() return result except RuntimeError as e: if "Placeholder storage" in str(e) or "MPS" in str(e): print(f"MPS操作失败，回退到CPU: {op.__name__}") tensor_cpu = tensor.cpu() result = op(tensor_cpu, *args, **kwargs) return result.to(original_device) raise # 使用示例 x = torch.randn(100, 100, device=device) # 安全执行操作 result = safe_mps_operation(x, torch.linalg.svd) print(f"SVD结果: U={result[0].shape}, S={result[1].shape}, V={result[2].shape}")

方案四：管理 MPS 内存

使用 MPS 内存管理功能来减少内存压力：

import torch device = torch.device('mps') # 获取当前MPS内存使用情况 def print_mps_memory(): if hasattr(torch.mps, 'current_allocated_memory'): allocated = torch.mps.current_allocated_memory() print(f"MPS已分配内存: {allocated / 1024**3:.2f} GB") if hasattr(torch.mps, 'driver_allocated_memory'): driver_allocated = torch.mps.driver_allocated_memory() print(f"MPS驱动分配内存: {driver_allocated / 1024**3:.2f} GB") # 训练前 print("训练前:") print_mps_memory() # 训练代码 model = torch.nn.Sequential( torch.nn.Linear(1000, 2000), torch.nn.ReLU(), torch.nn.Linear(2000, 1000) ).to(device) x = torch.randn(128, 1000, device=device) output = model(x) loss = output.sum() loss.backward() # 训练后 print("\n训练后:") print_mps_memory() # 清理MPS缓存 if hasattr(torch.mps, 'empty_cache'): torch.mps.empty_cache() print("\n清理缓存后:") print_mps_memory()

方案五：使用MPS_FALLBACK环境变量

设置环境变量让 PyTorch 自动将不支持的 MPS 操作回退到 CPU：

import os # 方法1: 在Python中设置（必须在import torch之前） os.environ['PYTORCH_ENABLE_MPS_FALLBACK'] = '1' import torch device = torch.device('mps') # 现在不支持的MPS操作会自动回退到CPU x = torch.randn(100, 100, device=device) try: result = torch.linalg.svd(x) print(f"SVD结果设备: {result[0].device}") # 可能是cpu except Exception as e: print(f"错误: {e}")

或者在命令行中设置：

PYTORCH_ENABLE_MPS_FALLBACK=1 python train.py

完整修复代码

以下是一个完整的、健壮的 MPS 训练框架，包含了所有必要的错误处理和内存管理：

""" Apple Silicon MPS 设备安全训练框架 解决 RuntimeError: Placeholder storage has not been allocated on MPS device """ import torch import torch.nn as nn import os import gc import warnings from typing import Optional, Tuple class MPSTrainingManager: """MPS设备训练管理器""" def __init__(self, enable_fallback: bool = True, verbose: bool = True): """ Args: enable_fallback: 是否启用CPU回退机制 verbose: 是否打印详细信息 """ self.verbose = verbose self.device = None self.fallback_device = torch.device('cpu') # 设置环境变量 if enable_fallback: os.environ['PYTORCH_ENABLE_MPS_FALLBACK'] = '1' # 检测可用设备 self._detect_device() # 训练状态 self.peak_memory = 0 def _detect_device(self): """检测并设置最佳计算设备""" if torch.backends.mps.is_available(): self.device = torch.device('mps') if self.verbose: print(f"[INFO] 使用MPS设备 (Apple Silicon GPU)") elif torch.cuda.is_available(): self.device = torch.device('cuda') if self.verbose: print(f"[INFO] 使用CUDA设备: {torch.cuda.get_device_name(0)}") else: self.device = torch.device('cpu') if self.verbose: print(f"[INFO] 使用CPU设备") def safe_forward(self, model: nn.Module, x: torch.Tensor) -> torch.Tensor: """安全的前向传播，支持MPS错误回退""" x = x.to(self.device) try: output = model(x) if self.device.type == 'mps': torch.mps.synchronize() return output except RuntimeError as e: if "Placeholder storage" in str(e) or "MPS" in str(e): warnings.warn(f"MPS前向传播失败，回退到CPU: {e}") model_cpu = model.cpu() x_cpu = x.cpu() output = model_cpu(x_cpu) model.to(self.device) return output.to(self.device) raise def safe_backward(self, loss: torch.Tensor) -> bool: """安全的反向传播""" try: loss.backward() if self.device.type == 'mps': torch.mps.synchronize() return True except RuntimeError as e: if "Placeholder storage" in str(e) or "MPS" in str(e): warnings.warn(f"MPS反向传播失败，回退到CPU: {e}") return False raise def move_to_device(self, obj): """将模型或tensor安全转移到设备""" if isinstance(obj, torch.Tensor): return obj.to(self.device) elif isinstance(obj, torch.nn.Module): return obj.to(self.device) return obj def get_memory_info(self) -> dict: """获取内存使用信息""" info = {} if self.device.type == 'mps': if hasattr(torch.mps, 'current_allocated_memory'): info['mps_allocated'] = torch.mps.current_allocated_memory() if hasattr(torch.mps, 'driver_allocated_memory'): info['mps_driver_allocated'] = torch.mps.driver_allocated_memory() elif self.device.type == 'cuda': info['cuda_allocated'] = torch.cuda.memory_allocated() info['cuda_reserved'] = torch.cuda.memory_reserved() return info def print_memory_info(self, tag: str = ""): """打印内存使用信息""" if not self.verbose: return info = self.get_memory_info() if not info: return parts = [f"[{tag}]"] if tag else [] if 'mps_allocated' in info: parts.append(f"MPS已分配: {info['mps_allocated'] / 1024**2:.1f} MB") if 'mps_driver_allocated' in info: parts.append(f"MPS驱动分配: {info['mps_driver_allocated'] / 1024**2:.1f} MB") if 'cuda_allocated' in info: parts.append(f"CUDA已分配: {info['cuda_allocated'] / 1024**3:.2f} GB") print(" | ".join(parts)) def cleanup(self): """清理内存""" gc.collect() if self.device.type == 'mps': torch.mps.synchronize() if hasattr(torch.mps, 'empty_cache'): torch.mps.empty_cache() elif self.device.type == 'cuda': torch.cuda.synchronize() torch.cuda.empty_cache() if self.verbose: self.print_memory_info("清理后") def train_step(self, model: nn.Module, optimizer: torch.optim.Optimizer, criterion, x: torch.Tensor, y: torch.Tensor) -> Tuple[torch.Tensor, bool]: """执行一个训练步骤""" x = x.to(self.device) y = y.to(self.device) optimizer.zero_grad() # 前向传播 output = self.safe_forward(model, x) loss = criterion(output, y) # 反向传播 success = self.safe_backward(loss) if success: optimizer.step() return loss, success def test_mps_basic(): """测试MPS基本功能""" print("=" * 60) print("MPS基本功能测试") print("=" * 60) if not torch.backends.mps.is_available(): print("[SKIP] MPS不可用") return device = torch.device('mps') tests = [ ("Tensor创建", lambda: torch.randn(100, 100, device=device)), ("矩阵乘法", lambda: torch.mm( torch.randn(100, 100, device=device), torch.randn(100, 100, device=device) )), ("卷积操作", lambda: nn.functional.conv2d( torch.randn(1, 3, 32, 32, device=device), torch.randn(16, 3, 3, 3, device=device), padding=1 )), ("反向传播", lambda: _test_backward(device)), ] for name, test_func in tests: try: result = test_func() torch.mps.synchronize() print(f" [PASS] {name}") except RuntimeError as e: print(f" [FAIL] {name}: {e}") def _test_backward(device): x = torch.randn(50, 50, device=device, requires_grad=True) y = (x ** 2).sum() y.backward() assert x.grad is not None def full_training_example(): """完整的训练示例""" print("\n" + "=" * 60) print("完整训练示例") print("=" * 60) # 初始化训练管理器 manager = MPSTrainingManager(enable_fallback=True, verbose=True) # 创建模型 model = nn.Sequential( nn.Flatten(), nn.Linear(28 * 28, 256), nn.ReLU(), nn.Dropout(0.2), nn.Linear(256, 128), nn.ReLU(), nn.Dropout(0.2), nn.Linear(128, 10) ).to(manager.device) optimizer = torch.optim.Adam(model.parameters(), lr=0.001) criterion = nn.CrossEntropyLoss() manager.print_memory_info("模型创建后") # 模拟训练 num_epochs = 5 batch_size = 64 for epoch in range(num_epochs): epoch_loss = 0.0 num_batches = 10 for i in range(num_batches): # 生成模拟数据 x = torch.randn(batch_size, 1, 28, 28) y = torch.randint(0, 10, (batch_size,)) # 训练步骤 loss, success = manager.train_step( model, optimizer, criterion, x, y ) epoch_loss += loss.item() avg_loss = epoch_loss / num_batches print(f"Epoch {epoch + 1}/{num_epochs}, Loss: {avg_loss:.4f}") # 定期清理 if (epoch + 1) % 2 == 0: manager.cleanup() manager.print_memory_info("训练完成后") manager.cleanup() print("\n训练完成!") if __name__ == "__main__": # 运行基本测试 test_mps_basic() # 运行完整训练示例 full_training_example()

常见陷阱与注意事项

1. 不要假设所有操作都支持 MPS

# 错误做法：假设所有操作都支持MPS x = torch.randn(100, 100, device='mps') result = some_complex_operation(x) # 可能不支持MPS # 正确做法：使用回退机制 os.environ['PYTORCH_ENABLE_MPS_FALLBACK'] = '1' import torch x = torch.randn(100, 100, device='mps') result = some_complex_operation(x) # 不支持时自动回退到CPU

2. 注意 DataLoader 的num_workers设置

# 在MPS设备上使用DataLoader时，注意worker配置 from torch.utils.data import DataLoader # 错误做法：使用过多worker可能导致MPS内存问题 dataloader = DataLoader(dataset, batch_size=64, num_workers=8) # 正确做法：减少worker数量或使用persistent_workers dataloader = DataLoader( dataset, batch_size=64, num_workers=2, # 减少worker数量 persistent_workers=True # 避免频繁创建/销毁worker )

3. 注意 tensor 的requires_grad设置

# 在MPS设备上，确保requires_grad的tensor被正确处理 x = torch.randn(100, 100, device='mps', requires_grad=True) # 某些操作可能需要先detach x_detached = x.detach() result = some_operation(x_detached) # 需要梯度时再恢复 x.requires_grad_(True)

4. 使用torch.mps.synchronize()而非torch.cuda.synchronize()

# 错误做法：在MPS设备上使用CUDA同步 torch.cuda.synchronize() # 在没有CUDA的Mac上会报错 # 正确做法：使用MPS同步 torch.mps.synchronize()

5. 注意 MPS 内存限制

Apple Silicon 使用统一内存，CPU 和 GPU 共享同一块物理内存。大型模型可能超出可用内存：

# 检查可用内存 import psutil available_memory = psutil.virtual_memory().available print(f"可用内存: {available_memory / 1024**3:.1f} GB") # 根据可用内存调整batch_size # M1 (8GB): batch_size = 32 # M1 Pro (16GB): batch_size = 64 # M1 Max (32GB): batch_size = 128

6. 混合精度训练在 MPS 上的限制

# MPS对混合精度训练的支持有限 # PyTorch 2.0+开始支持MPS上的autocast，但功能可能不完整 # 谨慎使用autocast with torch.autocast('mps', enabled=True): output = model(x) # 某些操作可能不支持autocast，导致错误

7. 保存和加载 MPS 模型

# 保存MPS模型时，先转移到CPU model = model.to('mps') # ... 训练 ... # 保存前转移到CPU model_cpu = model.cpu() torch.save(model_cpu.state_dict(), 'model.pth') # 加载时先在CPU上加载，再转移到MPS model.load_state_dict(torch.load('model.pth', map_location='cpu')) model = model.to('mps')

8. 使用torch.no_grad()减少内存使用

# 在推理时使用no_grad减少内存 model.eval() with torch.no_grad(): x = torch.randn(100, 100, device='mps') output = model(x) torch.mps.synchronize()

总结

RuntimeError: Placeholder storage has not been allocated on MPS device错误是 Apple Silicon MPS 后端的一个已知问题，主要与 MPS 的内存管理机制和算子支持不完整有关。解决此问题的核心策略包括：

1. 升级 PyTorch：使用最新版本的 PyTorch，新版本修复了大量 MPS 相关的 bug
2. 启用 CPU 回退：设置PYTORCH_ENABLE_MPS_FALLBACK=1让不支持的算子自动回退到 CPU
3. 同步操作：在关键操作后调用torch.mps.synchronize()确保操作完成
4. 管理内存：使用torch.mps.empty_cache()清理 MPS 缓存，监控内存使用
5. 正确保存/加载：保存模型前先转移到 CPU，加载时使用map_location='cpu'

关键要点：

• MPS 后端仍在持续完善中，使用最新版 PyTorch 可以避免大部分问题
• PYTORCH_ENABLE_MPS_FALLBACK=1是最简单的解决方案，让不支持的算子自动回退到 CPU
• Apple Silicon 使用统一内存架构，需要注意 CPU 和 GPU 之间的内存分配平衡
• 使用torch.mps.synchronize()而非 CUDA 版本来同步 MPS 操作
• 保存模型时先转移到 CPU，避免 MPS 相关的序列化问题

通过合理使用 MPS 后端和适当的错误处理策略，可以在 Apple Silicon Mac 上获得显著的 GPU 加速效果。