summaryrefslogtreecommitdiff
path: root/erts/emulator/beam/erl_thr_progress.h
blob: 3adc8e32c69a10fac64f8a3a2ead1d6aede69180 (plain)
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
/*
 * %CopyrightBegin%
 *
 * Copyright Ericsson AB 2011-2021. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * %CopyrightEnd%
 */

/*
 * Description: Thread progress information. Used by lock free algorithms
 *              to determine when all involved threads are guaranteed to
 *              have passed a specific point of execution.
 *
 *              Usage instructions can be found in ert_thr_progress.c
 *
 * Author: 	Rickard Green
 */

#ifndef ERL_THR_PROGRESS_H__TSD_TYPE__
#define ERL_THR_PROGRESS_H__TSD_TYPE__

#include "sys.h"

void erts_thr_progress_block(void);
void erts_thr_progress_unblock(void);
int erts_thr_progress_is_blocking(void);

typedef Uint64 ErtsThrPrgrVal;

#define ERTS_THR_PRGR_WAKEUP_DATA_SIZE 4 /* Need to be an even power of 2. */

typedef struct {
    ErtsThrPrgrVal next;
    ErtsThrPrgrVal current;
    int chk_next_ix;
    struct {
	int current;
	int waiting;
    } umrefc_ix;
} ErtsThrPrgrLeaderState;

typedef struct {
    int id;
    int is_managed;
    int is_blocking;
#ifdef ERTS_ENABLE_LOCK_CHECK
    int is_delaying; /* managed is always delaying */
#endif
    int is_temporary;

    /* --- Part below only for registered threads --- */

    ErtsThrPrgrVal wakeup_request[ERTS_THR_PRGR_WAKEUP_DATA_SIZE];

    /* --- Part below only for managed threads --- */

    int leader; /* Needs to be first in the managed threads part */
    int active;
    int is_deep_sleeper;
    ErtsThrPrgrVal confirmed;
    ErtsThrPrgrLeaderState leader_state;
} ErtsThrPrgrData;

int erts_thr_progress_fatal_error_block(ErtsThrPrgrData *tmp_tpd_bufp);
void erts_thr_progress_fatal_error_wait(SWord timeout);


typedef struct ErtsThrPrgrLaterOp_ ErtsThrPrgrLaterOp;
struct ErtsThrPrgrLaterOp_ {
    ErtsThrPrgrVal later;
    void (*func)(void *);
    void *data;
    ErtsThrPrgrLaterOp *next;
};

#endif

#if !defined(ERL_THR_PROGRESS_H__) && !defined(ERL_THR_PROGRESS_TSD_TYPE_ONLY)
#define ERL_THR_PROGRESS_H__

#include "erl_threads.h"
#include "erl_process.h"


/* ERTS_THR_PRGR_VAL_FIRST should only be used when initializing... */
#define ERTS_THR_PRGR_VAL_FIRST ((ErtsThrPrgrVal) 0)
#define ERTS_THR_PRGR_VAL_WAITING (~((ErtsThrPrgrVal) 0))
#define ERTS_THR_PRGR_INVALID (~((ErtsThrPrgrVal) 0))

extern erts_tsd_key_t erts_thr_prgr_data_key__;

#define ERTS_THR_PRGR_ATOMIC erts_atomic64_t

typedef struct {
    void *arg;
    void (*wakeup)(void *);
    void (*prepare_wait)(void *);
    void (*wait)(void *);
    void (*finalize_wait)(void *);
} ErtsThrPrgrCallbacks;

typedef struct {
    ERTS_THR_PRGR_ATOMIC current;
} ErtsThrPrgr;

typedef int ErtsThrPrgrDelayHandle;
#define ERTS_THR_PRGR_DHANDLE_MANAGED ((ErtsThrPrgrDelayHandle) -1)
/* ERTS_THR_PRGR_DHANDLE_MANAGED implies managed thread */
#define ERTS_THR_PRGR_DHANDLE_INVALID ((ErtsThrPrgrDelayHandle) -2)

extern ErtsThrPrgr erts_thr_prgr__;

void erts_thr_progress_pre_init(void);
void erts_thr_progress_init(int no_schedulers, int managed, int unmanaged);
ErtsThrPrgrData *erts_thr_progress_register_managed_thread(
    ErtsSchedulerData *esdp, ErtsThrPrgrCallbacks *, int, int);
void erts_thr_progress_register_unmanaged_thread(ErtsThrPrgrCallbacks *);
void erts_thr_progress_unregister_unmanaged_thread(void);
void erts_thr_progress_active(ErtsThrPrgrData *, int on);
void erts_thr_progress_wakeup(ErtsThrPrgrData *,
			      ErtsThrPrgrVal value);
int erts_thr_progress_update(ErtsThrPrgrData *);
int erts_thr_progress_leader_update(ErtsThrPrgrData *);
void erts_thr_progress_prepare_wait(ErtsThrPrgrData *);
void erts_thr_progress_finalize_wait(ErtsThrPrgrData *);
ErtsThrPrgrDelayHandle erts_thr_progress_unmanaged_delay__(void);
void erts_thr_progress_unmanaged_continue__(int umrefc_ix);
ErtsThrPrgrData *erts_thr_progress_data(void);

void erts_thr_progress_dbg_print_state(void);

ERTS_GLB_INLINE ErtsThrPrgrData *erts_thr_prgr_data(ErtsSchedulerData *esdp);

ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_prgr_read_nob__(ERTS_THR_PRGR_ATOMIC *atmc);
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_prgr_read_acqb__(ERTS_THR_PRGR_ATOMIC *atmc);
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_prgr_read_mb__(ERTS_THR_PRGR_ATOMIC *atmc);

ERTS_GLB_INLINE int erts_thr_progress_is_managed_thread(void);
ERTS_GLB_INLINE ErtsThrPrgrDelayHandle erts_thr_progress_unmanaged_delay(void);
ERTS_GLB_INLINE void erts_thr_progress_unmanaged_continue(ErtsThrPrgrDelayHandle handle);
#ifdef ERTS_ENABLE_LOCK_CHECK
ERTS_GLB_INLINE int erts_thr_progress_lc_is_delaying(void);
#endif
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_progress_current_to_later__(ErtsThrPrgrVal val);
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_progress_later(ErtsSchedulerData *);
ERTS_GLB_INLINE ErtsThrPrgrVal erts_thr_progress_current(void);
ERTS_GLB_INLINE int erts_thr_progress_has_passed__(ErtsThrPrgrVal val1, ErtsThrPrgrVal val2);
ERTS_GLB_INLINE int erts_thr_progress_has_reached_this(ErtsThrPrgrVal this_, ErtsThrPrgrVal val);
ERTS_GLB_INLINE int erts_thr_progress_equal(ErtsThrPrgrVal val1,
					    ErtsThrPrgrVal val2);
ERTS_GLB_INLINE int erts_thr_progress_cmp(ErtsThrPrgrVal val1, ErtsThrPrgrVal val2);
ERTS_GLB_INLINE int erts_thr_progress_has_reached(ErtsThrPrgrVal val);

#if ERTS_GLB_INLINE_INCL_FUNC_DEF

ERTS_GLB_INLINE ErtsThrPrgrData *
erts_thr_prgr_data(ErtsSchedulerData *esdp) {
    if (esdp) {
        return &esdp->thr_progress_data;
    } else {
        return erts_thr_progress_data();
    }
}

ERTS_GLB_INLINE ErtsThrPrgrVal
erts_thr_prgr_read_nob__(ERTS_THR_PRGR_ATOMIC *atmc)
{
    return (ErtsThrPrgrVal) erts_atomic64_read_nob(atmc);
}

ERTS_GLB_INLINE ErtsThrPrgrVal
erts_thr_prgr_read_acqb__(ERTS_THR_PRGR_ATOMIC *atmc)
{
    return (ErtsThrPrgrVal) erts_atomic64_read_acqb(atmc);
}

ERTS_GLB_INLINE ErtsThrPrgrVal
erts_thr_prgr_read_mb__(ERTS_THR_PRGR_ATOMIC *atmc)
{
    return (ErtsThrPrgrVal) erts_atomic64_read_mb(atmc);
}

ERTS_GLB_INLINE int
erts_thr_progress_is_managed_thread(void)
{
    ErtsThrPrgrData *tpd = (ErtsThrPrgrData*)erts_tsd_get(erts_thr_prgr_data_key__);
    return tpd && tpd->is_managed;
}

ERTS_GLB_INLINE ErtsThrPrgrDelayHandle
erts_thr_progress_unmanaged_delay(void)
{
    if (erts_thr_progress_is_managed_thread())
	return ERTS_THR_PRGR_DHANDLE_MANAGED; /* Nothing to do */
    else
	return erts_thr_progress_unmanaged_delay__();
}

ERTS_GLB_INLINE void
erts_thr_progress_unmanaged_continue(ErtsThrPrgrDelayHandle handle)
{
    ASSERT(handle != ERTS_THR_PRGR_DHANDLE_MANAGED
	   || erts_thr_progress_is_managed_thread());
    if (handle != ERTS_THR_PRGR_DHANDLE_MANAGED)
	erts_thr_progress_unmanaged_continue__(handle);
}

#ifdef ERTS_ENABLE_LOCK_CHECK

ERTS_GLB_INLINE int
erts_thr_progress_lc_is_delaying(void)
{
    ErtsThrPrgrData *tpd = (ErtsThrPrgrData *)erts_tsd_get(erts_thr_prgr_data_key__);
    return tpd && tpd->is_delaying;
}

#endif

ERTS_GLB_INLINE ErtsThrPrgrVal
erts_thr_progress_current_to_later__(ErtsThrPrgrVal val)
{
    if (val == (ERTS_THR_PRGR_VAL_WAITING-((ErtsThrPrgrVal)2)))
	return ((ErtsThrPrgrVal) 0);
    else if (val == (ERTS_THR_PRGR_VAL_WAITING-((ErtsThrPrgrVal)1)))
	return ((ErtsThrPrgrVal) 1);
    else
	return val + ((ErtsThrPrgrVal) 2);
}

ERTS_GLB_INLINE ErtsThrPrgrVal
erts_thr_progress_later(ErtsSchedulerData *esdp)
{
    ErtsThrPrgrData *tpd;
    ErtsThrPrgrVal val;
    if (esdp) {
	tpd = &esdp->thr_progress_data;
    managed_thread:
	val = tpd->confirmed;
	ERTS_THR_MEMORY_BARRIER;
    }
    else {
	tpd = (ErtsThrPrgrData *)erts_tsd_get(erts_thr_prgr_data_key__);
	if (tpd && tpd->is_managed)
	    goto managed_thread;
	val = erts_thr_prgr_read_mb__(&erts_thr_prgr__.current);
    }
    ASSERT(val != ERTS_THR_PRGR_VAL_WAITING);
    return erts_thr_progress_current_to_later__(val);
}

ERTS_GLB_INLINE ErtsThrPrgrVal
erts_thr_progress_current(void)
{
    if (erts_thr_progress_is_managed_thread())
	return erts_thr_prgr_read_nob__(&erts_thr_prgr__.current);
    else
	return erts_thr_prgr_read_acqb__(&erts_thr_prgr__.current);
}

ERTS_GLB_INLINE int
erts_thr_progress_has_passed__(ErtsThrPrgrVal val1, ErtsThrPrgrVal val0)
{
    if ((((((ErtsThrPrgrVal) 1) << 63) & val1)
	 ^ ((((ErtsThrPrgrVal) 1) << 63) & val0)) != 0) {
	/* May have wrapped... */
	if (val1 < (((ErtsThrPrgrVal) 1) << 62)
	    && val0 > (((ErtsThrPrgrVal) 3) << 62)) {
	    /*
	     * 'val1' has wrapped but 'val0' has not yet wrapped. While in
	     * these ranges 'current' is considered later than 'val0'.
	     */
	    return 1;
	}
    }
    return val1 > val0;
}

ERTS_GLB_INLINE int
erts_thr_progress_has_reached_this(ErtsThrPrgrVal this_, ErtsThrPrgrVal val)
{
    if (this_ == val)
	return 1;
    return erts_thr_progress_has_passed__(this_, val);
}

ERTS_GLB_INLINE int
erts_thr_progress_equal(ErtsThrPrgrVal val1, ErtsThrPrgrVal val2)
{
    return val1 == val2 && val1 != ERTS_THR_PRGR_INVALID;
}

ERTS_GLB_INLINE int
erts_thr_progress_cmp(ErtsThrPrgrVal val1, ErtsThrPrgrVal val2)
{
    if (val1 == val2)
	return 0;
    if (erts_thr_progress_has_passed__(val1, val2))
	return 1;
    else
	return -1;
}	

ERTS_GLB_INLINE int
erts_thr_progress_has_reached(ErtsThrPrgrVal val)
{
    ErtsThrPrgrVal current = erts_thr_progress_current();
    return erts_thr_progress_has_reached_this(current, val);
}

#endif


#endif