[PATCH 0/2] Add maximum task accounting

[PATCH 0/2] Add maximum task accounting

[Mark Hatle]

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(-)

-- 
1.8.3.1

[PATCH 1/2] runqueue.py: Minor cleanup for RunQueueStats and users

[Mark Hatle]

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.

Signed-off-by: Mark Hatle <mark.hatle@windriver.com>
---
 lib/bb/runqueue.py | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/lib/bb/runqueue.py b/lib/bb/runqueue.py
index f2e52cf..a937a0b 100644
--- a/lib/bb/runqueue.py
+++ b/lib/bb/runqueue.py
@@ -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):
         """
-- 
1.8.3.1

[PATCH 2/2] runqueue.py: Initial implementation of per task process limits

[Mark Hatle]

On 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.

Signed-off-by: Mark Hatle <mark.hatle@windriver.com>
---
 lib/bb/runqueue.py | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/lib/bb/runqueue.py b/lib/bb/runqueue.py
index a937a0b..2d9e18d 100644
--- a/lib/bb/runqueue.py
+++ b/lib/bb/runqueue.py
@@ -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]
-- 
1.8.3.1

Re: [PATCH 2/2] runqueue.py: Initial implementation of per task process limits

[akuster]

On 05/14/2018 07:21 AM, Mark Hatle wrote:
> 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
> 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.
>
> Signed-off-by: Mark Hatle <mark.hatle@windriver.com>
> ---
>  lib/bb/runqueue.py | 21 +++++++++++++++++++++
>  1 file changed, 21 insertions(+)
>
> diff --git a/lib/bb/runqueue.py b/lib/bb/runqueue.py
> index a937a0b..2d9e18d 100644
> --- a/lib/bb/runqueue.py
> +++ b/lib/bb/runqueue.py
> @@ -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

[Mark Hatle]

On 5/14/18 9:49 AM, akuster wrote:
> 
> 
> On 05/14/2018 07:21 AM, Mark Hatle wrote:
>> 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

> Thanks for fixing this.
> Armin
>> 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.
>>
>> Signed-off-by: Mark Hatle <mark.hatle@windriver.com>
>> ---
>>  lib/bb/runqueue.py | 21 +++++++++++++++++++++
>>  1 file changed, 21 insertions(+)
>>
>> diff --git a/lib/bb/runqueue.py b/lib/bb/runqueue.py
>> index a937a0b..2d9e18d 100644
>> --- a/lib/bb/runqueue.py
>> +++ b/lib/bb/runqueue.py
>> @@ -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]
>