GNOME.org
 

[Notes] [Git][BuildGrid/buildgrid][mablanch/75-requests-multiplexing] 6 commits: Do not duplicate jobs for same action

Title: GitLab

Martin Blanchard pushed to branch mablanch/75-requests-multiplexing at BuildGrid / buildgrid

Commits:

9 changed files:

Changes:

  • buildgrid/server/bots/instance.py
    ... ... @@ -123,7 +123,7 @@ class BotsInterface:
    123 123 
                 # Job does not exist, remove from bot.
    124 124 
                 return None
    125 125
    126 
    -        self._scheduler.update_job_lease(lease)
    126 
    +        self._scheduler.update_job_lease_state(lease.id, lease)
    127 127
    128 128 
             if lease_state == LeaseState.COMPLETED:
    129 129 
                 return None
    ... ... @@ -161,7 +161,7 @@ class BotsInterface:
    161 161 
                     self.__logger.error("Assigned lease id=[%s],"
    162 162 
                                         " not found on bot with name=[%s] and id=[%s]."
    163 163 
                                         " Retrying job", lease_id, bot_session.name, bot_session.bot_id)
    164 
    -                self._scheduler.retry_job(lease_id)
    164 
    +                self._scheduler.retry_job_lease(lease_id)
    165 165
    166 166 
         def _close_bot_session(self, name):
    167 167 
             """ Before removing the session, close any leases and
    ... ... @@ -174,7 +174,7 @@ class BotsInterface:
    174 174
    175 175 
             self.__logger.debug("Attempting to close [%s] with name: [%s]", bot_id, name)
    176 176 
             for lease_id in self._assigned_leases[name]:
    177 
    -            self._scheduler.retry_job(lease_id)
    177 
    +            self._scheduler.retry_job_lease(lease_id)
    178 178 
             self._assigned_leases.pop(name)
    179 179
    180 180 
             self.__logger.debug("Closing bot session: [%s]", name)

  • buildgrid/server/execution/instance.py
    ... ... @@ -23,9 +23,7 @@ import logging
    23 23
    24 24 
     from buildgrid._exceptions import FailedPreconditionError, InvalidArgumentError, NotFoundError
    25 25 
     from buildgrid._protos.build.bazel.remote.execution.v2.remote_execution_pb2 import Action
    26 
    -
    27 
    -from ..job import Job
    28 
    -from ...utils import get_hash_type
    26 
    +from buildgrid.utils import get_hash_type
    29 27
    30 28
    31 29 
     class ExecutionInstance:
    ... ... @@ -46,44 +44,45 @@ class ExecutionInstance:
    46 44 
         def hash_type(self):
    47 45 
             return get_hash_type()
    48 46
    49 
    -    def execute(self, action_digest, skip_cache_lookup, peer=None, message_queue=None):
    47 
    +    def execute(self, action_digest, skip_cache_lookup):
    50 48 
             """ Sends a job for execution.
    51 49 
             Queues an action and creates an Operation instance to be associated with
    52 50 
             this action.
    53 51 
             """
    54 
    -
    55 52 
             action = self._storage.get_message(action_digest, Action)
    56 53
    57 54 
             if not action:
    58 55 
                 raise FailedPreconditionError("Could not get action from storage.")
    59 56
    60 
    -        job = Job(action, action_digest)
    61 
    -        if peer is not None and message_queue is not None:
    62 
    -            job.register_operation_client(peer, message_queue)
    63 
    -
    64 
    -        self._scheduler.queue_job(job, skip_cache_lookup)
    57 
    +        return self._scheduler.queue_job_operation(action, action_digest, skip_cache_lookup)
    65 58
    66 
    -        return job.operation
    67 
    -
    68 
    -    def register_operation_client(self, job_name, peer, message_queue):
    59 
    +    def register_operation_client(self, operation_name, peer, message_queue):
    69 60 
             try:
    70 
    -            self._scheduler.job_name(job_name, peer, message_queue)
    61 
    +            return self._scheduler.register_job_operation_client(operation_name,
    62 
    +                                                                 peer, message_queue)
    71 63
    72 64 
             except NotFoundError:
    73 
    -            raise InvalidArgumentError("Operation name does not exist: [{}]".format(job_name))
    65 
    +            raise InvalidArgumentError("Operation name does not exist: [{}]"
    66 
    +                                       .format(operation_name))
    74 67
    75 
    -    def unregister_operation_client(self, job_name, peer):
    68 
    +    def unregister_operation_client(self, operation_name, peer):
    76 69 
             try:
    77 
    -            self._scheduler.unregister_operation_client(job_name, peer)
    70 
    +            self._scheduler.unregister_job_operation_client(operation_name, peer)
    78 71
    79 72 
             except NotFoundError:
    80 
    -            raise InvalidArgumentError("Operation name does not exist: [{}]".format(job_name))
    73 
    +            raise InvalidArgumentError("Operation name does not exist: [{}]"
    74 
    +                                       .format(operation_name))
    75 
    +
    76 
    +    def stream_operation_updates(self, message_queue):
    77 
    +        error, operation = message_queue.get()
    78 
    +        if error is not None:
    79 
    +            raise error
    80 
    +
    81 
    +        while not operation.done:
    82 
    +            yield operation
    81 83
    82 
    -    def stream_operation_updates(self, message_queue, operation_name):
    83 
    -        job = message_queue.get()
    84 
    -        while not job.operation.done:
    85 
    -            yield job.operation
    86 
    -            job = message_queue.get()
    87 
    -            job.check_operation_status()
    84 
    +            error, operation = message_queue.get()
    85 
    +            if error is not None:
    86 
    +                raise error
    88 87
    89 
    -        yield job.operation
    88 
    +        yield operation

  • buildgrid/server/execution/service.py
    ... ... @@ -97,13 +97,14 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
    97 97 
             try:
    98 98 
                 instance = self._get_instance(instance_name)
    99 99
    100 
    -            operation = instance.execute(request.action_digest,
    101 
    -                                         request.skip_cache_lookup,
    102 
    -                                         peer=peer,
    103 
    -                                         message_queue=message_queue)
    100 
    +            job_name = instance.execute(request.action_digest,
    101 
    +                                        request.skip_cache_lookup)
    102 
    +
    103 
    +            operation_name = instance.register_operation_client(job_name,
    104 
    +                                                                peer, message_queue)
    104 105
    105 106 
                 context.add_callback(partial(self._rpc_termination_callback,
    106 
    -                                         peer, instance_name, operation.name))
    107 
    +                                         peer, instance_name, operation_name))
    107 108
    108 109 
                 if self._is_instrumented:
    109 110 
                     if peer not in self.__peers:
    ... ... @@ -112,16 +113,13 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
    112 113 
                     else:
    113 114 
                         self.__peers[peer] += 1
    114 115
    115 
    -            instanced_op_name = "{}/{}".format(instance_name, operation.name)
    116 
    +            operation_full_name = "{}/{}".format(instance_name, operation_name)
    116 117
    117 
    -            self.__logger.info("Operation name: [%s]", instanced_op_name)
    118 
    +            self.__logger.info("Operation name: [%s]", operation_full_name)
    118 119
    119 
    -            for operation in instance.stream_operation_updates(message_queue,
    120 
    -                                                               operation.name):
    121 
    -                op = operations_pb2.Operation()
    122 
    -                op.CopyFrom(operation)
    123 
    -                op.name = instanced_op_name
    124 
    -                yield op
    120 
    +            for operation in instance.stream_operation_updates(message_queue):
    121 
    +                operation.name = operation_full_name
    122 
    +                yield operation
    125 123
    126 124 
             except InvalidArgumentError as e:
    127 125 
                 self.__logger.error(e)
    ... ... @@ -159,8 +157,8 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
    159 157 
             try:
    160 158 
                 instance = self._get_instance(instance_name)
    161 159
    162 
    -            instance.register_operation_client(operation_name,
    163 
    -                                               peer, message_queue)
    160 
    +            operation_name = instance.register_operation_client(operation_name,
    161 
    +                                                                peer, message_queue)
    164 162
    165 163 
                 context.add_callback(partial(self._rpc_termination_callback,
    166 164 
                                              peer, instance_name, operation_name))
    ... ... @@ -172,12 +170,11 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
    172 170 
                     else:
    173 171 
                         self.__peers[peer] += 1
    174 172
    175 
    -            for operation in instance.stream_operation_updates(message_queue,
    176 
    -                                                               operation_name):
    177 
    -                op = operations_pb2.Operation()
    178 
    -                op.CopyFrom(operation)
    179 
    -                op.name = request.name
    180 
    -                yield op
    173 
    +            operation_full_name = "{}/{}".format(instance_name, operation_name)
    174 
    +
    175 
    +            for operation in instance.stream_operation_updates(message_queue):
    176 
    +                operation.name = operation_full_name
    177 
    +                yield operation
    181 178
    182 179 
             except InvalidArgumentError as e:
    183 180 
                 self.__logger.error(e)
    ... ... @@ -212,10 +209,10 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
    212 209
    213 210 
         # --- Private API ---
    214 211
    215 
    -    def _rpc_termination_callback(self, peer, instance_name, job_name):
    212 
    +    def _rpc_termination_callback(self, peer, instance_name, operation_name):
    216 213 
             instance = self._get_instance(instance_name)
    217 214
    218 
    -        instance.unregister_operation_client(job_name, peer)
    215 
    +        instance.unregister_operation_client(operation_name, peer)
    219 216
    220 217 
             if self._is_instrumented:
    221 218 
                 if self.__peers[peer] > 1:

  • buildgrid/server/job.py
    ... ... @@ -20,7 +20,7 @@ import uuid
    20 20 
     from google.protobuf import duration_pb2, timestamp_pb2
    21 21
    22 22 
     from buildgrid._enums import LeaseState, OperationStage
    23 
    -from buildgrid._exceptions import CancelledError
    23 
    +from buildgrid._exceptions import CancelledError, NotFoundError
    24 24 
     from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
    25 25 
     from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
    26 26 
     from buildgrid._protos.google.longrunning import operations_pb2
    ... ... @@ -35,30 +35,32 @@ class Job:
    35 35 
             self._name = str(uuid.uuid4())
    36 36 
             self._priority = priority
    37 37 
             self._action = remote_execution_pb2.Action()
    38 
    -        self._operation = operations_pb2.Operation()
    39 38 
             self._lease = None
    40 39
    41 40 
             self.__execute_response = None
    42 41 
             self.__operation_metadata = remote_execution_pb2.ExecuteOperationMetadata()
    42 
    +        self.__operations_by_name = {}
    43 
    +        self.__operations_by_peer = {}
    43 44
    44 45 
             self.__queued_timestamp = timestamp_pb2.Timestamp()
    45 46 
             self.__queued_time_duration = duration_pb2.Duration()
    46 47 
             self.__worker_start_timestamp = timestamp_pb2.Timestamp()
    47 48 
             self.__worker_completed_timestamp = timestamp_pb2.Timestamp()
    48 49
    49 
    -        self.__operation_message_queues = {}
    50 
    -        self.__operation_cancelled = False
    50 
    +        self.__operations_message_queues = {}
    51 
    +        self.__operations_cancelled = set()
    51 52 
             self.__lease_cancelled = False
    53 
    +        self.__job_cancelled = False
    52 54
    53 55 
             self.__operation_metadata.action_digest.CopyFrom(action_digest)
    54 56 
             self.__operation_metadata.stage = OperationStage.UNKNOWN.value
    55 57
    56 58 
             self._action.CopyFrom(action)
    57 59 
             self._do_not_cache = self._action.do_not_cache
    58 
    -        self._operation.name = self._name
    59 
    -        self._operation.done = False
    60 60 
             self._n_tries = 0
    61 61
    62 
    +        self._done = False
    63 
    +
    62 64 
         def __eq__(self, other):
    63 65 
             if isinstance(other, Job):
    64 66 
                 return self.name == other.name
    ... ... @@ -78,12 +80,14 @@ class Job:
    78 80 
             return self._priority
    79 81
    80 82 
         @property
    81 
    -    def do_not_cache(self):
    82 
    -        return self._do_not_cache
    83 
    +    def done(self):
    84 
    +        return self._done
    85 
    +
    86 
    +    # --- Public API: REAPI ---
    83 87
    84 88 
         @property
    85 
    -    def action(self):
    86 
    -        return self._action
    89 
    +    def do_not_cache(self):
    90 
    +        return self._do_not_cache
    87 91
    88 92 
         @property
    89 93 
         def action_digest(self):
    ... ... @@ -97,56 +101,49 @@ class Job:
    97 101 
                 return None
    98 102
    99 103 
         @property
    100 
    -    def holds_cached_action_result(self):
    104 
    +    def holds_cached_result(self):
    101 105 
             if self.__execute_response is not None:
    102 106 
                 return self.__execute_response.cached_result
    103 107 
             else:
    104 108 
                 return False
    105 109
    106 
    -    @property
    107 
    -    def operation(self):
    108 
    -        return self._operation
    109 
    -
    110 
    -    @property
    111 
    -    def operation_stage(self):
    112 
    -        return OperationStage(self.__operation_metadata.state)
    113 
    -
    114 
    -    @property
    115 
    -    def lease(self):
    116 
    -        return self._lease
    117 
    -
    118 
    -    @property
    119 
    -    def lease_state(self):
    120 
    -        if self._lease is not None:
    121 
    -            return LeaseState(self._lease.state)
    122 
    -        else:
    123 
    -            return None
    110 
    +    def set_cached_result(self, action_result):
    111 
    +        """Allows specifying an action result form the action cache for the job.
    124 112
    125 
    -    @property
    126 
    -    def lease_cancelled(self):
    127 
    -        return self.__lease_cancelled
    113 
    +        Note:
    114 
    +            This won't trigger any :class:`Operation` stage transition.
    128 115
    129 
    -    @property
    130 
    -    def n_tries(self):
    131 
    -        return self._n_tries
    116 
    +        Args:
    117 
    +            action_result (ActionResult): The result from cache.
    118 
    +        """
    119 
    +        self.__execute_response = remote_execution_pb2.ExecuteResponse()
    120 
    +        self.__execute_response.result.CopyFrom(action_result)
    121 
    +        self.__execute_response.cached_result = True
    132 122
    133 123 
         @property
    134 124 
         def n_clients(self):
    135 
    -        return len(self.__operation_message_queues)
    125 
    +        return len(self.__operations_message_queues)
    136 126
    137 127 
         def register_operation_client(self, peer, message_queue):
    138 128 
             """Subscribes to the job's :class:`Operation` stage changes.
    139 129
    140 
    -        Queues this :object:`Job` instance.
    141 
    -
    142 130 
             Args:
    143 131 
                 peer (str): a unique string identifying the client.
    144 132 
                 message_queue (queue.Queue): the event queue to register.
    133 
    +
    134 
    +        Returns:
    135 
    +            str: The name of the subscribed :class:`Operation`.
    145 136 
             """
    146 
    -        if peer not in self.__operation_message_queues:
    147 
    -            self.__operation_message_queues[peer] = message_queue
    137 
    +        if peer not in self.__operations_by_peer:
    138 
    +            operation = self.__operations_by_peer[peer]
    139 
    +        else:
    140 
    +            operation = self.create_operation(peer)
    141 
    +
    142 
    +        self.__operations_message_queues[peer] = message_queue
    148 143
    149 
    -        message_queue.put(self)
    144 
    +        self._send_operations_updates(peers=[peer])
    145 
    +
    146 
    +        return self._operation.name
    150 147
    151 148 
         def unregister_operation_client(self, peer):
    152 149 
             """Unsubscribes to the job's :class:`Operation` stage change.
    ... ... @@ -154,21 +151,168 @@ class Job:
    154 151 
             Args:
    155 152 
                 peer (str): a unique string identifying the client.
    156 153 
             """
    157 
    -        if peer not in self.__operation_message_queues:
    158 
    -            del self.__operation_message_queues[peer]
    154 
    +        if peer in self.__operations_message_queues:
    155 
    +            del self.__operations_message_queues[peer]
    159 156
    160 
    -    def set_cached_result(self, action_result):
    161 
    -        """Allows specifying an action result form the action cache for the job.
    157 
    +        # Drop the operation if nobody is watching it anymore:
    158 
    +        if peer in self.__operations_by_peer:
    159 
    +            operation_name = self.__operations_by_peer[peer].name
    160 
    +
    161 
    +            if operation_name not in [operation.name for operation in
    162 
    +                                      self.__operations_by_peer.values()]:
    163 
    +                del self.__operations_by_name[operation_name]
    164 
    +
    165 
    +            del self.__operations_by_peer[peer]
    166 
    +
    167 
    +    def create_operation(self, peer):
    168 
    +        """Generates a new :class:`Operation` for `peer`.
    169 
    +
    170 
    +        Args:
    171 
    +            peer (str): a unique string identifying the client.
    162 172 
             """
    163 
    -        self.__execute_response = remote_execution_pb2.ExecuteResponse()
    164 
    -        self.__execute_response.result.CopyFrom(action_result)
    165 
    -        self.__execute_response.cached_result = True
    173 
    +        if peer in self.__operations_by_peer:
    174 
    +            return self.__operations_by_peer[peer]
    175 
    +
    176 
    +        new_operation = operations_pb2.Operation()
    177 
    +        # Copy state from first existing and non cancelled operation:
    178 
    +        for operation in self.__operation_by_name.values():
    179 
    +            if operation.name not in self.__operations_cancelled:
    180 
    +                new_operation.CopyFrom(operation)
    181 
    +                break
    182 
    +
    183 
    +        new_operation.name = str(uuid.uuid4())
    184 
    +
    185 
    +        self.__operation_by_name[new_operation.name] = new_operation
    186 
    +        self.__operations_by_peer[peer] = new_operation
    187 
    +
    188 
    +        return new_operation
    189 
    +
    190 
    +    def get_operation(self, operation_name):
    191 
    +        """Returns a copy of the the job's :class:`Operation`.
    192 
    +
    193 
    +        Args:
    194 
    +            operation_name (str): the operation's name.
    195 
    +
    196 
    +        Raises:
    197 
    +            NotFoundError: If no operation with `operation_name` exists.
    198 
    +        """
    199 
    +        try:
    200 
    +            operation = self.__operations_by_name[operation_name]
    201 
    +
    202 
    +        except KeyError:
    203 
    +            raise NotFoundError("Operation name does not exist: [{}]"
    204 
    +                                .format(operation_name))
    205 
    +
    206 
    +        return self._copy_operation(operation)
    207 
    +
    208 
    +    def update_operation_stage(self, stage):
    209 
    +        """Operates a stage transition for the job's :class:`Operation`.
    210 
    +
    211 
    +        Args:
    212 
    +            stage (OperationStage): the operation stage to transition to.
    213 
    +        """
    214 
    +        if stage.value == self.__operation_metadata.stage:
    215 
    +            return
    216 
    +
    217 
    +        self.__operation_metadata.stage = stage.value
    218 
    +
    219 
    +        if self.__operation_metadata.stage == OperationStage.QUEUED.value:
    220 
    +            if self.__queued_timestamp.ByteSize() == 0:
    221 
    +                self.__queued_timestamp.GetCurrentTime()
    222 
    +            self._n_tries += 1
    223 
    +
    224 
    +        elif self.__operation_metadata.stage == OperationStage.EXECUTING.value:
    225 
    +            queue_in, queue_out = self.__queued_timestamp.ToDatetime(), datetime.now()
    226 
    +            self.__queued_time_duration.FromTimedelta(queue_out - queue_in)
    227 
    +
    228 
    +        elif self.__operation_metadata.stage == OperationStage.COMPLETED.value:
    229 
    +            self._done = True
    230 
    +
    231 
    +        operations = [operation for operation in self.__operations_by_name.values()
    232 
    +                      if operation.name not in self.__operations_cancelled]
    233 
    +
    234 
    +        for operation in operations:
    235 
    +            if self.__execute_response is not None:
    236 
    +                operation.response.Pack(self.__execute_response)
    237 
    +            operation.metadata.Pack(self.__operation_metadata)
    238 
    +
    239 
    +        self._send_operations_updates()
    240 
    +
    241 
    +    def cancel_operation(self, peer):
    242 
    +        """Triggers a job's :class:`Operation` cancellation.
    243 
    +
    244 
    +        This may cancel any job's :class:`Lease` that may have been issued.
    245 
    +
    246 
    +        Args:
    247 
    +            peer (str): a unique string identifying the client.
    248 
    +        """
    249 
    +        operations, peers = set(), set()
    250 
    +        if peer in self.__operations_by_peer:
    251 
    +            operations.add(self.__operations_by_peer[peer].name)
    252 
    +            peers.add(peer)
    253 
    +
    254 
    +        else:
    255 
    +            operations.update(self.__operations_by_name.keys())
    256 
    +            peers.update(self.__operations_by_peer.keys())
    257 
    +
    258 
    +        operations = operations - self.__operations_cancelled
    259 
    +        if not operations:
    260 
    +            return
    261 
    +
    262 
    +        operation_metadata = remote_execution_pb2.ExecuteOperationMetadata()
    263 
    +        operation_metadata.CopyFrom(self.__operation_metadata)
    264 
    +        operation_metadata.stage = OperationStage.COMPLETED.value
    265 
    +
    266 
    +        execute_response = remote_execution_pb2.ExecuteResponse()
    267 
    +        if self.__execute_response is not None:
    268 
    +            execute_response.CopyFrom(self.__execute_response)
    269 
    +        execute_response.status.code = code_pb2.CANCELLED
    270 
    +        execute_response.status.message = "Operation cancelled by client."
    271 
    +
    272 
    +        for operation_name in operations:
    273 
    +            operation = self.__operations_by_name[operation_name]
    274 
    +
    275 
    +            operation.metadata.Pack(operation_metadata)
    276 
    +            operation.response.Pack(execute_response)
    277 
    +            operation.done = True
    278 
    +
    279 
    +            self.__operations_cancelled.add(operation_name)
    280 
    +
    281 
    +        operations = set(self.__operations_by_name.keys())
    282 
    +        # Job is cancelled if all the operation are:
    283 
    +        self.__job_cancelled = bool(operations - self.__operations_cancelled)
    284 
    +
    285 
    +        if self.__job_cancelled and self._lease is not None:
    286 
    +            self.cancel_lease()
    287 
    +
    288 
    +        self._send_operations_updates(peers=peers, notify_cancelled=True)
    289 
    +
    290 
    +    # --- Public API: RWAPI ---
    291 
    +
    292 
    +    @property
    293 
    +    def lease(self):
    294 
    +        return self._lease
    295 
    +
    296 
    +    @property
    297 
    +    def lease_state(self):
    298 
    +        if self._lease is not None:
    299 
    +            return LeaseState(self._lease.state)
    300 
    +        else:
    301 
    +            return None
    302 
    +
    303 
    +    @property
    304 
    +    def lease_cancelled(self):
    305 
    +        return self.__lease_cancelled
    306 
    +
    307 
    +    @property
    308 
    +    def n_tries(self):
    309 
    +        return self._n_tries
    166 310
    167 311 
         def create_lease(self):
    168 312 
             """Emits a new :class:`Lease` for the job.
    169 313
    170 314 
             Only one :class:`Lease` can be emitted for a given job. This method
    171 
    -        should only be used once, any furhter calls are ignored.
    315 
    +        should only be used once, any further calls are ignored.
    172 316 
             """
    173 317 
             if self.__operation_cancelled:
    174 318 
                 return None
    ... ... @@ -183,14 +327,14 @@ class Job:
    183 327 
             return self._lease
    184 328
    185 329 
         def update_lease_state(self, state, status=None, result=None):
    186 
    -        """Operates a state transition for the job's current :class:Lease.
    330 
    +        """Operates a state transition for the job's current :class:`Lease`.
    187 331
    188 332 
             Args:
    189 333 
                 state (LeaseState): the lease state to transition to.
    190 
    -            status (google.rpc.Status): the lease execution status, only
    191 
    -                required if `state` is `COMPLETED`.
    192 
    -            result (google.protobuf.Any): the lease execution result, only
    193 
    -                required if `state` is `COMPLETED`.
    334 
    +            status (google.rpc.Status, optional): the lease execution status,
    335 
    +                only required if `state` is `COMPLETED`.
    336 
    +            result (google.protobuf.Any, optional): the lease execution result,
    337 
    +                only required if `state` is `COMPLETED`.
    194 338 
             """
    195 339 
             if state.value == self._lease.state:
    196 340 
                 return
    ... ... @@ -231,72 +375,50 @@ class Job:
    231 375 
                 self.__execute_response.status.CopyFrom(status)
    232 376
    233 377 
         def cancel_lease(self):
    234 
    -        """Triggers a job's :class:Lease cancellation.
    378 
    +        """Triggers a job's :class:`Lease` cancellation.
    235 379
    236 
    -        This will not cancel the job's :class:Operation.
    380 
    +        Note:
    381 
    +            This will not cancel the job's :class:`Operation`.
    237 382 
             """
    238 383 
             self.__lease_cancelled = True
    239 384 
             if self._lease is not None:
    240 385 
                 self.update_lease_state(LeaseState.CANCELLED)
    241 386
    242 
    -    def update_operation_stage(self, stage):
    243 
    -        """Operates a stage transition for the job's :class:Operation.
    244 
    -
    245 
    -        Args:
    246 
    -            stage (OperationStage): the operation stage to transition to.
    247 
    -        """
    248 
    -        if stage.value == self.__operation_metadata.stage:
    249 
    -            return
    250 
    -
    251 
    -        self.__operation_metadata.stage = stage.value
    252 
    -
    253 
    -        if self.__operation_metadata.stage == OperationStage.QUEUED.value:
    254 
    -            if self.__queued_timestamp.ByteSize() == 0:
    255 
    -                self.__queued_timestamp.GetCurrentTime()
    256 
    -            self._n_tries += 1
    257 
    -
    258 
    -        elif self.__operation_metadata.stage == OperationStage.EXECUTING.value:
    259 
    -            queue_in, queue_out = self.__queued_timestamp.ToDatetime(), datetime.now()
    260 
    -            self.__queued_time_duration.FromTimedelta(queue_out - queue_in)
    261 
    -
    262 
    -        elif self.__operation_metadata.stage == OperationStage.COMPLETED.value:
    263 
    -            if self.__execute_response is not None:
    264 
    -                self._operation.response.Pack(self.__execute_response)
    265 
    -            self._operation.done = True
    266 
    -
    267 
    -        self._operation.metadata.Pack(self.__operation_metadata)
    387 
    +    # --- Public API: Monitoring ---
    268 388
    269 
    -        for message_queue in self.__operation_message_queues.values():
    270 
    -            message_queue.put(self)
    389 
    +    def query_queue_time(self):
    390 
    +        return self.__queued_time_duration.ToTimedelta()
    271 391
    272 
    -    def check_operation_status(self):
    273 
    -        """Reports errors on unexpected job's :class:Operation state.
    392 
    +    def query_n_retries(self):
    393 
    +        return self._n_tries - 1 if self._n_tries > 0 else 0
    274 394
    275 
    -        Raises:
    276 
    -            CancelledError: if the job's :class:Operation was cancelled.
    277 
    -        """
    278 
    -        if self.__operation_cancelled:
    279 
    -            raise CancelledError(self.__execute_response.status.message)
    395 
    +    # --- Private API ---
    280 396
    281 
    -    def cancel_operation(self):
    282 
    -        """Triggers a job's :class:Operation cancellation.
    397 
    +    def _copy_operation(self, operation):
    398 
    +        """Simply duplicates a given :class:`Lease` object."""
    399 
    +        new_operation = operations_pb2.Operation()
    283 400
    284 
    -        This will also cancel any job's :class:Lease that may have been issued.
    285 
    -        """
    286 
    -        self.__operation_cancelled = True
    287 
    -        if self._lease is not None:
    288 
    -            self.cancel_lease()
    401 
    +        new_operation.CopyFrom(operation)
    289 402
    290 
    -        self.__execute_response = remote_execution_pb2.ExecuteResponse()
    291 
    -        self.__execute_response.status.code = code_pb2.CANCELLED
    292 
    -        self.__execute_response.status.message = "Operation cancelled by client."
    403 
    +        return new_operation
    293 404
    294 
    -        self.update_operation_stage(OperationStage.COMPLETED)
    405 
    +    def _send_operations_updates(self, peers=None, notify_cancelled=False):
    406 
    +        """Sends :class:`Operation` stage change messages to watchers."""
    407 
    +        for peer, message_queue in self.__operations_message_queues.items():
    408 
    +            if peer not in self.__operations_by_peer:
    409 
    +                continue
    410 
    +            elif peers and peer not in peers:
    411 
    +                continue
    295 412
    296 
    -    # --- Public API: Monitoring ---
    413 
    +            operation = self.__operations_by_peer[peer]
    414 
    +            if not notify_cancelled and operation.name in self.__operations_cancelled:
    415 
    +                continue
    297 416
    298 
    -    def query_queue_time(self):
    299 
    -        return self.__queued_time_duration.ToTimedelta()
    417 
    +            # Messages are pairs of (Exception, Operation,):
    418 
    +            if operation.name not in self.__operations_cancelled:
    419 
    +                message = (None, self._copy_operation(operation),)
    420 
    +            else:
    421 
    +                message = (CancelledError(self.__execute_response.status.message),
    422 
    +                           self._copy_operation(operation),)
    300 423
    301 
    -    def query_n_retries(self):
    302 
    -        return self._n_tries - 1 if self._n_tries > 0 else 0
    424 
    +            message_queue.put(message)

  • buildgrid/server/operations/instance.py
    ... ... @@ -21,7 +21,7 @@ An instance of the LongRunningOperations Service.
    21 21
    22 22 
     import logging
    23 23
    24 
    -from buildgrid._exceptions import InvalidArgumentError
    24 
    +from buildgrid._exceptions import InvalidArgumentError, NotFoundError
    25 25 
     from buildgrid._protos.google.longrunning import operations_pb2
    26 26
    27 27
    ... ... @@ -39,39 +39,40 @@ class OperationsInstance:
    39 39 
         def register_instance_with_server(self, instance_name, server):
    40 40 
             server.add_operations_instance(self, instance_name)
    41 41
    42 
    -    def get_operation(self, name):
    43 
    -        job = self._scheduler.jobs.get(name)
    42 
    +    def get_operation(self, job_name):
    43 
    +        try:
    44 
    +            operation = self._scheduler.get_job_operation(job_name)
    44 45
    45 
    -        if job is None:
    46 
    -            raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
    46 
    +        except NotFoundError:
    47 
    +            raise InvalidArgumentError("Operation name does not exist: [{}]".format(job_name))
    47 48
    48 
    -        else:
    49 
    -            return job.operation
    49 
    +        return operation
    50 50
    51 51 
         def list_operations(self, list_filter, page_size, page_token):
    52 52 
             # TODO: Pages
    53 53 
             # Spec says number of pages and length of a page are optional
    54 54 
             response = operations_pb2.ListOperationsResponse()
    55 
    +
    55 56 
             operations = []
    56 
    -        for job in self._scheduler.list_jobs():
    57 
    -            op = operations_pb2.Operation()
    58 
    -            op.CopyFrom(job.operation)
    59 
    -            operations.append(op)
    57 
    +        for job_name in self._scheduler.list_current_jobs():
    58 
    +            operation = self._scheduler.get_job_operation(job_name)
    59 
    +            operations.append(operation)
    60 60
    61 61 
             response.operations.extend(operations)
    62 62
    63 63 
             return response
    64 64
    65 
    -    def delete_operation(self, name):
    65 
    +    def delete_operation(self, job_name):
    66 66 
             try:
    67 
    -            self._scheduler.jobs.pop(name)
    67 
    +            # TODO: Unregister the caller client
    68 
    +            pass
    68 69
    69 
    -        except KeyError:
    70 
    -            raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
    70 
    +        except NotFoundError:
    71 
    +            raise InvalidArgumentError("Operation name does not exist: [{}]".format(job_name))
    71 72
    72 
    -    def cancel_operation(self, name):
    73 
    +    def cancel_operation(self, job_name):
    73 74 
             try:
    74 
    -            self._scheduler.cancel_job_operation(name)
    75 
    +            self._scheduler.cancel_job_operation(job_name)
    75 76
    76 
    -        except KeyError:
    77 
    -            raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
    77 
    +        except NotFoundError:
    78 
    +            raise InvalidArgumentError("Operation name does not exist: [{}]".format(job_name))

  • buildgrid/server/scheduler.py
    ... ... @@ -25,6 +25,7 @@ import logging
    25 25
    26 26 
     from buildgrid._enums import LeaseState, OperationStage
    27 27 
     from buildgrid._exceptions import NotFoundError
    28 
    +from buildgrid.server.job import Job
    28 29
    29 30
    30 31 
     class Scheduler:
    ... ... @@ -42,7 +43,11 @@ class Scheduler:
    42 43 
             self.__retries_count = 0
    43 44
    44 45 
             self._action_cache = action_cache
    45 
    -        self.jobs = {}
    46 
    +
    47 
    +        self.__jobs_by_action = {}
    48 
    +        self.__jobs_by_operation = {}
    49 
    +        self.__jobs_by_name = {}
    50 
    +
    46 51 
             self.__queue = deque()
    47 52
    48 53 
             self._is_instrumented = monitor
    ... ... @@ -52,55 +57,81 @@ class Scheduler:
    52 57
    53 58 
         # --- Public API ---
    54 59
    55 
    -    def register_operation_client(self, job_name, peer, message_queue):
    60 
    +    def list_current_jobs(self):
    61 
    +        """Returns a list of the :class:`Job` objects currently managed."""
    62 
    +        return self.__jobs_by_name.keys()
    63 
    +
    64 
    +    # --- Public API: REAPI ---
    65 
    +
    66 
    +    def register_job_operation_client(self, operation_name, peer, message_queue):
    56 67 
             """Subscribes to one of the job's :class:`Operation` stage changes.
    57 68
    58 69 
             Args:
    59 
    -            job_name (str): name of the job subscribe to.
    70 
    +            operation_name (str): name of the operation to subscribe to.
    60 71 
                 peer (str): a unique string identifying the client.
    61 72 
                 message_queue (queue.Queue): the event queue to register.
    62 73
    74 
    +        Returns:
    75 
    +            str: The name of the subscribed :class:`Operation`.
    76 
    +
    63 77 
             Raises:
    64 
    -            NotFoundError: If no job with `job_name` exists.
    78 
    +            NotFoundError: If no operation with `operation_name` exists.
    65 79 
             """
    66 80 
             try:
    67 
    -            job = self.jobs[job_name]
    81 
    +            job = self.__jobs_by_name[operation_name]
    82 
    +
    68 83 
             except KeyError:
    69 
    -            raise NotFoundError('No job named {} found.'.format(job_name))
    84 
    +            raise NotFoundError("Job name does not exist: [{}]"
    85 
    +                                .format(operation_name))
    70 86
    71 
    -        job.register_operation_client(peer, message_queue)
    87 
    +        return job.register_operation_client(peer, message_queue)
    72 88
    73 
    -    def unregister_operation_client(self, job_name, peer):
    89 
    +    def unregister_job_operation_client(self, operation_name, peer):
    74 90 
             """Unsubscribes to one of the job's :class:`Operation` stage change.
    75 91
    76 92 
             Args:
    77 
    -            job_name (str): name of the job to unsubscribe from.
    93 
    +            operation_name (str): name of the operation to unsubscribe from.
    78 94 
                 peer (str): a unique string identifying the client.
    79 95
    80 96 
             Raises:
    81 
    -            NotFoundError: If no job with `job_name` exists.
    97 
    +            NotFoundError: If no operation with `operation_name` exists.
    82 98 
             """
    83 99 
             try:
    84 
    -            job = self.jobs[job_name]
    100 
    +            job = self.__jobs_by_name[operation_name]
    101 
    +
    85 102 
             except KeyError:
    86 
    -            raise NotFoundError('No job named {} found.'.format(job_name))
    103 
    +            raise NotFoundError("Operation name does not exist: [{}]"
    104 
    +                                .format(operation_name))
    87 105
    88 106 
             job.unregister_operation_client(peer)
    89 107
    90 
    -        if job.n_clients == 0 and job.operation.done:
    91 
    -            del self.jobs[job.name]
    108 
    +        if job.n_clients == 0 and job.done:
    109 
    +            del self.__jobs_by_action[job.action_digest]
    110 
    +            del self.__jobs_by_name[job.name]
    92 111
    93 112 
                 if self._is_instrumented:
    94 
    -                self.__operations_by_stage[OperationStage.CACHE_CHECK].discard(job_name)
    95 
    -                self.__operations_by_stage[OperationStage.QUEUED].discard(job_name)
    96 
    -                self.__operations_by_stage[OperationStage.EXECUTING].discard(job_name)
    97 
    -                self.__operations_by_stage[OperationStage.COMPLETED].discard(job_name)
    113 
    +                self.__operations_by_stage[OperationStage.CACHE_CHECK].discard(job.name)
    114 
    +                self.__operations_by_stage[OperationStage.QUEUED].discard(job.name)
    115 
    +                self.__operations_by_stage[OperationStage.EXECUTING].discard(job.name)
    116 
    +                self.__operations_by_stage[OperationStage.COMPLETED].discard(job.name)
    117 
    +
    118 
    +                self.__leases_by_state[LeaseState.PENDING].discard(job.name)
    119 
    +                self.__leases_by_state[LeaseState.ACTIVE].discard(job.name)
    120 
    +                self.__leases_by_state[LeaseState.COMPLETED].discard(job.name)
    98 121
    99 
    -                self.__leases_by_state[LeaseState.PENDING].discard(job_name)
    100 
    -                self.__leases_by_state[LeaseState.ACTIVE].discard(job_name)
    101 
    -                self.__leases_by_state[LeaseState.COMPLETED].discard(job_name)
    122 
    +    def queue_job_operation(self, action, action_digest, priority=0, skip_cache_lookup=False):
    123 
    +        """Inserts a newly created job into the execution queue.
    102 124
    103 
    -    def queue_job(self, job, skip_cache_lookup=False):
    125 
    +        Args:
    126 
    +            action (Action): the given action to queue for execution.
    127 
    +            action_digest (Digest): the digest of the given action.
    128 
    +            priority (int): the execution job's priority.
    129 
    +            skip_cache_lookup (bool): whether or not to look for pre-computed
    130 
    +                result for the given action.
    131 
    +
    132 
    +        Returns:
    133 
    +            str: the newly created operation's name.
    134 
    +        """
    104 135 
             def __queue_job(jobs_queue, new_job):
    105 136 
                 index = 0
    106 137 
                 for queued_job in reversed(jobs_queue):
    ... ... @@ -113,12 +144,29 @@ class Scheduler:
    113 144
    114 145 
                 jobs_queue.insert(index, new_job)
    115 146
    116 
    -        self.jobs[job.name] = job
    147 
    +        if action_digest.hash in self.__jobs_by_action:
    148 
    +            job = self.__jobs_by_action[action_digest.hash]
    149 
    +
    150 
    +            # Reschedule if priority is now greater:
    151 
    +            if priority < job.priority:
    152 
    +                job.priority = priority
    153 
    +
    154 
    +                if job in self.__queue:
    155 
    +                    self.__queue.remove(job)
    156 
    +                    __queue_job(self.__queue, job)
    157 
    +
    158 
    +            return job.name
    159 
    +
    160 
    +        job = Job(action, action_digest, priority=priority)
    161 
    +
    162 
    +        self.__jobs_by_action[job.action_digest.hash] = job
    163 
    +        self.__jobs_by_name[job.name] = job
    117 164
    118 165 
             operation_stage = None
    119 166 
             if self._action_cache is not None and not skip_cache_lookup:
    120 167 
                 try:
    121 168 
                     action_result = self._action_cache.get_action_result(job.action_digest)
    169 
    +
    122 170 
                 except NotFoundError:
    123 171 
                     operation_stage = OperationStage.QUEUED
    124 172 
                     __queue_job(self.__queue, job)
    ... ... @@ -136,24 +184,47 @@ class Scheduler:
    136 184
    137 185 
             self._update_job_operation_stage(job.name, operation_stage)
    138 186
    139 
    -    def retry_job(self, job_name):
    140 
    -        job = self.jobs[job_name]
    187 
    +        return job.name
    141 188
    142 
    -        operation_stage = None
    143 
    -        if job.n_tries >= self.MAX_N_TRIES:
    144 
    -            # TODO: Decide what to do with these jobs
    145 
    -            operation_stage = OperationStage.COMPLETED
    146 
    -            # TODO: Mark these jobs as done
    189 
    +    def get_job_operation(self, operation_name):
    190 
    +        """Retrieves a job's :class:`Operation` by name.
    147 191
    148 
    -        else:
    149 
    -            operation_stage = OperationStage.QUEUED
    150 
    -            job.update_lease_state(LeaseState.PENDING)
    151 
    -            self.__queue.append(job)
    192 
    +        Args:
    193 
    +            operation_name (str): name of the operation to query.
    152 194
    153 
    -        self._update_job_operation_stage(job_name, operation_stage)
    195 
    +        Raises:
    196 
    +            NotFoundError: If no operation with `operation_name` exists.
    197 
    +        """
    198 
    +        try:
    199 
    +            job = self.__jobs_by_name[operation_name]
    200 
    +        except KeyError:
    201 
    +            raise NotFoundError("Operation name does not exist: [{}]"
    202 
    +                                .format(operation_name))
    203 
    +
    204 
    +        return job.get_operation()
    205 
    +
    206 
    +    def cancel_job_operation(self, operation_name):
    207 
    +        """"Cancels a job's :class:`Operation` by name.
    208 
    +
    209 
    +        Note:
    210 
    +            This will also cancel any :class:`Lease` that may have been issued
    211 
    +            for that job.
    212 
    +
    213 
    +        Args:
    214 
    +            operation_name (str): name of the operation to cancel.
    215 
    +
    216 
    +        Raises:
    217 
    +            NotFoundError: If no operation with `operation_name` exists.
    218 
    +        """
    219 
    +        try:
    220 
    +            job = self.__jobs_by_name[operation_name]
    221 
    +        except KeyError:
    222 
    +            raise NotFoundError("Operation name does not exist: [{}]"
    223 
    +                                .format(operation_name))
    224 
    +
    225 
    +        job.cancel_operation()
    154 226
    155 
    -    def list_jobs(self):
    156 
    -        return self.jobs.values()
    227 
    +    # --- Public API: RWAPI ---
    157 228
    158 229 
         def request_job_leases(self, worker_capabilities):
    159 230 
             """Generates a list of the highest priority leases to be run.
    ... ... @@ -179,18 +250,24 @@ class Scheduler:
    179 250
    180 251 
             return None
    181 252
    182 
    -    def update_job_lease(self, lease):
    253 
    +    def update_job_lease_state(self, job_name, lease):
    183 254 
             """Requests a state transition for a job's current :class:Lease.
    184 255
    256 
    +        Note:
    257 
    +            This may trigger a job's :class:`Operation` stage transition.
    258 
    +
    185 259 
             Args:
    186 260 
                 job_name (str): name of the job to query.
    187 
    -            lease_state (LeaseState): the lease state to transition to.
    188 
    -            lease_status (google.rpc.Status): the lease execution status, only
    189 
    -                required if `lease_state` is `COMPLETED`.
    190 
    -            lease_result (google.protobuf.Any): the lease execution result, only
    191 
    -                required if `lease_state` is `COMPLETED`.
    261 
    +            lease (Lease): the lease holding the new state.
    262 
    +
    263 
    +        Raises:
    264 
    +            NotFoundError: If no job with `job_name` exists.
    192 265 
             """
    193 
    -        job = self.jobs[lease.id]
    266 
    +        try:
    267 
    +            job = self.__jobs_by_name[job_name]
    268 
    +        except KeyError:
    269 
    +            raise NotFoundError("Job name does not exist: [{}]".format(job_name))
    270 
    +
    194 271 
             lease_state = LeaseState(lease.state)
    195 272
    196 273 
             operation_stage = None
    ... ... @@ -226,29 +303,69 @@ class Scheduler:
    226 303 
                     self.__leases_by_state[LeaseState.ACTIVE].discard(lease.id)
    227 304 
                     self.__leases_by_state[LeaseState.COMPLETED].add(lease.id)
    228 305
    229 
    -        self._update_job_operation_stage(lease.id, operation_stage)
    306 
    +        self._update_job_operation_stage(job_name, operation_stage)
    307 
    +
    308 
    +    def retry_job_lease(self, job_name):
    309 
    +        """Re-queues a job on lease execution failure.
    310 
    +
    311 
    +        Note:
    312 
    +            This may trigger a job's :class:`Operation` stage transition.
    313 
    +
    314 
    +        Args:
    315 
    +            job_name (str): name of the job to query.
    316 
    +
    317 
    +        Raises:
    318 
    +            NotFoundError: If no job with `job_name` exists.
    319 
    +        """
    320 
    +        try:
    321 
    +            job = self.__jobs_by_name[job_name]
    322 
    +        except KeyError:
    323 
    +            raise NotFoundError("Job name does not exist: [{}]".format(job_name))
    324 
    +
    325 
    +        operation_stage = None
    326 
    +        if job.n_tries >= self.MAX_N_TRIES:
    327 
    +            # TODO: Decide what to do with these jobs
    328 
    +            operation_stage = OperationStage.COMPLETED
    329 
    +            # TODO: Mark these jobs as done
    330 
    +
    331 
    +        else:
    332 
    +            operation_stage = OperationStage.QUEUED
    333 
    +            job.update_lease_state(LeaseState.PENDING)
    334 
    +            self.__queue.append(job)
    335 
    +
    336 
    +        self._update_job_operation_stage(job_name, operation_stage)
    230 337
    231 338 
         def get_job_lease(self, job_name):
    232 
    -        """Returns the lease associated to job, if any have been emitted yet."""
    233 
    -        return self.jobs[job_name].lease
    339 
    +        """Returns the lease associated to job, if any have been emitted yet.
    234 340
    235 
    -    def get_job_lease_cancelled(self, job_name):
    236 
    -        """Returns true if the lease is cancelled"""
    237 
    -        return self.jobs[job_name].lease_cancelled
    341 
    +        Args:
    342 
    +            job_name (str): name of the job to query.
    238 343
    239 
    -    def get_job_operation(self, job_name):
    240 
    -        """Returns the operation associated to job."""
    241 
    -        return self.jobs[job_name].operation
    344 
    +        Raises:
    345 
    +            NotFoundError: If no job with `job_name` exists.
    346 
    +        """
    347 
    +        try:
    348 
    +            job = self.__jobs_by_name[job_name]
    349 
    +        except KeyError:
    350 
    +            raise NotFoundError("Job name does not exist: [{}]".format(job_name))
    242 351
    243 
    -    def cancel_job_operation(self, job_name):
    244 
    -        """"Cancels the underlying operation of a given job.
    352 
    +        return job.lease
    245 353
    246 
    -        This will also cancel any job's lease that may have been issued.
    354 
    +    def get_job_lease_cancelled(self, job_name):
    355 
    +        """Returns true if the lease is cancelled.
    247 356
    248 357 
             Args:
    249 
    -            job_name (str): name of the job holding the operation to cancel.
    358 
    +            job_name (str): name of the job to query.
    359 
    +
    360 
    +        Raises:
    361 
    +            NotFoundError: If no job with `job_name` exists.
    250 362 
             """
    251 
    -        self.jobs[job_name].cancel_operation()
    363 
    +        try:
    364 
    +            job = self.__jobs_by_name[job_name]
    365 
    +        except KeyError:
    366 
    +            raise NotFoundError("Job name does not exist: [{}]".format(job_name))
    367 
    +
    368 
    +        return job.lease_cancelled
    252 369
    253 370 
         # --- Public API: Monitoring ---
    254 371
    ... ... @@ -298,11 +415,11 @@ class Scheduler:
    298 415 
                 self.__build_metadata_queues.append(message_queue)
    299 416
    300 417 
         def query_n_jobs(self):
    301 
    -        return len(self.jobs)
    418 
    +        return len(self.__jobs_by_name)
    302 419
    303 420 
         def query_n_operations(self):
    304 421 
             # For now n_operations == n_jobs:
    305 
    -        return len(self.jobs)
    422 
    +        return len(self.__jobs_by_operation)
    306 423
    307 424 
         def query_n_operations_by_stage(self, operation_stage):
    308 425 
             try:
    ... ... @@ -313,7 +430,7 @@ class Scheduler:
    313 430 
             return 0
    314 431
    315 432 
         def query_n_leases(self):
    316 
    -        return len(self.jobs)
    433 
    +        return len(self.__jobs_by_name)
    317 434
    318 435 
         def query_n_leases_by_state(self, lease_state):
    319 436 
             try:
    ... ... @@ -340,7 +457,7 @@ class Scheduler:
    340 457 
                 job_name (str): name of the job to query.
    341 458 
                 operation_stage (OperationStage): the stage to transition to.
    342 459 
             """
    343 
    -        job = self.jobs[job_name]
    460 
    +        job = self.__jobs_by_name[job_name]
    344 461
    345 462 
             if operation_stage == OperationStage.CACHE_CHECK:
    346 463 
                 job.update_operation_stage(OperationStage.CACHE_CHECK)
    ... ... @@ -389,7 +506,7 @@ class Scheduler:
    389 506
    390 507 
                     self.__queue_time_average = average_order, average_time
    391 508
    392 
    -                if not job.holds_cached_action_result:
    509 
    +                if not job.holds_cached_result:
    393 510 
                         execution_metadata = job.action_result.execution_metadata
    394 511 
                         context_metadata = {'job-is': job.name}
    395 512

  • docs/source/conf.py
    ... ... @@ -182,3 +182,11 @@ texinfo_documents = [
    182 182 
          author, 'BuildGrid', 'One line description of project.',
    183 183 
          'Miscellaneous'),
    184 184 
     ]
    185 
    +
    186 
    +# -- Options for the autodoc extension ----------------------------------------
    187 
    +
    188 
    +# This value selects if automatically documented members are sorted
    189 
    +# alphabetical (value 'alphabetical'), by member type (value 'groupwise') or
    190 
    +# by source order (value 'bysource'). The default is alphabetical.
    191 
    +autodoc_member_order = 'bysource'
    192 
    +

  • tests/integration/bots_service.py
    ... ... @@ -25,7 +25,6 @@ import pytest
    25 25
    26 26 
     from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
    27 27 
     from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
    28 
    -from buildgrid.server import job
    29 28 
     from buildgrid.server.controller import ExecutionController
    30 29 
     from buildgrid.server.job import LeaseState
    31 30 
     from buildgrid.server.bots import service
    ... ... @@ -159,7 +158,8 @@ def test_post_bot_event_temp(context, instance):
    159 158 
     def _inject_work(scheduler, action=None, action_digest=None):
    160 159 
         if not action:
    161 160 
             action = remote_execution_pb2.Action()
    161 
    +
    162 162 
         if not action_digest:
    163 163 
             action_digest = remote_execution_pb2.Digest()
    164 
    -    j = job.Job(action, action_digest)
    165 
    -    scheduler.queue_job(j, True)
    164 
    +
    165 
    +    scheduler.queue_job_operation(action, action_digest, skip_cache_lookup=True)

  • tests/integration/execution_service.py
    ... ... @@ -106,13 +106,13 @@ def test_no_action_digest_in_storage(instance, context):
    106 106
    107 107
    108 108 
     def test_wait_execution(instance, controller, context):
    109 
    -    j = job.Job(action, action_digest)
    109 
    +    j = controller.execution_instance._scheduler.queue_job_operation(action,
    110 
    +                                                                     action_digest,
    111 
    +                                                                     skip_cache_lookup=True)
    110 112 
         j._operation.done = True
    111 113
    112 114 
         request = remote_execution_pb2.WaitExecutionRequest(name=j.name)
    113 115
    114 
    -    controller.execution_instance._scheduler.jobs[j.name] = j
    115 
    -
    116 116 
         action_result_any = any_pb2.Any()
    117 117 
         action_result = remote_execution_pb2.ActionResult()
    118 118 
         action_result_any.Pack(action_result)

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.com. If you'd like to receive fewer emails, you can adjust your notification settings.

    [Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]