Proposal: concept of supervisor type task slots for resolving parallel task deadlocks

[himanshug]

Motivation

ParallelIndexSupervisorTask is a supervisor style task that delegates work to one or more spawned subtasks. Since both, supervisor task and subtasks, are using same task slot pool, there is possibility of a deadlock.
For example, say a druid cluster has 4 task slots and 4 ParallelIndexSupervisorTask tasks are submitted simultaneously and started running on available 4 task slots. Subtasks spawned by the supervisor tasks would never be able to run and supervisor tasks would keep on waiting.

It is also discussed in #8061 (comment) .

Proposed changes

Add a method boolean Task.isSupervisor() to Task interface which returns true if the task is a supervisor task that spawns subtasks to delegate work. ParallelIndexSupervisorTask would return true while all other current task impls would return false.

Add a druid.worker.supervisorCapacity configuration on middleManagers, which designates available slots to run supervisor tasks. This config is similar to druid.worker.capacity which designates available non-supervisor task slots.

[Http]RemoteTaskRunner code would be updated to recognize that supervisor tasks consume slot from supervisorCapacity and not capacity .

Rationale

A potential alternative is to use #7066 to send all supervisor tasks to a dedicated set of middleManagers which only get supervisor tasks.

Operational impact

User of ParallelIndexSupervisorTask would need to set property druid.worker.supervisorCapacity on middleManagers.

Test plan (optional)

will run it on a staging cluster.

Future work (optional)

for reliability: supervisor tasks could be treated further specially be imposing a "always restartable" restriction on them and also not failing them if middleManager running them crashed .

[jihoonson]

@himanshug thank you for raising a proposal!

Add a method boolean Task.isSupervisor() to Task interface which returns true if the task is a supervisor task that spawns subtasks to delegate work. ParallelIndexSupervisorTask would return true while all other current task impls would return false.

ParallelIndexSupervisorTask can has two different modes, i.e., sequential and parallel modes. If it runs in the sequential mode, the supervisor task will behave just like the index task and process all data by itself. I guess you can call isParallelMode() to check this.

Add a druid.worker.supervisorCapacity configuration on middleManagers, which designates available slots to run supervisor tasks. This config is similar to druid.worker.capacity which designates available non-supervisor task slots.

Wondering if we can add an overlord configuration instead of middleManager one, like druid.indexer.supervisorCapacityRatio or druid.indexer.supervisorMaxCapacity, which automatically reserve the number of task slots for supervisor tasks out of the total task capacity reported by middleManagers. One advantage of this way is people can configure only overlords instead of middleManagers which are supposed to be more than overlords in the cluster. Or do you want a sort of special middleManager where every supervisor task runs on it?

[himanshug]

@jihoonson thanks

I guess you can call isParallelMode() to check this.

Yes ParallelIndexSupervisorTask.isSupervisor() impl would be return isParallelMode() or something very close.

Both type of configurations have their merits I think. You already listed the advantages for overlord only configuration.
I was thinking druid.worker.supervisorCapacity for following reasons.

• this is my primary reason: supervisor tasks generally consume way less cpu/memory than the subtasks , so a worker can offer more of those slots.
• this allows segregating supervisor tasks to a dedicated pool of MMs if needed, this is secondary for me and I am not planning to really use this advantage.

one slight concern(may not really happen) is about race conditions with overlord config where it computed a number for available supervisor capacity, but then couple of worker disappeared and overlord used the old number for available supervisor capacity.

It is also possible to support both kind of configurations, this is not a one-or-the-other situation really. Only disadvantage I see is that users will have more options and hence need more mental bandwidth to tune things.

what do you think?

[jihoonson]

It is also possible to support both kind of configurations, this is not a one-or-the-other situation really. Only disadvantage I see is that users will have more options and hence need more mental bandwidth to tune things.

I think it makes sense for a short term solution. In the long run, we need a better task scheduler. Maybe makes sense to add something similar to #6993.

[himanshug]

thanks, alright, I am gonna make the changes in the proposal with worker config this week as this is a blocker for us running parallel task and our cluster is currently running with very ugly hack that I would like to get rid of .

[himanshug]

For now, I am going to use #7066 for segregating execution of supervisor tasks onto a dedicated set of MMs as it has been merged.
As I am implementing the proposal here, it seems redundant. Only additional feature enabled by this proposal would be to allow single MM segregate resources for supervisors vs other tasks, which doesn't seem super important at this point.