<commit_before># Copyright 2014 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.

"""High level tasks execution scheduling API.

This is the interface closest to the HTTP handlers.
"""

import collections
import datetime
import logging
import math
import random
import time

from google.appengine.api import datastore_errors
from google.appengine.ext import ndb

from components import auth
from components import datastore_utils
from components import pubsub
from components import utils

import ts_mon_metrics

from server import bot_management
from server import config
from server import external_scheduler
from server import pools_config
from server import service_accounts
from server import task_pack
from server import task_queues
from server import task_request
from server import task_result
from server import task_to_run


### Private stuff.


_PROBABILITY_OF_QUICK_COMEBACK = 0.05

# When falling back from external scheduler, requests that belong to any
# external scheduler are ignored for this duration at the beginning of their
# life. This number should be larger than the bot polling period.
_ES_FALLBACK_SLACK = datetime.timedelta(minutes=6)


class Error(Exception):
  pass


def _secs_to_ms(value):
  """Converts a seconds value in float to the number of ms as an integer."""
  return int(round(value * 1000.))


def _expire_task_tx(now, request, to_run_key, result_summary_key, capacity,
                    es_cfg):
  """Expires a to_run_key and look for a TaskSlice fallback.

  Called as a ndb transaction by _expire_task().

  2 concurrent GET, one PUT. Optionally with an additional serialized GET.
  """
  to_run_future = to_run_key.get_async()
  result_summary_future = result_summary_key.get_async()
  to_run = to_run_future.get_result()
  if not to_run or not to_run.is_reapable:
    result_summary_future.get_result()
    return None, None

  # record expiration delay
  to_run.expiration_delay = max(0, (now - to_run.expiration_ts).total_seconds())

  # In any case, dequeue the TaskToRun.
  to_run.queue_number = None
  to_run.expiration_ts = None
  result_summary = result_summary_future.get_result()
  to_put = [to_run, result_summary]
  # Check if there's a TaskSlice fallback that could be reenqueued.
  new_to_run = None
  offset = result_summary.current_task_slice+1
  rest = request.num_task_slices - offset
  for index in range(rest):
    # Use the lookup created just before the transaction. There's a small race
    # condition in here but we're willing to accept it.
    if len(capacity) > index and capacity[index]:
      # Enqueue a new TasktoRun for this next TaskSlice, it has capacity!
      new_to_run = task_to_run.new_task_to_run(request, index + offset)
      result_summary.current_task_slice = index+offset
      to_put.append(new_to_run)
      break
    if len(capacity) <= index:
      # crbug.com/1030504
      # This invalid situation probably come from the invalid task-expire task
      # parameters (task_id, try_number, task_slice_index), and the task-expire
      # task will be enqueued with a valid parameters in a cron job later.
      # So ignoring here should not be a problem.
      logging.warning(
          'crbug.com/1030504: invalid capacity length or slice index. '
          'index=%d, capacity=%s\n'
          'TaskResultSummary: task_id=%s, current_task_slice=%d, '
          'num_task_slices=%d\n'
          'TaskToRun: task_id=%s, task_slice_index=%d, try_number=%d',
          index, capacity,
          result_summary.task_id, result_summary.current_task_slice,
          request.num_task_slices,
          to_run.task_id, to_run.task_slice_index, to_run.try_number)

  if not new_to_run:
    # There's no fallback, giving up.
    if result_summary.try_number:
      # It's a retry that is being expired, i.e. the first try had BOT_DIED.
      # Keep the old state. That requires an additional pipelined GET but that
      # shouldn't be the common case.
      run_result = result_summary.run_result_key.get()
      result_summary.set_from_run_result(run_result, request)
    else:
      result_summary.state = task_result.State.EXPIRED
    result_summary.abandoned_ts = now
    result_summary.completed_ts = now
    result_summary.expiration_delay = max(
        0, (now - request.expiration_ts).total_seconds())
  result_summary.modified_ts = now

  futures = ndb.put_multi_async(to_put)
  _maybe_taskupdate_notify_via_tq(
      result_summary, request, es_cfg, transactional=True)
  for f in futures:
    f.check_success()

  return result_summary, new_to_run
<commit_msg>[swarming] log task expiration delay when it's negative value<commit_after># Copyright 2014 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.

"""High level tasks execution scheduling API.

This is the interface closest to the HTTP handlers.
"""

import collections
import datetime
import logging
import math
import random
import time

from google.appengine.api import datastore_errors
from google.appengine.ext import ndb

from components import auth
from components import datastore_utils
from components import pubsub
from components import utils

import ts_mon_metrics

from server import bot_management
from server import config
from server import external_scheduler
from server import pools_config
from server import service_accounts
from server import task_pack
from server import task_queues
from server import task_request
from server import task_result
from server import task_to_run


### Private stuff.


_PROBABILITY_OF_QUICK_COMEBACK = 0.05

# When falling back from external scheduler, requests that belong to any
# external scheduler are ignored for this duration at the beginning of their
# life. This number should be larger than the bot polling period.
_ES_FALLBACK_SLACK = datetime.timedelta(minutes=6)


class Error(Exception):
  pass


def _secs_to_ms(value):
  """Converts a seconds value in float to the number of ms as an integer."""
  return int(round(value * 1000.))


def _expire_task_tx(now, request, to_run_key, result_summary_key, capacity,
                    es_cfg):
  """Expires a to_run_key and look for a TaskSlice fallback.

  Called as a ndb transaction by _expire_task().

  2 concurrent GET, one PUT. Optionally with an additional serialized GET.
  """
  to_run_future = to_run_key.get_async()
  result_summary_future = result_summary_key.get_async()
  to_run = to_run_future.get_result()
  if not to_run or not to_run.is_reapable:
    result_summary_future.get_result()
    return None, None

  # record expiration delay
  delay = (now - to_run.expiration_ts).total_seconds()
  to_run.expiration_delay = max(0, delay)

  if delay <= 0:
    logging.warning(
        '_expire_task_tx: slice_expiration_delay is expected to be > 0, '
        'but was %s sec. run_id=%s, now=%s, expiration_ts=%s', delay,
        to_run.task_id, now, to_run.expiration_ts)

  # In any case, dequeue the TaskToRun.
  to_run.queue_number = None
  to_run.expiration_ts = None
  result_summary = result_summary_future.get_result()
  to_put = [to_run, result_summary]
  # Check if there's a TaskSlice fallback that could be reenqueued.
  new_to_run = None
  offset = result_summary.current_task_slice+1
  rest = request.num_task_slices - offset
  for index in range(rest):
    # Use the lookup created just before the transaction. There's a small race
    # condition in here but we're willing to accept it.
    if len(capacity) > index and capacity[index]:
      # Enqueue a new TasktoRun for this next TaskSlice, it has capacity!
      new_to_run = task_to_run.new_task_to_run(request, index + offset)
      result_summary.current_task_slice = index+offset
      to_put.append(new_to_run)
      break
    if len(capacity) <= index:
      # crbug.com/1030504
      # This invalid situation probably come from the invalid task-expire task
      # parameters (task_id, try_number, task_slice_index), and the task-expire
      # task will be enqueued with a valid parameters in a cron job later.
      # So ignoring here should not be a problem.
      logging.warning(
          'crbug.com/1030504: invalid capacity length or slice index. '
          'index=%d, capacity=%s\n'
          'TaskResultSummary: task_id=%s, current_task_slice=%d, '
          'num_task_slices=%d\n'
          'TaskToRun: task_id=%s, task_slice_index=%d, try_number=%d',
          index, capacity,
          result_summary.task_id, result_summary.current_task_slice,
          request.num_task_slices,
          to_run.task_id, to_run.task_slice_index, to_run.try_number)

  if not new_to_run:
    # There's no fallback, giving up.
    if result_summary.try_number:
      # It's a retry that is being expired, i.e. the first try had BOT_DIED.
      # Keep the old state. That requires an additional pipelined GET but that
      # shouldn't be the common case.
      run_result = result_summary.run_result_key.get()
      result_summary.set_from_run_result(run_result, request)
    else:
      result_summary.state = task_result.State.EXPIRED
    result_summary.abandoned_ts = now
    result_summary.completed_ts = now
    delay = (now - request.expiration_ts).total_seconds()
    result_summary.expiration_delay = max(0, delay)
    if delay <= 0:
      logging.warning(
          '_expire_task_tx: expiration_delay is expected to be > 0, '
          'but was %s sec. task_id=%s, now=%s, expiration_ts=%s', delay,
          result_summary.task_id, now, request.expiration_ts)

  result_summary.modified_ts = now

  futures = ndb.put_multi_async(to_put)
  _maybe_taskupdate_notify_via_tq(
      result_summary, request, es_cfg, transactional=True)
  for f in futures:
    f.check_success()

  return result_summary, new_to_run

{'fraction_non_alphanumeric': 0.06866214986348468, 'fraction_numerical': 0.006876327232278289, 'mean_word_length': 3.7870280735721202, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4881442', 'n_tokens_mistral': 3106, 'n_tokens_neox': 2994, 'n_words': 983}