Sparkcore-Go: Distributed Task Execution

Github Link: Sparkcore-Go

In session one, I successfully implemented RDD lineage building, stage planning, and partition-level tasks in a single Go process. Goroutines provide concurrency, and channels coordinate that concurrency at the task execution level. But all goroutines share the same memory, function registry, filesystem access, and process lifecycle. An unrecovered panic in one goroutine can bring down the entire application.

In this session, I replace the single process with distributed execution to establish process and network boundaries. I introduce a physical scheduler and multiple independent worker processes that communicate with the coordinator over HTTP.

The coordinator shares the same Go process as the scheduler, but deliberately handles HTTP endpoints, JSON decoding, request validation, and error responses to avoid mixing network handling with scheduler placement logic.

Scheduling Tasks

Let’s start with the scheduler and scheduling policies. Here, I mimic Spark’s FIFO scheduling policies with similar capacity-matching rules, but simplify some internal mechanisms and use a queue-based scheduling model. I set the following three rules:

1. Select the oldest task set with pending tasks.

Spark schedules by active stages and priority pools rather than strictly picking the “oldest” task set globally. The scheduler first appends each accepted task set to s.queue (scheduler/fifo_task_scheduler.go, line 136), then scans that queue in insertion order (line 239).

// Append task sets in admission order.
s.queue = append(s.queue, set)

// Visit them in the same order.
for _, set := range s.queue {
    if set.terminal {
        continue
    }
    if err := set.ctx.Err(); err != nil {
        s.finish(set, err)
        continue
    }
    // Assign pending tasks.
}

2. Select its lowest pending partition.

The scheduler prepares task sets and sorts their tasks by ascending PartitionID (scheduler/fifo_task_scheduler.go, line 188). It then selects set.tasks[set.pending] and advances the cursor (line 247).

// Sort once when preparing the task set.
sort.Slice(set.tasks, func(i, j int) bool {
    return set.tasks[i].task.PartitionID < set.tasks[j].task.PartitionID
})

// Take the next pending partition during assignment.
task := set.tasks[set.pending]
set.pending++

3. Assign work up to the offering worker’s available capacity.

The scheduler starts with the worker’s reported free slots. Some slots may already be reserved for tasks whose assignments are still on the way, so the worker hasn’t included them in its heartbeat yet. It subtracts those reservations to avoid assigning more work than the worker can handle and checks that the result stays within the worker’s total unreserved capacity.

For example, if a worker reports two free slots but one assignment is still on the way, the scheduler can assign only one more task. In the code below, seen contains the active attempt IDs reported by the worker, so reservations already reflected in its free-slot count are not subtracted again.

Then it assigns tasks while slots remain, reserving one slot and reducing available for each assignment (scheduler/fifo_task_scheduler.go, line 232).

available := free
for attemptID := range worker.reserved {
    if !seen[attemptID] {
        available--
    }
}
available = min(available, worker.TotalSlots-len(worker.reserved))

for available > 0 && set.pending < len(set.tasks) {
    // Select task and create assignment...
    worker.reserved[identity.ID] = struct{}{}
    assignments = append(assignments, assignment)
    available--
}

The Worker Lifecycle

The worker, meanwhile, has its own essential workflow, from reporting its status to claiming tasks from the scheduler. Let’s take a high-level look at the worker’s lifecycle (worker/runtime.go, line 92).

On startup, the worker sends its ID and total slot capacity to the coordinator as a one-time initial registration. It then sends a heartbeat immediately and periodically afterward, containing free slots and active attempt IDs to report its status and request work.

Upon receiving tasks, the worker validates the assignments, marks them active locally, and launches each in a goroutine. LocalRunner.RunTask executes the partition pipeline. Once execution completes, the worker sends either the task’s output or its failure to the coordinator interface, which we’ll discuss in the next section. The coordinator forwards it to the scheduler.

After the report is acknowledged, the worker releases local capacity and repeats the process. The worker only shuts down when canceled or when communication fails.

Communicating over HTTP

For this project, the immediate messages are registration, heartbeats, assignments, and results. They fit the request/response model well.

I choose to use a RESTful API with HTTP and JSON for rapid prototyping, allowing me to concentrate on the more complicated scheduling behavior: reservations, stale attempts, cancellation, and worker failure. For production, I would use gRPC with Protocol Buffers as the payload for more efficient and type-safe internal service-to-service communication.

Here is the list of endpoints that handle communication between workers and the scheduler:

This wraps up my distributed task execution implementation. Here’s a screenshot showing how a coordinator interacts with two worker processes, each offering two slots, through initial registration, resource offers, task assignments, and result reporting.

Moving tasks across processes meant tracking both the worker’s reported capacity and assignments it hadn’t received yet.

I used manual commands to spin up the services here. A future improvement is to package the coordinator and workers with Docker Compose, replacing separate terminal commands with a reproducible setup that includes service networking, shared input files, and combined logs.