azure-batch/azure/batch/custom/patch.py
def _bulk_add_tasks(self, results_queue, chunk_tasks_to_add):
        """Adds a chunk of tasks to the job

        Retry chunk if body exceeds the maximum request size and retry tasks
        if failed due to server errors.

        :param results_queue: Queue to place the return value of the request
        :type results_queue: collections.deque
        :param chunk_tasks_to_add: Chunk of at most 100 tasks with retry details
        :type chunk_tasks_to_add: list[~TrackedCloudTask]
        """

        try:
            add_collection_response = self._original_add_collection(
                self._client,
                self._job_id,
                chunk_tasks_to_add,
                self._task_add_collection_options,
                self._custom_headers,
                self._raw)
        except BatchErrorException as e:
            # In case of a chunk exceeding the MaxMessageSize split chunk in half
            # and resubmit smaller chunk requests
            # TODO: Replace string with constant variable once available in SDK
            if e.error.code == "RequestBodyTooLarge":  # pylint: disable=no-member
                # In this case the task is misbehaved and will not be able to be added due to:
                #   1) The task exceeding the max message size
                #   2) A single cell of the task exceeds the per-cell limit, or
                #   3) Sum of all cells exceeds max row limit
                if len(chunk_tasks_to_add) == 1:
                    failed_task = chunk_tasks_to_add.pop()
                    self.errors.appendleft(e)
                    _LOGGER.error("Failed to add task with ID %s due to the body"
                                  " exceeding the maximum request size", failed_task.id)
                else:
                    # Assumption: Tasks are relatively close in size therefore if one batch exceeds size limit
                    # we should decrease the initial task collection size to avoid repeating the error
                    # Midpoint is lower bounded by 1 due to above base case
                    midpoint = int(len(chunk_tasks_to_add) / 2)
                    # Restrict one thread at a time to do this compare and set,
                    # therefore forcing max_tasks_per_request to be strictly decreasing
                    with self._max_tasks_lock:
                        if midpoint < self._max_tasks_per_request:
                            self._max_tasks_per_request = midpoint
                            _LOGGER.info("Amount of tasks per request reduced from %s to %s due to the"
                                         " request body being too large", str(self._max_tasks_per_request),
                                         str(midpoint))

                    # Not the most efficient solution for all cases, but the goal of this is to handle this
                    # exception and have it work in all cases where tasks are well behaved
                    # Behavior retries as a smaller chunk and
                    # appends extra tasks to queue to be picked up by another thread .
                    self.tasks_to_add.extendleft(chunk_tasks_to_add[midpoint:])
                    self._bulk_add_tasks(results_queue, chunk_tasks_to_add[:midpoint])
            # Retry server side errors
            elif 500 <= e.response.status_code <= 599:
                self.tasks_to_add.extendleft(chunk_tasks_to_add)
            else:
                # Re-add to pending queue as unknown status / don't have result
                self.tasks_to_add.extendleft(chunk_tasks_to_add)
                # Unknown State - don't know if tasks failed to add or were successful
                self.errors.appendleft(e)
        except Exception as e:  # pylint: disable=broad-except
            # Re-add to pending queue as unknown status / don't have result
            self.tasks_to_add.extendleft(chunk_tasks_to_add)
            # Unknown State - don't know if tasks failed to add or were successful
            self.errors.appendleft(e)
        else:
            try:
                add_collection_response = add_collection_response.output
            except AttributeError:
                pass

            for task_result in add_collection_response.value:  # pylint: disable=no-member
                if task_result.status == TaskAddStatus.server_error:
                    # Server error will be retried
                    with self._pending_queue_lock:
                        for task in chunk_tasks_to_add:
                            if task.id == task_result.task_id:
                                self.tasks_to_add.appendleft(task)
                elif (task_result.status == TaskAddStatus.client_error
                        and not task_result.error.code == "TaskExists"):
                    # Client error will be recorded unless Task already exists
                    self.failure_tasks.appendleft(task_result)
                else:
                    results_queue.appendleft(task_result)
def _bulk_add_tasks(self, results_queue, chunk_tasks_to_add):
        try:
            add_collection_response = self._original_add_collection(
                self._client,
                self._job_id,
                chunk_tasks_to_add,
                self._task_add_collection_options,
                self._custom_headers,
                self._raw)
        except BatchErrorException as e:
            # In case of a chunk exceeding the MaxMessageSize split chunk in half
            # and resubmit smaller chunk requests
            # TODO: Replace string with constant variable once available in SDK
            if e.error.code == "RequestBodyTooLarge":  # pylint: disable=no-member
                # In this case the task is misbehaved and will not be able to be added due to:
                #   1) The task exceeding the max message size
                #   2) A single cell of the task exceeds the per-cell limit, or
                #   3) Sum of all cells exceeds max row limit
                if len(chunk_tasks_to_add) == 1:
                    failed_task = chunk_tasks_to_add.pop()
                    self.errors.appendleft(e)
                    _LOGGER.error("Failed to add task with ID %s due to the body"
                                  " exceeding the maximum request size", failed_task.id)
                else:
                    # Assumption: Tasks are relatively close in size therefore if one batch exceeds size limit
                    # we should decrease the initial task collection size to avoid repeating the error
                    # Midpoint is lower bounded by 1 due to above base case
                    midpoint = int(len(chunk_tasks_to_add) / 2)
                    # Restrict one thread at a time to do this compare and set,
                    # therefore forcing max_tasks_per_request to be strictly decreasing
                    with self._max_tasks_lock:
                        if midpoint < self._max_tasks_per_request:
                            self._max_tasks_per_request = midpoint
                            _LOGGER.info("Amount of tasks per request reduced from %s to %s due to the"
                                         " request body being too large", str(self._max_tasks_per_request),
                                         str(midpoint))

                    # Not the most efficient solution for all cases, but the goal of this is to handle this
                    # exception and have it work in all cases where tasks are well behaved
                    # Behavior retries as a smaller chunk and
                    # appends extra tasks to queue to be picked up by another thread .
                    self.tasks_to_add.extendleft(chunk_tasks_to_add[midpoint:])
                    self._bulk_add_tasks(results_queue, chunk_tasks_to_add[:midpoint])
            # Retry server side errors
            elif 500 <= e.response.status_code <= 599:
                self.tasks_to_add.extendleft(chunk_tasks_to_add)
            else:
                # Re-add to pending queue as unknown status / don't have result
                self.tasks_to_add.extendleft(chunk_tasks_to_add)
                # Unknown State - don't know if tasks failed to add or were successful
                self.errors.appendleft(e)
        except Exception as e:  # pylint: disable=broad-except
            # Re-add to pending queue as unknown status / don't have result
            self.tasks_to_add.extendleft(chunk_tasks_to_add)
            # Unknown State - don't know if tasks failed to add or were successful
            self.errors.appendleft(e)
        else:
            try:
                add_collection_response = add_collection_response.output
            except AttributeError:
                pass

            for task_result in add_collection_response.value:  # pylint: disable=no-member
                if task_result.status == TaskAddStatus.server_error:
                    # Server error will be retried
                    with self._pending_queue_lock:
                        for task in chunk_tasks_to_add:
                            if task.id == task_result.task_id:
                                self.tasks_to_add.appendleft(task)
                elif (task_result.status == TaskAddStatus.client_error
                        and not task_result.error.code == "TaskExists"):
                    # Client error will be recorded unless Task already exists
                    self.failure_tasks.appendleft(task_result)
                else:
                    results_queue.appendleft(task_result)
def _bulk_add_tasks ( self , results_queue , chunk_tasks_to_add ) : try : add_collection_response = self . _original_add_collection ( self . _client , self . _job_id , chunk_tasks_to_add , self . _task_add_collection_options , self . _custom_headers , self . _raw ) except BatchErrorException as e : # In case of a chunk exceeding the MaxMessageSize split chunk in half # and resubmit smaller chunk requests # TODO: Replace string with constant variable once available in SDK if e . error . code == "RequestBodyTooLarge" : # pylint: disable=no-member # In this case the task is misbehaved and will not be able to be added due to: #   1) The task exceeding the max message size #   2) A single cell of the task exceeds the per-cell limit, or #   3) Sum of all cells exceeds max row limit if len ( chunk_tasks_to_add ) == 1 : failed_task = chunk_tasks_to_add . pop ( ) self . errors . appendleft ( e ) _LOGGER . error ( "Failed to add task with ID %s due to the body" " exceeding the maximum request size" , failed_task . id ) else : # Assumption: Tasks are relatively close in size therefore if one batch exceeds size limit # we should decrease the initial task collection size to avoid repeating the error # Midpoint is lower bounded by 1 due to above base case midpoint = int ( len ( chunk_tasks_to_add ) / 2 ) # Restrict one thread at a time to do this compare and set, # therefore forcing max_tasks_per_request to be strictly decreasing with self . _max_tasks_lock : if midpoint < self . _max_tasks_per_request : self . _max_tasks_per_request = midpoint _LOGGER . info ( "Amount of tasks per request reduced from %s to %s due to the" " request body being too large" , str ( self . _max_tasks_per_request ) , str ( midpoint ) ) # Not the most efficient solution for all cases, but the goal of this is to handle this # exception and have it work in all cases where tasks are well behaved # Behavior retries as a smaller chunk and # appends extra tasks to queue to be picked up by another thread . self . tasks_to_add . extendleft ( chunk_tasks_to_add [ midpoint : ] ) self . _bulk_add_tasks ( results_queue , chunk_tasks_to_add [ : midpoint ] ) # Retry server side errors elif 500 <= e . response . status_code <= 599 : self . tasks_to_add . extendleft ( chunk_tasks_to_add ) else : # Re-add to pending queue as unknown status / don't have result self . tasks_to_add . extendleft ( chunk_tasks_to_add ) # Unknown State - don't know if tasks failed to add or were successful self . errors . appendleft ( e ) except Exception as e : # pylint: disable=broad-except # Re-add to pending queue as unknown status / don't have result self . tasks_to_add . extendleft ( chunk_tasks_to_add ) # Unknown State - don't know if tasks failed to add or were successful self . errors . appendleft ( e ) else : try : add_collection_response = add_collection_response . output except AttributeError : pass for task_result in add_collection_response . value : # pylint: disable=no-member if task_result . status == TaskAddStatus . server_error : # Server error will be retried with self . _pending_queue_lock : for task in chunk_tasks_to_add : if task . id == task_result . task_id : self . tasks_to_add . appendleft ( task ) elif ( task_result . status == TaskAddStatus . client_error and not task_result . error . code == "TaskExists" ) : # Client error will be recorded unless Task already exists self . failure_tasks . appendleft ( task_result ) else : results_queue . appendleft ( task_result )
Adds a chunk of tasks to the job

        Retry chunk if body exceeds the maximum request size and retry tasks
        if failed due to server errors.

        :param results_queue: Queue to place the return value of the request
        :type results_queue: collections.deque
        :param chunk_tasks_to_add: Chunk of at most 100 tasks with retry details
        :type chunk_tasks_to_add: list[~TrackedCloudTask]
Adds a chunk of tasks to the job