<commit_before># Copyright 2012 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.

"""This module defines Isolate Server backend url handlers."""

import binascii
import datetime
import json
import logging
import time
import zlib

import webapp2
from google.appengine import runtime
from google.appengine.api import datastore_errors
from google.appengine.api import memcache
from google.appengine.ext import ndb

import config
import gcs
import mapreduce_jobs
import model
import stats
import template
from components import decorators
from components import utils


### Utility


class Accumulator(object):
  """Accumulates output from a generator."""
  def __init__(self, source):
    self.accumulated = []
    self._source = source

  def __iter__(self):
    for i in self._source:
      self.accumulated.append(i)
      yield i
      del i


def _split_payload(request, chunk_size, max_chunks):
  """Splits a binary payload into elements of |chunk_size| length.

  Returns each chunks.
  """
  content = request.request.body
  if len(content) % chunk_size:
    msg = (
        'Payload must be in increments of %d bytes, had %d bytes total, last '
        'chunk was of length %d' % (
              chunk_size,
              len(content),
              len(content) % chunk_size))
    logging.error(msg)
    request.abort(400, detail=msg)

  count = len(content) / chunk_size
  if count > max_chunks:
    msg = (
        'Requested more than %d hash digests in a single request, '
        'aborting' % count)
    logging.warning(msg)
    request.abort(400, detail=msg)

  return [content[i * chunk_size: (i + 1) * chunk_size] for i in xrange(count)]


def _payload_to_hashes(request, namespace):
  """Converts a raw payload into hashes as bytes."""
  h = model.get_hash(namespace)
  return _split_payload(request, h.digest_size, model.MAX_KEYS_PER_DB_OPS)


def _throttle_futures(futures, limit):
  finished = []
  if len(futures) <= limit:
    return futures, finished
  while len(futures) > limit:
    ndb.Future.wait_any(futures)
    tmp = []
    for f in futures:
      if f.done():
        f.get_result()
        finished.append(f)
      else:
        tmp.append(f)
    futures = tmp
  return futures, finished


def _incremental_delete(query, delete_async):
  """Applies |delete| to objects in a query asynchrously.

  This function is itself synchronous. It runs the query for at most 9 minutes
  to try to have enough time to finish the deletion. Swallows most frequent
  exceptions.

  Arguments:
  - query: iterator of items to process.
  - delete_async: callback that accepts an object to delete and returns a
        ndb.Future.

  Returns the number of objects found.
  """
  start = time.time()
  try:
    count = 0
    deleted_count = 0
    futures = []
    for item in query:
      count += 1
      if not (count % 1000):
        logging.debug('Found %d items', count)
      futures.append(delete_async(item))
      deleted_count += 1

      # Throttle.
      futures, _ = _throttle_futures(futures, 50)

      # Time to stop querying.
      if (time.time() - start) > 9*60:
        break

    for f in futures:
      f.get_result()
  except (
      datastore_errors.BadRequestError,
      datastore_errors.InternalError,
      datastore_errors.Timeout,
      datastore_errors.TransactionFailedError,
      runtime.DeadlineExceededError) as e:
    logging.info('Silencing exception %s', e)

  # Try to do the best it can.
  for f in futures:
    try:
      f.get_result()
    except (
        datastore_errors.BadRequestError,
        datastore_errors.InternalError,
        datastore_errors.Timeout,<commit_msg>[isolate] Throttle deletion more; 25 instead of 50<commit_after># Copyright 2012 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.

"""This module defines Isolate Server backend url handlers."""

import binascii
import datetime
import json
import logging
import time
import zlib

import webapp2
from google.appengine import runtime
from google.appengine.api import datastore_errors
from google.appengine.api import memcache
from google.appengine.ext import ndb

import config
import gcs
import mapreduce_jobs
import model
import stats
import template
from components import decorators
from components import utils


### Utility


class Accumulator(object):
  """Accumulates output from a generator."""
  def __init__(self, source):
    self.accumulated = []
    self._source = source

  def __iter__(self):
    for i in self._source:
      self.accumulated.append(i)
      yield i
      del i


def _split_payload(request, chunk_size, max_chunks):
  """Splits a binary payload into elements of |chunk_size| length.

  Returns each chunks.
  """
  content = request.request.body
  if len(content) % chunk_size:
    msg = (
        'Payload must be in increments of %d bytes, had %d bytes total, last '
        'chunk was of length %d' % (
              chunk_size,
              len(content),
              len(content) % chunk_size))
    logging.error(msg)
    request.abort(400, detail=msg)

  count = len(content) / chunk_size
  if count > max_chunks:
    msg = (
        'Requested more than %d hash digests in a single request, '
        'aborting' % count)
    logging.warning(msg)
    request.abort(400, detail=msg)

  return [content[i * chunk_size: (i + 1) * chunk_size] for i in xrange(count)]


def _payload_to_hashes(request, namespace):
  """Converts a raw payload into hashes as bytes."""
  h = model.get_hash(namespace)
  return _split_payload(request, h.digest_size, model.MAX_KEYS_PER_DB_OPS)


def _throttle_futures(futures, limit):
  finished = []
  if len(futures) <= limit:
    return futures, finished
  while len(futures) > limit:
    ndb.Future.wait_any(futures)
    tmp = []
    for f in futures:
      if f.done():
        f.get_result()
        finished.append(f)
      else:
        tmp.append(f)
    futures = tmp
  return futures, finished


def _incremental_delete(query, delete_async):
  """Applies |delete| to objects in a query asynchrously.

  This function is itself synchronous. It runs the query for at most 9 minutes
  to try to have enough time to finish the deletion. Swallows most frequent
  exceptions.

  Arguments:
  - query: iterator of items to process.
  - delete_async: callback that accepts an object to delete and returns a
        ndb.Future.

  Returns the number of objects found.
  """
  start = time.time()
  try:
    count = 0
    deleted_count = 0
    futures = []
    for item in query:
      count += 1
      if not (count % 1000):
        logging.debug('Found %d items', count)
      futures.append(delete_async(item))
      deleted_count += 1

      # Throttle.
      futures, _ = _throttle_futures(futures, 25)

      # Time to stop querying.
      if (time.time() - start) > 9*60:
        break

    for f in futures:
      f.get_result()
  except (
      datastore_errors.BadRequestError,
      datastore_errors.InternalError,
      datastore_errors.Timeout,
      datastore_errors.TransactionFailedError,
      runtime.DeadlineExceededError) as e:
    logging.info('Silencing exception %s', e)

  # Try to do the best it can.
  for f in futures:
    try:
      f.get_result()
    except (
        datastore_errors.BadRequestError,
        datastore_errors.InternalError,
        datastore_errors.Timeout,
{'fraction_non_alphanumeric': 0.08505257553573806, 'fraction_numerical': 0.007986157327299348, 'mean_word_length': 2.9883227176220806, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, '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': '6798973', 'n_tokens_mistral': 2389, 'n_tokens_neox': 2279, 'n_words': 756}