Logging.

Author: ioquatix

ext/io/event/worker_pool.c

@@ -143,58 +143,89 @@ static void *worker_wait_and_execute(void *_worker) {
 	struct IO_Event_WorkerPool_Worker *worker = (struct IO_Event_WorkerPool_Worker *)_worker;
 	struct IO_Event_WorkerPool *pool = worker->pool;
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: starting worker=%p pool=%p\n", (void*)worker, (void*)pool);
+	
 	while (true) {
 		struct IO_Event_WorkerPool_Work *work = NULL;
 
+		fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: acquiring mutex worker=%p\n", (void*)worker);
 		pthread_mutex_lock(&pool->mutex);
 
 		// Wait for work, shutdown, or interruption
 		while (!pool->work_queue && !pool->shutdown && !worker->interrupted) {
+			fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: waiting for work worker=%p\n", (void*)worker);
 			pthread_cond_wait(&pool->work_available, &pool->mutex);
+			fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: woke up worker=%p\n", (void*)worker);
 		}
 
 		if (pool->shutdown || worker->interrupted) {
+			fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: shutdown=%d interrupted=%d worker=%p\n", 
+				pool->shutdown, worker->interrupted, (void*)worker);
 			pthread_mutex_unlock(&pool->mutex);
 			break;
 		}
 
 		work = dequeue_work(pool);
+		fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: dequeued work=%p worker=%p\n", (void*)work, (void*)worker);
 
 		pthread_mutex_unlock(&pool->mutex);
 
 		// Execute work WITHOUT GVL (this is the whole point!)
 		if (work) {
+			fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: executing work=%p blocking_operation=%p worker=%p\n", 
+				(void*)work, (void*)work->blocking_operation, (void*)worker);
+			
 			worker->current_blocking_operation = work->blocking_operation;
 			rb_fiber_scheduler_blocking_operation_execute(work->blocking_operation);
 			worker->current_blocking_operation = NULL;
+			
+			fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: work execution completed work=%p worker=%p\n", 
+				(void*)work, (void*)worker);
 		}
 
+		fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: returning work=%p worker=%p\n", (void*)work, (void*)worker);
 		return work;
 	}
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_wait_and_execute: returning NULL (shutdown) worker=%p\n", (void*)worker);
 	return NULL; // Shutdown signal
 }
 
 static VALUE worker_thread_func(void *_worker) {
 	struct IO_Event_WorkerPool_Worker *worker = (struct IO_Event_WorkerPool_Worker *)_worker;
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_thread_func: starting worker=%p\n", (void*)worker);
+	
 	while (true) {
+		fprintf(stderr, "[WorkerPool DEBUG] worker_thread_func: calling rb_thread_call_without_gvl worker=%p\n", (void*)worker);
+		
 		// Wait for work and execute it without holding GVL
-		struct IO_Event_WorkerPool_Work *work = (struct IO_Event_WorkerPool_Work *)rb_thread_call_without_gvl(worker_wait_and_execute, worker, worker_unblock_func, worker);
+		void *result = rb_thread_call_without_gvl(worker_wait_and_execute, worker, worker_unblock_func, worker);
+		struct IO_Event_WorkerPool_Work *work = (struct IO_Event_WorkerPool_Work *)result;
+		
+		fprintf(stderr, "[WorkerPool DEBUG] worker_thread_func: rb_thread_call_without_gvl returned work=%p (raw=%p) worker=%p\n", (void*)work, result, (void*)worker);
 
 		if (!work) {
-			// Shutdown signal received
+			fprintf(stderr, "[WorkerPool DEBUG] worker_thread_func: shutdown signal received worker=%p\n", (void*)worker);
 			break;
 		}
 
+		fprintf(stderr, "[WorkerPool DEBUG] worker_thread_func: setting work->completed=true work=%p worker=%p\n", (void*)work, (void*)worker);
+		
 		// Protected by GVL:
 		work->completed = true;
 		worker->pool->completed_count++;
 
+		fprintf(stderr, "[WorkerPool DEBUG] worker_thread_func: calling rb_fiber_scheduler_unblock work=%p scheduler=%p blocker=%p fiber=%p worker=%p\n", 
+			(void*)work, (void*)work->scheduler, (void*)work->blocker, (void*)work->fiber, (void*)worker);
+		
 		// Work was executed without GVL, now unblock the waiting fiber (we have GVL here)
 		rb_fiber_scheduler_unblock(work->scheduler, work->blocker, work->fiber);
+		
+		fprintf(stderr, "[WorkerPool DEBUG] worker_thread_func: rb_fiber_scheduler_unblock completed work=%p worker=%p\n", (void*)work, (void*)worker);
 	}
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_thread_func: exiting worker=%p\n", (void*)worker);
 	return Qnil;
 }
 
@@ -287,9 +318,12 @@ static VALUE worker_pool_initialize(int argc, VALUE *argv, VALUE self) {
 static VALUE worker_pool_work_begin(VALUE _work) {
 	struct IO_Event_WorkerPool_Work *work = (void*)_work;
 
-	if (DEBUG) fprintf(stderr, "worker_pool_work_begin:rb_fiber_scheduler_block work=%p\n", work);
+	fprintf(stderr, "[WorkerPool DEBUG] worker_pool_work_begin: work=%p scheduler=%p blocker=%p fiber=%p\n", 
+		work, (void*)work->scheduler, (void*)work->blocker, (void*)work->fiber);
+	
 	rb_fiber_scheduler_block(work->scheduler, work->blocker, Qnil);
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_pool_work_begin: completed work=%p\n", work);
 	return Qnil;
 }
 
@@ -298,6 +332,8 @@ 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);
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_pool_call: starting pool=%p\n", (void*)pool);
+	
 	if (pool->shutdown) {
 		rb_raise(rb_eRuntimeError, "Worker pool is shut down!");
 	}
@@ -319,6 +355,9 @@ static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) {
 		rb_raise(rb_eArgError, "Invalid blocking operation!");
 	}
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_pool_call: blocking_operation=%p fiber=%p scheduler=%p\n", 
+		(void*)blocking_operation, (void*)fiber, (void*)scheduler);
+	
 	// Create work item
 	struct IO_Event_WorkerPool_Work work = {
 		.blocking_operation = blocking_operation,
@@ -328,27 +367,40 @@ static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) {
 		.fiber = fiber,
 		.next = NULL
 	};
+	
+	fprintf(stderr, "[WorkerPool DEBUG] worker_pool_call: created work item work=%p on stack\n", (void*)&work);
 
 	// Enqueue work:
 	pthread_mutex_lock(&pool->mutex);
 	enqueue_work(pool, &work);
 	pthread_cond_signal(&pool->work_available);
 	pthread_mutex_unlock(&pool->mutex);
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_pool_call: enqueued work, starting wait loop work=%p\n", (void*)&work);
+	
 	// Block the current fiber until work is completed:
 	int state;
+	int iteration = 0;
 	while (true) {
+		iteration++;
+		fprintf(stderr, "[WorkerPool DEBUG] worker_pool_call: iteration %d, work.completed=%d work=%p\n", 
+			iteration, work.completed, (void*)&work);
+		
 		rb_protect(worker_pool_work_begin, (VALUE)&work, &state);
 
 		if (work.completed) {
+			fprintf(stderr, "[WorkerPool DEBUG] worker_pool_call: work completed, breaking work=%p\n", (void*)&work);
 			break;
 		} else {
-			if (DEBUG) fprintf(stderr, "worker_pool_call:rb_fiber_scheduler_blocking_operation_cancel\n");
+			fprintf(stderr, "[WorkerPool DEBUG] worker_pool_call: work not completed, cancelling blocking_operation=%p\n", 
+				(void*)blocking_operation);
 			rb_fiber_scheduler_blocking_operation_cancel(blocking_operation);
 			// The work was not completed, we need to wait for it to be completed.
 		}
 	}
 
+	fprintf(stderr, "[WorkerPool DEBUG] worker_pool_call: exiting work=%p state=%d\n", (void*)&work, state);
+	
 	if (state) {
 		rb_jump_tag(state);
 	} else {