Python os.setpgid 函数

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

本综合指南探讨了 Python 的 os.setpgid 函数，该函数管理类 Unix 系统中的进程组。我们将介绍进程组、会话领导者以及进程控制的实际示例。

基本定义

os.setpgid 函数为指定的进程设置进程组 ID。它用于将进程组织成组以便进行信号处理。

主要参数：pid（要修改的进程 ID），pgid（新的进程组 ID）。成功时返回 None，失败时引发 OSError。仅限 Unix 系统。

创建新的进程组

此示例演示如何创建子进程并将其放置在一个新的进程组中。父进程保留在其原始组中。

new_process_group.py

import os
import time

pid = os.fork()

if pid == 0:  # Child process
    print(f"Child PID: {os.getpid()}, Original PGID: {os.getpgid(0)}")
    os.setpgid(0, 0)  # Create new process group with child as leader
    print(f"New PGID: {os.getpgid(0)}")
    time.sleep(5)
else:  # Parent process
    print(f"Parent PID: {os.getpid()}, PGID: {os.getpgid(0)}")
    time.sleep(1)
    print(f"Child's PGID after change: {os.getpgid(pid)}")

子进程创建一个新的进程组，并将自身作为领导者。父进程可以在更改后验证子进程的新组 ID。

进程组领导者通常是管理相关进程组的 shell 或会话领导者。

加入现有进程组

此示例演示如何使进程加入现有进程组，而不是创建新的进程组。目标组必须在同一会话中。

join_process_group.py

import os
import time

# Create two child processes
pid1 = os.fork()

if pid1 == 0:  # First child
    print(f"Child1 PID: {os.getpid()}, PGID: {os.getpgid(0)}")
    time.sleep(10)
else:
    pid2 = os.fork()
    
    if pid2 == 0:  # Second child
        print(f"Child2 PID: {os.getpid()}, Original PGID: {os.getpgid(0)}")
        # Join first child's process group
        os.setpgid(0, pid1)
        print(f"New PGID: {os.getpgid(0)}")
        time.sleep(5)
    else:  # Parent
        print(f"Parent PID: {os.getpid()}")
        time.sleep(1)
        print(f"Child1 PGID: {os.getpgid(pid1)}")
        print(f"Child2 PGID: {os.getpgid(pid2)}")

第二个子进程加入第一个子进程的进程组。现在，两个子进程都将在同一组中，共享信号处理特征。

此技术对于创建进程层次结构很有用，在进程层次结构中，相关进程应一起接收信号。

使用 setpgid 进行错误处理

此示例演示了使用 os.setpgid 时的正确错误处理，包括权限检查和无效进程场景。

error_handling.py

import os
import sys
import errno

try:
    # Attempt to set process group for non-existent process
    os.setpgid(99999, 0)
except OSError as e:
    if e.errno == errno.ESRCH:
        print("Error: No such process (ESRCH)")
    elif e.errno == errno.EPERM:
        print("Error: Permission denied (EPERM)")
    elif e.errno == errno.EACCES:
        print("Error: Access denied (EACCES)")
    else:
        print(f"Unexpected error: {e}")

# Valid but restricted operation
pid = os.fork()
if pid == 0:  # Child
    try:
        # Try to change parent's process group
        os.setpgid(os.getppid(), 0)
    except OSError as e:
        print(f"Failed to change parent's PGID: {e}")
    sys.exit()
else:  # Parent
    os.waitpid(pid, 0)

第一次尝试失败，出现 ESRCH（没有这样的进程）。由于 EPERM（权限被拒绝），子进程无法修改父进程的组。

在管理进程组时，正确的错误处理至关重要，因为操作通常具有严格的权限要求。

会话领导者和 setpgid

此示例探讨了会话和进程组之间的关系，演示了 os.setpgid 如何与会话领导者交互。

session_leader.py

import os
import time

# Create a new session
if os.fork() == 0:  # Child becomes session leader
    print(f"New session leader PID: {os.getpid()}")
    print(f"SID: {os.getsid(0)}")
    os.setsid()  # Create new session
    
    # Now try to set process group
    try:
        os.setpgid(0, 0)
        print("Successfully created new process group")
    except OSError as e:
        print(f"Failed to setpgid: {e}")
    
    time.sleep(5)
    sys.exit()
else:  # Parent
    time.sleep(1)
    print(f"Parent SID: {os.getsid(0)}")

子进程使用 os.setsid() 创建一个新的会话，然后尝试创建一个新的进程组。会话领导者对于进程组管理有特殊的规则。

会话领导者的进程组 ID 始终等于其 PID，并且尝试更改它将失败并显示 EPERM。

进程组继承

此示例演示了在 fork() 期间如何继承进程组以及 os.setpgid 如何修改此行为。

inheritance.py

import os
import time

print(f"Original process PGID: {os.getpgid(0)}")

pid = os.fork()

if pid == 0:  # Child
    print(f"Child inherited PGID: {os.getpgid(0)}")
    
    # Change process group immediately after fork
    os.setpgid(0, 0)
    print(f"Child new PGID: {os.getpgid(0)}")
    
    time.sleep(5)
else:  # Parent
    # Parent can also change child's group (must happen before exec)
    time.sleep(0.1)  # Ensure child runs first
    try:
        os.setpgid(pid, pid)
        print(f"Parent set child's PGID to: {os.getpgid(pid)}")
    except OSError as e:
        print(f"Failed to change child's PGID: {e}")

子进程最初继承父进程的进程组。父进程或子进程都可以更改组，但必须在调用 exec() 之前完成。

父进程和子进程之间的这种协调对于 shell 实现中的正确进程组管理至关重要。

使用进程组进行信号处理

此示例演示了进程组如何影响信号传递，演示了如何将信号发送到整个进程组。

signal_handling.py

import os
import signal
import time

def handler(signum, frame):
    print(f"Received signal {signum} in PID {os.getpid()}")

# Set up signal handler
signal.signal(signal.SIGUSR1, handler)

pid = os.fork()

if pid == 0:  # Child
    # Create new process group
    os.setpgid(0, 0)
    print(f"Child in new PGID: {os.getpgid(0)}")
    while True:
        time.sleep(1)
else:  # Parent
    time.sleep(1)  # Let child set up
    
    # Send signal to entire process group
    print(f"Sending SIGUSR1 to group {os.getpgid(pid)}")
    os.kill(-os.getpgid(pid), signal.SIGUSR1)
    
    time.sleep(1)
    os.kill(pid, signal.SIGTERM)

父进程使用负 PID 向整个进程组发送信号。如果两个进程在同一组中，则都会收到信号。

进程组允许对相关进程进行协调的信号处理，这对于 shell 中的作业控制特别有用。

安全注意事项

权限要求：需要适当的权限才能修改进程组
竞争条件：进程状态可能在检查和 setpgid 之间发生变化
会话限制：只能在同一会话中移动进程
时序约束：必须在子进程中的 exec 之前设置 setpgid
平台限制：特定于 Unix 的功能

最佳实践

错误处理：始终检查 OSError 异常
原子操作：仔细协调父/子组更改
会话意识：了解会话领导者的限制
信号规划：在设计进程组时要考虑到信号处理
文档：清楚地记录进程组关系

资料来源

作者

我的名字是 Jan Bodnar，我是一位热情的程序员，拥有丰富的编程经验。我从 2007 年开始撰写编程文章。到目前为止，我已经撰写了超过 1,400 篇文章和 8 本电子书。我拥有超过十年的编程教学经验。

列出所有 Python 教程。