vm: write async signals to pipe for multiplexer

also factor out safepoint logic into its own file

GNUmakefile

@@ -59,6 +59,7 @@ ifdef CONFIG
 		vm/primitives.o \
 		vm/quotations.o \
 		vm/run.o \
+		vm/safepoints.o \
 		vm/sampling_profiler.o \
 		vm/strings.o \
 		vm/to_tenured_collector.o \

Nmakefile

@@ -63,6 +63,7 @@ DLL_OBJS = $(PLAF_DLL_OBJS) \
 	vm\primitives.obj \
 	vm\quotations.obj \
 	vm\run.obj \
+	vm\safepoints.obj \
 	vm\sampling_profiler.obj \
 	vm\strings.obj \
 	vm\to_tenured_collector.obj \

vm/errors.cpp

@@ -89,7 +89,7 @@ void factor_vm::not_implemented_error()
 void factor_vm::memory_protection_error(cell addr)
 {
 	if(code->safepoint_p(addr))
-		handle_safepoint();
+		safepoint.handle_safepoint();
 	else if(faulting_p)
 		fatal_error("Double fault", 0);
 	else if(ctx->datastack_seg->underflow_p(addr))
@@ -167,28 +167,4 @@ void fp_signal_handler_impl()
 {
 	current_vm()->fp_signal_handler_impl();
 }
-
-void factor_vm::enqueue_safepoint_fep()
-{
-	if (fep_p)
-		fatal_error("Low-level debugger interrupted", 0);
-	atomic::store(&safepoint_fep_p, true);
-	code->guard_safepoint();
-}
-
-void factor_vm::handle_safepoint()
-{
-	code->unguard_safepoint();
-	if (atomic::load(&safepoint_fep_p))
-	{
-		if (atomic::load(&sampling_profiler_p))
-			end_sampling_profiler();
-		std::cout << "Interrupted\n";
-		factorbug();
-		atomic::store(&safepoint_fep_p, false);
-	}
-	else if (sampling_profiler_p)
-		record_sample();
-}
-
 }

vm/master.hpp

@@ -127,6 +127,7 @@ namespace factor
 #include "callbacks.hpp"
 #include "dispatch.hpp"
 #include "entry_points.hpp"
+#include "safepoints.hpp"
 #include "vm.hpp"
 #include "allot.hpp"
 #include "tagged.hpp"

vm/objects.hpp

@@ -104,6 +104,9 @@ enum special_object {
 	OBJ_VM_COMPILER = 72,     /* version string of the compiler we were built with */
 
 	OBJ_WAITING_CALLBACKS = 73,
+
+	OBJ_SIGNAL_PIPE = 74,     /* file descriptor for pipe used to communicate signals
+	                          only used on unix */
 };
 
 /* save-image-and-exit discards special objects that are filled in on startup

vm/os-unix.cpp

@@ -146,13 +146,6 @@ void factor_vm::dispatch_signal(void *uap, void (handler)())
 	UAP_SET_TOC_POINTER(uap, (cell)FUNCTION_TOC_POINTER(handler));
 }
 
-void factor_vm::enqueue_safepoint_signal(cell signal)
-{
-	/* to be implemented, see #297
-	code->guard_safepoint();
-	*/
-}
-
 void factor_vm::start_sampling_profiler_timer()
 {
 	struct itimerval timer;
@@ -191,7 +184,7 @@ void enqueue_signal_handler(int signal, siginfo_t *siginfo, void *uap)
 {
 	factor_vm *vm = current_vm_p();
 	if (vm)
-		vm->enqueue_safepoint_signal(signal);
+		vm->safepoint.enqueue_signal(signal);
 	else
 		fatal_error("Foreign thread received signal", signal);
 }
@@ -200,7 +193,7 @@ void fep_signal_handler(int signal, siginfo_t *siginfo, void *uap)
 {
 	factor_vm *vm = current_vm_p();
 	if (vm)
-		vm->enqueue_safepoint_fep();
+		vm->safepoint.enqueue_fep();
 	else
 		fatal_error("Foreign thread received signal", signal);
 }
@@ -209,10 +202,10 @@ void sample_signal_handler(int signal, siginfo_t *siginfo, void *uap)
 {
 	factor_vm *vm = current_vm_p();
 	if (vm)
-		vm->enqueue_safepoint_sample(1, (cell)UAP_PROGRAM_COUNTER(uap), false);
+		vm->safepoint.enqueue_samples(1, (cell)UAP_PROGRAM_COUNTER(uap), false);
 	else if (thread_vms.size() == 1) {
 		factor_vm *the_only_vm = thread_vms.begin()->second;
-		the_only_vm->enqueue_safepoint_sample(1, (cell)UAP_PROGRAM_COUNTER(uap), true);
+		the_only_vm->safepoint.enqueue_samples(1, (cell)UAP_PROGRAM_COUNTER(uap), true);
 	}
 }
 
@@ -255,8 +248,33 @@ static void init_sigaction_with_handler(struct sigaction *act,
 	act->sa_flags = SA_SIGINFO | SA_ONSTACK;
 }
 
+static void safe_pipe(int *in, int *out)
+{
+	int filedes[2];
+
+	if(pipe(filedes) < 0)
+		fatal_error("Error opening pipe",errno);
+
+	*in = filedes[0];
+	*out = filedes[1];
+
+	if(fcntl(*in,F_SETFD,FD_CLOEXEC) < 0)
+		fatal_error("Error with fcntl",errno);
+
+	if(fcntl(*out,F_SETFD,FD_CLOEXEC) < 0)
+		fatal_error("Error with fcntl",errno);
+}
+
+static void init_signal_pipe(factor_vm *vm)
+{
+	safe_pipe(&vm->signal_pipe_input, &vm->signal_pipe_output);
+	vm->special_objects[OBJ_SIGNAL_PIPE] = tag_fixnum(vm->signal_pipe_output);
+}
+
 void factor_vm::unix_init_signals()
 {
+	init_signal_pipe(this);
+
 	signal_callstack_seg = new segment(callstack_size,false);
 
 	stack_t signal_callstack;
@@ -417,23 +435,6 @@ void *stdin_loop(void *arg)
 	return NULL;
 }
 
-void safe_pipe(int *in, int *out)
-{
-	int filedes[2];
-
-	if(pipe(filedes) < 0)
-		fatal_error("Error opening pipe",errno);
-
-	*in = filedes[0];
-	*out = filedes[1];
-
-	if(fcntl(*in,F_SETFD,FD_CLOEXEC) < 0)
-		fatal_error("Error with fcntl",errno);
-
-	if(fcntl(*out,F_SETFD,FD_CLOEXEC) < 0)
-		fatal_error("Error with fcntl",errno);
-}
-
 void open_console()
 {
 	safe_pipe(&control_read,&control_write);
@@ -442,4 +443,14 @@ void open_console()
 	start_thread(stdin_loop,NULL);
 }
 
+void safepoint_state::report_signal(int fd) volatile
+{
+	int signal = (int)atomic::load(&queued_signal);
+	if (signal != 0)
+	{
+		safe_write(fd, &signal, sizeof(int));
+		atomic::store(&queued_signal, 0);
+	}
+}
+
 }

vm/os-windows.cpp

@@ -169,6 +169,7 @@ void factor_vm::move_file(const vm_char *path1, const vm_char *path2)
 }
 
 void factor_vm::init_signals() {}
+void safepoint_state::report_signal(int fd) volatile {}
 
 THREADHANDLE start_thread(void *(*start_routine)(void *), void *args)
 {
@@ -288,7 +289,7 @@ static BOOL WINAPI ctrl_handler(DWORD dwCtrlType)
 		threads. */
 		assert(thread_vms.size() == 1);
 		factor_vm *vm = thread_vms.begin()->second;
-		vm->enqueue_safepoint_fep();
+		vm->safepoint.enqueue_fep();
 		return TRUE;
 	}
 	default:
@@ -340,7 +341,7 @@ void factor_vm::sampler_thread_loop()
 			suscount = ResumeThread(thread);
 			assert(suscount == 1);
 
-			enqueue_safepoint_sample(samples, context.EIP, false);
+			safepoint.enqueue_samples(samples, context.EIP, false);
 		}
 	}
 }

vm/safepoints.cpp

@@ -0,0 +1,62 @@
+#include "master.hpp"
+
+namespace factor
+{
+
+void safepoint_state::enqueue_safepoint() volatile
+{
+	parent->code->guard_safepoint();
+}
+
+void safepoint_state::enqueue_fep() volatile
+{
+	if (parent->fep_p)
+		fatal_error("Low-level debugger interrupted", 0);
+	atomic::store(&fep_p, true);
+	enqueue_safepoint();
+}
+
+void safepoint_state::enqueue_signal(cell signal) volatile
+{
+	atomic::store(&queued_signal, signal);
+	enqueue_safepoint();
+}
+
+void safepoint_state::enqueue_samples(cell samples, cell pc, bool foreign_thread_p) volatile
+{
+	if (atomic::load(&parent->sampling_profiler_p))
+	{
+		atomic::add(&sample_counts.sample_count, samples);
+		if (foreign_thread_p)
+			atomic::add(&sample_counts.foreign_thread_sample_count, samples);
+		else {
+			if (atomic::load(&parent->current_gc_p))
+				atomic::fetch_add(&sample_counts.gc_sample_count, samples);
+			if (atomic::load(&parent->current_jit_count) > 0)
+				atomic::fetch_add(&sample_counts.jit_sample_count, samples);
+			if (!parent->code->seg->in_segment_p(pc))
+				atomic::fetch_add(&sample_counts.foreign_sample_count, samples);
+		}
+		enqueue_safepoint();
+	}
+}
+
+void safepoint_state::handle_safepoint() volatile
+{
+	parent->code->unguard_safepoint();
+
+	report_signal(parent->signal_pipe_input);
+
+	if (atomic::load(&fep_p))
+	{
+		if (atomic::load(&parent->sampling_profiler_p))
+			parent->end_sampling_profiler();
+		std::cout << "Interrupted\n";
+		parent->factorbug();
+		atomic::store(&fep_p, false);
+	}
+	else if (atomic::load(&parent->sampling_profiler_p))
+		parent->record_sample();
+}
+
+}

vm/safepoints.hpp

@@ -0,0 +1,32 @@
+namespace factor
+{
+
+struct safepoint_state
+{
+	factor_vm *parent;
+
+	cell fep_p;
+	cell queued_signal;
+	profiling_sample_count sample_counts;
+
+	safepoint_state(factor_vm *parent) :
+		parent(parent),
+		fep_p(false),
+		queued_signal(0),
+		sample_counts()
+	{
+	}
+
+	void handle_safepoint() volatile;
+
+	void enqueue_safepoint() volatile;
+	void enqueue_samples(cell samples, cell pc, bool foreign_thread_p) volatile;
+	void enqueue_fep() volatile;
+
+	// os-*.cpp
+	void enqueue_signal(cell signal) volatile;
+	void report_signal(int fd) volatile;
+
+};
+
+}

vm/sampling_profiler.cpp

@@ -42,7 +42,7 @@ profiling_sample::profiling_sample(factor_vm *vm,
 
 void factor_vm::record_sample()
 {
-	profiling_sample_count counts = safepoint_sample_counts.record_counts();
+	profiling_sample_count counts = safepoint.sample_counts.record_counts();
 	if (!counts.empty())
 		samples.push_back(profiling_sample(this,
 			counts, special_objects[OBJ_CURRENT_THREAD]));
@@ -86,7 +86,7 @@ void factor_vm::clear_samples()
 void factor_vm::start_sampling_profiler(fixnum rate)
 {
 	samples_per_second = rate;
-	safepoint_sample_counts.clear();
+	safepoint.sample_counts.clear();
 	clear_samples();
 	samples.reserve(10*rate);
 	sample_callstacks.reserve(100*rate);
@@ -152,23 +152,4 @@ void factor_vm::primitive_clear_samples()
 	clear_samples();
 }
 
-void factor_vm::enqueue_safepoint_sample(cell samples, cell pc, bool foreign_thread_p)
-{
-	if (atomic::load(&sampling_profiler_p))
-	{
-		atomic::fetch_add(&safepoint_sample_counts.sample_count, samples);
-		if (foreign_thread_p)
-			atomic::fetch_add(&safepoint_sample_counts.foreign_thread_sample_count, samples);
-		else {
-			if (atomic::load(&current_gc_p))
-				atomic::fetch_add(&safepoint_sample_counts.gc_sample_count, samples);
-			if (atomic::load(&current_jit_count) > 0)
-				atomic::fetch_add(&safepoint_sample_counts.jit_sample_count, samples);
-			if (!code->seg->in_segment_p(pc))
-				atomic::fetch_add(&safepoint_sample_counts.foreign_sample_count, samples);
-		}
-		code->guard_safepoint();
-	}
-}
-
 }

vm/vm.cpp

@@ -10,7 +10,8 @@ factor_vm::factor_vm(THREADHANDLE thread) :
 	c_to_factor_func(NULL),
 	counting_profiler_p(false),
 	sampling_profiler_p(false),
-	safepoint_fep_p(false),
+	signal_pipe_input(0),
+	signal_pipe_output(0),
 	gc_off(false),
 	current_gc(NULL),
 	current_gc_p(false),
@@ -22,7 +23,8 @@ factor_vm::factor_vm(THREADHANDLE thread) :
 	full_output(false),
 	last_nano_count(0),
 	signal_callstack_seg(NULL),
-	faulting_p(false)
+	faulting_p(false),
+	safepoint(this)
 {
 	primitive_reset_dispatch_stats();
 }

vm/vm.hpp

@@ -36,8 +36,8 @@ struct factor_vm
 	// ^^^^^^
 	//
 
-        /* Handle to the main thread we run in */
+	/* Handle to the main thread we run in */
-        THREADHANDLE thread;
+	THREADHANDLE thread;
 
 	/* Data stack and retain stack sizes */
 	cell datastack_size, retainstack_size, callstack_size;
@@ -74,12 +74,13 @@ struct factor_vm
 	cell signal_number;
 	cell signal_fault_addr;
 	unsigned int signal_fpu_status;
-	volatile cell safepoint_fep_p;
+
+	/* Pipe used to notify Factor multiplexer of signals */
+	int signal_pipe_input, signal_pipe_output;
 
 	/* State kept by the sampling profiler */
 	std::vector<profiling_sample> samples;
 	std::vector<cell> sample_callstacks;
-	volatile profiling_sample_count safepoint_sample_counts;
 
 	/* GC is off during heap walking */
 	bool gc_off;
@@ -144,6 +145,9 @@ struct factor_vm
 	/* Are we already handling a fault? Used to catch double memory faults */
 	bool faulting_p;
 
+	/* Safepoint state */
+	volatile safepoint_state safepoint;
+
 	// contexts
 	context *new_context();
 	void init_context(context *ctx);
@@ -208,7 +212,6 @@ struct factor_vm
 	void start_sampling_profiler(fixnum rate);
 	void end_sampling_profiler();
 	void set_sampling_profiler(fixnum rate);
-	void enqueue_safepoint_sample(cell samples, cell pc, bool foreign_thread_p);
 	void primitive_sampling_profiler();
 	void primitive_get_samples();
 	void primitive_clear_samples();
@@ -225,8 +228,6 @@ struct factor_vm
 	void memory_signal_handler_impl();
 	void synchronous_signal_handler_impl();
 	void fp_signal_handler_impl();
-	void enqueue_safepoint_fep();
-	void handle_safepoint();
 
 	// bignum
 	int bignum_equal_p(bignum * x, bignum * y);
@@ -751,7 +752,6 @@ struct factor_vm
 
   #else  // UNIX
 	void dispatch_signal(void *uap, void (handler)());
-	void enqueue_safepoint_signal(cell signal);
 	void unix_init_signals();
   #endif