深入操作系统 Socket 底层：增加锁 / 引用计数并发安全 + UDP 适配

米尔的可达鸭

504人浏览 · 2026-07-15 10:00:00

米尔的可达鸭 · 2026-07-15 10:00:00 发布

增加锁 / 引用计数并发安全 + UDP 适配深度详解

整体定位

完成 backlog全连接队列 + shutdown半关闭 + recv_exact/send_all 后，接下来两大核心任务：

锁 + 引用计数：解决多线程/软中断/定时器并发访问 Socket 对象导致崩溃、脏读、野指针（并发安全底座）
UDP完整适配：实现无连接/连接模式UDP语义、sendto/recvfrom、ICMP差错联动（补齐两种核心传输模型）

一、锁 + 引用计数：并发安全改造

背景问题

真实内核不是单线程：

用户主线程调用 send/recv/close
软中断/网卡收包线程修改 Socket rx_buf / 状态 / 等待队列
定时器线程处理超时、TIME_WAIT、僵死连接
多线程/回调同时修改同一个 SocketSk、fd表、accept队列，会产生：
- 缓冲区数据错乱、半包损坏
- 野指针：close销毁Socket后还有回调/线程继续访问它
- 等待队列状态错乱、虚假唤醒、死锁
- fd表并发修改导致fd重复分配、内存泄漏

1. 引用计数（refcount）核心原理

核心目标

延迟销毁 SocketSk 对象，直到没有任何线程/回调持有引用，防止野指针访问

新增 refcount 字段：记录当前有多少使用者（主线程、等待线程、回调、定时器）引用该Socket
- refcount > 0：Socket 对象不能直接释放内存
- refcount == 0：确认无引用，才真正释放内存
- 配套：sk_hold() 增加引用，sk_release() 减少引用
- close() 不再立刻销毁对象/回收fd，仅标记关闭状态、递减引用计数，等待refcount归零再彻底释放
- 必须是原子计数（多线程无锁安全计数），避免计数本身被并发篡改
引用计数生命周期管理：
- 创建时初始化为1（创建者持有）
- 每个阻塞等待线程进入前必须 sk_hold()，唤醒退出后必须 sk_release()
- epoll监听事件时增加引用，事件移除时减少引用
- 定时器回调注册时增加引用，回调执行完毕或取消时减少引用
- 异步I/O回调注册时增加引用，回调完成时减少引用

示例定义

// C语言实现（真实内核使用 atomic_t）
#include <stdatomic.h>
#include <pthread.h>

typedef struct SocketSk {
    // ...原有字段
    atomic_int refcount;          // 原子引用计数
    pthread_mutex_t ref_lock;     // 保护refcount和is_destroyed
    int is_destroyed;             // 销毁标记
    pthread_mutex_t sk_lock;      // Socket内部状态锁
    // 其他字段...
} SocketSk;

// 增加引用计数（持有引用）
int sk_hold(SocketSk* sk) {
    if (!sk) return 0;
    
    pthread_mutex_lock(&sk->ref_lock);
    if (!sk->is_destroyed) {
        // 原子递增，返回旧值（可用于调试）
        int old = atomic_fetch_add(&sk->refcount, 1);
        pthread_mutex_unlock(&sk->ref_lock);
        return 1;
    }
    pthread_mutex_unlock(&sk->ref_lock);
    return 0;  // Socket已销毁，无法持有
}

// 减少引用计数，计数为0则标记销毁
int sk_release(SocketSk* sk) {
    if (!sk) return 0;
    
    int need_free = 0;
    pthread_mutex_lock(&sk->ref_lock);
    
    // 原子递减并获取新值
    int new_count = atomic_fetch_sub(&sk->refcount, 1) - 1;
    
    if (new_count == 0) {
        // 最后一个引用释放，标记销毁
        need_free = 1;
        sk->is_destroyed = 1;
        
        // 注意：这里只标记，不立即释放内存
        // 实际内存释放由调用者负责（如close_new函数）
    } else if (new_count < 0) {
        // 引用计数异常：不应小于0
        // 内核中会触发BUG_ON或panic
        // 学习版可记录错误日志
    }
    
    pthread_mutex_unlock(&sk->ref_lock);
    return need_free;
}

// 获取当前引用计数（调试用）
int sk_refcount(SocketSk* sk) {
    if (!sk) return 0;
    return atomic_load(&sk->refcount);
}

2. 并发测试验证

多线程同时调用 send/recv/close/accept 做压力测试
检查是否出现：数据损坏、卡死、反复崩溃、内存持续上涨（内存泄漏）
检测：引用计数是否正确增减，无永久占用Socket

3. 锁的典型死锁场景与解决方案

在多线程并发访问Socket时，锁顺序不当是导致死锁的常见原因。以下是一个典型的死锁场景及其解决方案：

死锁场景：锁顺序反转

场景描述：
线程A和线程B同时操作两个Socket对象（sk1和sk2），但获取锁的顺序相反：

线程A：先获取sk1的锁，再获取sk2的锁
线程B：先获取sk2的锁，再获取sk1的锁

当两个线程同时执行时，可能出现：

线程A获取sk1锁
线程B获取sk2锁
线程A等待sk2锁（被线程B持有）
线程B等待sk1锁（被线程A持有）
形成死锁，两个线程永久阻塞

示例代码：

// 线程A的执行路径
void thread_a_func(SocketSk* sk1, SocketSk* sk2) {
    pthread_mutex_lock(&sk1->sk_lock);    // 获取sk1锁
    // ... 操作sk1 ...
    pthread_mutex_lock(&sk2->sk_lock);    // 尝试获取sk2锁（可能阻塞）
    // ... 操作sk2 ...
    pthread_mutex_unlock(&sk2->sk_lock);
    pthread_mutex_unlock(&sk1->sk_lock);
}

// 线程B的执行路径  
void thread_b_func(SocketSk* sk1, SocketSk* sk2) {
    pthread_mutex_lock(&sk2->sk_lock);    // 获取sk2锁
    // ... 操作sk2 ...
    pthread_mutex_lock(&sk1->sk_lock);    // 尝试获取sk1锁（可能阻塞）
    // ... 操作sk1 ...
    pthread_mutex_unlock(&sk1->sk_lock);
    pthread_mutex_unlock(&sk2->sk_lock);
}

解决方案

1. 锁排序（Lock Ordering）

原理：为所有Socket对象定义全局排序规则（如按内存地址、fd值排序），所有线程按相同顺序获取锁
实现：

// 按Socket指针地址排序获取锁
void lock_two_sockets(SocketSk* sk1, SocketSk* sk2) {
    SocketSk* first = sk1;
    SocketSk* second = sk2;
    
    // 确保按地址顺序获取锁
    if ((uintptr_t)sk1 > (uintptr_t)sk2) {
        first = sk2;
        second = sk1;
    }
    
    pthread_mutex_lock(&first->sk_lock);
    pthread_mutex_lock(&second->sk_lock);
    
    // 返回实际顺序，便于后续操作
    if (first == sk1) {
        // sk1先于sk2
    } else {
        // sk2先于sk1
    }
}

2. 尝试锁（Try-Lock）与回退

原理：使用 pthread_mutex_trylock() 尝试获取锁，失败时释放已持有锁并重试
实现：

int lock_with_retry(SocketSk* sk1, SocketSk* sk2, int max_retries) {
    int retries = 0;
    
    while (retries < max_retries) {
        // 尝试获取第一个锁
        if (pthread_mutex_trylock(&sk1->sk_lock) != 0) {
            usleep(1000); // 短暂等待
            retries++;
            continue;
        }
        
        // 尝试获取第二个锁
        if (pthread_mutex_trylock(&sk2->sk_lock) != 0) {
            // 获取第二个锁失败，释放第一个锁
            pthread_mutex_unlock(&sk1->sk_lock);
            usleep(1000);
            retries++;
            continue;
        }
        
        // 成功获取两个锁
        return 1;
    }
    
    return 0; // 超过重试次数
}

3. 死锁检测与恢复

原理：记录锁获取顺序，检测潜在死锁环路
实现思路：
1. 为每个线程维护已获取锁的列表
2. 获取新锁前检查是否形成环路（A等B，B等A）
3. 检测到死锁时，强制释放某些锁或终止线程

4. 避免嵌套锁（Lock Hierarchy）

原理：设计清晰的锁层次结构，禁止低层次锁持有期间获取高层次锁
示例层次：
1. 全局FD表锁（最高层）
2. Socket内部状态锁（中间层）
3. 缓冲区锁（最底层）

5. 使用读写锁优化

适用场景：读多写少的场景
实现：

// 使用读写锁替代互斥锁
pthread_rwlock_t sk_rwlock;

// 读操作（共享锁）
pthread_rwlock_rdlock(&sk_rwlock);
// ... 读取操作 ...
pthread_rwlock_unlock(&sk_rwlock);

// 写操作（独占锁）
pthread_rwlock_wrlock(&sk_rwlock);
// ... 写入操作 ...
pthread_rwlock_unlock(&sk_rwlock);