MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

深入理解操作系统的并发控制

2022-08-176.5k 阅读

操作系统并发控制的基础概念

在操作系统环境下，多个程序常常会同时运行，这些程序在执行过程中可能会共享一些资源，比如内存空间、文件句柄、打印机等。并发控制的目的就是确保这些共享资源在多个并发执行的程序之间能够被正确地访问和使用，避免出现数据不一致、资源争用等问题。

进程与线程

进程是操作系统进行资源分配和调度的基本单位，每个进程都有独立的地址空间、代码段、数据段等。而线程则是进程内的一个执行单元，多个线程共享进程的资源，它们可以并发执行。例如，在一个浏览器进程中，可能有一个线程负责页面渲染，另一个线程负责网络请求。

在多线程编程中，由于多个线程共享进程资源，当多个线程同时访问和修改共享数据时，就可能出现数据不一致的情况。比如下面这段简单的Java代码：

public class Counter {
    private int count = 0;

    public void increment() {
        count++;
    }

    public int getCount() {
        return count;
    }
}

public class ThreadExample {
    public static void main(String[] args) {
        Counter counter = new Counter();
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                counter.increment();
            }
        });
        Thread thread2 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                counter.increment();
            }
        });
        thread1.start();
        thread2.start();
        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Expected count: 2000, Actual count: " + counter.getCount());
    }
}

在上述代码中，预期的结果是 count 最终为2000，但实际运行结果往往小于2000。这是因为 count++ 操作不是原子的，它包含读取、增加和写入三个步骤。当两个线程同时执行 count++ 时，可能会出现读取到相同的值，然后都进行增加操作，最后写入相同的值，导致数据不一致。

并发与并行

并发是指在一段时间内，系统能够处理多个任务，这些任务在宏观上是同时执行的，但在微观上可能是交替执行的。例如，单核CPU的操作系统通过时间片轮转的方式，让多个进程交替使用CPU，从用户角度看好像多个进程在同时运行。

并行则是指在同一时刻，多个任务真正地同时执行，这需要多核CPU等硬件支持。例如，在一个多核CPU系统中，每个核心可以同时运行一个任务，实现真正的并行处理。

共享资源与竞态条件

共享资源是指多个进程或线程都可以访问的资源，如内存、文件、数据库等。当多个并发执行的程序访问和修改共享资源时，如果没有适当的控制，就可能出现竞态条件。

竞态条件的产生

竞态条件是指多个进程或线程在访问和修改共享资源时，最终的结果取决于这些进程或线程执行的相对顺序。以之前的 Counter 类为例，由于 count++ 操作的非原子性，导致不同线程执行顺序不同时，最终 count 的值也会不同，这就是竞态条件的一种表现。

竞态条件的危害

竞态条件可能导致数据不一致，影响程序的正确性。在一个银行转账的场景中，如果两个线程同时对账户进行取款和存款操作，没有合适的并发控制，可能会导致账户余额计算错误，给用户造成经济损失。

并发控制的基本方法

为了避免竞态条件，保证共享资源的正确访问，操作系统提供了多种并发控制方法。

互斥锁（Mutex）

互斥锁是一种最基本的并发控制工具，它的作用是保证在同一时刻只有一个进程或线程能够访问共享资源。互斥锁只有两种状态：锁定（locked）和解锁（unlocked）。当一个线程获取了互斥锁（将其状态设为锁定），其他线程就无法获取，直到该线程释放互斥锁（将其状态设为解锁）。

下面是使用C++ 和POSIX线程库实现的互斥锁示例：

#include <iostream>
#include <thread>
#include <mutex>

std::mutex mtx;
int count = 0;

void increment() {
    mtx.lock();
    count++;
    mtx.unlock();
}

int main() {
    std::thread threads[10];
    for (int i = 0; i < 10; i++) {
        threads[i] = std::thread(increment);
    }
    for (auto& thread : threads) {
        thread.join();
    }
    std::cout << "Final count: " << count << std::endl;
    return 0;
}

在上述代码中，通过 std::mutex 实现了对共享变量 count 的互斥访问。mtx.lock() 用于获取互斥锁，mtx.unlock() 用于释放互斥锁，这样就保证了 count++ 操作在同一时刻只有一个线程能够执行，避免了竞态条件。

信号量（Semaphore）

信号量是一个整型变量，它的值表示当前可用的资源数量。信号量可以用来控制同时访问共享资源的进程或线程数量。例如，假设有一个打印机资源，同一时间只能允许3个进程使用，就可以通过一个初始值为3的信号量来实现。

下面是使用POSIX信号量的C代码示例：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <semaphore.h>

sem_t semaphore;
int count = 0;

void* increment(void* arg) {
    sem_wait(&semaphore);
    count++;
    sem_post(&semaphore);
    return NULL;
}

int main() {
    sem_init(&semaphore, 0, 1);
    pthread_t threads[10];
    for (int i = 0; i < 10; i++) {
        pthread_create(&threads[i], NULL, increment, NULL);
    }
    for (int i = 0; i < 10; i++) {
        pthread_join(threads[i], NULL);
    }
    sem_destroy(&semaphore);
    printf("Final count: %d\n", count);
    return 0;
}

在这个例子中，sem_wait(&semaphore) 会将信号量的值减1，如果信号量的值为0，则线程会阻塞等待。sem_post(&semaphore) 会将信号量的值加1。通过这种方式，信号量可以控制同时访问共享资源的线程数量。

读写锁（Read - Write Lock）

读写锁用于区分对共享资源的读操作和写操作。它允许多个线程同时进行读操作，因为读操作不会修改共享资源，不会产生竞态条件。但是，当有一个线程进行写操作时，其他线程既不能进行读操作也不能进行写操作，以保证数据的一致性。

下面是使用POSIX读写锁的C代码示例：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

pthread_rwlock_t rwlock;
int sharedData = 0;

void* reader(void* arg) {
    pthread_rwlock_rdlock(&rwlock);
    printf("Reader read: %d\n", sharedData);
    pthread_rwlock_unlock(&rwlock);
    return NULL;
}

void* writer(void* arg) {
    pthread_rwlock_wrlock(&rwlock);
    sharedData++;
    printf("Writer wrote: %d\n", sharedData);
    pthread_rwlock_unlock(&rwlock);
    return NULL;
}

int main() {
    pthread_rwlock_init(&rwlock, NULL);
    pthread_t readerThreads[5], writerThreads[3];
    for (int i = 0; i < 5; i++) {
        pthread_create(&readerThreads[i], NULL, reader, NULL);
    }
    for (int i = 0; i < 3; i++) {
        pthread_create(&writerThreads[i], NULL, writer, NULL);
    }
    for (int i = 0; i < 5; i++) {
        pthread_join(readerThreads[i], NULL);
    }
    for (int i = 0; i < 3; i++) {
        pthread_join(writerThreads[i], NULL);
    }
    pthread_rwlock_destroy(&rwlock);
    return 0;
}

在上述代码中，pthread_rwlock_rdlock(&rwlock) 用于获取读锁，pthread_rwlock_wrlock(&rwlock) 用于获取写锁。读锁允许多个线程同时获取，而写锁则会阻止其他线程获取读锁或写锁。

高级并发控制机制

除了基本的互斥锁、信号量和读写锁外，操作系统还提供了一些高级的并发控制机制。

条件变量（Condition Variable）

条件变量通常与互斥锁一起使用，用于线程之间的同步。它允许线程在满足特定条件时才继续执行。例如，在一个生产者 - 消费者模型中，消费者线程需要等待生产者线程生产出数据后才能消费。

下面是使用C++ 和条件变量的示例代码：

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>

std::mutex mtx;
std::condition_variable cv;
std::queue<int> dataQueue;
bool finished = false;

void producer() {
    for (int i = 0; i < 10; i++) {
        std::unique_lock<std::mutex> lock(mtx);
        dataQueue.push(i);
        std::cout << "Produced: " << i << std::endl;
        lock.unlock();
        cv.notify_one();
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    std::unique_lock<std::mutex> lock(mtx);
    finished = true;
    cv.notify_all();
}

void consumer() {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [] { return!dataQueue.empty() || finished; });
        if (dataQueue.empty() && finished) {
            break;
        }
        int value = dataQueue.front();
        dataQueue.pop();
        std::cout << "Consumed: " << value << std::endl;
        lock.unlock();
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
    }
}

int main() {
    std::thread producerThread(producer);
    std::thread consumerThread(consumer);
    producerThread.join();
    consumerThread.join();
    return 0;
}

在这个例子中，cv.wait(lock, [] { return!dataQueue.empty() || finished; }); 表示消费者线程等待条件变量 cv，直到队列 dataQueue 不为空或者 finished 为 true。cv.notify_one() 和 cv.notify_all() 用于通知等待在条件变量上的线程。

屏障（Barrier）

屏障用于同步多个线程，使得所有线程在到达某个点时，都必须等待其他线程到达该点后才能继续执行。这在一些需要多个线程协同完成某个阶段任务的场景中非常有用。

下面是使用OpenMP实现屏障的C代码示例：

#include <stdio.h>
#include <omp.h>

int main() {
    #pragma omp parallel num_threads(4)
    {
        printf("Thread %d started\n", omp_get_thread_num());
        #pragma omp barrier
        printf("Thread %d passed barrier\n", omp_get_thread_num());
    }
    return 0;
}

在上述代码中，#pragma omp barrier 表示一个屏障点，所有线程在到达这个点时，必须等待其他线程也到达该点，然后才能继续执行后续的代码。

操作系统内核中的并发控制

操作系统内核管理着系统的各种资源，如CPU、内存、设备等，因此内核中的并发控制至关重要。

内核中的锁机制

内核中广泛使用锁来保护共享资源。例如，自旋锁（Spinlock）是一种在多处理器系统内核中常用的锁。当一个线程尝试获取自旋锁时，如果锁已经被占用，线程不会进入睡眠状态，而是在原地不断尝试获取锁，直到锁被释放。自旋锁适用于锁的持有时间较短的情况，因为线程自旋的过程中会占用CPU资源，如果锁的持有时间过长，自旋会浪费大量CPU时间。

下面是一个简单的自旋锁实现的伪代码：

typedef struct {
    int locked;
} spinlock_t;

void spin_lock(spinlock_t *lock) {
    while (test_and_set(&lock->locked)) {
        // 自旋等待
    }
}

void spin_unlock(spinlock_t *lock) {
    lock->locked = 0;
}

在实际的内核代码中，自旋锁的实现会更加复杂，需要考虑硬件平台的特性和原子操作的实现。

中断处理与并发控制

操作系统内核需要处理各种中断，如硬件中断、软件中断等。在处理中断时，也会涉及到并发控制问题。例如，当中断处理程序访问内核共享资源时，可能会与正在运行的内核线程产生竞态条件。为了避免这种情况，内核通常会在进入中断处理程序时禁止本地中断，在离开中断处理程序时再重新启用中断。

下面是一个简单的示例代码，展示如何在中断处理程序中保护共享资源：

#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/interrupt.h>

static DEFINE_SPINLOCK(my_spinlock);
static int shared_variable = 0;

irqreturn_t my_interrupt_handler(int irq, void *dev_id) {
    unsigned long flags;
    spin_lock_irqsave(&my_spinlock, flags);
    // 访问和修改共享资源
    shared_variable++;
    spin_unlock_irqrestore(&my_spinlock, flags);
    return IRQ_HANDLED;
}

static int __init my_module_init(void) {
    int ret;
    ret = request_irq(IRQ_NUM, my_interrupt_handler, IRQF_SHARED, "my_irq", NULL);
    if (ret) {
        printk(KERN_ERR "Failed to request IRQ\n");
        return ret;
    }
    return 0;
}

static void __exit my_module_exit(void) {
    free_irq(IRQ_NUM, NULL);
}

module_init(my_module_init);
module_exit(my_module_exit);
MODULE_LICENSE("GPL");

在上述代码中，spin_lock_irqsave(&my_spinlock, flags) 会禁止本地中断并保存中断状态，spin_unlock_irqrestore(&my_spinlock, flags) 会恢复中断状态并释放自旋锁，从而保护了共享变量 shared_variable 在中断处理程序中的安全访问。

并发控制中的死锁问题

死锁是并发控制中一个严重的问题，它是指两个或多个进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。

死锁的产生条件

  1. 互斥条件：资源在某一时刻只能被一个进程或线程所占有。例如，打印机在同一时间只能被一个进程使用。
  2. 占有并等待条件：进程已经占有了一些资源，但又请求新的资源，并且在等待新资源的同时不释放已占有的资源。比如，进程A已经占有了资源R1，又请求资源R2，在等待R2的过程中不释放R1。
  3. 不可剥夺条件：资源只能由占有它的进程或线程主动释放，不能被其他进程或线程强行剥夺。例如，进程B占有了资源R3，其他进程不能直接从B手中夺走R3。
  4. 循环等待条件：存在一个进程或线程的循环链，链中的每个进程或线程都在等待下一个进程或线程所占有的资源。例如，进程C占有资源R4，进程D占有资源R5，进程C等待进程D释放R5，进程D等待进程C释放R4，形成循环等待。

死锁的检测与预防

  1. 死锁预防：通过破坏死锁产生的四个条件中的一个或多个来预防死锁。例如，采用资源分配图算法，当一个进程请求资源时，系统检查分配资源后是否会形成死锁，如果会则拒绝分配。
  2. 死锁检测：操作系统定期检查系统中是否存在死锁。一种常用的方法是通过资源分配图算法，找出系统中的循环等待关系，从而检测出死锁。当检测到死锁时，操作系统可以采取一些措施，如终止一个或多个进程，释放它们所占有的资源，以打破死锁。

下面是一个简单的死锁检测算法的伪代码：

// 假设资源分配图为G = (V, E)，V为顶点集（进程和资源），E为边集（请求边和分配边）
bool detect_deadlock() {
    // 初始化每个顶点的入度
    int inDegree[V.size()];
    for (int i = 0; i < V.size(); i++) {
        inDegree[i] = 0;
    }
    for (auto edge : E) {
        inDegree[edge.to]++;
    }
    std::queue<int> queue;
    for (int i = 0; i < V.size(); i++) {
        if (inDegree[i] == 0) {
            queue.push(i);
        }
    }
    int count = 0;
    while (!queue.empty()) {
        int vertex = queue.front();
        queue.pop();
        count++;
        for (auto edge : E) {
            if (edge.from == vertex) {
                inDegree[edge.to]--;
                if (inDegree[edge.to] == 0) {
                    queue.push(edge.to);
                }
            }
        }
    }
    return count != V.size();
}

在上述伪代码中，通过计算资源分配图中每个顶点的入度，将入度为0的顶点放入队列，然后不断从队列中取出顶点并更新其他顶点的入度。如果最终处理的顶点数不等于顶点总数，则说明存在死锁。

并发控制在现代操作系统中的应用

随着计算机硬件技术的发展，多核处理器已经成为主流，现代操作系统需要更加高效的并发控制机制来充分利用多核资源。

多核心处理器上的并发控制优化

现代操作系统在多核处理器上采用了多种优化策略。例如，对于自旋锁，操作系统会根据处理器核心数和锁的持有时间等因素，动态调整自旋的次数。如果处理器核心数较多且锁的持有时间较短，适当增加自旋次数可以减少线程上下文切换的开销；反之，如果锁的持有时间较长，则减少自旋次数，避免浪费CPU资源。

另外，操作系统还会采用一些更细粒度的锁机制，如分段锁（Segmented Lock）。在管理大型数据结构时，将数据结构分成多个段，每个段使用独立的锁进行保护。这样，不同的线程可以同时访问不同段的数据，提高并发性能。

分布式系统中的并发控制

在分布式系统中，多个节点通过网络连接，共同完成任务。分布式系统中的并发控制面临更多挑战，如网络延迟、节点故障等。为了保证数据一致性，分布式系统通常采用分布式锁、共识算法等机制。

例如，在使用Zookeeper实现分布式锁时，多个客户端可以通过在Zookeeper中创建临时顺序节点来竞争锁。Zookeeper会保证节点的顺序性，客户端通过判断自己创建的节点是否是最小序号的节点来决定是否获取到锁。如果没有获取到锁，则监听前一个序号的节点，当前一个节点被删除时，客户端会收到通知，然后再次尝试获取锁。

下面是一个简单的使用Zookeeper客户端库（如Curator）实现分布式锁的Java代码示例：

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class DistributedLockExample {
    private static final String ZOOKEEPER_SERVERS = "localhost:2181";
    private static final String LOCK_PATH = "/distributed_lock";

    public static void main(String[] args) {
        CuratorFramework client = CuratorFrameworkFactory.newClient(ZOOKEEPER_SERVERS, new ExponentialBackoffRetry(1000, 3));
        client.start();
        InterProcessMutex lock = new InterProcessMutex(client, LOCK_PATH);
        try {
            if (lock.acquire(5, java.util.concurrent.TimeUnit.SECONDS)) {
                try {
                    System.out.println("Acquired lock");
                    // 执行临界区代码
                } finally {
                    lock.release();
                    System.out.println("Released lock");
                }
            } else {
                System.out.println("Failed to acquire lock");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            client.close();
        }
    }
}

在上述代码中，InterProcessMutex 类提供了分布式锁的功能。lock.acquire(5, java.util.concurrent.TimeUnit.SECONDS) 尝试在5秒内获取锁，如果获取成功则执行临界区代码，最后通过 lock.release() 释放锁。

通过以上对操作系统并发控制的深入理解，我们可以更好地编写高效、正确的并发程序，充分利用现代计算机系统的多核资源，同时避免因并发访问带来的各种问题。无论是在单机系统还是分布式系统中，合理运用并发控制机制都是实现高性能、高可靠性应用的关键。