1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
/* Implementation for gpr_cancellable */
#include <grpc/support/atm.h>
#include <grpc/support/sync.h>
#include <grpc/support/time.h>
void gpr_cancellable_init(gpr_cancellable *c) {
gpr_mu_init(&c->mu);
c->cancelled = 0;
c->waiters.next = &c->waiters;
c->waiters.prev = &c->waiters;
c->waiters.mu = NULL;
c->waiters.cv = NULL;
}
void gpr_cancellable_destroy(gpr_cancellable *c) { gpr_mu_destroy(&c->mu); }
int gpr_cancellable_is_cancelled(gpr_cancellable *c) {
return gpr_atm_acq_load(&c->cancelled) != 0;
}
/* Threads in gpr_cv_cancellable_wait(cv, mu, ..., c) place themselves on a
linked list c->waiters of gpr_cancellable_list_ before waiting on their
condition variables. They check for cancellation while holding *mu. Thus,
to wake a thread from gpr_cv_cancellable_wait(), it suffices to:
- set c->cancelled
- acquire and release *mu
- gpr_cv_broadcast(cv)
However, gpr_cancellable_cancel() may not use gpr_mu_lock(mu), since the
caller may already hold *mu---a possible deadlock. (If we knew the caller
did not hold *mu, care would still be needed, because c->mu follows *mu in
the locking order, so *mu could not be acquired while holding c->mu---which
is needed to iterate over c->waiters.)
Therefore, gpr_cancellable_cancel() uses gpr_mu_trylock() rather than
gpr_mu_lock(), and retries until either gpr_mu_trylock() succeeds or the
thread leaves gpr_cv_cancellable_wait() for other reasons. In the first
case, gpr_cancellable_cancel() removes the entry from the waiters list; in
the second, the waiting thread removes itself from the list.
A one-entry cache of mutexes and condition variables processed is kept to
avoid doing the same work again and again if many threads are blocked in the
same place. However, it's important to broadcast on a condition variable if
the corresponding mutex has been locked successfully, even if the condition
variable has been signalled before. */
void gpr_cancellable_cancel(gpr_cancellable *c) {
if (!gpr_cancellable_is_cancelled(c)) {
int failures;
int backoff = 1;
do {
struct gpr_cancellable_list_ *l;
struct gpr_cancellable_list_ *nl;
gpr_mu *omu = 0; /* one-element cache of a processed gpr_mu */
gpr_cv *ocv = 0; /* one-element cache of a processd gpr_cv */
gpr_mu_lock(&c->mu);
gpr_atm_rel_store(&c->cancelled, 1);
failures = 0;
for (l = c->waiters.next; l != &c->waiters; l = nl) {
nl = l->next;
if (omu != l->mu) {
omu = l->mu;
if (gpr_mu_trylock(l->mu)) {
gpr_mu_unlock(l->mu);
l->next->prev = l->prev; /* remove *l from list */
l->prev->next = l->next;
/* allow unconditional dequeue in gpr_cv_cancellable_wait() */
l->next = l;
l->prev = l;
ocv = 0; /* force broadcast */
} else {
failures++;
}
}
if (ocv != l->cv) {
ocv = l->cv;
gpr_cv_broadcast(l->cv);
}
}
gpr_mu_unlock(&c->mu);
if (failures != 0) {
if (backoff < 10) {
volatile int i;
for (i = 0; i != (1 << backoff); i++) {
}
backoff++;
} else {
gpr_event ev;
gpr_event_init(&ev);
gpr_event_wait(&ev,
gpr_time_add(gpr_now(), gpr_time_from_micros(1000)));
}
}
} while (failures != 0);
}
}
int gpr_cv_cancellable_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline,
gpr_cancellable *c) {
gpr_int32 timeout;
gpr_mu_lock(&c->mu);
timeout = gpr_cancellable_is_cancelled(c);
if (!timeout) {
struct gpr_cancellable_list_ le;
le.mu = mu;
le.cv = cv;
le.next = c->waiters.next;
le.prev = &c->waiters;
le.next->prev = ≤
le.prev->next = ≤
gpr_mu_unlock(&c->mu);
timeout = gpr_cv_wait(cv, mu, abs_deadline);
gpr_mu_lock(&c->mu);
le.next->prev = le.prev;
le.prev->next = le.next;
if (!timeout) {
timeout = gpr_cancellable_is_cancelled(c);
}
}
gpr_mu_unlock(&c->mu);
return timeout;
}
|
