Add maximum task accounting
cover Add maximum task accounting
On my own machine, a 192 thread machine, if I start a new project without a local download cache, I end of DoSing my machine. I wanted a way to create a maximum number of tasks, of any given type, so that I could limit the number of concurrent do_fetch commands.
The first patch is a cleanup, while it's not required for the task accounting I did discover that a few operations were happening in the wrong order -- this could cause a problem if anyone is trying to use the RunQueueStats in new ways.
The second patch is the one that actually implements the max-task limits. As noted in the comment, while theoretically you could set this in a recipe, the behavior will be unpredictable, it's meant to be set in the global configuration space, i.e. the local.conf.
I've been using these patches successfully for the path few months.
Mark Hatle (2):
runqueue.py: Minor cleanup for RunQueueStats and users runqueue.py: Initial implementation of per task process limits
lib/bb/runqueue.py | 39 ++++++++++++++++++++++++++++++--------- 1 file changed, 30 insertions(+), 9 deletions(-)
Signature · 1 line
1.8.3.1
PATCH 1/2 runqueue.py: Minor cleanup for RunQueueStats and users
The RunQueueStats:taskCompleted and RunQueueStats:taskSkipped can take multiple arguments. However, nowehere in bitbake are multiple arguments used. Change this to match the behavior of the other APIs where it needs to be called once for each task.
Additionally, these two functions were usually called in tandem, however in the wrong order. It really doesn't matter as there is no specific preemption point between the calls. But the taskSkipped should be called first to increment the 'active' count, and then taskCompleted called to decrement it.
--- lib/bb/runqueue.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-)
lib/bb/runqueue.py +9 -9
@@ -94,13 +94,13 @@ class RunQueueStats: self.active = self.active - 1 self.failed = self.failed + 1 - def taskCompleted(self, number = 1): - self.active = self.active - number - self.completed = self.completed + number + def taskCompleted(self): + self.active = self.active - 1 + self.completed = self.completed + 1 - def taskSkipped(self, number = 1): - self.active = self.active + number - self.skipped = self.skipped + number + def taskSkipped(self): + self.active = self.active + 1 + self.skipped = self.skipped + 1 def taskActive(self): self.active = self.active + 1
@@ -1896,8 +1896,8 @@ class RunQueueExecuteTasks(RunQueueExecute): self.setbuildable(task) bb.event.fire(runQueueTaskSkipped(task, self.stats, self.rq, reason), self.cfgData) self.task_completeoutright(task) - self.stats.taskCompleted() self.stats.taskSkipped() + self.stats.taskCompleted() def execute(self): """
@@ -2342,8 +2342,8 @@ class RunQueueExecuteScenequeue(RunQueueExecute): def task_failoutright(self, task): self.runq_running.add(task) self.runq_buildable.add(task) - self.stats.taskCompleted() self.stats.taskSkipped() + self.stats.taskCompleted() self.scenequeue_notcovered.add(task) self.scenequeue_updatecounters(task, True)
@@ -2351,8 +2351,8 @@ class RunQueueExecuteScenequeue(RunQueueExecute): self.runq_running.add(task) self.runq_buildable.add(task) self.task_completeoutright(task) - self.stats.taskCompleted() self.stats.taskSkipped() + self.stats.taskCompleted() def execute(self): """
Signature · 1 line
1.8.3.1
PATCH 2/2 runqueue.py: Initial implementation of per task process limits
In reply to Mark HatleOn high core machines, in do_fetch, it is possible to DDoS your own machine. A method to limit any arbitrary task type to a certain number of simultaneous threads is needed. (Similar to how BB_NUMBER_THREADS works in the general case.) The format of this new limitation is:
do_fetch[number_threads] = "2"
This should be set globally. If it is set in individual recipes it could result in unpredictable behavior.
Note: a value for number_threads > BB_NUMBER_THREADS will have no effect.
--- lib/bb/runqueue.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+)
lib/bb/runqueue.py +21 -0
@@ -134,6 +134,7 @@ class RunQueueScheduler(object): self.prio_map = [self.rqdata.runtaskentries.keys()] self.buildable = [] + self.skip_maxthread = {} self.stamps = {} for tid in self.rqdata.runtaskentries: (mc, fn, taskname, taskfn) = split_tid_mcfn(tid)
@@ -150,8 +151,25 @@ class RunQueueScheduler(object): self.buildable = [x for x in self.buildable if x not in self.rq.runq_running] if not self.buildable: return None + + # Filter out tasks that have a max number of threads that have been exceeded + skip_buildable = {} + for running in self.rq.runq_running.difference(self.rq.runq_complete): + rtaskname = taskname_from_tid(running) + if rtaskname not in self.skip_maxthread: + self.skip_maxthread[rtaskname] = self.rq.cfgData.getVarFlag(rtaskname, "number_threads") + if not self.skip_maxthread[rtaskname]: + continue + if rtaskname in skip_buildable: + skip_buildable[rtaskname] += 1 + else: + skip_buildable[rtaskname] = 1 + if len(self.buildable) == 1: tid = self.buildable[0] + taskname = taskname_from_tid(tid) + if taskname in skip_buildable and skip_buildable[taskname] >= int(self.skip_maxthread[taskname]): + return None stamp = self.stamps[tid] if stamp not in self.rq.build_stamps.values(): return tid
@@ -164,6 +182,9 @@ class RunQueueScheduler(object): best = None bestprio = None for tid in self.buildable: + taskname = taskname_from_tid(tid) + if taskname in skip_buildable and skip_buildable[taskname] >= int(self.skip_maxthread[taskname]): + continue prio = self.rev_prio_map[tid] if bestprio is None or bestprio > prio: stamp = self.stamps[tid]
Signature · 1 line
1.8.3.1
Re: [PATCH 2/2] runqueue.py: Initial implementation of per task process limits
Quoted 1 line · Mark Hatle
On high core machines, in
do_fetch, it is possible to DDoS your own machine.
If this is the case, isn't a CVE warranted? Also, which version does this issue affect?
Thanks for fixing this.
Armin
Quoted 64 lines
A method to limit any arbitrary task type to a certain number of simultaneous threads is needed. (Similar to how
BB_NUMBER_THREADSworks in the general case.) The format of this new limitation is:do_fetch[number_threads] = "2"This should be set globally. If it is set in individual recipes it could result in unpredictable behavior.
Note: a value for
number_threads>BB_NUMBER_THREADSwill have no effect.Signed-off-by Mark Hatle <[email protected]>--- lib/bb/runqueue.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+)lib/bb/runqueue.py +21
-0@@ -134,6 +134,7 @@ class RunQueueScheduler(object): self.prio_map = [self.rqdata.runtaskentries.keys()] self.buildable = [] + self.skip_maxthread = {} self.stamps = {} for tid in self.rqdata.runtaskentries: (mc, fn, taskname, taskfn) = split_tid_mcfn(tid)@@ -150,8 +151,25 @@ class RunQueueScheduler(object): self.buildable = [x for x in self.buildable if x not in self.rq.runq_running] if not self.buildable: return None + + # Filter out tasks that have a max number of threads that have been exceeded + skip_buildable = {} + for running in self.rq.runq_running.difference(self.rq.runq_complete): + rtaskname = taskname_from_tid(running) + if rtaskname not in self.skip_maxthread: + self.skip_maxthread[rtaskname] = self.rq.cfgData.getVarFlag(rtaskname, "number_threads") + if not self.skip_maxthread[rtaskname]: + continue + if rtaskname in skip_buildable: + skip_buildable[rtaskname] += 1 + else: + skip_buildable[rtaskname] = 1 + if len(self.buildable) == 1: tid = self.buildable[0] + taskname = taskname_from_tid(tid) + if taskname in skip_buildable and skip_buildable[taskname] >= int(self.skip_maxthread[taskname]): + return None stamp = self.stamps[tid] if stamp not in self.rq.build_stamps.values(): return tid@@ -164,6 +182,9 @@ class RunQueueScheduler(object): best = None bestprio = None for tid in self.buildable: + taskname = taskname_from_tid(tid) + if taskname in skip_buildable and skip_buildable[taskname] >= int(self.skip_maxthread[taskname]): + continue prio = self.rev_prio_map[tid] if bestprio is None or bestprio > prio: stamp = self.stamps[tid]
Re: [PATCH 2/2] runqueue.py: Initial implementation of per task process limits
Quoted 6 lines · akuster
Quoted 1 line · Mark Hatle
On high core machines, in
do_fetch, it is possible to DDoS your own machine.If this is the case, isn't a CVE warranted? Also, which version does this issue affect?
No, cause you are doing it to yourself..
Run a do_fetch, and ask 192 external machines all to send you data at the same time.. depending on ISP, rate limiting, etc.. it's very possible for you to (temporarily) DDoS yourself... no different then if you fork bomb your own machine, or use curl recursively.. :)
--Mark
Quoted 67 lines
Thanks for fixing this.
ArminQuoted 64 lines
A method to limit any arbitrary task type to a certain number of simultaneous threads is needed. (Similar to how
BB_NUMBER_THREADSworks in the general case.) The format of this new limitation is:do_fetch[number_threads] = "2"This should be set globally. If it is set in individual recipes it could result in unpredictable behavior.
Note: a value for
number_threads>BB_NUMBER_THREADSwill have no effect.Signed-off-by Mark Hatle <[email protected]>--- lib/bb/runqueue.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+)lib/bb/runqueue.py +21
-0@@ -134,6 +134,7 @@ class RunQueueScheduler(object): self.prio_map = [self.rqdata.runtaskentries.keys()] self.buildable = [] + self.skip_maxthread = {} self.stamps = {} for tid in self.rqdata.runtaskentries: (mc, fn, taskname, taskfn) = split_tid_mcfn(tid)@@ -150,8 +151,25 @@ class RunQueueScheduler(object): self.buildable = [x for x in self.buildable if x not in self.rq.runq_running] if not self.buildable: return None + + # Filter out tasks that have a max number of threads that have been exceeded + skip_buildable = {} + for running in self.rq.runq_running.difference(self.rq.runq_complete): + rtaskname = taskname_from_tid(running) + if rtaskname not in self.skip_maxthread: + self.skip_maxthread[rtaskname] = self.rq.cfgData.getVarFlag(rtaskname, "number_threads") + if not self.skip_maxthread[rtaskname]: + continue + if rtaskname in skip_buildable: + skip_buildable[rtaskname] += 1 + else: + skip_buildable[rtaskname] = 1 + if len(self.buildable) == 1: tid = self.buildable[0] + taskname = taskname_from_tid(tid) + if taskname in skip_buildable and skip_buildable[taskname] >= int(self.skip_maxthread[taskname]): + return None stamp = self.stamps[tid] if stamp not in self.rq.build_stamps.values(): return tid@@ -164,6 +182,9 @@ class RunQueueScheduler(object): best = None bestprio = None for tid in self.buildable: + taskname = taskname_from_tid(tid) + if taskname in skip_buildable and skip_buildable[taskname] >= int(self.skip_maxthread[taskname]): + continue prio = self.rev_prio_map[tid] if bestprio is None or bestprio > prio: stamp = self.stamps[tid]