From bbe3369836ca3155646750e58ad88e0c36313c9a Mon Sep 17 00:00:00 2001 From: Xavier-Do Date: Fri, 17 Mar 2023 11:55:15 +0100 Subject: [PATCH] [IMP] runbot: global state computation One of the most problematic concurrency issue is when multiple children tries to write the state/result on the parent concurrently. Multiple partial solutions are possible here: - avoid to write if the state doesn't changes - avoid to write on a build belonging to another host A maybe overkill solution would be to add a message queue for an host, signaling that one of the child state changed. An intermediate solution would be to let the host check the state of the children while there are some of them, and update the local build state assynchronously himself. We can actualy use the 'waiting' global state to now if we need to continue to check the build state and result. While a build is not done or running, we need to check all children result and state on case they were updated. One corner case is when rebuilding a child: a new child is added but the parent is maybe not in the 'waiting' global state anymore. If this is the case, we need to recusivelly change the state of the parents to waiting so that they will update again. --- runbot/models/build.py | 138 +++++++++++++++++++++---------------- runbot/models/host.py | 7 +- runbot/models/runbot.py | 1 - runbot/tests/test_build.py | 19 +++-- 4 files changed, 95 insertions(+), 70 deletions(-) diff --git a/runbot/models/build.py b/runbot/models/build.py index 7e4cb489..9266606f 100644 --- a/runbot/models/build.py +++ b/runbot/models/build.py @@ -163,9 +163,9 @@ class BuildResult(models.Model): trigger_id = fields.Many2one('runbot.trigger', related='params_id.trigger_id', store=True, index=True) # state machine - global_state = fields.Selection(make_selection(state_order), string='Status', compute='_compute_global_state', store=True, recursive=True) + global_state = fields.Selection(make_selection(state_order), string='Status', default='pending', required=True) local_state = fields.Selection(make_selection(state_order), string='Build Status', default='pending', required=True, index=True) - global_result = fields.Selection(make_selection(result_order), string='Result', compute='_compute_global_result', store=True, recursive=True) + global_result = fields.Selection(make_selection(result_order), string='Result', default='ok') local_result = fields.Selection(make_selection(result_order), string='Build Result', default='ok') requested_action = fields.Selection([('wake_up', 'To wake up'), ('deathrow', 'To kill')], string='Action requested', index=True) @@ -254,20 +254,6 @@ class BuildResult(models.Model): build.log_list = ','.join({step.name for step in build.params_id.config_id.step_ids() if step._has_log()}) # TODO replace logic, add log file to list when executed (avoid 404, link log on docker start, avoid fake is_docker_step) - @api.depends('children_ids.global_state', 'local_state') - def _compute_global_state(self): - for record in self: - waiting_score = record._get_state_score('waiting') - children_ids = [child for child in record.children_ids if not child.orphan_result] - if record._get_state_score(record.local_state) > waiting_score and children_ids: # if finish, check children - children_state = record._get_youngest_state([child.global_state for child in children_ids]) - if record._get_state_score(children_state) > waiting_score: - record.global_state = record.local_state - else: - record.global_state = 'waiting' - else: - record.global_state = record.local_state - @api.depends('gc_delay', 'job_end') def _compute_gc_date(self): icp = self.env['ir.config_parameter'].sudo() @@ -299,22 +285,6 @@ class BuildResult(models.Model): def _get_state_score(self, result): return state_order.index(result) - @api.depends('children_ids.global_result', 'local_result', 'children_ids.orphan_result') - def _compute_global_result(self): - for record in self: - if record.local_result and record._get_result_score(record.local_result) >= record._get_result_score('ko'): - record.global_result = record.local_result - else: - children_ids = [child for child in record.children_ids if not child.orphan_result] - if children_ids: - children_result = record._get_worst_result([child.global_result for child in children_ids], max_res='ko') - if record.local_result: - record.global_result = record._get_worst_result([record.local_result, children_result]) - else: - record.global_result = children_result - else: - record.global_result = record.local_result - def _get_worst_result(self, results, max_res=False): results = [result for result in results if result] # filter Falsy values index = max([self._get_result_score(result) for result in results]) if results else 0 @@ -340,37 +310,50 @@ class BuildResult(models.Model): }) return [values] + @api.model_create_multi + def create(self, vals_list): + for values in vals_list: + if 'local_state' in values: + values['global_state'] = values['local_state'] + if 'local_result' in values: + values['global_result'] = values['local_result'] + records = super().create(vals_list) + if records.parent_id: + records.parent_id._update_globals() + return records + def write(self, values): # some validation to ensure db consistency if 'local_state' in values: if values['local_state'] == 'done': self.filtered(lambda b: b.local_state != 'done').commit_export_ids.unlink() - local_result = values.get('local_result') - for build in self: - if local_result and local_result != self._get_worst_result([build.local_result, local_result]): # dont write ok on a warn/error build - if len(self) == 1: - values.pop('local_result') - else: - raise ValidationError('Local result cannot be set to a less critical level') - init_global_results = self.mapped('global_result') - init_global_states = self.mapped('global_state') - init_local_states = self.mapped('local_state') + # don't update if the value doesn't change to avoid triggering concurrent updates + def minimal_update(records, field_name): + updated = self.browse() + if field_name in values: + field_result = values.pop(field_name) + updated = records.filtered(lambda b: (b[field_name] != field_result)) + if updated: + super(BuildResult, updated).write({field_name: field_result}) + return updated - res = super(BuildResult, self).write(values) - for init_global_result, build in zip(init_global_results, self): - if init_global_result != build.global_result: - build._github_status() + # local result is a special case since we don't only want to avoid an update if the value didn't change, but also if the value is less than the previous one + # example: don't write 'ok' if result is 'ko' or 'warn' + updated = self.browse() + if 'local_result' in values: + to_update = self.filtered(lambda b: (self._get_result_score(values['local_result']) > self._get_result_score(b.local_result))) + updated = minimal_update(to_update, 'local_result') + updated |= minimal_update(self, 'local_state') + updated._update_globals() + parents_to_update = minimal_update(self, 'global_result').parent_id + parents_to_update |= minimal_update(self, 'global_state').parent_id + parents_to_update |= minimal_update(self, 'orphan_result').parent_id + parents_to_update._notify_global_update() - for init_local_state, build in zip(init_local_states, self): - if init_local_state not in ('done', 'running') and build.local_state in ('done', 'running'): - build.build_end = now() - - for init_global_state, build in zip(init_global_states, self): - if init_global_state not in ('done', 'running') and build.global_state in ('done', 'running'): - build._github_status() - - return res + if values: + super().write(values) + return True def _add_child(self, param_values, orphan=False, description=False, additionnal_commit_links=False): @@ -476,8 +459,6 @@ class BuildResult(models.Model): self.orphan_result = True new_build = self.create(values) - if self.parent_id: - new_build._github_status() user = request.env.user if request else self.env.user new_build._log('rebuild', 'Rebuild initiated by %s%s' % (user.name, (' :%s' % message) if message else '')) @@ -676,6 +657,48 @@ class BuildResult(models.Model): build.write({'requested_action': False, 'local_state': 'done'}) continue + def _notify_global_update(self): + for record in self: + if not record.host_id: + record._update_globals() + else: + self.env['runbot.host.message'].create({ + 'host_id': record.host_id.id, + 'build_id': record.id, + 'message': 'global_updated', + }) + + def _update_globals(self): + for record in self: + children = record.children_ids.filtered(lambda child: not child.orphan_result) + global_result = record.local_result + if children: + child_result = record._get_worst_result(children.mapped('global_result'), max_res='ko') + global_result = record._get_worst_result([record.local_result, child_result]) + if global_result != record.global_result: + record.global_result = global_result + if not record.parent_id: + record._github_status() # failfast + + init_state = record.global_state + testing_children = any(child.global_state not in ('running', 'done') for child in children) + global_state = record.local_state + if testing_children: + child_state = 'waiting' + global_state = record._get_youngest_state([record.local_state, child_state]) + + if global_state != record.global_state: + record.global_state = global_state + + ending_build = init_state not in ('done', 'running') and record.global_state in ('done', 'running') + + if ending_build: + if not record.local_result: # Set 'ok' result if no result set (no tests job on build) + record.local_result = 'ok' + record.build_end = now() + if not record.parent_id: + record._github_status() + def _schedule(self): """schedule the build""" icp = self.env['ir.config_parameter'].sudo() @@ -958,7 +981,6 @@ class BuildResult(models.Model): v['local_result'] = result build.write(v) self.env.cr.commit() - build._github_status() self.invalidate_cache() def _ask_kill(self, lock=True, message=None): diff --git a/runbot/models/host.py b/runbot/models/host.py index cb3088f7..dbbbbebd 100644 --- a/runbot/models/host.py +++ b/runbot/models/host.py @@ -287,7 +287,12 @@ class MessageQueue(models.Model): def _process(self): records = self - # todo consume messages here + global_updates = records.filtered(lambda r: r.message == 'global_updated') + global_updates.build_id._update_globals() + records -= global_updates + # ask_kills = records.filtered(lambda r: r.message == 'ask_kill') + # ask_kills.build_id._ask_kill() + # records -= ask_kills if records: for record in records: self.env['runbot.runbot'].warning(f'Host {record.host_id.name} got an unexpected message {record.message}') diff --git a/runbot/models/runbot.py b/runbot/models/runbot.py index 445134a6..a79c2194 100644 --- a/runbot/models/runbot.py +++ b/runbot/models/runbot.py @@ -275,7 +275,6 @@ class Runbot(models.AbstractModel): _logger.warning('Removing %s from pull_info_failures', pr_number) del pull_info_failures[pr_number] - return manager.get('sleep', default_sleep) def _scheduler_loop_turn(self, host, sleep=5): diff --git a/runbot/tests/test_build.py b/runbot/tests/test_build.py index d977904c..67751656 100644 --- a/runbot/tests/test_build.py +++ b/runbot/tests/test_build.py @@ -173,10 +173,9 @@ class TestBuildResult(RunbotCase): # test a bulk write, that one cannot change from 'ko' to 'ok' builds = self.Build.browse([build.id, other.id]) - with self.assertRaises(ValidationError): - builds.write({'local_result': 'warn'}) - # self.assertEqual(build.local_result, 'warn') - # self.assertEqual(other.local_result, 'ko') + builds.write({'local_result': 'warn'}) + self.assertEqual(build.local_result, 'warn') + self.assertEqual(other.local_result, 'ko') def test_markdown_description(self): @@ -385,16 +384,16 @@ class TestBuildResult(RunbotCase): with self.assertQueries([]): # no change should be triggered build1_2.local_state = "testing" - # with self.assertQueries(['''UPDATE "runbot_build" SET "global_state"=%s,"local_state"=%s,"write_date"=%s,"write_uid"=%s WHERE id IN %s''']): - build1.local_state = 'done' - build1.flush() + with self.assertQueries(['''UPDATE "runbot_build" SET "global_state"=%s,"local_state"=%s,"write_date"=%s,"write_uid"=%s WHERE id IN %s''']): + build1.local_state = 'done' + build1.flush() self.assertEqual('waiting', build1.global_state) self.assertEqual('testing', build1_1.global_state) - # with self.assertQueries([]): # write the same value, no update should be triggered - build1.local_state = 'done' - build1.flush() + with self.assertQueries([]): # write the same value, no update should be triggered + build1.local_state = 'done' + build1.flush() build1_1.local_state = 'done'