[SPARK-21225][CORE] Considering CPUS_PER_TASK when allocating task slots for each WorkerOffer

[JackYangzg]

JIRA Issue:https://issues.apache.org/jira/browse/SPARK-21225
In the function "resourceOffers", It declare a variable "tasks" for storage the tasks which have allocated a executor. It declared like this:
val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores))
But, I think this code only conside a situation for that one task per core. If the user set "spark.task.cpus" as 2 or 3, It really don't need so much Mem. I think It can motify as follow:
val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores / CPUS_PER_TASK))
to instead.
Motify like this the other earning is that it's more easy to understand the way how the tasks allocate offers.

[srowen]

Please use math.ceil(o.cores.toDouble / CPUS_PER_TASK).toInt
How much difference can this make? you're saving at most about 8 bytes x # cores per offer.

[JackYangzg]

I have motify to use math.ceil(o.cores.toDouble / CPUS_PER_TASK).toInt);
and each offer can save o.cores*(1-1/CPUS_PER_TASK)*length(TaskDescription) bytes.

[JackYangzg]

I have motify to use math.ceil(o.cores.toDouble/CPUS_PER_TASK).toInt);
and each offer can save o.cores*(1-1/CPUS_PER_TASK)*length(TaskDescription) bytes.

[srowen]

It doesn't save a TaskDescription's size; it saves the size of a reference only. that's why I'm not sure this is worthwhile, but at the same time, doesn't really hurt.

[JackYangzg]

yes, It's only save a reference.
I have read the related code, and It only affect this code
if (availableCpus(i) >= CPUS_PER_TASK) { try { for (task <- taskSet.resourceOffer(execId, host, maxLocality)) { **tasks(i) += task** val tid = task.taskId taskIdToTaskSetManager(tid) = taskSet taskIdToExecutorId(tid) = execId executorIdToRunningTaskIds(execId).add(tid) availableCpus(i) -= CPUS_PER_TASK assert(availableCpus(i) >= 0) launchedTask = true } }
obvious, it safe

[jerryshao]

From my understanding, this looks like a bug here, we didn't consider CPU_PER_TASK configuration. Instead of saving memory, I think this PR is more like fixing a bug here. As for saving memory, yes it does, but I don't think performance will be affected a lot.

[jerryshao]

Besides is it enough to use o.cores / CPUS_PER_TASK? I'm not sure why do we need to use ceil, for example if we have 10 cores in work offer and CPUS_PER_TASK is 3, then 3 slots should be enough? Please correct me if I'm wrong.

[JackYangzg]

@jerryshao I agree with you point, it enough to use o.cores / CPUS_PER_TASK to instead。like the example you list, the left one core is not enough to allocate for a task.
In fact, to deal with this
val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores / CPUS_PER_TASK))
is easy to understand the way how to allocate offers for the task .
@srowen

[jerryshao]

Can you please change your PR title and description to reflect the real issue here?

[JackYangzg]

@jerryshao Thank you, I have changed.

[SparkQA]

Test build #3816 has finished for PR 18435 at commit b745dab.

• This patch fails due to an unknown error code, -10.
• This patch merges cleanly.
• This patch adds no public classes.

[JackYangzg]

From the jenkins, The Execption is caused by lost of the file "target/unit-tests.log" ,I think it not caused by this PR and all the tests look like passed. Can you reset the test for me, thanks you @srowen

[jerryshao]

@jiangxb1987 @cloud-fan Can you please review this JIRA? The changes should be safe from my understanding.

Also can we change the title to: Considering CPUS_PER_TASK when allocating task slots for each WorkerOffer. I think that is more clear to reflect the issue here.

[JackYangzg]

@jerryshao Ok

[srowen]

This is only changing the default size of a collection, so I think it's safe to merge.

[SparkQA]

Test build #3819 has finished for PR 18435 at commit b745dab.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

LGTM, merging to master!