在认清生活的真相之后依然热爱生活

1
  template <Writer<T> W> [[gnu::hot]] bool push_with_writer(W writer) noexcept {
2
    const size_t head = prod_.head.load(std::memory_order_relaxed);
3
    const size_t next_head = nextIndex(head);
4

5
    if (next_head == prod_.cached_tail) [[unlikely]] {
6
      prod_.cached_tail = cons_.tail.load(load_order);
7
      if (next_head == prod_.cached_tail) [[unlikely]] {
8
        return false;
9
      }
10
    }
11

12
#ifdef ENABLE_PREFETCH
13
    __builtin_prefetch(&buffer_[head], 1, 1);
14
#endif
15

16
    writer(&buffer_[head]);
17
    prod_.head.store(next_head, store_order);
18
    return true;
19
  }

1
    // 简单的完美转发模板
2
  template <typename U>
3
  bool push(U &&value) noexcept
4
    requires(!Writer<U, T>)
5
  {
6
    return push_with_writer([value = std::forward<U>(value)](T *buffer) {
7
        *buffer = std::move(value);
8
    });
9
  }
10
    // 直接使用writer版本
11
  template <typename U>
12
  bool push(U writer) noexcept
13
    requires(Writer<U, T>)
14
  {
15
    return push_with_writer(writer);
16
  }

1
  template <Reader<T> R> [[gnu::hot]] bool pop(R reader) noexcept {
2
    const size_t tail = cons_.tail.load(std::memory_order_relaxed);
3

4
    if (tail == cons_.cached_head) [[unlikely]] {
5
      cons_.cached_head = prod_.head.load(load_order);
6
      if (tail == cons_.cached_head) [[unlikely]] {
7
        return false;
8
      }
9
    }
10

11
    // 让 reader 直接读取缓冲区位置
12
    reader(&buffer_[tail]);
13
    cons_.tail.store(nextIndex(tail), store_order);
14
    return true;
15
  }
16

17
  bool pop(T &value) noexcept {
18
    return pop([&value](const T *buffer) { value = *buffer; });
19
  }

1
  // 批量 Writer 操作
2
  template <BulkWriter<T> W>
3
  size_t push_bulk(W writer, size_t max_count) noexcept {
4
    const size_t head = prod_.head.load(std::memory_order_relaxed);
5
    size_t tail;
6

7
    tail = prod_.cached_tail;
8

9
    size_t available = available_space(head, tail);
10
    if (available < max_count) [[unlikely]] {
11
      tail = cons_.tail.load(load_order);
12
      prod_.cached_tail = tail;
13
      available = available_space(head, tail);
14
    }
15

16
    const size_t can_write = std::min(max_count, available);
17
    if (can_write == 0) [[unlikely]] {
18
      return 0;
19
    }
20

21
    const size_t end_of_buffer = Capacity - head;
22

23
    if (can_write <= end_of_buffer) [[likely]] {
24
      // 预取
25
      T *dest = &buffer_[head];
26
      for (size_t i = 0; i < can_write; i += CacheLineSize / sizeof(T)) {
27
        __builtin_prefetch(dest + i + CacheLineSize / sizeof(T), 1, 0);
28
      }
29
      // 连续写入
30
      writer(&buffer_[head], can_write, 0);
31
    } else {
32
      // 分两段写入
33
      const size_t part1 = end_of_buffer;
34
      writer(&buffer_[head], part1, 0);
35
      const size_t part2 = can_write - part1;
36
      writer(&buffer_[0], part2, part1);
37
    }
38
    prod_.head.store(nextIndex(head, can_write), store_order);
39

40
    return can_write;
41
  }
42

43
  size_t push_bulk(const T *data, size_t count) noexcept {
44
    return push_bulk(
45
        [data](T *buffer, size_t n, size_t offset) {
46
          std::memcpy(buffer, data + offset, n * sizeof(T));
47
          return n;
48
        },
49
        count);
50
  }
51

52
  // 批量 Reader 操作
53
  template <BulkReader<T> R>
54
  size_t pop_bulk(R reader, size_t max_count) noexcept {
55
    const size_t tail = cons_.tail.load(std::memory_order_relaxed);
56
    size_t head;
57

58
    head = cons_.cached_head;
59

60
    size_t available = available_space(tail, head + 1);
61
    if (available < max_count) [[unlikely]] {
62
      head = prod_.head.load(load_order);
63
      cons_.cached_head = head;
64
      available = available_space(tail, head + 1);
65
    }
66

67
    const size_t can_read = std::min(max_count, available);
68
    if (can_read == 0) [[unlikely]] {
69
      return 0;
70
    }
71

72
    const size_t end_of_buffer = Capacity - tail;
73

74
    if (can_read <= end_of_buffer) [[likely]] {
75
      // 预取
76
      const T *src = &buffer_[tail];
77
      for (size_t i = 0; i < can_read; i += CacheLineSize / sizeof(T)) {
78
        __builtin_prefetch(src + i + CacheLineSize / sizeof(T), 0, 0);
79
      }
80
      // 连续读取
81
      reader(&buffer_[tail], can_read, 0);
82
    } else {
83
      // 分两段读取
84
      const size_t part1 = end_of_buffer;
85
      reader(&buffer_[tail], part1, 0);
86
      const size_t part2 = can_read - part1;
87
      reader(&buffer_[0], part2, part1);
88
    }
89
    if (can_read > 0) {
90
      cons_.tail.store(nextIndex(tail, can_read), store_order);
91
    }
92
    return can_read;
93
  }
94

95
  size_t pop_bulk(T *data, size_t count) noexcept {
96
    return pop_bulk(
97
        [data](const T *buffer, size_t n, size_t offset) {
98
          std::memcpy(data + offset, buffer, n * sizeof(T));
99
          return n;
100
        },
101
        count);
102
  }

1
// 单线程performance
2
template <typename Q> void test_performance_spsc_(Q &q, const size_t N) {
3
  for (size_t i = 0; i < N; ++i) {
4
    while (!q.push(i)) {
5
    }
6
    int v;
7
    while (!q.pop(v)) {
8
    }
9
  }
10
}
11

12
// 多线程performance(不可避免的有创建线程的开销，我们通过足够大的N来减少这个影响)
13
template <typename Q> void test_performance_mt_(Q &q, const size_t N) {
14
  std::atomic<bool> done{false};
15
  std::thread prod([&] {
16
    for (int i = 0; i < N; ++i) {
17
      while (!q.push(i))
18
        std::this_thread::yield();
19
    }
20
    done = true;
21
  });
22

23
  std::thread cons([&] {
24
    int v, cnt = 0;
25
    while (!done || cnt < N) {
26
      if (q.pop(v))
27
        ++cnt;
28
      else
29
        std::this_thread::yield();
30
    }
31
  });
32

33
  prod.join();
34
  cons.join();
35
}
36
// 多线程bulk操作测试
37
template <size_t BATCH, typename Q>
38
void test_bulk_multithread_(Q &q, const size_t N) {
39
  std::atomic<bool> done{false};
40

41
  std::thread prod([&] {
42
    int arr[BATCH];
43
    for (int i = 0; i < N; i += BATCH) {
44
      int this_batch = std::min(BATCH, N - i);
45
      for (int j = 0; j < this_batch; ++j)
46
        arr[j] = i + j;
47
      size_t pushed = 0;
48
      while (pushed < this_batch)
49
        pushed += q.push_bulk(arr + pushed, this_batch - pushed);
50
    }
51
    done = true;
52
  });
53

54
  std::thread cons([&] {
55
    int arr[BATCH];
56
    int cnt = 0;
57
    while (!done || cnt < N) {
58
      size_t popped = q.pop_bulk(arr, BATCH);
59
      for (size_t j = 0; j < popped; ++j) {
60
        ASSERT_EQ(arr[j], cnt);
61
        ++cnt;
62
      }
63
      if (popped == 0)
64
        std::this_thread::yield();
65
    }
66
  });
67

68
  prod.join();
69
  cons.join();
70
}