__sync_xxx 系列函数似乎我的理解有很大问题

zylthinking

以下代码铁定异常， 概率很大， 但我目前理解理解不了为什么

1. 
   
2. #include <stddef.h>
   
3. 
   
4. struct lognode {
   
5.     const char* oplog;
   
6.     size_t bytes;
   
7.     struct lognode* next;
   
8. };
   
9. 
   
10. static struct lognode root = {0};
    
11. static struct lognode* head = &root;
    
12. static struct lognode* tail = &root;
    
13. 
    
14. void lognode_put(struct lognode* node)
    
15. {
    
16.     struct lognode* before = NULL;
    
17.     struct lognode* last;
    
18.     do {
    
19.         last = tail;
    
20.         struct lognode* before = __sync_val_compare_and_swap(&(last->next), NULL, node);
    
21.     } while (before != NULL);
    
22. 
    
23.     __sync_val_compare_and_swap(&tail, last, node);
    
24. }
    
25. 
    
26. struct lognode* lognode_get()
    
27. {
    
28.     struct lognode* node = head->next;
    
29.     head->next = NULL;
    
30.     //tail = head;
    
31.     __sync_lock_test_and_set(&tail, head);
    
32.     return node;
    
33. }
    
34. 
    
35. 
    
36. #include <stdio.h>
    
37. #include <assert.h>
    
38. #include <pthread.h>
    
39. 
    
40. #define nrp 10000
    
41. 
    
42. void* writer(void* any)
    
43. {
    
44.     int n = nrp;
    
45.     for (int i = 0; i < n; ++i) {
    
46.         struct lognode* nodep = (struct lognode *) malloc(sizeof(struct lognode));
    
47.         assert(nodep != NULL);
    
48.         lognode_put(nodep);
    
49.         //usleep(100);
    
50.     }   
    
51. 
    
52.     printf("done\n");
    
53.     return NULL;
    
54. }
    
55. 
    
56. 
    
57. int main(int argc, char** argv)
    
58. {
    
59.     int n = 0;
    
60.     for (int i = 0; i < 1; ++i) {
    
61.         pthread_t tid;
    
62.         if(0 == pthread_create(&tid, NULL, writer, NULL)) {
    
63.             ++n;
    
64.         }   
    
65.     }   
    
66. 
    
67.     int nr = nrp * n;
    
68.     printf("%d writer, nr = %d\n", n, nr);
    
69.         
    
70.     while (nr > 0) {
    
71.         struct lognode* nodep = lognode_get();
    
72.         struct lognode* cur = nodep;
    
73.         while (cur != NULL) {
    
74.             nodep = nodep->next;
    
75.             free(cur);
    
76.             cur = nodep;
    
77.             --nr;
    
78.         }   
    
79.         if (nr > 100000 && 0 == nr % 100000) {
    
80.             printf("%d\n", nr);
    
81.         } else if (nr > 10000 && 0 == nr % 10000) {
    
82.             printf("%d\n", nr);
    
83.         } else if (nr > 1000 && 0 == n % 1000) {
    
84.             printf("%d\n", nr);
    
85.         }   
    
86.     }   
    
87. 
    
88.     return 0;
    
89. }
    
90.   
    

复制代码

zhouzhenghui

如果没有理解错，楼主的代码实际上是想实现一个类似半工状态的无锁列表，put操作也就是生产者添加节点到列表尾，get操作也就是消费者把列表整个取下，再等新的列表建立，如此循环。

其中put操作，示例中代码只有一个生产者，但似乎有意使用多个生产者。多个会有竞争，一个的话因为get操作的方式，也仍然会跟消费者竞争。尾节点附加操作，竞争失败循环，看上去似乎没问题，但实际上存在取tail->next操作，这个不是原子操作的一部分。并且成功后替换tail指针仍然存在竞争（多个生产者情况下）。也就是说附加尾节点与替换尾指针整个不是原子操作，并且没有避开多处潜在竞争问题，所以是错误的。

继续看get的操作，示例代码的消费者似乎只有一个，并且对tail的操作是原子化的。但如上述，因为修改tail的操作会跟put的第一步操作取tail->next以及修改tail->next这个不是原子化的操作存有竞争，所以非常错误。另外据文档显示这里使用的__sync_lock_test_and_set可能会有问题，而且一般也不是必要的（对于简单赋值操作，必要的是内存栏栅），不过x86应该是可以的。

可以看看这篇文章http://coolshell.cn/articles/8239.html，注意后面的回复。

irp

1, #45, malloc之后，nodep->next应该初始化，否则#73,74就会crash, 显然你expect next == NULL是结束条件，但是lognode_put仅仅queue这个node，没有nullize next

w_anthony

我觉得具体逻辑先不说，但是19行的before和15、20行的before明显不是同一个

zylthinking

zylthinking 发表于 2013-08-15 19:42
以下代码铁定异常， 概率很大， 但我目前理解理解不了为什么

研究了一天， 现在如果只有  put 没有 get 的情况下， 可以实现正确写了， 但若存在 get, 则会出问题， 似乎哪里还是有问题， 继续研究

1. 
   
2. #define  rmb()  __asm__ volatile("lfence":::"memory")
   
3. #define  wmb()  __asm__ volatile("sfence":::"memory")
   
4. 
   
5. struct lognode {
   
6.     struct lognode* next;
   
7. };
   
8. 
   
9. static struct lognode root = {0};
   
10. static struct lognode* head = &root;
    
11. static struct lognode** tail = &(root.next);
    
12. 
    
13. void lognode_put(struct lognode* node)
    
14. {
    
15.     while (node != __sync_val_compare_and_swap(tail, NULL, node));
    
16. 
    
17.     struct lognode** last = tail;
    
18.     rmb();
    
19.     if (*last != node) {
    
20.         return;
    
21.     }
    
22. 
    
23.     __sync_bool_compare_and_swap(&tail, last, &(node->next));
    
24. }
    
25. 
    
26. struct lognode* lognode_get()
    
27. {
    
28.     struct lognode* node = head->next;
    
29.     head->next = NULL;
    
30.     wmb();
    
31.     //tail = &(root.next);
    
32.     __sync_lock_test_and_set(&tail, &(root.next));
    
33.     return node;
    
34. }

复制代码

测试代码

1. 
   
2. #define nrp 10000
   
3. #define nrt 1000
   
4. 
   
5. static volatile int run = 0;
   
6. static struct lognode* pool = NULL;
   
7. static volatile int indx = 0;
   
8. 
   
9. void* writer(void* any)
   
10. {
    
11.     while (run == 0) {
    
12.         usleep(100);
    
13.     }
    
14. 
    
15.     for (int i = 0; i < nrp; ++i) {
    
16.         int idx = __sync_fetch_and_add(&indx, 1);
    
17.         struct lognode* nodep = &pool[idx];
    
18.         nodep->next = NULL;
    
19.         lognode_put(nodep);
    
20.     }
    
21. 
    
22.     __sync_sub_and_fetch(&run, 1);
    
23.     return NULL;
    
24. }
    
25. 
    
26. #define use_get
    
27. 
    
28. int main(int argc, char** argv)
    
29. {
    
30.     int n = 0;
    
31.     for (int i = 0; i < nrt; ++i) {
    
32.         pthread_t tid;
    
33.         if(0 == pthread_create(&tid, NULL, writer, NULL)) {
    
34.             ++n;
    
35.             pthread_detach(tid);
    
36.         }
    
37.     }
    
38. 
    
39.     int nr = nrp * n;
    
40.     pool = (struct lognode *) malloc(sizeof(struct lognode) * nr);
    
41.     if (pool == 0) {
    
42.         printf("memory\n");
    
43.         return 0;
    
44.     }
    
45.     run = n;
    
46.     printf("run = %d\n", run);
    
47. 
    
48.     printf("%d writer, nr = %d\n", n, nr);
    
49.     time_t t1 = time(0);
    
50. 
    
51.     while (run > 0) {
    
52. #ifdef use_get
    
53.         struct lognode* nodep = lognode_get();
    
54.         struct lognode* cur = nodep;
    
55.         while (cur != NULL) {
    
56.             nodep = nodep->next;
    
57.             cur = nodep;
    
58.             --nr;
    
59.         }
    
60. #endif
    
61. 
    
62.         time_t t2 = time(0);
    
63.         if (t2 > t1 + 3) {
    
64.             printf("%d %d\n", nr, indx);
    
65.             t1 = t2;
    
66.         }
    
67.     }
    
68. 
    
69. #ifndef use_get
    
70.     struct lognode* nodep = lognode_get();
    
71.     struct lognode* cur = nodep;
    
72.     while (cur != NULL) {
    
73.         nodep = nodep->next;
    
74.         cur = nodep;
    
75.         --nr;
    
76.     }
    
77. #endif
    
78.     printf("done %d\n", nr);
    
79.     return 0;
    
80. }
    

复制代码

starwing83

zyl，想问问你，你这个是用户态的程序吧？如果是单核的CPU，那么你的while循环相当于一个spinlock，会耗尽当前的时间片，这样做会不会反而降低效率？只有在多核的情况下才有实用价值吧？

zylthinking

starwing83 发表于 2013-08-17 08:44
zyl，想问问你，你这个是用户态的程序吧？如果是单核的CPU，那么你的while循环相当于一个spinlock，会耗尽当 ...

当然是多核， 单核不让出 CPU 那不是傻了， 再轮也进不去啊

zhouzhenghui

正如starwing指出的那样，修改后的版本虽然put没有问题，但实际上存在活锁，就是必须等待另外一方做完事情才可以继续，这一般来说是不好的，虽然在kernel类似的spinlock会这么做。比较好的无锁是在其他人没有结束必须先结束的事务操作时，帮助对方结束。

楼主所需代码大约如下：

1. static struct lognode root = {0};
   
2. static struct lognode* head = &root;
   
3. static struct lognode* tail = &root;
   
4. 
   
5. void lognode_put(struct lognode* node)
   
6. {
   
7.     struct lognode* last;
   
8.     do {
   
9.         last = tail;
   
10.         struct lognode* before = __sync_val_compare_and_swap(&(last->next), NULL, node);
    
11.         if (before != NULL)
    
12.            __sync_val_compare_and_swap(&tail, last, before);
    
13.     } while (before != NULL);
    
14. 
    
15.     __sync_val_compare_and_swap(&tail, last, node);
    
16. }
    
17. 
    
18. struct lognode* lognode_get()
    
19. {
    
20.     struct lognode temp = {0};
    
21.     tail = &temp;
    
22.     wmb();
    
23. 
    
24.     struct lognode* node = head->next;
    
25.     head->next = NULL;
    
26.    
    
27.     if (__sync_val_compare_and_swap(&tail, &temp, head) == &temp)
    
28.         if (temp.next) lognode_put(temp.next);
    
29.     else      
    
30.        head->next = temp.next;
    
31. 
    
32.     return node;
    
33. }

复制代码

此代码中的get只适合于单个消费者的情况，多个消费者留作作业了。

zylthinking

zhouzhenghui 发表于 2013-08-17 22:13
正如starwing指出的那样，修改后的版本虽然put没有问题，但实际上存在活锁，就是必须等待另外一方做完事情才 ...

你确定你的代码测试过吗，

1. static struct lognode root = {0};
   
2. static struct lognode* head = &root;
   
3. static struct lognode* tail = &root;
   
4. int nr_not_change_tail = 0;
   
5. int up = 0, get = 0;
   
6. 
   
7. 
   
8. void lognode_put(struct lognode* node)
   
9. {
   
10.     struct lognode* before;
    
11.     struct lognode* last;
    
12.     do {
    
13.         last = tail;
    
14.         before = __sync_val_compare_and_swap(&(last->next), NULL, node);
    
15.         if (before != NULL)
    
16.            __sync_val_compare_and_swap(&tail, last, before);
    
17.     } while (before != NULL);
    
18. 
    
19.     __sync_val_compare_and_swap(&tail, last, node);
    
20. }
    
21. 
    
22. struct lognode* lognode_get()
    
23. {
    
24.     struct lognode temp = {0};
    
25.     tail = &temp;
    
26.     wmb();
    
27. 
    
28.     struct lognode* node = head->next;
    
29.     head->next = NULL;
    
30. 
    
31.     if (__sync_val_compare_and_swap(&tail, &temp, head) == &temp)
    
32.         if (temp.next) lognode_put(temp.next);
    
33.     else
    
34.        head->next = temp.next;
    
35. 
    
36.     return node;
    
37. }
    
38. 
    
39. #define nrp 1000000
    
40. #define nrt 1
    
41. 
    
42. static volatile int run = 0;
    
43. static struct lognode* pool = NULL;
    
44. static volatile int indx = 0;
    
45. 
    
46. void* writer(void* any)
    
47. {
    
48.     while (run == 0) {
    
49.         usleep(100);
    
50.     }
    
51. 
    
52.     for (int i = 0; i < nrp; ++i) {
    
53.         int idx = __sync_fetch_and_add(&indx, 1);
    
54.         struct lognode* nodep = &pool[idx];
    
55.         nodep->next = NULL;
    
56.         lognode_put(nodep);
    
57.     }
    
58. 
    
59.     __sync_sub_and_fetch(&run, 1);
    
60.     return NULL;
    
61. }
    
62. 
    
63. static int fetched[1024 * 1024 * 32];
    
64. static int idx = 0;
    
65. 
    
66. int main(int argc, char** argv)
    
67. {
    
68.     int n = 0;
    
69.     for (int i = 0; i < nrt; ++i) {
    
70.         pthread_t tid;
    
71.         if(0 == pthread_create(&tid, NULL, writer, NULL)) {
    
72.             ++n;
    
73.             pthread_detach(tid);
    
74.         }
    
75.     }
    
76. 
    
77.     int nr = nrp * n;
    
78.     pool = (struct lognode *) malloc(sizeof(struct lognode) * nr);
    
79.     if (pool == 0) {
    
80.         printf("memory\n");
    
81.         return 0;
    
82.     }
    
83.     run = n;
    
84.     printf("run = %d\n", run);
    
85. 
    
86.     printf("%d writer, nr = %d\n", n, nr);
    
87.     time_t t1 = time(0);
    
88. 
    
89.     int fch = 0;
    
90.     while (run > 0) {
    
91. #if 1
    
92.         struct lognode* nodep = lognode_get();
    
93.         struct lognode* cur = nodep;
    
94.         while (cur != NULL) {
    
95.             //printf("%p feched\n", cur);
    
96.             nodep = nodep->next;
    
97.             cur = nodep;
    
98.             ++fch;
    
99.             --nr;
    
100.         }
     
101. 
     
102.         if (fch > 0) {
     
103.             fetched[idx++] = fch;
     
104.             fch = 0;
     
105.         }
     
106. #endif
     
107.         time_t t2 = time(0);
     
108.         if (t2 > t1 + 3) {
     
109.             printf("%d %d\n", nr, indx);
     
110.             t1 = t2;
     
111.         }
     
112.     }
     
113. 
     
114.     struct lognode* nodep = lognode_get();
     
115.     struct lognode* cur = nodep;
     
116.     while (cur != NULL) {
     
117.         nodep = nodep->next;
     
118.         cur = nodep;
     
119.         ++fch;
     
120.         --nr;
     
121.     }
     
122. 
     
123.     if (fch > 0) {
     
124.         fetched[idx++] = fch;
     
125.     }
     
126.     printf("idx = %d\n", idx);
     
127.     printf("done %d, nr_not_change_tail = %d, up = %d, get %d\n", nr, nr_not_change_tail, up, get);
     
128. return 0;
     
129.     n = 0;
     
130.     fch = 0;
     
131.     nr = 0;
     
132.     for (int i = 0; i < nrp; ++i) {
     
133.         ++fch;
     
134.         if (pool[i].next == NULL) {
     
135.             if(fch < fetched[n]) {
     
136.                 assert(0);
     
137.             } else if (fch > fetched[n]) {
     
138.                 if (nr < fch - fetched[n]) {
     
139.                     nr = fch - fetched[n];
     
140.                 }
     
141. 
     
142.                 //printf("%d: %d, %d\n", n, fch, fetched[n]);
     
143.             }
     
144.             ++n;
     
145.             fch = 0;
     
146.         }
     
147.     }
     
148.     printf("%d get indeed, %d\n", n, nr);
     
149. 
     
150.     cur = &pool[0];
     
151.     for (int i = 1; i < nrp; ++i) {
     
152.         ++fch;
     
153.         if (cur->next != NULL) {
     
154.             assert(cur->next == &pool[i]);
     
155.             //printf("%d:%d\n", fetched[nr], fch);
     
156.         }
     
157.         cur = &pool[i];
     
158.     }
     
159.     printf("it's linked  indeed\n");
     
160.     return 0;
     
161. }
     

复制代码

运行结果：
run = 1
1 writer, nr = 1000000
idx = 238635
done 622448, nr_not_change_tail = 0, up = 0, get 0

凡是 done 不为  0 的， 一律存在 bug

zhouzhenghui

的确没有测试（BTW：楼主的代码也太混乱了吧）

理论上来说，这里必须使用动态的head和tail，如下：

static struct lognode* head;
static struct lognode* tail;

int main()
{
    head = malloc(sizeof(struct lognode));
    head->next = NULL;
    tail = head;
    ...
}

然后，get方法表现为：

struct lognode* lognode_get()
{
    if (!head->next) return head;

    struct lognode *node = head->next;

    head = malloc(sizeof(struct lognode));
    head->next = NULL;

    lognode_put(head);

    return node;
}

如果一定要使用静态的头部，那需要不稳定逻辑避过，并且效率会受到影响：

struct lognode* lognode_get()
{
    if (head == tail) return head;

    struct lognode *node = head->next;
    if (!node) return head;

    head->next = NULL;
    lognode_put(head);

    return node;
}

注意get链表以head结束，即循环终止条件是cur != head.