Reactos 第6章 进程间通信（续）

第6章 进程间通信（续）

本章深入剖析Windows/ReactOS中进程间通信的同步对象和通信机制。

6.4 信号量（Semaphore）

信号量是一种用于控制并发访问的同步原语，通过计数器机制限制同时访问某资源的线程数。在进程间通信中，信号量常用于资源池管理和生产者-消费者模式。

6.4.1 KSEMAPHORE结构定义

typedef struct _KSEMAPHORE {
DISPATCHER_HEADER Header; // 调度器头部
LONG Limit; // 最大计数值
} KSEMAPHORE, *PKSEMAPHORE;

结构关系：

KSEMAPHORE结构
┌─────────────────────────────────────────────────────────────┐
│ KSEMAPHORE │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ DISPATCHER_HEADER │ │
│ │ Type : SemaphoreObject (类型标识) │ │
│ │ SignalState : 当前计数 (可正可负) │ │
│ │ WaitListHead : 等待线程列表 │ │
│ └─────────────────────────────────────────────────────┘ │
│ Limit : 最大允许的计数值 │
└─────────────────────────────────────────────────────────────┘

SignalState含义：

6.4.2 NtCreateSemaphore系统调用

NTSTATUS
NTAPI
NtCreateSemaphore(OUT PHANDLE SemaphoreHandle,
IN ACCESS_MASK DesiredAccess,
IN POBJECT_ATTRIBUTES ObjectAttributes OPTIONAL,
IN LONG InitialCount,
IN LONG MaximumCount)
{
PKSEMAPHORE Semaphore;
HANDLE hSemaphore;
KPROCESSOR_MODE PreviousMode = ExGetPreviousMode();
NTSTATUS Status;
PAGED_CODE();

// 参数验证
if ((MaximumCount <= 0) ||
(InitialCount < 0) ||
(InitialCount > MaximumCount))
{
return STATUS_INVALID_PARAMETER;
}

// 创建信号量对象
Status = ObCreateObject(PreviousMode,
ExSemaphoreObjectType,
ObjectAttributes,
PreviousMode,
NULL,
sizeof(KSEMAPHORE),
0,
0,
(PVOID*)&Semaphore);
if (!NT_SUCCESS(Status))
{
return Status;
}

// 初始化信号量
KeInitializeSemaphore(Semaphore, InitialCount, MaximumCount);

// 插入对象到对象管理器命名空间
Status = ObInsertObject((PVOID)Semaphore,
NULL,
DesiredAccess,
0,
NULL,
&hSemaphore);

// 返回句柄给调用者
if (NT_SUCCESS(Status))
{
*SemaphoreHandle = hSemaphore;
}

return Status;
}

源码位置：[ntoskrnl/ex/sem.c#L69](file:///d:/reactos/ntoskrnl/ex/sem.c#L69)

6.4.3 KeReleaseSemaphore实现分析

LONG
NTAPI
KeReleaseSemaphore(IN PKSEMAPHORE Semaphore,
IN KPRIORITY Increment,
IN LONG Adjustment,
IN BOOLEAN Wait)
{
LONG InitialState, State;
KIRQL OldIrql;
PKTHREAD CurrentThread;
ASSERT_SEMAPHORE(Semaphore);
ASSERT_IRQL_LESS_OR_EQUAL(DISPATCH_LEVEL);

// 获取调度器锁
OldIrql = KiAcquireDispatcherLock();

// 保存旧状态并计算新状态
InitialState = Semaphore->Header.SignalState;
State = InitialState + Adjustment;

// 检查是否超过限制
if ((Semaphore->Limit < State) || (InitialState > State))
{
KiReleaseDispatcherLock(OldIrql);
ExRaiseStatus(STATUS_SEMAPHORE_LIMIT_EXCEEDED);
}

// 设置新状态
Semaphore->Header.SignalState = State;

// 如果之前无可用资源且有等待者，唤醒它们
if (!(InitialState) && !(IsListEmpty(&Semaphore->Header.WaitListHead)))
{
KiWaitTest(&Semaphore->Header, Increment);
}

// 处理等待请求
if (Wait == FALSE)
{
KiReleaseDispatcherLock(OldIrql);
}
else
{
CurrentThread = KeGetCurrentThread();
CurrentThread->WaitNext = TRUE;
CurrentThread->WaitIrql = OldIrql;
}

return InitialState;
}

源码位置：[ntoskrnl/ke/semphobj.c#L54](file:///d:/reactos/ntoskrnl/ke/semphobj.c#L54)

6.4.4 跨进程信号量的使用

信号量通过命名对象实现跨进程共享：

跨进程信号量共享
┌─────────────────────────────────────────────────────────────────────┐
│ │
│ 进程A创建命名信号量 │
│ hSemaphore = CreateSemaphore(NULL, 3, 5, "Global\\\\ResourcePool");│
│ │ │
│ ▼ │
│ 对象管理器在全局命名空间创建信号量对象 │
│ │
│ 进程B打开同一信号量 │
│ hSemaphore = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, │
│ "Global\\\\ResourcePool"); │
│ │ │
│ ▼ │
│ 获取对同一信号量对象的引用 │
│ │
│ 两个进程共享同一个信号量计数器 │
│ 进程A ReleaseSemaphore(1) → 计数+1 │
│ 进程B WaitForSingleObject → 计数-1 │
│ │
└─────────────────────────────────────────────────────────────────────┘

6.4.5 完整示例代码

#include <windows.h>
#include <stdio.h>

#define MAX_CONCURRENT_THREADS 3

// 跨进程信号量示例
void CrossProcessSemaphoreExample()
{
// 创建命名信号量
HANDLE hSemaphore = CreateSemaphore(
NULL, // 默认安全属性
MAX_CONCURRENT_THREADS, // 初始计数
MAX_CONCURRENT_THREADS, // 最大计数
L"Global\\\\MySemaphore" // 全局名称
);

if (hSemaphore == NULL)
{
printf("CreateSemaphore failed: %d\\n", GetLastError());
return;
}

// 检查是否已存在
if (GetLastError() == ERROR_ALREADY_EXISTS)
{
printf("Semaphore already exists (other process created it)\\n");
}

// 创建工作线程
const int numThreads = 5;
HANDLE hThreads[numThreads];

for (int i = 0; i < numThreads; i++)
{
hThreads[i] = CreateThread(NULL, 0,
[](LPVOID hSem) -> DWORD {
HANDLE hSemaphore = (HANDLE)hSem;

// 获取信号量（最多3个线程同时执行）
printf("Thread %d waiting for semaphore…\\n",
GetCurrentThreadId());

WaitForSingleObject(hSemaphore, INFINITE);

printf("Thread %d acquired semaphore\\n",
GetCurrentThreadId());

// 模拟工作
Sleep(2000);

printf("Thread %d releasing semaphore\\n",
GetCurrentThreadId());

// 释放信号量
ReleaseSemaphore(hSemaphore, 1, NULL);

return 0;
}, hSemaphore, 0, NULL);
}

// 等待所有线程完成
WaitForMultipleObjects(numThreads, hThreads, TRUE, INFINITE);

// 清理
for (int i = 0; i < numThreads; i++)
CloseHandle(hThreads[i]);
CloseHandle(hSemaphore);
}

// 资源池管理示例
void ResourcePoolExample()
{
const int POOL_SIZE = 5;
HANDLE hSemaphore = CreateSemaphore(NULL, POOL_SIZE, POOL_SIZE, NULL);

// 模拟资源获取
void AcquireResource()
{
WaitForSingleObject(hSemaphore, INFINITE);
// 使用资源…
}

void ReleaseResource()
{
ReleaseSemaphore(hSemaphore, 1, NULL);
}

CloseHandle(hSemaphore);
}

// 生产者-消费者模式示例
void ProducerConsumerWithSemaphore()
{
const int BUFFER_SIZE = 10;

// 空闲槽位信号量
HANDLE hEmptySlots = CreateSemaphore(NULL, BUFFER_SIZE, BUFFER_SIZE, NULL);
// 已填充槽位信号量
HANDLE hFullSlots = CreateSemaphore(NULL, 0, BUFFER_SIZE, NULL);
// 互斥体保护缓冲区
HANDLE hMutex = CreateMutex(NULL, FALSE, NULL);

// 生产者
void Producer()
{
for (int i = 0; i < 20; i++)
{
WaitForSingleObject(hEmptySlots, INFINITE); // 等待空闲槽位
WaitForSingleObject(hMutex, INFINITE);

// 放入数据
printf("Produced: %d\\n", i);

ReleaseMutex(hMutex);
ReleaseSemaphore(hFullSlots, 1, NULL); // 增加已填充槽位
}
}

// 消费者
void Consumer()
{
for (int i = 0; i < 20; i++)
{
WaitForSingleObject(hFullSlots, INFINITE); // 等待已填充槽位
WaitForSingleObject(hMutex, INFINITE);

// 取出数据
printf("Consumed: %d\\n", i);

ReleaseMutex(hMutex);
ReleaseSemaphore(hEmptySlots, 1, NULL); // 增加空闲槽位
}
}

// 创建线程
HANDLE hProducer = CreateThread(NULL, 0, [](LPVOID) -> DWORD {
Producer(); return 0; }, NULL, 0, NULL);
HANDLE hConsumer = CreateThread(NULL, 0, [](LPVOID) -> DWORD {
Consumer(); return 0; }, NULL, 0, NULL);

WaitForSingleObject(hProducer, INFINITE);
WaitForSingleObject(hConsumer, INFINITE);

CloseHandle(hProducer);
CloseHandle(hConsumer);
CloseHandle(hEmptySlots);
CloseHandle(hFullSlots);
CloseHandle(hMutex);
}

6.5 互斥门（Mutant）

互斥门（Mutant）是一种支持递归获取的互斥同步原语，通过所有权机制确保同一时刻只有一个线程可以访问共享资源。

6.5.1 KMUTANT结构定义

typedef struct _KMUTANT {
DISPATCHER_HEADER Header; // 调度器头部
LIST_ENTRY MutantListEntry; // 线程的互斥体列表项
PKTHREAD OwnerThread; // 当前所有者线程
BOOLEAN Abandoned; // 是否已被遗弃
UCHAR ApcDisable; // APC禁用计数
} KMUTANT, *PKMUTANT;

结构关系：

KMUTANT结构
┌─────────────────────────────────────────────────────────────┐
│ KMUTANT │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ DISPATCHER_HEADER │ │
│ │ Type : MutantObject │ │
│ │ SignalState : 锁计数 (>0未锁定, 0锁定, <0递归) │ │
│ │ WaitListHead : 等待线程列表 │ │
│ └─────────────────────────────────────────────────────┘ │
│ MutantListEntry : 链入所有者线程的互斥体列表 │
│ OwnerThread : 当前持有互斥体的线程 │
│ Abandoned : 是否被遗弃 │
│ ApcDisable : APC禁用计数 │
└─────────────────────────────────────────────────────────────┘

6.5.2 NtCreateMutant系统调用

NTSTATUS
NTAPI
NtCreateMutant(OUT PHANDLE MutantHandle,
IN ACCESS_MASK DesiredAccess,
IN POBJECT_ATTRIBUTES ObjectAttributes OPTIONAL,
IN BOOLEAN InitialOwner)
{
KPROCESSOR_MODE PreviousMode = ExGetPreviousMode();
HANDLE hMutant;
PKMUTANT Mutant;
NTSTATUS Status;
PAGED_CODE();

// 创建互斥体对象
Status = ObCreateObject(PreviousMode,
ExMutantObjectType,
ObjectAttributes,
PreviousMode,
NULL,
sizeof(KMUTANT),
0,
0,
(PVOID*)&Mutant);

if (NT_SUCCESS(Status))
{
// 初始化内核互斥体
KeInitializeMutant(Mutant, InitialOwner);

// 插入对象到命名空间
Status = ObInsertObject((PVOID)Mutant,
NULL,
DesiredAccess,
0,
NULL,
&hMutant);

if (NT_SUCCESS(Status))
{
*MutantHandle = hMutant;
}
}

return Status;
}

源码位置：[ntoskrnl/ex/mutant.c#L79](file:///d:/reactos/ntoskrnl/ex/mutant.c#L79)

6.5.3 KeReleaseMutant实现分析

LONG
NTAPI
KeReleaseMutant(IN PKMUTANT Mutant,
IN KPRIORITY Increment,
IN BOOLEAN Abandon,
IN BOOLEAN Wait)
{
KIRQL OldIrql;
LONG PreviousState;
PKTHREAD CurrentThread = KeGetCurrentThread();
BOOLEAN EnableApc = FALSE;

// 获取调度器锁
OldIrql = KiAcquireDispatcherLock();

// 保存旧状态
PreviousState = Mutant->Header.SignalState;

if (Abandon == FALSE)
{
// 正常释放，检查所有权
if (Mutant->OwnerThread != CurrentThread)
{
KiReleaseDispatcherLock(OldIrql);
ExRaiseStatus(Mutant->Abandoned ? STATUS_ABANDONED :
STATUS_MUTANT_NOT_OWNED);
}

// 增加信号状态（减少锁计数）
Mutant->Header.SignalState++;
}
else
{
// 遗弃处理
Mutant->Header.SignalState = 1;
Mutant->Abandoned = TRUE;
}

// 如果信号状态变为1（完全释放）
if (Mutant->Header.SignalState == 1)
{
// 从线程的互斥体列表移除
if (PreviousState <= 0)
{
RemoveEntryList(&Mutant->MutantListEntry);
EnableApc = Mutant->ApcDisable;
}

// 清除所有者
Mutant->OwnerThread = NULL;

// 唤醒等待者
if (!IsListEmpty(&Mutant->Header.WaitListHead))
{
KiWaitTest(&Mutant->Header, Increment);
}
}

// 处理等待请求
if (Wait == FALSE)
{
KiReleaseDispatcherLock(OldIrql);
}
else
{
CurrentThread->WaitNext = TRUE;
CurrentThread->WaitIrql = OldIrql;
}

// 重新启用APC
if (EnableApc) KeLeaveCriticalRegion();

return PreviousState;
}

源码位置：[ntoskrnl/ke/mutex.c#L98](file:///d:/reactos/ntoskrnl/ke/mutex.c#L98)

6.5.4 命名互斥体的跨进程同步

命名互斥体跨进程同步
┌─────────────────────────────────────────────────────────────────────┐
│ │
│ 场景：防止多个进程同时运行 │
│ │
│ 进程A启动时创建命名互斥体 │
│ hMutex = CreateMutex(NULL, TRUE, "Global\\\\SingleInstance"); │
│ │ │
│ ▼ │
│ GetLastError() == ERROR_ALREADY_EXISTS? │
│ │ │
│ ├─► YES: 已有实例运行，退出 │
│ └─► NO: 成为第一个实例 │
│ │
│ 进程B尝试启动 │
│ hMutex = CreateMutex(NULL, TRUE, "Global\\\\SingleInstance"); │
│ │ │
│ ▼ │
│ GetLastError() == ERROR_ALREADY_EXISTS → 退出 │
│ │
│ 进程A退出时释放互斥体 │
│ ReleaseMutex(hMutex); │
│ │
└─────────────────────────────────────────────────────────────────────┘

6.5.5 完整示例代码

#include <windows.h>
#include <stdio.h>

// 单实例应用程序示例
BOOL EnsureSingleInstance()
{
// 创建命名互斥体
HANDLE hMutex = CreateMutex(
NULL, // 默认安全属性
TRUE, // 初始拥有者
L"Global\\\\MySingleInstanceApp" // 全局名称
);

if (hMutex == NULL)
{
printf("CreateMutex failed: %d\\n", GetLastError());
return FALSE;
}

// 检查是否已存在
if (GetLastError() == ERROR_ALREADY_EXISTS)
{
printf("Another instance is already running\\n");
CloseHandle(hMutex);
return FALSE;
}

// 成功获取互斥体，继续运行
printf("First instance running\\n");

// 程序运行期间持有互斥体
// …

// 退出时释放
ReleaseMutex(hMutex);
CloseHandle(hMutex);

return TRUE;
}

// 跨进程互斥示例
void CrossProcessMutexExample()
{
// 创建命名互斥体
HANDLE hMutex = CreateMutex(
NULL, // 默认安全属性
FALSE, // 初始不拥有
L"Global\\\\SharedResource" // 全局名称
);

if (hMutex == NULL)
{
printf("CreateMutex failed: %d\\n", GetLastError());
return;
}

// 等待互斥体
printf("Waiting for mutex…\\n");
DWORD result = WaitForSingleObject(hMutex, 5000);

switch (result)
{
case WAIT_OBJECT_0:
printf("Acquired mutex\\n");

// 执行临界区操作
printf("Performing shared operation…\\n");
Sleep(3000);

printf("Releasing mutex\\n");
ReleaseMutex(hMutex);
break;

case WAIT_TIMEOUT:
printf("Timeout waiting for mutex\\n");
break;

case WAIT_ABANDONED:
printf("Mutex was abandoned by another process\\n");
// 获得了互斥体，但资源状态可能不一致
ReleaseMutex(hMutex);
break;

default:
printf("Wait failed: %d\\n", GetLastError());
}

CloseHandle(hMutex);
}

// 递归互斥体示例
void RecursiveMutexExample()
{
HANDLE hMutex = CreateMutex(NULL, FALSE, NULL);

// 第一次获取
WaitForSingleObject(hMutex, INFINITE);
printf("First acquire\\n");

// 第二次获取（递归）
WaitForSingleObject(hMutex, INFINITE);
printf("Second acquire (recursive)\\n");

// 第一次释放
ReleaseMutex(hMutex);
printf("First release\\n");

// 第二次释放
ReleaseMutex(hMutex);
printf("Second release\\n");

CloseHandle(hMutex);
}

6.6 事件（Event）

事件是一种用于线程间通知的同步原语，支持通知事件和同步事件两种模式，广泛用于进程间的状态同步。

6.6.1 KEVENT结构定义

typedef struct _KEVENT {
DISPATCHER_HEADER Header; // 调度器头部
} KEVENT, *PKEVENT;

事件是最简单的同步对象，仅包含调度器头部。

类型区分：

6.6.2 NtCreateEvent系统调用

NTSTATUS
NTAPI
NtCreateEvent(OUT PHANDLE EventHandle,
IN ACCESS_MASK DesiredAccess,
IN POBJECT_ATTRIBUTES ObjectAttributes OPTIONAL,
IN EVENT_TYPE EventType,
IN BOOLEAN InitialState)
{
KPROCESSOR_MODE PreviousMode = ExGetPreviousMode();
PKEVENT Event;
HANDLE hEvent;
NTSTATUS Status;
PAGED_CODE();

// 验证事件类型
if ((EventType != NotificationEvent) &&
(EventType != SynchronizationEvent))
{
return STATUS_INVALID_PARAMETER;
}

// 创建事件对象
Status = ObCreateObject(PreviousMode,
ExEventObjectType,
ObjectAttributes,
PreviousMode,
NULL,
sizeof(KEVENT),
0,
0,
(PVOID*)&Event);
if (!NT_SUCCESS(Status))
{
return Status;
}

// 初始化事件
KeInitializeEvent(Event, EventType, InitialState);

// 插入对象
Status = ObInsertObject((PVOID)Event,
NULL,
DesiredAccess,
0,
NULL,
&hEvent);

if (NT_SUCCESS(Status))
{
*EventHandle = hEvent;
}

return Status;
}

源码位置：[ntoskrnl/ex/event.c#L96](file:///d:/reactos/ntoskrnl/ex/event.c#L96)

6.6.3 命名事件与跨进程同步

命名事件跨进程同步模式
┌─────────────────────────────────────────────────────────────────────┐
│ │
│ 场景：进程A完成工作后通知进程B │
│ │
│ 进程A创建命名事件（通知事件） │
│ hEvent = CreateEvent(NULL, TRUE, FALSE, "Global\\\\WorkDone"); │
│ │
│ 进程B打开事件并等待 │
│ hEvent = OpenEvent(EVENT_ALL_ACCESS, FALSE, │
│ "Global\\\\WorkDone"); │
│ WaitForSingleObject(hEvent, INFINITE); │
│ │
│ 进程A完成工作后设置事件 │
│ SetEvent(hEvent); │
│ │ │
│ ▼ │
│ 进程B被唤醒，继续执行 │
│ │
│ 进程B处理完成后复位事件 │
│ ResetEvent(hEvent); │
│ │
└─────────────────────────────────────────────────────────────────────┘

6.6.4 完整示例代码

#include <windows.h>
#include <stdio.h>

// 跨进程事件通知示例
void CrossProcessEventNotification()
{
// 创建命名通知事件
HANDLE hEvent = CreateEvent(
NULL, // 默认安全属性
TRUE, // 手动复位（通知事件）
FALSE, // 初始无信号
L"Global\\\\WorkComplete" // 全局名称
);

if (hEvent == NULL)
{
printf("CreateEvent failed: %d\\n", GetLastError());
return;
}

// 模拟工作完成
printf("Performing work…\\n");
Sleep(3000);
printf("Work complete, signaling event\\n");

// 通知所有等待进程
SetEvent(hEvent);

// 保持事件信号状态一段时间
Sleep(5000);

// 复位事件
ResetEvent(hEvent);

CloseHandle(hEvent);
}

// 同步事件示例（自动复位）
void SynchronizationEventExample()
{
// 创建同步事件（自动复位）
HANDLE hEvent = CreateEvent(
NULL, // 默认安全属性
FALSE, // 自动复位
FALSE, // 初始无信号
NULL // 无名事件
);

// 创建多个等待线程
const int numThreads = 3;
HANDLE hThreads[numThreads];

for (int i = 0; i < numThreads; i++)
{
hThreads[i] = CreateThread(NULL, 0,
[](LPVOID hEv) -> DWORD {
HANDLE hEvent = (HANDLE)hEv;

printf("Thread %d waiting\\n", GetCurrentThreadId());
WaitForSingleObject(hEvent, INFINITE);
printf("Thread %d woke up\\n", GetCurrentThreadId());

// 事件已自动复位，其他线程继续等待
return 0;
}, hEvent, 0, NULL);
}

// 设置事件 – 只唤醒一个线程
Sleep(100);
printf("Setting event (wake one thread)\\n");
SetEvent(hEvent);

// 再次设置事件
Sleep(100);
printf("Setting event (wake another thread)\\n");
SetEvent(hEvent);

// 再次设置事件
Sleep(100);
printf("Setting event (wake third thread)\\n");
SetEvent(hEvent);

WaitForMultipleObjects(numThreads, hThreads, TRUE, INFINITE);

for (int i = 0; i < numThreads; i++)
CloseHandle(hThreads[i]);
CloseHandle(hEvent);
}

// 多个事件等待示例
void MultipleEventWaitExample()
{
HANDLE hEvents[3];
for (int i = 0; i < 3; i++)
hEvents[i] = CreateEvent(NULL, TRUE, FALSE, NULL);

// 设置第二个事件
SetEvent(hEvents[1]);

// 等待任一事件
DWORD result = WaitForMultipleObjects(3, hEvents, FALSE, INFINITE);

if (result >= WAIT_OBJECT_0 && result < WAIT_OBJECT_0 + 3)
{
printf("Event %d was signaled\\n", result – WAIT_OBJECT_0);
}

// 复位所有事件
for (int i = 0; i < 3; i++)
{
ResetEvent(hEvents[i]);
CloseHandle(hEvents[i]);
}
}

// PulseEvent示例
void PulseEventExample()
{
HANDLE hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);

// 创建等待线程
HANDLE hThread = CreateThread(NULL, 0,
[](LPVOID hEv) -> DWORD {
HANDLE hEvent = (HANDLE)hEv;
printf("Thread waiting…\\n");

DWORD result = WaitForSingleObject(hEvent, 5000);
if (result == WAIT_OBJECT_0)
printf("Thread woke up during pulse\\n");
else if (result == WAIT_TIMEOUT)
printf("Thread missed the pulse\\n");

return 0;
}, hEvent, 0, NULL);

// 等待线程开始等待
Sleep(100);

// PulseEvent：短暂设置为有信号然后立即复位
printf("Pulsing event…\\n");
PulseEvent(hEvent);

WaitForSingleObject(hThread, INFINITE);
CloseHandle(hThread);
CloseHandle(hEvent);
}

6.7 命名管道（Named Pipe）和邮件槽（Mailslot）

命名管道和邮件槽是Windows提供的两种进程间通信机制，适用于不同的通信场景。

6.7.1 命名管道概述

命名管道是一种面向连接的、可靠的字节流通信机制，支持双向通信。

管道类型：

6.7.2 CreateNamedPipe实现

HANDLE
WINAPI
CreateNamedPipeA(IN LPCSTR lpName,
IN DWORD dwOpenMode,
IN DWORD dwPipeMode,
IN DWORD nMaxInstances,
IN DWORD nOutBufferSize,
IN DWORD nInBufferSize,
IN DWORD nDefaultTimeOut,
IN LPSECURITY_ATTRIBUTES lpSecurityAttributes)
{
UNICODE_STRING Name;
NTSTATUS Status;
OBJECT_ATTRIBUTES ObjectAttributes;
HANDLE PipeHandle;
IO_STATUS_BLOCK IoStatus;

// 转换名称
RtlInitUnicodeString(&Name, lpName);

// 初始化对象属性
InitializeObjectAttributes(&ObjectAttributes,
&Name,
OBJ_CASE_INSENSITIVE,
NULL,
lpSecurityAttributes ?
lpSecurityAttributes->lpSecurityDescriptor : NULL);

// 创建管道
Status = NtCreateNamedPipeFile(&PipeHandle,
dwOpenMode,
&ObjectAttributes,
&IoStatus,
FILE_SHARE_READ | FILE_SHARE_WRITE,
FILE_CREATE,
FILE_SYNCHRONOUS_IO_NONALERT,
dwPipeMode,
nMaxInstances,
nOutBufferSize,
nInBufferSize,
nDefaultTimeOut);

if (!NT_SUCCESS(Status))
{
SetLastError(BaseNtStatusToWin32Error(Status));
return INVALID_HANDLE_VALUE;
}

return PipeHandle;
}

关键参数：

6.7.3 ConnectNamedPipe与等待连接

BOOL
WINAPI
ConnectNamedPipe(IN HANDLE hNamedPipe,
OUT LPOVERLAPPED lpOverlapped)
{
NTSTATUS Status;
IO_STATUS_BLOCK IoStatus;

if (lpOverlapped)
{
// 异步连接
Status = NtFsControlFile(hNamedPipe,
lpOverlapped->hEvent,
NULL,
NULL,
&IoStatus,
FSCTL_PIPE_LISTEN,
NULL,
0,
NULL,
0);
}
else
{
// 同步连接
Status = NtFsControlFile(hNamedPipe,
NULL,
NULL,
NULL,
&IoStatus,
FSCTL_PIPE_LISTEN,
NULL,
0,
NULL,
0);
}

if (!NT_SUCCESS(Status))
{
SetLastError(BaseNtStatusToWin32Error(Status));
return FALSE;
}

return TRUE;
}

源码位置：[dll/win32/kernel32/client/file/npipe.c](file:///d:/reactos/dll/win32/kernel32/client/file/npipe.c)

6.7.4 邮件槽概述

邮件槽是一种单向、不可靠的数据报通信机制，支持一对多的广播通信。

邮件槽特点：

• 单向通信：服务器创建，客户端发送
• 广播能力：支持域内广播
• 不可靠：消息可能丢失
• 大小限制：最大消息大小由服务器设置

6.7.5 CreateMailslot实现

HANDLE
WINAPI
CreateMailslotW(IN LPCWSTR lpName,
IN DWORD nMaxMessageSize,
IN DWORD lReadTimeout,
IN LPSECURITY_ATTRIBUTES lpSecurityAttributes)
{
OBJECT_ATTRIBUTES ObjectAttributes;
UNICODE_STRING MailslotName;
HANDLE MailslotHandle;
NTSTATUS Status;
LARGE_INTEGER DefaultTimeOut;
IO_STATUS_BLOCK Iosb;

// 转换名称
RtlDosPathNameToNtPathName_U(lpName, &MailslotName, NULL, NULL);

// 初始化对象属性
InitializeObjectAttributes(&ObjectAttributes,
&MailslotName,
OBJ_CASE_INSENSITIVE,
NULL,
lpSecurityAttributes ?
lpSecurityAttributes->lpSecurityDescriptor : NULL);

// 设置超时
if (lReadTimeout == MAILSLOT_WAIT_FOREVER)
{
DefaultTimeOut.QuadPart = 0xFFFFFFFFFFFFFFFFLL;
}
else
{
DefaultTimeOut.QuadPart = lReadTimeout * –10000LL;
}

// 创建邮件槽
Status = NtCreateMailslotFile(&MailslotHandle,
GENERIC_READ | SYNCHRONIZE | WRITE_DAC,
&ObjectAttributes,
&Iosb,
FILE_WRITE_THROUGH,
0,
nMaxMessageSize,
&DefaultTimeOut);

RtlFreeHeap(RtlGetProcessHeap(), 0, MailslotName.Buffer);

if (!NT_SUCCESS(Status))
{
SetLastError(BaseNtStatusToWin32Error(Status));
return INVALID_HANDLE_VALUE;
}

return MailslotHandle;
}

源码位置：[dll/win32/kernel32/client/file/mailslot.c](file:///d:/reactos/dll/win32/kernel32/client/file/mailslot.c)

6.7.6 命名管道与邮件槽对比

6.7.7 完整示例代码

#include <windows.h>
#include <stdio.h>

// 命名管道服务器示例
void NamedPipeServer()
{
HANDLE hPipe;
char buffer[1024];
DWORD bytesRead;

// 创建命名管道
hPipe = CreateNamedPipe(
L"\\\\\\\\.\\\\pipe\\\\MyPipe",
PIPE_ACCESS_DUPLEX, // 双向访问
PIPE_TYPE_MESSAGE | // 消息模式
PIPE_READMODE_MESSAGE |
PIPE_WAIT, // 阻塞模式
PIPE_UNLIMITED_INSTANCES, // 无限实例
1024, // 输出缓冲区
1024, // 输入缓冲区
5000, // 默认超时
NULL); // 默认安全属性

if (hPipe == INVALID_HANDLE_VALUE)
{
printf("CreateNamedPipe failed: %d\\n", GetLastError());
return;
}

printf("Pipe created, waiting for client…\\n");

// 等待客户端连接
if (!ConnectNamedPipe(hPipe, NULL))
{
printf("ConnectNamedPipe failed: %d\\n", GetLastError());
CloseHandle(hPipe);
return;
}

printf("Client connected\\n");

// 读取客户端消息
if (!ReadFile(hPipe, buffer, sizeof(buffer), &bytesRead, NULL))
{
printf("ReadFile failed: %d\\n", GetLastError());
CloseHandle(hPipe);
return;
}

printf("Received: %s\\n", buffer);

// 发送响应
const char* response = "Hello from server!";
DWORD bytesWritten;
WriteFile(hPipe, response, strlen(response) + 1, &bytesWritten, NULL);

// 断开连接
DisconnectNamedPipe(hPipe);
CloseHandle(hPipe);
}

// 命名管道客户端示例
void NamedPipeClient()
{
HANDLE hPipe;
char buffer[1024];
DWORD bytesRead;

// 连接到命名管道
while (TRUE)
{
hPipe = CreateFile(
L"\\\\\\\\.\\\\pipe\\\\MyPipe",
GENERIC_READ | GENERIC_WRITE,
0,
NULL,
OPEN_EXISTING,
0,
NULL);

if (hPipe != INVALID_HANDLE_VALUE)
break;

if (GetLastError() != ERROR_PIPE_BUSY)
{
printf("CreateFile failed: %d\\n", GetLastError());
return;
}

// 等待管道可用
WaitNamedPipe(L"\\\\\\\\.\\\\pipe\\\\MyPipe", NMPWAIT_WAIT_FOREVER);
}

printf("Connected to server\\n");

// 发送消息
const char* message = "Hello from client!";
DWORD bytesWritten;
WriteFile(hPipe, message, strlen(message) + 1, &bytesWritten, NULL);

// 读取响应
ReadFile(hPipe, buffer, sizeof(buffer), &bytesRead, NULL);
printf("Server response: %s\\n", buffer);

CloseHandle(hPipe);
}

// 邮件槽服务器示例
void MailslotServer()
{
HANDLE hMailslot;
char buffer[1024];
DWORD bytesRead;
DWORD messageSize;

// 创建邮件槽
hMailslot = CreateMailslot(
L"\\\\\\\\.\\\\mailslot\\\\MyMailslot",
0, // 无最大消息大小限制
MAILSLOT_WAIT_FOREVER, // 无限等待
NULL); // 默认安全属性

if (hMailslot == INVALID_HANDLE_VALUE)
{
printf("CreateMailslot failed: %d\\n", GetLastError());
return;
}

printf("Mailslot created, waiting for messages…\\n");

// 读取消息
while (TRUE)
{
if (!ReadFile(hMailslot, buffer, sizeof(buffer), &bytesRead, NULL))
{
printf("ReadFile failed: %d\\n", GetLastError());
break;
}

printf("Received %d bytes: %s\\n", bytesRead, buffer);
}

CloseHandle(hMailslot);
}

// 邮件槽客户端示例
void MailslotClient()
{
HANDLE hMailslot;
DWORD bytesWritten;

// 打开邮件槽
hMailslot = CreateFile(
L"\\\\\\\\.\\\\mailslot\\\\MyMailslot",
GENERIC_WRITE,
FILE_SHARE_READ,
NULL,
OPEN_EXISTING,
0,
NULL);

if (hMailslot == INVALID_HANDLE_VALUE)
{
printf("CreateFile failed: %d\\n", GetLastError());
return;
}

// 发送消息
const char* message = "Hello from mailslot client!";
WriteFile(hMailslot, message, strlen(message) + 1, &bytesWritten, NULL);

printf("Sent %d bytes\\n", bytesWritten);

CloseHandle(hMailslot);
}

// 邮件槽广播示例
void MailslotBroadcast()
{
HANDLE hMailslot;
DWORD bytesWritten;

// 使用域广播地址
hMailslot = CreateFile(
L"\\\\\\\\*\\\\mailslot\\\\MyBroadcastSlot",
GENERIC_WRITE,
FILE_SHARE_READ,
NULL,
OPEN_EXISTING,
0,
NULL);

if (hMailslot == INVALID_HANDLE_VALUE)
{
printf("CreateFile failed: %d\\n", GetLastError());
return;
}

// 广播消息
const char* message = "Broadcast message to all servers!";
WriteFile(hMailslot, message, strlen(message) + 1, &bytesWritten, NULL);

printf("Broadcast sent\\n");

CloseHandle(hMailslot);
}

总结

核心要点回顾

本章代码索引

本节完。