summaryrefslogtreecommitdiff
path: root/ractor_core.h
blob: f0bdec212f20772bf6e75a37a6ddec5011697ff0 (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
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
#include "ruby/ruby.h"
#include "ruby/ractor.h"
#include "vm_core.h"
#include "id_table.h"
#include "vm_debug.h"

#ifndef RACTOR_CHECK_MODE
#define RACTOR_CHECK_MODE (0 || VM_CHECK_MODE || RUBY_DEBUG)
#endif

enum rb_ractor_basket_type {
    basket_type_none,
    basket_type_ref,
    basket_type_copy,
    basket_type_move,
    basket_type_will,
    basket_type_deleted,
    basket_type_reserved,
};

struct rb_ractor_basket {
    bool exception;
    enum rb_ractor_basket_type type;
    VALUE v;
    VALUE sender;
};

struct rb_ractor_queue {
    struct rb_ractor_basket *baskets;
    int start;
    int cnt;
    int size;
    unsigned int serial;
    unsigned int reserved_cnt;
};

struct rb_ractor_waiting_list {
    int cnt;
    int size;
    rb_ractor_t **ractors;
};

struct rb_ractor_sync {
    // ractor lock
    rb_nativethread_lock_t lock;
#if RACTOR_CHECK_MODE > 0
    VALUE locked_by;
#endif
    rb_nativethread_cond_t cond;

    // communication
    struct rb_ractor_queue  incoming_queue;
    struct rb_ractor_waiting_list taking_ractors;

    bool incoming_port_closed;
    bool outgoing_port_closed;

    struct ractor_wait {
        enum ractor_wait_status {
            wait_none      = 0x00,
            wait_receiving = 0x01,
            wait_taking    = 0x02,
            wait_yielding  = 0x04,
            wait_moving    = 0x08,
        } status;

        enum ractor_wakeup_status {
            wakeup_none,
            wakeup_by_send,
            wakeup_by_yield,
            wakeup_by_take,
            wakeup_by_close,
            wakeup_by_interrupt,
            wakeup_by_retry,
        } wakeup_status;

        struct rb_ractor_basket yielded_basket;
        struct rb_ractor_basket taken_basket;
    } wait;
};

struct rb_ractor_struct {
    struct rb_ractor_pub pub;

    struct rb_ractor_sync sync;
    VALUE receiving_mutex;
    bool yield_atexit;

    // vm wide barrier synchronization
    rb_nativethread_cond_t barrier_wait_cond;

    // thread management
    struct {
        struct ccan_list_head set;
        unsigned int cnt;
        unsigned int blocking_cnt;
        unsigned int sleeper;
        rb_global_vm_lock_t gvl;
        rb_execution_context_t *running_ec;
        rb_thread_t *main;
    } threads;
    VALUE thgroup_default;

    VALUE name;
    VALUE loc;

    // created
    //   | ready to run
    // ====================== inserted to vm->ractor
    //   v
    // blocking <---+ all threads are blocking
    //   |          |
    //   v          |
    // running -----+
    //   | all threads are terminated.
    // ====================== removed from vm->ractor
    //   v
    // terminated
    //
    // status is protected by VM lock (global state)

    enum ractor_status {
        ractor_created,
        ractor_running,
        ractor_blocking,
        ractor_terminated,
    } status_;

    struct ccan_list_node vmlr_node;

    // ractor local data

    st_table *local_storage;
    struct rb_id_table *idkey_local_storage;

    VALUE r_stdin;
    VALUE r_stdout;
    VALUE r_stderr;
    VALUE verbose;
    VALUE debug;

    rb_ractor_newobj_cache_t newobj_cache;

    // gc.c rb_objspace_reachable_objects_from
    struct gc_mark_func_data_struct {
        void *data;
        void (*mark_func)(VALUE v, void *data);
    } *mfd;
}; // rb_ractor_t is defined in vm_core.h


static inline VALUE
rb_ractor_self(const rb_ractor_t *r)
{
    return r->pub.self;
}

rb_ractor_t *rb_ractor_main_alloc(void);
void rb_ractor_main_setup(rb_vm_t *vm, rb_ractor_t *main_ractor, rb_thread_t *main_thread);
void rb_ractor_atexit(rb_execution_context_t *ec, VALUE result);
void rb_ractor_atexit_exception(rb_execution_context_t *ec);
void rb_ractor_teardown(rb_execution_context_t *ec);
void rb_ractor_receive_parameters(rb_execution_context_t *ec, rb_ractor_t *g, int len, VALUE *ptr);
void rb_ractor_send_parameters(rb_execution_context_t *ec, rb_ractor_t *g, VALUE args);

VALUE rb_thread_create_ractor(rb_ractor_t *g, VALUE args, VALUE proc); // defined in thread.c

rb_global_vm_lock_t *rb_ractor_gvl(rb_ractor_t *);
int rb_ractor_living_thread_num(const rb_ractor_t *);
VALUE rb_ractor_thread_list(rb_ractor_t *r);
bool rb_ractor_p(VALUE rv);

void rb_ractor_living_threads_init(rb_ractor_t *r);
void rb_ractor_living_threads_insert(rb_ractor_t *r, rb_thread_t *th);
void rb_ractor_living_threads_remove(rb_ractor_t *r, rb_thread_t *th);
void rb_ractor_blocking_threads_inc(rb_ractor_t *r, const char *file, int line); // TODO: file, line only for RUBY_DEBUG_LOG
void rb_ractor_blocking_threads_dec(rb_ractor_t *r, const char *file, int line); // TODO: file, line only for RUBY_DEBUG_LOG

void rb_ractor_vm_barrier_interrupt_running_thread(rb_ractor_t *r);
void rb_ractor_terminate_interrupt_main_thread(rb_ractor_t *r);
void rb_ractor_terminate_all(void);
bool rb_ractor_main_p_(void);
void rb_ractor_finish_marking(void);
void rb_ractor_atfork(rb_vm_t *vm, rb_thread_t *th);

VALUE rb_ractor_ensure_shareable(VALUE obj, VALUE name);

void rb_ractor_ensure_main_ractor(const char *msg);

RUBY_SYMBOL_EXPORT_BEGIN
bool rb_ractor_shareable_p_continue(VALUE obj);

// THIS FUNCTION SHOULD NOT CALL WHILE INCREMENTAL MARKING!!
// This function is for T_DATA::free_func
void rb_ractor_local_storage_delkey(rb_ractor_local_key_t key);

RUBY_SYMBOL_EXPORT_END

static inline bool
rb_ractor_main_p(void)
{
    if (ruby_single_main_ractor) {
        return true;
    }
    else {
        return rb_ractor_main_p_();
    }
}

static inline bool
rb_ractor_status_p(rb_ractor_t *r, enum ractor_status status)
{
    return r->status_ == status;
}

static inline void
rb_ractor_sleeper_threads_inc(rb_ractor_t *r)
{
    r->threads.sleeper++;
}

static inline void
rb_ractor_sleeper_threads_dec(rb_ractor_t *r)
{
    r->threads.sleeper--;
}

static inline void
rb_ractor_sleeper_threads_clear(rb_ractor_t *r)
{
    r->threads.sleeper = 0;
}

static inline int
rb_ractor_sleeper_thread_num(rb_ractor_t *r)
{
    return r->threads.sleeper;
}

static inline void
rb_ractor_thread_switch(rb_ractor_t *cr, rb_thread_t *th)
{
    if (cr->threads.running_ec != th->ec) {
        if (0) {
            ruby_debug_printf("rb_ractor_thread_switch ec:%p->%p\n",
                              (void *)cr->threads.running_ec, (void *)th->ec);
        }
    }
    else {
        return;
    }

    if (cr->threads.running_ec != th->ec) {
        th->running_time_us = 0;
    }

    cr->threads.running_ec = th->ec;

    VM_ASSERT(cr == GET_RACTOR());
}

static inline void
rb_ractor_set_current_ec(rb_ractor_t *cr, rb_execution_context_t *ec)
{
#ifdef RB_THREAD_LOCAL_SPECIFIER
  #ifdef __APPLE__
    rb_current_ec_set(ec);
  #else
    ruby_current_ec = ec;
  #endif
#else
    native_tls_set(ruby_current_ec_key, ec);
#endif

    if (cr->threads.running_ec != ec) {
        if (0) {
            ruby_debug_printf("rb_ractor_set_current_ec ec:%p->%p\n",
                              (void *)cr->threads.running_ec, (void *)ec);
        }
    }
    else {
        VM_ASSERT(0); // should be different
    }

    cr->threads.running_ec = ec;
}

void rb_vm_ractor_blocking_cnt_inc(rb_vm_t *vm, rb_ractor_t *cr, const char *file, int line);
void rb_vm_ractor_blocking_cnt_dec(rb_vm_t *vm, rb_ractor_t *cr, const char *file, int line);

static inline uint32_t
rb_ractor_id(const rb_ractor_t *r)
{
    return r->pub.id;
}

#if RACTOR_CHECK_MODE > 0
uint32_t rb_ractor_current_id(void);

static inline void
rb_ractor_setup_belonging_to(VALUE obj, uint32_t rid)
{
    VALUE flags = RBASIC(obj)->flags & 0xffffffff; // 4B
    RBASIC(obj)->flags = flags | ((VALUE)rid << 32);
}

static inline void
rb_ractor_setup_belonging(VALUE obj)
{
    rb_ractor_setup_belonging_to(obj, rb_ractor_current_id());
}

static inline uint32_t
rb_ractor_belonging(VALUE obj)
{
    if (SPECIAL_CONST_P(obj) || RB_OBJ_SHAREABLE_P(obj)) {
        return 0;
    }
    else {
        return RBASIC(obj)->flags >> 32;
    }
}

static inline VALUE
rb_ractor_confirm_belonging(VALUE obj)
{
    uint32_t id = rb_ractor_belonging(obj);

    if (id == 0) {
        if (UNLIKELY(!rb_ractor_shareable_p(obj))) {
            rp(obj);
            rb_bug("id == 0 but not shareable");
        }
    }
    else if (UNLIKELY(id != rb_ractor_current_id())) {
        if (rb_ractor_shareable_p(obj)) {
            // ok
        }
        else {
            rp(obj);
            rb_bug("rb_ractor_confirm_belonging object-ractor id:%u, current-ractor id:%u", id, rb_ractor_current_id());
        }
    }
    return obj;
}
#else
#define rb_ractor_confirm_belonging(obj) obj
#endif