操作系统中的进程、线程与协程

1
/* 单调时钟，返回秒 */
2
static double now(void)
3
{
4
  struct timespec ts;
5
  clock_gettime(CLOCK_MONOTONIC, &ts);
6
  return (double)ts.tv_sec + (double)ts.tv_nsec * 1e-9;
7
}
8

9
/* fork + 子进程退出 + wait，除以次数 */
10
static double bench_fork(int n)
11
{
12
  double t0 = now();
13
  for (int i = 0; i < n; i++) {
14
    pid_t p = fork();
15
    if (p == 0)
16
      _exit(0);
17
    int st;
18
    waitpid(p, &st, 0);
19
  }
20
  return (now() - t0) / n;
21
}
22

23
/* 两个线程绑到同一个核，用两条管道来回传一个字节 */
24
static double bench_pp_thread(long n)
25
{
26
  if (pipe(pp_a) || pipe(pp_b)) {
27
    perror("pipe");
28
    exit(1);
29
  }
30
  pp_n = n;
31
  pthread_t t;
32
  pthread_create(&t, NULL, pp_peer, NULL);
33
  pin0(); /* 把自己也绑到 0 号核 */
34
  char c = 'x';
35
  double t0 = now();
36
  for (long i = 0; i < n; i++) {
37
    if (write(pp_a[1], &c, 1) != 1)
38
      break;
39
    if (read(pp_b[0], &c, 1) != 1)
40
      break;
41
  }
42
  double dt = now() - t0;
43
  pthread_join(t, NULL);
44
  return dt / n / 2.0; /* 一次往返 = 两次切换 */
45
}

1
#include <stdio.h>
2
#include <sys/wait.h>
3
#include <unistd.h>
4

5
int main(void)
6
{
7
  int x = 1;
8
  printf("fork 之前:  x=%d, pid=%d\n", x, getpid());
9
  fflush(stdout);
10
  pid_t p = fork();
11
  if (p == 0) {
12
    x = 42;
13
    printf("子进程里:   x=%d, pid=%d\n", x, getpid());
14
    fflush(stdout);
15
    _exit(0);
16
  }
17
  wait(NULL);
18
  printf("父进程里:   x=%d, pid=%d\n", x, getpid());
19
  return 0;
20
}

1
/* 从 /proc/self/status 里读一个 KB 单位的字段 */
2
static long vm(const char *field)
3
{
4
  FILE *f = fopen("/proc/self/status", "r");
5
  if (!f)
6
    return -1;
7
  char line[256];
8
  long v = -1;
9
  size_t fl = strlen(field);
10
  while (fgets(line, sizeof line, f)) {
11
    if (!strncmp(line, field, fl)) {
12
      sscanf(line + fl, "%ld", &v);
13
      break;
14
    }
15
  }
16
  fclose(f);
17
  return v;
18
}
19

20
int main(void)
21
{
22
  int NT = 2000;
23
  pthread_t *ts = malloc(sizeof(pthread_t) * NT);
24
  long v0 = vm("VmSize:"), r0 = vm("VmRSS:");
25
  /* waiter 阻塞在条件变量上 */
26
  for (int i = 0; i < NT; i++)
27
    pthread_create(&ts[i], NULL, waiter, NULL);
28
  long v1 = vm("VmSize:"), r1 = vm("VmRSS:");
29
  printf("vm_per_thread_kb,%.1f\n", (double)(v1 - v0) / NT);
30
  printf("rss_per_thread_kb,%.1f\n", (double)(r1 - r0) / NT);
31
  /* 唤醒 g_go 并 join 所有线程，略 */
32
  return 0;
33
}

1
#include <pthread.h>
2
#include <stdatomic.h>
3
#include <stdio.h>
4

5
/* volatile 强制每次都读内存、写内存，让读-改-写的窗口暴露出来 */
6
static volatile long plain = 0;
7
static atomic_long atom = 0;
8

9
static void *bump_plain(void *a)
10
{
11
  (void)a;
12
  for (long i = 0; i < 1000000; i++) {
13
    long v = plain;
14
    plain = v + 1;
15
  }
16
  return NULL;
17
}
18

19
static void *bump_atom(void *a)
20
{
21
  (void)a;
22
  for (long i = 0; i < 1000000; i++)
23
    atomic_fetch_add(&atom, 1);
24
  return NULL;
25
}

1
#define _XOPEN_SOURCE 700
2
#include <stdio.h>
3
#include <ucontext.h>
4

5
static ucontext_t main_ctx, a_ctx, b_ctx;
6
static char stk_a[65536], stk_b[65536];
7

8
static void co_a(void)
9
{
10
  for (;;) {
11
    printf("A");
12
    fflush(stdout);
13
    swapcontext(&a_ctx, &main_ctx); /* 暂停 A，回到 main */
14
  }
15
}
16

17
static void co_b(void)
18
{
19
  for (;;) {
20
    printf("B");
21
    fflush(stdout);
22
    swapcontext(&b_ctx, &main_ctx);
23
  }
24
}
25

26
int main(void)
27
{
28
  getcontext(&a_ctx);
29
  a_ctx.uc_stack.ss_sp = stk_a;
30
  a_ctx.uc_stack.ss_size = sizeof stk_a;
31
  a_ctx.uc_link = NULL;
32
  makecontext(&a_ctx, co_a, 0);
33

34
  getcontext(&b_ctx);
35
  b_ctx.uc_stack.ss_sp = stk_b;
36
  b_ctx.uc_stack.ss_size = sizeof stk_b;
37
  b_ctx.uc_link = NULL;
38
  makecontext(&b_ctx, co_b, 0);
39

40
  for (int i = 0; i < 4; i++) {
41
    swapcontext(&main_ctx, &a_ctx);
42
    swapcontext(&main_ctx, &b_ctx);
43
  }
44
  printf("\n");
45
  return 0;
46
}

1
#include <coroutine>
2

3
struct Gen {
4
  struct promise_type {
5
    int value;
6
    std::suspend_always yield_value(int v) {
7
      value = v;
8
      return {};
9
    }
10
    Gen get_return_object() {
11
      return Gen{std::coroutine_handle<promise_type>::from_promise(*this)};
12
    }
13
    std::suspend_always initial_suspend() noexcept { return {}; }
14
    std::suspend_always final_suspend() noexcept { return {}; }
15
    void return_void() {}
16
    void unhandled_exception() {}
17
  };
18
  std::coroutine_handle<promise_type> h;
19
  bool next() {
20
    h.resume();
21
    return !h.done();
22
  }
23
};
24

25
Gen gen(long n) {
26
  for (long i = 0; i < n; i++) co_yield static_cast<int>(i);
27
}

1
   6           RETURN_GENERATOR
2
               POP_TOP
3
       L1:     RESUME                   0
4

5
   7           LOAD_GLOBAL              0 (asyncio)
6
               LOAD_ATTR                2 (sleep)
7
               PUSH_NULL
8
               LOAD_SMALL_INT           0
9
               CALL                     1
10
               GET_AWAITABLE            0
11
               LOAD_CONST               1 (None)
12
       L2:     SEND                     3 (to L5)
13
       L3:     YIELD_VALUE              1
14
       L4:     RESUME                   3
15
               JUMP_BACKWARD_NO_INTERRUPT 5 (to L2)
16
       L5:     END_SEND
17
               POP_TOP

1
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
2

3
const RAW: RawWaker = RawWaker::new(std::ptr::null(), &VTABLE);
4
const VTABLE: RawWakerVTable = RawWakerVTable::new(|_| RAW, |_| {}, |_| {}, |_| {});
5

6
fn noop_waker() -> Waker {
7
    unsafe { Waker::from_raw(RAW) }
8
}
9

10
/// 最小执行器：不接任何运行时，轮询到完成
11
fn block_on<F: Future>(f: F) -> F::Output {
12
    let w = noop_waker();
13
    let mut cx = Context::from_waker(&w);
14
    let mut f = Box::pin(f);
15
    loop {
16
        match f.as_mut().poll(&mut cx) {
17
            Poll::Ready(v) => return v,
18
            Poll::Pending => std::hint::spin_loop(),
19
        }
20
    }
21
}

1
struct Spin {
2
    n: u64,
3
    target: u64,
4
}
5

6
impl Future for Spin {
7
    type Output = u64;
8
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u64> {
9
        self.n += 1;
10
        if self.n >= self.target {
11
            Poll::Ready(self.n)
12
        } else {
13
            cx.waker().wake_by_ref();
14
            Poll::Pending
15
        }
16
    }
17
}
18

19
let t0 = Instant::now();
20
block_on(Spin {
21
    n: 0,
22
    target: 100_000_000,
23
});
24
let per_poll = t0.elapsed().as_secs_f64() / 100_000_000.0;

1
#define _GNU_SOURCE
2
#include <pthread.h>
3
#include <sched.h>
4
#include <stdio.h>
5
#include <stdlib.h>
6
#include <string.h>
7
#include <sys/wait.h>
8
#include <time.h>
9
#include <unistd.h>
10

11
static double now(void)
12
{
13
  struct timespec ts;
14
  clock_gettime(CLOCK_MONOTONIC, &ts);
15
  return (double)ts.tv_sec + (double)ts.tv_nsec * 1e-9;
16
}
17

18
/* 把自己绑到 0 号核，减少迁移噪声 */
19
static void pin0(void)
20
{
21
  cpu_set_t s;
22
  CPU_ZERO(&s);
23
  CPU_SET(0, &s);
24
  sched_setaffinity(0, sizeof s, &s);
25
}
26

27
static void *nop(void *a)
28
{
29
  (void)a;
30
  return NULL;
31
}
32

33
static double bench_fork(int n)
34
{
35
  double t0 = now();
36
  for (int i = 0; i < n; i++) {
37
    pid_t p = fork();
38
    if (p == 0)
39
      _exit(0);
40
    int st;
41
    waitpid(p, &st, 0);
42
  }
43
  return (now() - t0) / n;
44
}
45

46
static double bench_thread(int n)
47
{
48
  double t0 = now();
49
  for (int i = 0; i < n; i++) {
50
    pthread_t t;
51
    pthread_create(&t, NULL, nop, NULL);
52
    pthread_join(t, NULL);
53
  }
54
  return (now() - t0) / n;
55
}
56

57
/* two threads, same core, handing off one byte each way */
58
static int pp_a[2], pp_b[2];
59
static long pp_n;
60

61
static void *pp_peer(void *a)
62
{
63
  (void)a;
64
  pin0();
65
  char c;
66
  for (long i = 0; i < pp_n; i++) {
67
    if (read(pp_a[0], &c, 1) != 1)
68
      break;
69
    if (write(pp_b[1], &c, 1) != 1)
70
      break;
71
  }
72
  return NULL;
73
}
74

75
static double bench_pp_thread(long n)
76
{
77
  if (pipe(pp_a) || pipe(pp_b)) {
78
    perror("pipe");
79
    exit(1);
80
  }
81
  pp_n = n;
82
  pthread_t t;
83
  pthread_create(&t, NULL, pp_peer, NULL);
84
  pin0();
85
  char c = 'x';
86
  double t0 = now();
87
  for (long i = 0; i < n; i++) {
88
    if (write(pp_a[1], &c, 1) != 1)
89
      break;
90
    if (read(pp_b[0], &c, 1) != 1)
91
      break;
92
  }
93
  double dt = now() - t0;
94
  pthread_join(t, NULL);
95
  close(pp_a[0]);
96
  close(pp_a[1]);
97
  close(pp_b[0]);
98
  close(pp_b[1]);
99
  return dt / n / 2.0; /* two switches per round trip */
100
}
101

102
static double bench_pp_proc(long n)
103
{
104
  if (pipe(pp_a) || pipe(pp_b)) {
105
    perror("pipe");
106
    exit(1);
107
  }
108
  pid_t pid = fork();
109
  if (pid == 0) {
110
    pin0();
111
    char c;
112
    for (long i = 0; i < n; i++) {
113
      if (read(pp_a[0], &c, 1) != 1)
114
        break;
115
      if (write(pp_b[1], &c, 1) != 1)
116
        break;
117
    }
118
    _exit(0);
119
  }
120
  pin0();
121
  char c = 'x';
122
  double t0 = now();
123
  for (long i = 0; i < n; i++) {
124
    if (write(pp_a[1], &c, 1) != 1)
125
      break;
126
    if (read(pp_b[0], &c, 1) != 1)
127
      break;
128
  }
129
  double dt = now() - t0;
130
  int st;
131
  waitpid(pid, &st, 0);
132
  close(pp_a[0]);
133
  close(pp_a[1]);
134
  close(pp_b[0]);
135
  close(pp_b[1]);
136
  return dt / n / 2.0;
137
}
138

139
/* 从 /proc/self/status 里读一个 KB 单位的字段 */
140
static long vm(const char *field)
141
{
142
  FILE *f = fopen("/proc/self/status", "r");
143
  if (!f)
144
    return -1;
145
  char line[256];
146
  long v = -1;
147
  size_t fl = strlen(field);
148
  while (fgets(line, sizeof line, f)) {
149
    if (!strncmp(line, field, fl)) {
150
      sscanf(line + fl, "%ld", &v);
151
      break;
152
    }
153
  }
154
  fclose(f);
155
  return v;
156
}
157

158
static pthread_mutex_t g_mtx = PTHREAD_MUTEX_INITIALIZER;
159
static pthread_cond_t g_cv = PTHREAD_COND_INITIALIZER;
160
static int g_go = 0;
161

162
static void *waiter(void *a)
163
{
164
  (void)a;
165
  pthread_mutex_lock(&g_mtx);
166
  while (!g_go)
167
    pthread_cond_wait(&g_cv, &g_mtx);
168
  pthread_mutex_unlock(&g_mtx);
169
  return NULL;
170
}
171

172
static double best3(double (*fn)(int), int n)
173
{
174
  double b = 1e18;
175
  for (int r = 0; r < 3; r++) {
176
    double v = fn(n);
177
    if (v < b)
178
      b = v;
179
  }
180
  return b;
181
}
182

183
int main(void)
184
{
185
  bench_fork(50);
186
  bench_thread(200);
187

188
  printf("nproc,%ld\n", sysconf(_SC_NPROCESSORS_ONLN));
189
  printf("fork_us,%.2f\n", best3(bench_fork, 2000) * 1e6);
190
  printf("pthread_create_join_us,%.2f\n",
191
         best3(bench_thread, 20000) * 1e6);
192
  printf("ctx_switch_thread_us,%.3f\n", bench_pp_thread(200000) * 1e6);
193
  printf("ctx_switch_process_us,%.3f\n", bench_pp_proc(50000) * 1e6);
194

195
  pthread_attr_t attr;
196
  pthread_attr_init(&attr);
197
  size_t ss = 0;
198
  pthread_attr_getstacksize(&attr, &ss);
199
  printf("pthread_default_stack_kb,%zu\n", ss / 1024);
200

201
  int NT = 2000;
202
  long v0 = vm("VmSize:"), r0 = vm("VmRSS:");
203
  pthread_t *ts = malloc(sizeof(pthread_t) * NT);
204
  for (int i = 0; i < NT; i++)
205
    pthread_create(&ts[i], NULL, waiter, NULL);
206
  long v1 = vm("VmSize:"), r1 = vm("VmRSS:");
207
  printf("threads,%d\n", NT);
208
  printf("vm_delta_per_thread_kb,%.1f\n", (double)(v1 - v0) / NT);
209
  printf("rss_delta_per_thread_kb,%.1f\n", (double)(r1 - r0) / NT);
210
  pthread_mutex_lock(&g_mtx);
211
  g_go = 1;
212
  pthread_cond_broadcast(&g_cv);
213
  pthread_mutex_unlock(&g_mtx);
214
  for (int i = 0; i < NT; i++)
215
    pthread_join(ts[i], NULL);
216
  free(ts);
217
  return 0;
218
}

1
#include <chrono>
2
#include <coroutine>
3
#include <cstdint>
4
#include <cstdio>
5
#include <cstdlib>
6
#include <cstring>
7
#include <thread>
8
#include <vector>
9

10
static double now(void) {
11
  using namespace std::chrono;
12
  return duration<double>(steady_clock::now().time_since_epoch()).count();
13
}
14

15
static long vm(const char* field) {
16
  FILE* f = fopen("/proc/self/status", "r");
17
  if (!f) return -1;
18
  char line[256];
19
  long v = -1;
20
  size_t fl = strlen(field);
21
  while (fgets(line, sizeof line, f)) {
22
    if (!strncmp(line, field, fl)) {
23
      sscanf(line + fl, "%ld", &v);
24
      break;
25
    }
26
  }
27
  fclose(f);
28
  return v;
29
}
30

31
static double bench_std_thread(int n) {
32
  double t0 = now();
33
  for (int i = 0; i < n; i++) {
34
    std::thread t([]() {});
35
    t.join();
36
  }
37
  return (now() - t0) / n;
38
}
39

40
/* 最小的无栈生成器；每次 next() 恢复一次协程帧 */
41
struct Gen {
42
  struct promise_type {
43
    int value;
44
    std::suspend_always yield_value(int v) {
45
      value = v;
46
      return {};
47
    }
48
    Gen get_return_object() {
49
      return Gen{std::coroutine_handle<promise_type>::from_promise(*this)};
50
    }
51
    std::suspend_always initial_suspend() noexcept { return {}; }
52
    std::suspend_always final_suspend() noexcept { return {}; }
53
    void return_void() {}
54
    void unhandled_exception() {}
55
  };
56
  std::coroutine_handle<promise_type> h;
57
  bool next() {
58
    h.resume();
59
    return !h.done();
60
  }
61
};
62

63
static Gen gen(long n) {
64
  for (long i = 0; i < n; i++) co_yield static_cast<int>(i);
65
}
66

67
int main(void) {
68
  bench_std_thread(200);
69
  printf("cpp_std_thread_create_join_us,%.2f\n", bench_std_thread(20000) * 1e6);
70

71
  const long N = 100000000L;
72
  double t0 = now();
73
  long c = 0;
74
  Gen g = gen(N);
75
  while (g.next()) c++;
76
  double t1 = now();
77
  printf("cpp_coro_resumes,%ld\n", c);
78
  printf("cpp_coro_resume_ns,%.2f\n", (t1 - t0) / N * 1e9);
79

80
  volatile long s = 0;
81
  double t2 = now();
82
  for (long i = 0; i < N; i++) s += i;
83
  double t3 = now();
84
  printf("cpp_plain_loop_ns,%.2f\n", (t3 - t2) / N * 1e9);
85

86
  /* 大量挂起的协程帧：在堆上，不占线程栈 */
87
  long r0 = vm("VmRSS:");
88
  std::vector<std::coroutine_handle<Gen::promise_type>> v;
89
  v.reserve(100000);
90
  for (int i = 0; i < 100000; i++) v.push_back(gen(10).h);
91
  long r1 = vm("VmRSS:");
92
  printf("cpp_suspended_coroutines,100000\n");
93
  printf("cpp_frame_bytes_est,%.1f\n", (r1 - r0) * 1024.0 / 100000);
94
  for (auto h : v) h.destroy();
95
  return 0;
96
}

1
import asyncio
2
import multiprocessing
3
import os
4
import sys
5
import threading
6
import time
7
from collections.abc import Callable, Iterator
8

9
N: int = 20_000_000
10

11

12
def now() -> float:
13
    return time.perf_counter()
14

15

16
def best3(fn: Callable[..., float], *args: int) -> float:
17
    best = float("inf")
18
    for _ in range(3):
19
        best = min(best, fn(*args))
20
    return best
21

22

23
def thread_create_join(n: int) -> float:
24
    def nop() -> None:
25
        pass
26

27
    t0 = now()
28
    for _ in range(n):
29
        t = threading.Thread(target=nop)
30
        t.start()
31
        t.join()
32
    return (now() - t0) / n
33

34

35
def burn(n: int) -> int:
36
    s = 0
37
    for i in range(n):
38
        s += i * i
39
    return s
40

41

42
def gil_one() -> float:
43
    t0 = now()
44
    burn(N)
45
    return now() - t0
46

47

48
def gil_two_threads() -> float:
49
    t0 = now()
50
    ts = [threading.Thread(target=burn, args=(N,)) for _ in range(2)]
51
    for t in ts:
52
        t.start()
53
    for t in ts:
54
        t.join()
55
    return now() - t0
56

57

58
def gil_two_procs() -> float:
59
    t0 = now()
60
    ps = [multiprocessing.Process(target=burn, args=(N,)) for _ in range(2)]
61
    for p in ps:
62
        p.start()
63
    for p in ps:
64
        p.join()
65
    return now() - t0
66

67

68
async def _step() -> int:
69
    return 1
70

71

72
async def _tight(n: int) -> None:
73
    for _ in range(n):
74
        await asyncio.sleep(0)
75

76

77
async def _gather(n: int) -> None:
78
    await asyncio.gather(*(_step() for _ in range(n)))
79

80

81
def asyncio_tasks(n: int) -> float:
82
    t0 = now()
83
    asyncio.run(_gather(n))
84
    return now() - t0
85

86

87
def gen_resume(n: int) -> float:
88
    def gen(k: int) -> Iterator[int]:
89
        for i in range(k):
90
            yield i
91

92
    g = gen(n)
93
    t0 = now()
94
    c = 0
95
    for _ in g:
96
        c += 1
97
    return (now() - t0) / c
98

99

100
def main() -> None:
101
    thread_create_join(200)
102
    print("python,cpu_count,%d" % (os.cpu_count() or 1))
103
    print("python,switch_interval_s,%.4f" % sys.getswitchinterval())
104
    print(
105
        "python,thread_create_join_us,%.2f"
106
        % (best3(thread_create_join, 20000) * 1e6)
107
    )
108

109
    one = best3(gil_one)
110
    two = best3(gil_two_threads)
111
    mp = best3(gil_two_procs)
112
    print("python,burn_1thread_s,%.3f" % one)
113
    print("python,burn_2threads_s,%.3f" % two)
114
    print("python,burn_2procs_s,%.3f" % mp)
115
    print("python,gil_speedup_2threads,%.2fx" % (one * 2 / two))
116
    print("python,mp_speedup_2procs,%.2fx" % (one * 2 / mp))
117

118
    asyncio.run(_tight(1000))
119
    ns = 2_000_000
120
    t0 = now()
121
    asyncio.run(_tight(ns))
122
    dt = now() - t0
123
    print("python,asyncio_await_switch_ns,%.1f" % (dt / ns * 1e9))
124

125
    nt = 200_000
126
    dt = asyncio_tasks(nt)
127
    print("python,asyncio_task_create_run_us,%.3f" % (dt / nt * 1e6))
128

129
    print(
130
        "python,generator_resume_ns,%.1f"
131
        % (best3(gen_resume, 50_000_000) * 1e9)
132
    )
133

134

135
if __name__ == "__main__":
136
    main()