Class: IO::Event::WorkerPool

Inherits:

Object

• Object
• IO::Event::WorkerPool

Defined in:

ext/io/event/worker_pool.c

Class Method Summary

• .busy(*args) ⇒ Object

  This creates a cancellable blocking operation for testing.

Instance Method Summary

• #call(_blocking_operation) ⇒ Object
• #close ⇒ Object
• #initialize(*args) ⇒ Object constructor
• #statistics ⇒ Object

Constructor Details

#initialize(*args) ⇒ Object

# File 'ext/io/event/worker_pool.c', line 252

static VALUE worker_pool_initialize(int argc, VALUE *argv, VALUE self) {
	size_t maximum_worker_count = 1; // Default
	
	// Extract keyword arguments
	VALUE kwargs = Qnil;
	VALUE rb_maximum_worker_count = Qnil;
	
	rb_scan_args(argc, argv, "0:", &kwargs);
	
	if (!NIL_P(kwargs)) {
		VALUE kwvals[1];
		ID kwkeys[1] = {id_maximum_worker_count};
		rb_get_kwargs(kwargs, kwkeys, 0, 1, kwvals);
		rb_maximum_worker_count = kwvals[0];
	}
	
	if (!NIL_P(rb_maximum_worker_count)) {
		maximum_worker_count = NUM2SIZET(rb_maximum_worker_count);
		if (maximum_worker_count == 0) {
			rb_raise(rb_eArgError, "maximum_worker_count must be greater than 0!");
		}
	}
	
	// Get the pool that was allocated by worker_pool_allocate
	struct IO_Event_WorkerPool *pool;
	TypedData_Get_Struct(self, struct IO_Event_WorkerPool, &IO_Event_WorkerPool_type, pool);
	
	if (!pool) {
		rb_raise(rb_eRuntimeError, "WorkerPool allocation failed!");
	}
	
	pthread_mutex_init(&pool->mutex, NULL);
	pthread_cond_init(&pool->work_available, NULL);
	
	pool->work_queue = NULL;
	pool->work_queue_tail = NULL;
	pool->workers = NULL;
	pool->current_worker_count = 0;
	pool->maximum_worker_count = maximum_worker_count;
	pool->call_count = 0;
	pool->completed_count = 0;
	pool->cancelled_count = 0;
	pool->shutdown = false;
	
	// Create initial workers
	for (size_t i = 0; i < maximum_worker_count; i++) {
		if (create_worker_thread(self, pool) != 0) {
			// Just set the maximum_worker_count for debugging, don't fail completely
			// worker_pool_free(pool);
			// rb_raise(rb_eRuntimeError, "Failed to create workers");
			break;
		}
	}
	
	return self;
}

Class Method Details

.busy(*args) ⇒ Object

This creates a cancellable blocking operation for testing

# File 'ext/io/event/worker_pool_test.c', line 115

static VALUE worker_pool_test_busy(int argc, VALUE *argv, VALUE self) {
	double duration = 1.0;  // Default 1 second
	
	// Extract keyword arguments
	VALUE kwargs = Qnil;
	VALUE rb_duration = Qnil;
	
	rb_scan_args(argc, argv, "0:", &kwargs);
	
	if (!NIL_P(kwargs)) {
		VALUE kwvals[1];
		ID kwkeys[1] = {id_duration};
		rb_get_kwargs(kwargs, kwkeys, 0, 1, kwvals);
		rb_duration = kwvals[0];
	}
	
	if (!NIL_P(rb_duration)) {
		duration = NUM2DBL(rb_duration);
	}
	
	// Create pipe for cancellation
	int pipe_fds[2];
	if (pipe(pipe_fds) != 0) {
		rb_sys_fail("pipe creation failed");
	}
	
	// Stack allocate operation data
	struct BusyOperationData busy_data = {
		.read_fd = pipe_fds[0],
		.write_fd = pipe_fds[1],
		.cancelled = 0,
		.duration = duration,
		.start_time = 0,
		.end_time = 0,
		.operation_result = 0,
		.exception = Qnil
	};
	
	// Execute the blocking operation with exception handling using function pointers
	rb_rescue(
		busy_operation_execute,
		(VALUE)&busy_data,
		busy_operation_rescue,
		(VALUE)&busy_data
	);
	
	// Calculate elapsed time from the state stored in busy_data
	double elapsed = ((double)(busy_data.end_time - busy_data.start_time)) / CLOCKS_PER_SEC;
	
	// Create result hash using the state from busy_data
	VALUE result = rb_hash_new();
	rb_hash_aset(result, ID2SYM(rb_intern("duration")), DBL2NUM(duration));
	rb_hash_aset(result, ID2SYM(rb_intern("elapsed")), DBL2NUM(elapsed));
	
	// Determine result based on operation outcome
	if (busy_data.exception != Qnil) {
		rb_hash_aset(result, ID2SYM(rb_intern("result")), ID2SYM(rb_intern("exception")));
		rb_hash_aset(result, ID2SYM(rb_intern("cancelled")), Qtrue);
		rb_hash_aset(result, ID2SYM(rb_intern("exception")), busy_data.exception);
	} else if (busy_data.operation_result == -1) {
		rb_hash_aset(result, ID2SYM(rb_intern("result")), ID2SYM(rb_intern("cancelled")));
		rb_hash_aset(result, ID2SYM(rb_intern("cancelled")), Qtrue);
	} else if (busy_data.operation_result == 0) {
		rb_hash_aset(result, ID2SYM(rb_intern("result")), ID2SYM(rb_intern("completed")));
		rb_hash_aset(result, ID2SYM(rb_intern("cancelled")), Qfalse);
	} else {
		rb_hash_aset(result, ID2SYM(rb_intern("result")), ID2SYM(rb_intern("error")));
		rb_hash_aset(result, ID2SYM(rb_intern("cancelled")), Qfalse);
	}
	
	// Clean up pipe file descriptors
	close(pipe_fds[0]);
	close(pipe_fds[1]);
	
	return result;
}

Instance Method Details

#call(_blocking_operation) ⇒ Object

# File 'ext/io/event/worker_pool.c', line 358

static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) {
	struct IO_Event_WorkerPool *pool;
	TypedData_Get_Struct(self, struct IO_Event_WorkerPool, &IO_Event_WorkerPool_type, pool);
	
	if (pool->shutdown) {
		rb_raise(rb_eRuntimeError, "Worker pool is shut down!");
	}
	
	// Increment call count (protected by GVL)
	pool->call_count++;
	
	// Get current fiber and scheduler
	VALUE fiber = rb_fiber_current();
	VALUE scheduler = rb_fiber_scheduler_current();
	if (NIL_P(scheduler)) {
		rb_raise(rb_eRuntimeError, "WorkerPool requires a fiber scheduler!");
	}
	
	// Extract blocking operation handle
	rb_fiber_scheduler_blocking_operation_t *blocking_operation = rb_fiber_scheduler_blocking_operation_extract(_blocking_operation);
	
	if (!blocking_operation) {
		rb_raise(rb_eArgError, "Invalid blocking operation!");
	}
	
	// Create work item
	struct IO_Event_WorkerPool_Work work = {
		.blocking_operation = blocking_operation,
		.completed = false,
		.scheduler = scheduler,
		.blocker = self,
		.fiber = fiber,
		.next = NULL
	};
		
	// Enqueue work:
	pthread_mutex_lock(&pool->mutex);
	enqueue_work(pool, &work);
	pthread_cond_signal(&pool->work_available);
	pthread_mutex_unlock(&pool->mutex);
	
	// Block the current fiber until work is completed. If blocking exits via an
	// exception or another non-local jump, ensure the work is cancelled and fully
	// drained before Ruby restores and rethrows the original control-flow state.
	return rb_ensure(worker_pool_work_wait, (VALUE)&work, worker_pool_work_ensure, (VALUE)&work);
}

#close ⇒ Object

# File 'ext/io/event/worker_pool.c', line 416

static VALUE worker_pool_close(VALUE self) {
	struct IO_Event_WorkerPool *pool;
	TypedData_Get_Struct(self, struct IO_Event_WorkerPool, &IO_Event_WorkerPool_type, pool);
	
	if (!pool) {
		rb_raise(rb_eRuntimeError, "WorkerPool not initialized!");
	}
	
	if (pool->shutdown) {
		return Qnil; // Already closed
	}
	
	// Signal shutdown to all workers
	pthread_mutex_lock(&pool->mutex);
	pool->shutdown = true;
	pthread_cond_broadcast(&pool->work_available);
	pthread_mutex_unlock(&pool->mutex);
	
	// Wait for all worker threads to finish
	struct IO_Event_WorkerPool_Worker *worker = pool->workers;
	while (worker) {
		if (!NIL_P(worker->thread)) {
			rb_funcall(worker->thread, rb_intern("join"), 0);
		}
		worker = worker->next;
	}
	
	// Clean up worker structures
	worker = pool->workers;
	while (worker) {
		struct IO_Event_WorkerPool_Worker *next = worker->next;
		free(worker);
		worker = next;
	}
	pool->workers = NULL;
	pool->current_worker_count = 0;
	
	// Clean up mutex and condition variable
	pthread_mutex_destroy(&pool->mutex);
	pthread_cond_destroy(&pool->work_available);
	
	return Qnil;
}

#statistics ⇒ Object

# File 'ext/io/event/worker_pool.c', line 461

static VALUE worker_pool_statistics(VALUE self) {
	struct IO_Event_WorkerPool *pool;
	TypedData_Get_Struct(self, struct IO_Event_WorkerPool, &IO_Event_WorkerPool_type, pool);
	
	if (!pool) {
		rb_raise(rb_eRuntimeError, "WorkerPool not initialized!");
	}
	
	VALUE stats = rb_hash_new();
	rb_hash_aset(stats, ID2SYM(rb_intern("current_worker_count")), SIZET2NUM(pool->current_worker_count));
	rb_hash_aset(stats, ID2SYM(rb_intern("maximum_worker_count")), SIZET2NUM(pool->maximum_worker_count));
	rb_hash_aset(stats, ID2SYM(rb_intern("call_count")), SIZET2NUM(pool->call_count));
	rb_hash_aset(stats, ID2SYM(rb_intern("completed_count")), SIZET2NUM(pool->completed_count));
	rb_hash_aset(stats, ID2SYM(rb_intern("cancelled_count")), SIZET2NUM(pool->cancelled_count));
	rb_hash_aset(stats, ID2SYM(rb_intern("shutdown")), pool->shutdown ? Qtrue : Qfalse);
	
	// Count work items in queue (only if properly initialized)
	if (pool->maximum_worker_count > 0) {
		pthread_mutex_lock(&pool->mutex);
		size_t current_queue_size = 0;
		struct IO_Event_WorkerPool_Work *work = pool->work_queue;
		while (work) {
			current_queue_size++;
			work = work->next;
		}
		pthread_mutex_unlock(&pool->mutex);
		rb_hash_aset(stats, ID2SYM(rb_intern("current_queue_size")), SIZET2NUM(current_queue_size));
	} else {
		rb_hash_aset(stats, ID2SYM(rb_intern("current_queue_size")), SIZET2NUM(0));
	}
	
	return stats;
}