/home/travis/build/MoarVM/MoarVM/src/spesh/worker.c

Source
#include "moar.h"

/* The specialization worker thread receives logs from other threads about
 * calls and types that showed up at runtime. It uses this to produce
 * specialized versions of code. */

/* Enters the work loop. */
static void worker(MVMThreadContext *tc, MVMCallsite *callsite, MVMRegister *args) {
    MVMObject *updated_static_frames = MVM_repr_alloc_init(tc,
        tc->instance->boot_types.BOOTArray);
    MVMObject *previous_static_frames = MVM_repr_alloc_init(tc,
        tc->instance->boot_types.BOOTArray);

    tc->instance->speshworker_thread_id = tc->thread_obj->body.thread_id;

    MVMROOT2(tc, updated_static_frames, previous_static_frames, {
        while (1) {
            MVMObject *log_obj;
            MVMuint64 start_time;
            unsigned int interval_id;
            if (MVM_spesh_debug_enabled(tc))
                start_time = uv_hrtime();
            log_obj = MVM_repr_shift_o(tc, tc->instance->spesh_queue);
            if (MVM_spesh_debug_enabled(tc)) {
                MVM_spesh_debug_printf(tc,
                    "Received Logs\n"
                    "=============\n\n"
                    "Was waiting %dus for logs on the log queue.\n\n",
                    (int)((uv_hrtime() - start_time) / 1000));
            }

            if (tc->instance->main_thread->prof_data)
                MVM_profiler_log_spesh_start(tc);

            interval_id = MVM_telemetry_interval_start(tc, "spesh worker consuming a log");

            uv_mutex_lock(&(tc->instance->mutex_spesh_sync));
            tc->instance->spesh_working = 1;
            uv_mutex_unlock(&(tc->instance->mutex_spesh_sync));

            tc->instance->spesh_stats_version++;
            if (log_obj->st->REPR->ID == MVM_REPR_ID_MVMSpeshLog) {
                MVMSpeshLog *sl = (MVMSpeshLog *)log_obj;
                MVM_telemetry_interval_annotate((uintptr_t)sl->body.thread->body.tc, interval_id, "from this thread");
                MVMROOT(tc, sl, {
                    MVMThreadContext *stc;
                    MVMuint32 i;
                    MVMuint32 n;

                    /* Update stats, and if we're logging dump each of them. */
                    tc->instance->spesh_stats_version++;
                    if (MVM_spesh_debug_enabled(tc))
                        start_time = uv_hrtime();
                    MVM_spesh_stats_update(tc, sl, updated_static_frames);
                    n = MVM_repr_elems(tc, updated_static_frames);
                    if (MVM_spesh_debug_enabled(tc)) {
                        MVM_spesh_debug_printf(tc,
                            "Statistics Updated\n"
                            "==================\n"
                            "%d frames had their statistics updated in %dus.\n\n",
                            (int)n, (int)((uv_hrtime() - start_time) / 1000));
                        for (i = 0; i < n; i++) {
                            char *dump = MVM_spesh_dump_stats(tc, (MVMStaticFrame* )
                                MVM_repr_at_pos_o(tc, updated_static_frames, i));
                            MVM_spesh_debug_printf(tc, "%s==========\n\n", dump);
                            MVM_free(dump);
                        }
                    }
                    MVM_telemetry_interval_annotate((uintptr_t)n, interval_id, "stats for this many frames");
                    GC_SYNC_POINT(tc);

                    /* Form a specialization plan. */
                    if (MVM_spesh_debug_enabled(tc))
                        start_time = uv_hrtime();
                    tc->instance->spesh_plan = MVM_spesh_plan(tc, updated_static_frames);
                    if (MVM_spesh_debug_enabled(tc)) {
                        n = tc->instance->spesh_plan->num_planned;
                        MVM_spesh_debug_printf(tc,
                            "Specialization Plan\n"
                            "===================\n"
                            "%u specialization(s) will be produced (planned in %dus).\n\n",
                            n, (int)((uv_hrtime() - start_time) / 1000));
                        for (i = 0; i < n; i++) {
                            char *dump = MVM_spesh_dump_planned(tc,
                                &(tc->instance->spesh_plan->planned[i]));
                            MVM_spesh_debug_printf(tc, "%s==========\n\n", dump);
                            MVM_free(dump);
                        }
                    }
                    MVM_telemetry_interval_annotate((uintptr_t)tc->instance->spesh_plan->num_planned, interval_id,
                            "this many specializations planned");
                    GC_SYNC_POINT(tc);

                    /* Implement the plan and then discard it. */
                    n = tc->instance->spesh_plan->num_planned;
                    for (i = 0; i < n; i++) {
                        MVM_spesh_candidate_add(tc, &(tc->instance->spesh_plan->planned[i]));
                        GC_SYNC_POINT(tc);
                    }
                    MVM_spesh_plan_destroy(tc, tc->instance->spesh_plan);
                    tc->instance->spesh_plan = NULL;

                    /* Clear up stats that didn't get updated for a while,
                     * then add frames updated this time into the previously
                     * updated array. */
                    MVM_spesh_stats_cleanup(tc, previous_static_frames);
                    n = MVM_repr_elems(tc, updated_static_frames);
                    for (i = 0; i < n; i++)
                        MVM_repr_push_o(tc, previous_static_frames,
                            MVM_repr_at_pos_o(tc, updated_static_frames, i));

                    /* Clear updated static frames array. */
                    MVM_repr_pos_set_elems(tc, updated_static_frames, 0);

                    /* Allow the sending thread to produce more logs again,
                     * putting a new spesh log in place if needed. */
                    stc = sl->body.thread->body.tc;
                    if (stc && !sl->body.was_compunit_bumped)
                        if (MVM_incr(&(stc->spesh_log_quota)) == 0) {
                            stc->spesh_log = MVM_spesh_log_create(tc, sl->body.thread);
                            MVM_telemetry_timestamp(stc, "logging restored after quota had run out");
                        }

                    /* If needed, signal sending thread that it can continue. */
                    if (sl->body.block_mutex) {
                        uv_mutex_lock(sl->body.block_mutex);
                        MVM_store(&(sl->body.completed), 1);
                        uv_cond_signal(sl->body.block_condvar);
                        uv_mutex_unlock(sl->body.block_mutex);
                    }
                });
            }
            else {
                MVM_panic(1, "Unexpected object sent to specialization worker");
            }

            MVM_telemetry_interval_stop(tc, interval_id, "spesh worker finished");

            if (tc->instance->main_thread->prof_data)
                MVM_profiler_log_spesh_end(tc);

            uv_mutex_lock(&(tc->instance->mutex_spesh_sync));
            tc->instance->spesh_working = 0;
            uv_cond_broadcast(&(tc->instance->cond_spesh_sync));
            uv_mutex_unlock(&(tc->instance->mutex_spesh_sync));
        }
    });
}

void MVM_spesh_worker_setup(MVMThreadContext *tc) {
    if (tc->instance->spesh_enabled) {
        MVMObject *worker_entry_point;
        tc->instance->spesh_queue = MVM_repr_alloc_init(tc, tc->instance->boot_types.BOOTQueue);
        worker_entry_point = MVM_repr_alloc_init(tc, tc->instance->boot_types.BOOTCCode);
        ((MVMCFunction *)worker_entry_point)->body.func = worker;
        MVM_thread_run(tc, MVM_thread_new(tc, worker_entry_point, 1));
    }
}

Line	Count	Source
1		#include "moar.h"
2
3		/* The specialization worker thread receives logs from other threads about
4		* calls and types that showed up at runtime. It uses this to produce
5		* specialized versions of code. */
6
7		/* Enters the work loop. */
8	144	static void worker(MVMThreadContext tc, MVMCallsite callsite, MVMRegister *args) {
9	144	MVMObject *updated_static_frames = MVM_repr_alloc_init(tc,
10	144	tc->instance->boot_types.BOOTArray);
11	144	MVMObject *previous_static_frames = MVM_repr_alloc_init(tc,
12	144	tc->instance->boot_types.BOOTArray);
13	144
14	144	tc->instance->speshworker_thread_id = tc->thread_obj->body.thread_id;
15	144
16	144	MVMROOT2(tc, updated_static_frames, previous_static_frames, {
17	144	while (1) {
18	144	MVMObject *log_obj;
19	144	MVMuint64 start_time;
20	144	unsigned int interval_id;
21	144	if (MVM_spesh_debug_enabled(tc))
22	144	start_time = uv_hrtime();
23	144	log_obj = MVM_repr_shift_o(tc, tc->instance->spesh_queue);
24	144	if (MVM_spesh_debug_enabled(tc)) {
25	144	MVM_spesh_debug_printf(tc,
26	144	"Received Logs\n"
27	144	"=============\n\n"
28	144	"Was waiting %dus for logs on the log queue.\n\n",
29	144	(int)((uv_hrtime() - start_time) / 1000));
30	144	}
31	144
32	144	if (tc->instance->main_thread->prof_data)
33	144	MVM_profiler_log_spesh_start(tc);
34	144
35	144	interval_id = MVM_telemetry_interval_start(tc, "spesh worker consuming a log");
36	144
37	144	uv_mutex_lock(&(tc->instance->mutex_spesh_sync));
38	144	tc->instance->spesh_working = 1;
39	144	uv_mutex_unlock(&(tc->instance->mutex_spesh_sync));
40	144
41	144	tc->instance->spesh_stats_version++;
42	144	if (log_obj->st->REPR->ID == MVM_REPR_ID_MVMSpeshLog) {
43	144	MVMSpeshLog sl = (MVMSpeshLog )log_obj;
44	144	MVM_telemetry_interval_annotate((uintptr_t)sl->body.thread->body.tc, interval_id, "from this thread");
45	144	MVMROOT(tc, sl, {
46	144	MVMThreadContext *stc;
47	144	MVMuint32 i;
48	144	MVMuint32 n;
49	144
50	144	/* Update stats, and if we're logging dump each of them. */
51	144	tc->instance->spesh_stats_version++;
52	144	if (MVM_spesh_debug_enabled(tc))
53	144	start_time = uv_hrtime();
54	144	MVM_spesh_stats_update(tc, sl, updated_static_frames);
55	144	n = MVM_repr_elems(tc, updated_static_frames);
56	144	if (MVM_spesh_debug_enabled(tc)) {
57	144	MVM_spesh_debug_printf(tc,
58	144	"Statistics Updated\n"
59	144	"==================\n"
60	144	"%d frames had their statistics updated in %dus.\n\n",
61	144	(int)n, (int)((uv_hrtime() - start_time) / 1000));
62	144	for (i = 0; i < n; i++) {
63	144	char dump = MVM_spesh_dump_stats(tc, (MVMStaticFrame )
64	144	MVM_repr_at_pos_o(tc, updated_static_frames, i));
65	144	MVM_spesh_debug_printf(tc, "%s==========\n\n", dump);
66	144	MVM_free(dump);
67	144	}
68	144	}
69	144	MVM_telemetry_interval_annotate((uintptr_t)n, interval_id, "stats for this many frames");
70	144	GC_SYNC_POINT(tc);
71	144
72	144	/* Form a specialization plan. */
73	144	if (MVM_spesh_debug_enabled(tc))
74	144	start_time = uv_hrtime();
75	144	tc->instance->spesh_plan = MVM_spesh_plan(tc, updated_static_frames);
76	144	if (MVM_spesh_debug_enabled(tc)) {
77	144	n = tc->instance->spesh_plan->num_planned;
78	144	MVM_spesh_debug_printf(tc,
79	144	"Specialization Plan\n"
80	144	"===================\n"
81	144	"%u specialization(s) will be produced (planned in %dus).\n\n",
82	144	n, (int)((uv_hrtime() - start_time) / 1000));
83	144	for (i = 0; i < n; i++) {
84	144	char *dump = MVM_spesh_dump_planned(tc,
85	144	&(tc->instance->spesh_plan->planned[i]));
86	144	MVM_spesh_debug_printf(tc, "%s==========\n\n", dump);
87	144	MVM_free(dump);
88	144	}
89	144	}
90	144	MVM_telemetry_interval_annotate((uintptr_t)tc->instance->spesh_plan->num_planned, interval_id,
91	144	"this many specializations planned");
92	144	GC_SYNC_POINT(tc);
93	144
94	144	/* Implement the plan and then discard it. */
95	144	n = tc->instance->spesh_plan->num_planned;
96	144	for (i = 0; i < n; i++) {
97	144	MVM_spesh_candidate_add(tc, &(tc->instance->spesh_plan->planned[i]));
98	144	GC_SYNC_POINT(tc);
99	144	}
100	144	MVM_spesh_plan_destroy(tc, tc->instance->spesh_plan);
101	144	tc->instance->spesh_plan = NULL;
102	144
103	144	/* Clear up stats that didn't get updated for a while,
104	144	* then add frames updated this time into the previously
105	144	* updated array. */
106	144	MVM_spesh_stats_cleanup(tc, previous_static_frames);
107	144	n = MVM_repr_elems(tc, updated_static_frames);
108	144	for (i = 0; i < n; i++)
109	144	MVM_repr_push_o(tc, previous_static_frames,
110	144	MVM_repr_at_pos_o(tc, updated_static_frames, i));
111	144
112	144	/* Clear updated static frames array. */
113	144	MVM_repr_pos_set_elems(tc, updated_static_frames, 0);
114	144
115	144	/* Allow the sending thread to produce more logs again,
116	144	* putting a new spesh log in place if needed. */
117	144	stc = sl->body.thread->body.tc;
118	144	if (stc && !sl->body.was_compunit_bumped)
119	144	if (MVM_incr(&(stc->spesh_log_quota)) == 0) {
120	144	stc->spesh_log = MVM_spesh_log_create(tc, sl->body.thread);
121	144	MVM_telemetry_timestamp(stc, "logging restored after quota had run out");
122	144	}
123	144
124	144	/* If needed, signal sending thread that it can continue. */
125	144	if (sl->body.block_mutex) {
126	144	uv_mutex_lock(sl->body.block_mutex);
127	144	MVM_store(&(sl->body.completed), 1);
128	144	uv_cond_signal(sl->body.block_condvar);
129	144	uv_mutex_unlock(sl->body.block_mutex);
130	144	}
131	144	});
132	144	}
133	144	else {
134	144	MVM_panic(1, "Unexpected object sent to specialization worker");
135	144	}
136	144
137	144	MVM_telemetry_interval_stop(tc, interval_id, "spesh worker finished");
138	144
139	144	if (tc->instance->main_thread->prof_data)
140	144	MVM_profiler_log_spesh_end(tc);
141	144
142	144	uv_mutex_lock(&(tc->instance->mutex_spesh_sync));
143	144	tc->instance->spesh_working = 0;
144	144	uv_cond_broadcast(&(tc->instance->cond_spesh_sync));
145	144	uv_mutex_unlock(&(tc->instance->mutex_spesh_sync));
146	144	}
147	144	});
148	144	}
149
150	144	void MVM_spesh_worker_setup(MVMThreadContext *tc) {
151	144	if (tc->instance->spesh_enabled) {
152	144	MVMObject *worker_entry_point;
153	144	tc->instance->spesh_queue = MVM_repr_alloc_init(tc, tc->instance->boot_types.BOOTQueue);
154	144	worker_entry_point = MVM_repr_alloc_init(tc, tc->instance->boot_types.BOOTCCode);
155	144	((MVMCFunction *)worker_entry_point)->body.func = worker;
156	144	MVM_thread_run(tc, MVM_thread_new(tc, worker_entry_point, 1));
157	144	}
158	144	}

Coverage Report

Created: 2018-07-03 15:31