Python os.plock 函数

上次修改时间：2025 年 4 月 11 日

本综合指南探讨了 Python 的 os.plock 函数，该函数控制 Unix 系统上的进程锁定。我们将介绍锁定类型、权限要求和实际的内存管理示例。

基本定义

os.plock 函数将进程段锁定到内存中，防止交换。 它在 Unix 系统上可用，并且大多数操作都需要 root 权限。

关键参数：op（锁定操作类型）。常见值：PLOCK_PGIN、PLOCK_PGOUT、PLOCK_SHLOCK、PLOCK_EXLOCK。成功时返回 None。

锁定进程文本段

此示例演示如何将进程的文本段（可执行代码）锁定到内存中。 这可以防止交换程序指令。

import os

try:
    # Lock text segment (executable code) into memory
    os.plock(os.PLOCK_TEXT)
    print("Process text segment locked in memory")
except AttributeError:
    print("os.plock not available on this platform")
except PermissionError:
    print("Requires root privileges to lock memory")

此代码尝试将进程的可执行部分锁定在 RAM 中。 此操作需要 root 权限，并且特定于 Unix。

请注意，PLOCK_TEXT 并非在所有 Unix 变体上都可用。 即使在某些系统上具有适当的权限，操作也可能会失败。

锁定进程数据段

此示例演示如何将数据段（变量、堆）锁定到内存中。 这可以防止交换进程数据结构。

import os

try:
    # Lock data segment (variables, heap) into memory
    os.plock(os.PLOCK_DATA)
    print("Process data segment locked in memory")
    
    # Allocate some memory that will stay in RAM
    large_list = [0] * 1000000
except AttributeError:
    print("os.plock not available on this platform")
except PermissionError:
    print("Requires root privileges to lock memory")

锁定数据段后，任何已分配的内存（如 large_list）将保留在物理 RAM 中。 这对于实时应用程序非常有用。

请谨慎使用内存锁定，因为它会减少其他进程和系统缓存可用的 RAM。

解锁进程段

此示例演示如何解锁先前锁定的进程段。 内存锁定应该是临界区的临时锁定。

import os

try:
    # Lock both text and data segments
    os.plock(os.PLOCK_TEXT | os.PLOCK_DATA)
    print("Process segments locked in memory")
    
    # Critical section code here
    
    # Unlock segments when done
    os.plock(os.PLOCK_UNLOCK)
    print("Process segments unlocked")
except AttributeError:
    print("os.plock not available on this platform")
except PermissionError:
    print("Requires root privileges to lock memory")

该示例为临界区锁定内存，然后将其解锁。 PLOCK_UNLOCK 释放进程拥有的所有锁。

始终在完成后解锁内存，以避免系统资源耗尽。 不必要的锁定会降低整体系统性能。

共享内存锁定

此示例演示进程之间的共享内存锁定。 PLOCK_SHLOCK 提供对内存段的共享建议锁定。

import os
import sys

try:
    if len(sys.argv) > 1 and sys.argv[1] == "exclusive":
        # Request exclusive lock
        os.plock(os.PLOCK_EXLOCK)
        print("Exclusive memory lock acquired")
    else:
        # Request shared lock
        os.plock(os.PLOCK_SHLOCK)
        print("Shared memory lock acquired")
        
    # Process shared memory operations here
    input("Press Enter to release lock...")
    
    os.plock(os.PLOCK_UNLOCK)
except AttributeError:
    print("os.plock not available on this platform")
except PermissionError:
    print("Requires root privileges to lock memory")

该脚本接受命令行参数，以选择共享和独占锁定模式。 共享锁允许并发读取器。

独占锁 (PLOCK_EXLOCK) 阻止其他进程获取对同一内存段的任何锁。

防止页面调出

PLOCK_PGIN 阻止页面被调出，而 PLOCK_PGOUT 强制页面调出。 此示例演示了这两种操作。

import os
import time

try:
    # Prevent pages from being paged out
    os.plock(os.PLOCK_PGIN)
    print("Pages will remain in memory")
    
    # Allocate some memory
    data = bytearray(1024 * 1024)  # 1MB
    
    # Keep pages in memory for 10 seconds
    time.sleep(10)
    
    # Allow pages to be paged out
    os.plock(os.PLOCK_PGOUT)
    print("Pages may now be paged out")
except AttributeError:
    print("os.plock not available on this platform")
except PermissionError:
    print("Requires root privileges to control paging")

该示例首先阻止内存页被交换出去，分配内存，然后在延迟后允许交换。

PLOCK_PGIN/PGOUT 提供比段锁定更精细的控制，但在大多数系统上仍然需要 root 权限。

检查锁定支持

此示例演示如何在尝试使用 plock 之前检查其支持和可用操作。

import os
import sys

def check_plock_support():
    if not hasattr(os, 'plock'):
        print("os.plock not supported on this platform")
        return False
    
    required_constants = ['PLOCK_TEXT', 'PLOCK_DATA', 'PLOCK_SHLOCK']
    missing = [c for c in required_constants if not hasattr(os, c)]
    
    if missing:
        print(f"Missing required constants: {', '.join(missing)}")
        return False
    
    return True

if check_plock_support():
    print("os.plock is supported with basic operations")
    try:
        os.plock(os.PLOCK_TEXT)
        print("Successfully locked text segment")
        os.plock(os.PLOCK_UNLOCK)
    except PermissionError:
        print("Insufficient privileges to lock memory")
else:
    print("os.plock functionality not fully available")

该函数在尝试使用 plock 之前检查 plock 的可用性和所需的常量。 这可以防止 AttributeError 异常。

即使有支持检查，也始终处理 PermissionError，因为内存锁定通常需要提升的权限。

实时应用程序示例

此示例演示了一个简化的实时应用程序，该应用程序使用内存锁定来通过防止交换来确保一致的性能。

import os
import time

def realtime_task():
    # Critical real-time processing
    for i in range(10):
        print(f"Processing cycle {i}")
        time.sleep(0.1)

try:
    # Lock text and data segments
    os.plock(os.PLOCK_TEXT | os.PLOCK_DATA)
    print("Memory locked for real-time operation")
    
    # Allocate working memory
    buffer = bytearray(1024 * 512)  # 512KB working buffer
    
    # Execute real-time task
    realtime_task()
    
    # Release locks
    os.plock(os.PLOCK_UNLOCK)
    print("Memory locks released")
except (AttributeError, PermissionError) as e:
    print(f"Warning: Could not lock memory ({e})")
    print("Running without memory locks (may experience swapping delays)")
    realtime_task()

该示例为时间敏感的任务锁定内存，如果权限不足，则回退到未锁定的操作。

这种模式在实时系统中很常见，在这些系统中，一致的计时比绝对性能更重要。

安全注意事项

• 权限要求：大多数操作都需要 root 权限
• 资源限制：锁定的内存计入进程限制
• 系统影响：过度锁定会降低系统性能
• 平台支持：并非在所有 Unix 变体上都可用
• 替代机制：考虑使用 mlock() 以获得更多控制

最佳实践

• 最小化持续时间：仅在绝对必要时锁定内存
• 错误处理：始终处理 AttributeError 和 PermissionError
• 权限分离：锁定后删除权限
• 内存限制：在锁定之前检查 RLIMIT_MEMLOCK
• 记录假设：明确记录锁定要求

资料来源

• Python os.plock 文档
• Linux plock(2) 手册页

作者

我的名字是 Jan Bodnar，我是一位充满激情的程序员，拥有丰富的编程经验。 自 2007 年以来，我一直在撰写编程文章。 迄今为止，我已撰写了 1,400 多篇文章和 8 本电子书。 我在编程教学方面拥有超过十年的经验。

列出所有 Python 教程。