<commit_before># Copyright 2016 PerfKitBenchmarker Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Background tasks that propagate PKB thread context.

TODO(skschneider): Many of the threading module flaws have been corrected in
Python 3. When PKB switches to Python 3, this module can be simplified.

PKB tries its best to clean up provisioned resources upon SIGINT. By default,
Python raises a KeyboardInterrupt upon a SIGINT, but none of the built-in
threading module classes are designed to handle a KeyboardInterrupt very well:

- threading.Lock has an atomic acquire method that cannot be interrupted and
  hangs forever if the same thread tries to acquire twice. Its release method
  can be called by any thread but raises thread.error if an unacquired Lock is
  released.

- More complicated classes (threading.RLock, threading.Event, threading.Thread,
  Queue.Queue) use internal Locks in such a way that a KeyboardInterrupt can
  cause a thread that has acquired a Lock to jump out of its current action
  without releasing the Lock. For example, in the below code, a
  KeyboardInterrupt can be raised immediately after the acquire call but before
  entering the try block:
    lock.acquire()
    try:
      ...
    except:
      lock.release()

Taken together, this means that there is a possibility to leave an internal Lock
acquired, and when later cleanup steps on the same or different thread attempt
to acquire the Lock, they will hang forever, unresponsive to even a second
KeyboardInterrupt. A KeyboardInterrupt during Thread.start() or Thread.join()
can even trigger an unbalanced acquire on a global lock used to keep track of
active threads, so that later attempts to start or join any Thread will hang
forever.

While it would take a significant and impractical redesign of PKB's code to
completely eliminate any risk of deadlock following a KeyboardInterrupt, the
code in this module is designed to allow interrupting parallel tasks while
keeping the risk of deadlock low.
"""


import abc
from collections import deque
import ctypes
import functools
import logging
import os
import signal
import threading
import time
import traceback
from concurrent import futures

from perfkitbenchmarker import context
from perfkitbenchmarker import errors
from absl import flags
from perfkitbenchmarker import log_util
import six
from six.moves import queue
from six.moves import range
from six.moves import zip


# For situations where an interruptable wait is necessary, a loop of waits with
# long timeouts is used instead. This is because some of Python's built-in wait
# methods are non-interruptable without a timeout.
_LONG_TIMEOUT = 1000.

# Constants used for polling waits. See _WaitForCondition.
_WAIT_MIN_RECHECK_DELAY = 0.001  # 1 ms
_WAIT_MAX_RECHECK_DELAY = 0.050  # 50 ms

# Values sent to child threads that have special meanings.
_THREAD_STOP_PROCESSING = 0
_THREAD_WAIT_FOR_KEYBOARD_INTERRUPT = 1

# The default value for max_concurrent_threads.
MAX_CONCURRENT_THREADS = 200

# The default value is set in pkb.py. It is the greater of
# MAX_CONCURRENT_THREADS or the value passed to --num_vms. This is particularly
# important for the cluster_boot benchmark where we want to launch all of the
# VMs in parallel.
flags.DEFINE_integer(
    'max_concurrent_threads', None, 'Maximum number of concurrent threads to '
    'use when running a benchmark.')
FLAGS = flags.FLAGS


def _GetCallString(target_arg_tuple):
  """Returns the string representation of a function call."""
  target, args, kwargs = target_arg_tuple
  while isinstance(target, functools.partial):
    args = target.args + args
    inner_kwargs = target.keywords.copy()
    inner_kwargs.update(kwargs)
    kwargs = inner_kwargs
    target = target.func
  arg_strings = [str(a) for a in args]
  arg_strings.extend(['{0}={1}'.format(k, v) for k, v in six.iteritems(kwargs)])
  return '{0}({1})'.format(getattr(target, '__name__', target),
                           ', '.join(arg_strings))


def _WaitForCondition(condition_callback, timeout=None):
  """Waits until the specified callback returns a value that evaluates True.

  Similar to the threading.Condition.wait method that is the basis of most
  threading class wait routines. Polls the condition, starting with frequent
  checks but extending the delay between checks upon each failure.

  Args:
    condition_callback: Callable that returns a value that evaluates True to end
        the wait or evaluates False to continue the wait.
    timeout: Optional float. Number of seconds to wait before giving up. If
        provided, the condition is still checked at least once before giving up.
        If not provided, the wait does not time out.

  Returns:
    True if condition_callback returned a value that evaluated True. False if
    condition_callback did not return a value that evaluated True before the
    timeout.
  """
  deadline = None if timeout is None else time.time() + timeout
  delay = _WAIT_MIN_RECHECK_DELAY
  while True:
    if condition_callback():
      return True
    remaining_time = (_WAIT_MAX_RECHECK_DELAY if deadline is None
                      else deadline - time.time())
    if remaining_time <= 0:
      return False
    time.sleep(delay)
    delay = min(delay * 2, remaining_time, _WAIT_MAX_RECHECK_DELAY)


class _SingleReaderQueue(object):
  """Queue to which multiple threads write but from which only one thread reads.

  A lightweight substitute for the Queue.Queue class that does not use
  internal Locks.

  Gets are interruptable but depend on polling.
  """

  def __init__(self):
    self._deque = deque()

  def Get(self, timeout=None):
    if not _WaitForCondition(lambda: self._deque, timeout):
      raise queue.Empty
    return self._deque.popleft()

  def Put(self, item):
    self._deque.append(item)


class _NonPollingSingleReaderQueue(object):
  """Queue to which multiple threads write but from which only one thread reads.

  Uses a threading.Lock to implement a non-interruptable Get that does not poll
  and is therefore easier on CPU usage. The reader waits for items by acquiring
  the Lock, and writers release the Lock to signal that items have been written.
  """

  def __init__(self):
    self._deque = deque()
    self._lock = threading.Lock()
    self._lock.acquire()

  def _WaitForItem(self):
    self._lock.acquire()

  def _SignalAvailableItem(self):
    try:
      self._lock.release()
    except threading.ThreadError:
      pass

  def Get(self):
    while True:
      self._WaitForItem()
      if self._deque:<commit_msg>Linter fixes.<commit_after># Copyright 2016 PerfKitBenchmarker Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Background tasks that propagate PKB thread context.

TODO(skschneider): Many of the threading module flaws have been corrected in
Python 3. When PKB switches to Python 3, this module can be simplified.

PKB tries its best to clean up provisioned resources upon SIGINT. By default,
Python raises a KeyboardInterrupt upon a SIGINT, but none of the built-in
threading module classes are designed to handle a KeyboardInterrupt very well:

- threading.Lock has an atomic acquire method that cannot be interrupted and
  hangs forever if the same thread tries to acquire twice. Its release method
  can be called by any thread but raises thread.error if an unacquired Lock is
  released.

- More complicated classes (threading.RLock, threading.Event, threading.Thread,
  Queue.Queue) use internal Locks in such a way that a KeyboardInterrupt can
  cause a thread that has acquired a Lock to jump out of its current action
  without releasing the Lock. For example, in the below code, a
  KeyboardInterrupt can be raised immediately after the acquire call but before
  entering the try block:
    lock.acquire()
    try:
      ...
    except:
      lock.release()

Taken together, this means that there is a possibility to leave an internal Lock
acquired, and when later cleanup steps on the same or different thread attempt
to acquire the Lock, they will hang forever, unresponsive to even a second
KeyboardInterrupt. A KeyboardInterrupt during Thread.start() or Thread.join()
can even trigger an unbalanced acquire on a global lock used to keep track of
active threads, so that later attempts to start or join any Thread will hang
forever.

While it would take a significant and impractical redesign of PKB's code to
completely eliminate any risk of deadlock following a KeyboardInterrupt, the
code in this module is designed to allow interrupting parallel tasks while
keeping the risk of deadlock low.
"""


import abc
import collections
from concurrent import futures
import ctypes
import functools
import logging
import os
import signal
import threading
import time
import traceback

from absl import flags
from perfkitbenchmarker import context
from perfkitbenchmarker import errors
from perfkitbenchmarker import log_util
import six
from six.moves import queue
from six.moves import range
from six.moves import zip


# For situations where an interruptable wait is necessary, a loop of waits with
# long timeouts is used instead. This is because some of Python's built-in wait
# methods are non-interruptable without a timeout.
_LONG_TIMEOUT = 1000.

# Constants used for polling waits. See _WaitForCondition.
_WAIT_MIN_RECHECK_DELAY = 0.001  # 1 ms
_WAIT_MAX_RECHECK_DELAY = 0.050  # 50 ms

# Values sent to child threads that have special meanings.
_THREAD_STOP_PROCESSING = 0
_THREAD_WAIT_FOR_KEYBOARD_INTERRUPT = 1

# The default value for max_concurrent_threads.
MAX_CONCURRENT_THREADS = 200

# The default value is set in pkb.py. It is the greater of
# MAX_CONCURRENT_THREADS or the value passed to --num_vms. This is particularly
# important for the cluster_boot benchmark where we want to launch all of the
# VMs in parallel.
flags.DEFINE_integer(
    'max_concurrent_threads', None, 'Maximum number of concurrent threads to '
    'use when running a benchmark.')
FLAGS = flags.FLAGS


def _GetCallString(target_arg_tuple):
  """Returns the string representation of a function call."""
  target, args, kwargs = target_arg_tuple
  while isinstance(target, functools.partial):
    args = target.args + args
    inner_kwargs = target.keywords.copy()
    inner_kwargs.update(kwargs)
    kwargs = inner_kwargs
    target = target.func
  arg_strings = [str(a) for a in args]
  arg_strings.extend(['{0}={1}'.format(k, v) for k, v in six.iteritems(kwargs)])
  return '{0}({1})'.format(getattr(target, '__name__', target),
                           ', '.join(arg_strings))


def _WaitForCondition(condition_callback, timeout=None):
  """Waits until the specified callback returns a value that evaluates True.

  Similar to the threading.Condition.wait method that is the basis of most
  threading class wait routines. Polls the condition, starting with frequent
  checks but extending the delay between checks upon each failure.

  Args:
    condition_callback: Callable that returns a value that evaluates True to end
        the wait or evaluates False to continue the wait.
    timeout: Optional float. Number of seconds to wait before giving up. If
        provided, the condition is still checked at least once before giving up.
        If not provided, the wait does not time out.

  Returns:
    True if condition_callback returned a value that evaluated True. False if
    condition_callback did not return a value that evaluated True before the
    timeout.
  """
  deadline = None if timeout is None else time.time() + timeout
  delay = _WAIT_MIN_RECHECK_DELAY
  while True:
    if condition_callback():
      return True
    remaining_time = (_WAIT_MAX_RECHECK_DELAY if deadline is None
                      else deadline - time.time())
    if remaining_time <= 0:
      return False
    time.sleep(delay)
    delay = min(delay * 2, remaining_time, _WAIT_MAX_RECHECK_DELAY)


class _SingleReaderQueue(object):
  """Queue to which multiple threads write but from which only one thread reads.

  A lightweight substitute for the Queue.Queue class that does not use
  internal Locks.

  Gets are interruptable but depend on polling.
  """

  def __init__(self):
    self._deque = collections.deque()

  def Get(self, timeout=None):
    if not _WaitForCondition(lambda: self._deque, timeout):
      raise queue.Empty
    return self._deque.popleft()

  def Put(self, item):
    self._deque.append(item)


class _NonPollingSingleReaderQueue(object):
  """Queue to which multiple threads write but from which only one thread reads.

  Uses a threading.Lock to implement a non-interruptable Get that does not poll
  and is therefore easier on CPU usage. The reader waits for items by acquiring
  the Lock, and writers release the Lock to signal that items have been written.
  """

  def __init__(self):
    self._deque = collections.deque()
    self._lock = threading.Lock()
    self._lock.acquire()

  def _WaitForItem(self):
    self._lock.acquire()

  def _SignalAvailableItem(self):
    try:
      self._lock.release()
    except threading.ThreadError:
      pass

  def Get(self):
    while True:
      self._WaitForItem()
      if self._deque:
{'fraction_non_alphanumeric': 0.056790123456790124, 'fraction_numerical': 0.005079365079365079, 'mean_word_length': 3.9171002428026362, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5614067', 'n_tokens_mistral': 3893, 'n_tokens_neox': 3727, 'n_words': 1782}