05 - 操作系统基础¶

# Python多进程示例
import multiprocessing
import os
import time

def worker(name):
    print(f"Worker {name} started, PID: {os.getpid()}")
    time.sleep(2)
    print(f"Worker {name} finished")

if __name__ == "__main__":
    print(f"Main process PID: {os.getpid()}")

    # 创建进程
    processes = []
    for i in range(3):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()

    print("All workers finished")

# Python多线程示例
import threading
import time

def worker(name):
    print(f"Thread {name} started")
    time.sleep(2)
    print(f"Thread {name} finished")

# 创建线程
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("All threads finished")

# Python内存管理示例
import sys

# 查看对象内存占用
a = [1, 2, 3, 4, 5]
print(f"List size: {sys.getsizeof(a)} bytes")

# 内存池机制
# Python使用pymalloc管理小内存（<512字节）
# 大内存直接调用C的malloc

# 垃圾回收
import gc

# 查看垃圾回收器状态
print(gc.get_count())  # (年轻代, 中年代, 老年代)

# 手动触发垃圾回收
gc.collect()

# 循环引用示例
class Node:
    def __init__(self):
        self.ref = None

a = Node()
b = Node()
a.ref = b
b.ref = a

# 即使删除引用，循环引用对象也不会被立即回收
del a
del b

# 需要垃圾回收器检测循环引用
gc.collect()

# 简单的LRU缓存实现（基于操作系统页面置换思想）
from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity):
        self.capacity = capacity
        self.cache = OrderedDict()

    def get(self, key):
        if key not in self.cache:
            return -1
        # 移动到末尾（最近使用）
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        # 超出容量，移除最久未使用的
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

# 使用示例
cache = LRUCache(2)
cache.put(1, 1)
cache.put(2, 2)
print(cache.get(1))    # 返回 1
cache.put(3, 3)        # 移除 key 2
print(cache.get(2))    # 返回 -1

import os
import mmap

# 1. 普通文件IO
with open('test.txt', 'w') as f:  # with自动管理资源，确保文件正确关闭
    f.write('Hello World')

# 2. 内存映射文件（mmap）
# 适合大文件随机访问
with open('large_file.bin', 'r+b') as f:
    # 内存映射
    mmapped = mmap.mmap(f.fileno(), 0)
    # 像操作内存一样操作文件
    print(mmapped[:10])
    mmapped.close()

# 3. 零拷贝（Zero-copy）
# 使用sendfile系统调用，数据不经过用户空间
import socket

sock = socket.socket()
# 在Linux上可以使用sendfile
# sock.sendfile(open('file.bin', 'rb'))

# select/poll/epoll示例
import select
import socket

# 创建服务器socket
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('localhost', 8080))
server.listen(5)
server.setblocking(False)

# 使用select实现IO多路复用
inputs = [server]
outputs = []

while inputs:
    # 等待IO事件
    readable, writable, exceptional = select.select(inputs, outputs, inputs)

    for s in readable:
        if s is server:
            # 新连接
            connection, client_address = s.accept()
            connection.setblocking(False)
            inputs.append(connection)
        else:
            # 接收数据
            data = s.recv(1024)
            if data:
                # 处理数据
                pass
            else:
                # 连接关闭
                inputs.remove(s)
                s.close()

# 1. 管道 (Pipe)
import multiprocessing

def sender(conn):
    conn.send("Hello from sender")
    conn.close()

def receiver(conn):
    msg = conn.recv()
    print(f"Received: {msg}")

parent_conn, child_conn = multiprocessing.Pipe()
p1 = multiprocessing.Process(target=sender, args=(child_conn,))
p2 = multiprocessing.Process(target=receiver, args=(parent_conn,))

p1.start()
p2.start()
p1.join()
p2.join()

# 2. 共享内存
from multiprocessing import Process, Value, Array

def modify_shared(n, arr):
    n.value = 3.14159
    for i in range(len(arr)):
        arr[i] = -arr[i]

num = Value('d', 0.0)  # 共享双精度浮点数
arr = Array('i', range(10))  # 共享整数数组

p = Process(target=modify_shared, args=(num, arr))
p.start()
p.join()

print(num.value)
print(arr[:])

# 3. 消息队列
from multiprocessing import Queue

def worker(q):
    q.put("Message from worker")

q = Queue()
p = Process(target=worker, args=(q,))
p.start()
print(q.get())  # 阻塞等待消息
p.join()

import threading  # 线程池/多线程：并发执行任务
import time

# 1. 锁 (Lock)
lock = threading.Lock()
counter = 0

def increment():
    global counter
    for _ in range(100000):
        with lock:  # 获取锁
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Counter: {counter}")  # 应该是1000000

# 2. 信号量 (Semaphore)
# 控制同时访问的线程数量
semaphore = threading.Semaphore(3)  # 最多3个线程同时访问

def limited_resource():
    with semaphore:
        print(f"{threading.current_thread().name} accessing resource")
        time.sleep(1)

# 3. 条件变量 (Condition)
# 用于线程间协调
condition = threading.Condition()
shared_data = []

def producer():
    with condition:
        shared_data.append("data")
        condition.notify_all()  # 通知等待的线程

def consumer():
    with condition:
        while not shared_data:
            condition.wait()  # 等待数据
        data = shared_data.pop()
        print(f"Consumed: {data}")

# 4. 读写锁 (RLock)
# 允许多个读线程，但写线程独占
rw_lock = threading.RLock()

import multiprocessing
import time
from queue import Empty

class ProcessPool:
    """简单的进程池实现"""

    def __init__(self, num_workers):
        self.num_workers = num_workers
        self.task_queue = multiprocessing.Queue()
        self.result_queue = multiprocessing.Queue()
        self.workers = []

    def worker_func(self):
        """工作进程函数"""
        while True:
            try:  # try/except捕获异常
                task = self.task_queue.get(timeout=1)
                if task is None:  # 结束信号
                    break
                func, args = task
                result = func(*args)  # *args接收任意位置参数；**kwargs接收任意关键字参数
                self.result_queue.put(result)
            except Empty:
                continue

    def start(self):
        """启动工作进程"""
        for _ in range(self.num_workers):
            p = multiprocessing.Process(target=self.worker_func)
            p.start()
            self.workers.append(p)

    def submit(self, func, args=()):
        """提交任务"""
        self.task_queue.put((func, args))

    def get_result(self, timeout=None):
        """获取结果"""
        return self.result_queue.get(timeout=timeout)

    def stop(self):
        """停止所有工作进程"""
        for _ in self.workers:
            self.task_queue.put(None)
        for w in self.workers:
            w.join()

# 使用示例
def square(n):
    time.sleep(1)
    return n * n

pool = ProcessPool(4)
pool.start()

for i in range(10):
    pool.submit(square, (i,))

for _ in range(10):
    print(pool.get_result())

pool.stop()

import tracemalloc
import sys

def analyze_memory():
    """分析内存使用情况"""
    # 开始跟踪内存分配
    tracemalloc.start()

    # 执行一些操作
    data = [i for i in range(100000)]

    # 获取内存分配快照
    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.statistics('lineno')

    print("Top 10 memory allocations:")
    for stat in top_stats[:10]:  # 切片操作：[start:end:step]提取子序列
        print(stat)

    tracemalloc.stop()

# 对象引用计数分析
import gc

def analyze_references():
    """分析对象引用关系"""
    class Node:
        def __init__(self, name):
            self.name = name
            self.ref = None

    a = Node("A")
    b = Node("B")
    a.ref = b
    b.ref = a

    # 获取引用对象
    print(f"Referrers of a: {gc.get_referrers(a)}")
    print(f"Referents of a: {gc.get_referents(a)}")