Direct dispatch: bypass worker threads, execute blocking op inline

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: samuel-williams-shopify

ext/io/event/worker_pool.c

@@ -342,58 +342,10 @@ static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) {
 		rb_raise(rb_eArgError, "Invalid blocking operation!");
 	}
 
-	// Create work item.
-	// Store the VALUE alongside the raw pointer — the compacting GC may move the
-	// TypedData object while the calling fiber is suspended, making the raw pointer
-	// stale. Registering the VALUEs as precise GC roots ensures they are updated
-	// to the new address, allowing safe re-extraction in the worker thread.
-	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);
+	// Direct dispatch: execute the blocking operation inline without worker threads.
+	rb_fiber_scheduler_blocking_operation_execute(blocking_operation);
 
-	// Block the current fiber until work is completed:
-	int state = 0;
-	while (true) {
-		int current_state = 0;
-		rb_protect(worker_pool_work_begin, (VALUE)&work, &current_state);
-		if (DEBUG) fprintf(stderr, "-- worker_pool_call:work completed=%d, current_state=%d, state=%d\n", work.completed, current_state, state);
-		
-		// Store the first exception state:
-		if (!state) {
-			state = current_state;
-		}
-		
-		// If the work is still in the queue, we must wait for a worker to complete it (even if cancelled):
-		if (work.completed) {
-			// The work was completed, we can exit the loop:
-			break;
-		} else {
-			if (DEBUG) fprintf(stderr, "worker_pool_call:rb_fiber_scheduler_blocking_operation_cancel\n");
-			// Ensure the blocking operation is cancelled:
-			rb_fiber_scheduler_blocking_operation_cancel(work.blocking_operation);
-			
-			// The work was not completed, we need to wait for it to be completed, so we go around the loop again.
-		}
-	}
-	
-	if (DEBUG) fprintf(stderr, "<- worker_pool_call:work completed=%d, state=%d\n", work.completed, state);
-	
-	if (state) {
-		rb_jump_tag(state);
-	} else {
-		return Qtrue;
-	}
+	return Qtrue;
 }
 
 static VALUE worker_pool_allocate(VALUE klass) {