checker.py 25.8 KB
Newer Older
1
"""Checker Manager and Checker classes."""
2
import errno
3
import logging
4
import os
5 6
import sys
import tokenize
7 8 9 10 11 12

try:
    import multiprocessing
except ImportError:
    multiprocessing = None

13 14 15 16 17 18
try:
    import Queue as queue
except ImportError:
    import queue

from flake8 import defaults
19
from flake8 import exceptions
20
from flake8 import processor
21 22
from flake8 import utils

23 24
LOG = logging.getLogger(__name__)

25 26 27 28 29 30 31 32 33 34 35 36 37 38 39
SERIAL_RETRY_ERRNOS = set([
    # ENOSPC: Added by sigmavirus24
    # > On some operating systems (OSX), multiprocessing may cause an
    # > ENOSPC error while trying to trying to create a Semaphore.
    # > In those cases, we should replace the customized Queue Report
    # > class with pep8's StandardReport class to ensure users don't run
    # > into this problem.
    # > (See also: https://gitlab.com/pycqa/flake8/issues/74)
    errno.ENOSPC,
    # NOTE(sigmavirus24): When adding to this list, include the reasoning
    # on the lines before the error code and always append your error
    # code. Further, please always add a trailing `,` to reduce the visual
    # noise in diffs.
])

40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59

class Manager(object):
    """Manage the parallelism and checker instances for each plugin and file.

    This class will be responsible for the following:

    - Determining the parallelism of Flake8, e.g.:

      * Do we use :mod:`multiprocessing` or is it unavailable?

      * Do we automatically decide on the number of jobs to use or did the
        user provide that?

    - Falling back to a serial way of processing files if we run into an
      OSError related to :mod:`multiprocessing`

    - Organizing the results of each checker so we can group the output
      together and make our output deterministic.
    """

60
    def __init__(self, style_guide, arguments, checker_plugins):
61 62
        """Initialize our Manager instance.

63 64 65 66
        :param style_guide:
            The instantiated style guide for this instance of Flake8.
        :type style_guide:
            flake8.style_guide.StyleGuide
67 68 69 70 71 72 73 74
        :param list arguments:
            The extra arguments parsed from the CLI (if any)
        :param checker_plugins:
            The plugins representing checks parsed from entry-points.
        :type checker_plugins:
            flake8.plugins.manager.Checkers
        """
        self.arguments = arguments
75 76
        self.style_guide = style_guide
        self.options = style_guide.options
77 78
        self.checks = checker_plugins
        self.jobs = self._job_count()
79
        self.process_queue = None
80
        self.results_queue = None
81
        self.statistics_queue = None
82
        self.using_multiprocessing = self.jobs > 1
83 84
        self.processes = []
        self.checkers = []
85 86 87 88 89 90
        self.statistics = {
            'files': 0,
            'logical lines': 0,
            'physical lines': 0,
            'tokens': 0,
        }
91

92 93 94 95
        if self.using_multiprocessing:
            try:
                self.process_queue = multiprocessing.Queue()
                self.results_queue = multiprocessing.Queue()
96
                self.statistics_queue = multiprocessing.Queue()
97 98 99 100
            except OSError as oserr:
                if oserr.errno not in SERIAL_RETRY_ERRNOS:
                    raise
                self.using_multiprocessing = False
101

102 103 104 105 106 107 108 109 110 111 112
    @staticmethod
    def _cleanup_queue(q):
        while not q.empty():
            q.get_nowait()

    def _force_cleanup(self):
        if self.using_multiprocessing:
            for proc in self.processes:
                proc.join(0.2)
            self._cleanup_queue(self.process_queue)
            self._cleanup_queue(self.results_queue)
113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135
            self._cleanup_queue(self.statistics_queue)

    def _process_statistics(self):
        all_statistics = self.statistics
        if self.using_multiprocessing:
            total_number_of_checkers = len(self.checkers)
            statistics_gathered = 0
            while statistics_gathered < total_number_of_checkers:
                try:
                    statistics = self.statistics_queue.get(block=False)
                    statistics_gathered += 1
                except queue.Empty:
                    break

                for statistic in defaults.STATISTIC_NAMES:
                    all_statistics[statistic] += statistics[statistic]
        else:
            statistics_generator = (checker.statistics
                                    for checker in self.checkers)
            for statistics in statistics_generator:
                for statistic in defaults.STATISTIC_NAMES:
                    all_statistics[statistic] += statistics[statistic]
        all_statistics['files'] += len(self.checkers)
136

137
    def _job_count(self):
138
        # type: () -> int
139 140 141 142 143 144 145 146 147 148 149 150
        # First we walk through all of our error cases:
        # - multiprocessing library is not present
        # - we're running on windows in which case we know we have significant
        #   implemenation issues
        # - the user provided stdin and that's not something we can handle
        #   well
        # - we're processing a diff, which again does not work well with
        #   multiprocessing and which really shouldn't require multiprocessing
        # - the user provided some awful input
        if not multiprocessing:
            LOG.warning('The multiprocessing module is not available. '
                        'Ignoring --jobs arguments.')
151
            return 0
152

153 154
        if (utils.is_windows() and
                not utils.can_run_multiprocessing_on_windows()):
155 156
            LOG.warning('The --jobs option is not available on Windows due to'
                        ' a bug (https://bugs.python.org/issue27649) in '
157 158 159
                        'Python 2.7.11+ and 3.3+. We have detected that you '
                        'are running an unsupported version of Python on '
                        'Windows. Ignoring --jobs arguments.')
160
            return 0
161 162 163 164

        if utils.is_using_stdin(self.arguments):
            LOG.warning('The --jobs option is not compatible with supplying '
                        'input using - . Ignoring --jobs arguments.')
165
            return 0
166 167 168 169

        if self.options.diff:
            LOG.warning('The --diff option was specified with --jobs but '
                        'they are not compatible. Ignoring --jobs arguments.')
170
            return 0
171 172 173 174 175

        jobs = self.options.jobs
        if jobs != 'auto' and not jobs.isdigit():
            LOG.warning('"%s" is not a valid parameter to --jobs. Must be one '
                        'of "auto" or a numerical value, e.g., 4.', jobs)
176
            return 0
177 178 179 180 181 182 183 184 185

        # If the value is "auto", we want to let the multiprocessing library
        # decide the number based on the number of CPUs. However, if that
        # function is not implemented for this particular value of Python we
        # default to 1
        if jobs == 'auto':
            try:
                return multiprocessing.cpu_count()
            except NotImplementedError:
186
                return 0
187 188 189 190

        # Otherwise, we know jobs should be an integer and we can just convert
        # it to an integer
        return int(jobs)
191

192 193
    def _results(self):
        seen_done = 0
194
        LOG.info('Retrieving results')
195 196 197 198 199 200 201 202 203 204
        while True:
            result = self.results_queue.get()
            if result == 'DONE':
                seen_done += 1
                if seen_done >= self.jobs:
                    break
                continue

            yield result

205
    def _handle_results(self, filename, results):
206
        style_guide = self.style_guide
207
        reported_results_count = 0
208
        for (error_code, line_number, column, text, physical_line) in results:
209
            reported_results_count += style_guide.handle_error(
210 211 212 213 214 215 216
                code=error_code,
                filename=filename,
                line_number=line_number,
                column_number=column,
                text=text,
                physical_line=physical_line,
            )
217
        return reported_results_count
218

219 220 221
    def is_path_excluded(self, path):
        # type: (str) -> bool
        """Check if a path is excluded.
222

223 224 225 226 227 228 229 230
        :param str path:
            Path to check against the exclude patterns.
        :returns:
            True if there are exclude patterns and the path matches,
            otherwise False.
        :rtype:
            bool
        """
231 232 233 234 235
        if path == '-':
            if self.options.stdin_display_name == 'stdin':
                return False
            path = self.options.stdin_display_name

236 237 238 239 240
        exclude = self.options.exclude
        if not exclude:
            return False
        basename = os.path.basename(path)
        if utils.fnmatch(basename, exclude):
241
            LOG.debug('"%s" has been excluded', basename)
242 243 244 245
            return True

        absolute_path = os.path.abspath(path)
        match = utils.fnmatch(absolute_path, exclude)
246 247
        LOG.debug('"%s" has %sbeen excluded', absolute_path,
                  '' if match else 'not ')
248
        return match
249 250 251 252 253 254

    def make_checkers(self, paths=None):
        # type: (List[str]) -> NoneType
        """Create checkers for each file."""
        if paths is None:
            paths = self.arguments
255 256 257 258

        if not paths:
            paths = ['.']

259
        filename_patterns = self.options.filename
260 261 262 263 264

        # NOTE(sigmavirus24): Yes this is a little unsightly, but it's our
        # best solution right now.
        def should_create_file_checker(filename):
            """Determine if we should create a file checker."""
265 266
            matches_filename_patterns = utils.fnmatch(
                filename, filename_patterns
267
            )
268 269 270
            is_stdin = filename == '-'
            file_exists = os.path.exists(filename)
            return (file_exists and matches_filename_patterns) or is_stdin
271

272
        checks = self.checks.to_dictionary()
273
        self.checkers = [
274
            FileChecker(filename, checks, self.options)
275 276 277
            for argument in paths
            for filename in utils.filenames_from(argument,
                                                 self.is_path_excluded)
278 279 280 281 282 283 284 285
            # NOTE(sigmavirus24): If a user explicitly specifies something,
            # e.g, ``flake8 bin/script`` then we should run Flake8 against
            # that. Since should_create_file_checker looks to see if the
            # filename patterns match the filename, we want to skip that in
            # the event that the argument and the filename are identical.
            # If it was specified explicitly, the user intended for it to be
            # checked.
            if argument == filename or should_create_file_checker(filename)
286
        ]
287
        LOG.info('Checking %d files', len(self.checkers))
288

289
    def report(self):
290
        # type: () -> (int, int)
291 292 293 294
        """Report all of the errors found in the managed file checkers.

        This iterates over each of the checkers and reports the errors sorted
        by line number.
295 296 297 298 299

        :returns:
            A tuple of the total results found and the results reported.
        :rtype:
            tuple(int, int)
300
        """
301
        results_reported = results_found = 0
302
        for checker in self.checkers:
303
            results = sorted(checker.results, key=lambda tup: (tup[1], tup[2]))
304
            results_reported += self._handle_results(checker.display_name,
305
                                                     results)
306
            results_found += len(results)
307
        return (results_found, results_reported)
308

309 310
    def run_parallel(self):
        """Run the checkers in parallel."""
311
        LOG.info('Starting %d process workers', self.jobs)
312
        for i in range(self.jobs):
313
            proc = multiprocessing.Process(
314 315 316
                target=_run_checks_from_queue,
                args=(self.process_queue, self.results_queue,
                      self.statistics_queue)
317 318 319 320
            )
            proc.daemon = True
            proc.start()
            self.processes.append(proc)
321

322 323 324 325 326
        final_results = {}
        for (filename, results) in self._results():
            final_results[filename] = results

        for checker in self.checkers:
327
            filename = checker.display_name
328
            checker.results = sorted(final_results.get(filename, []),
329
                                     key=lambda tup: (tup[2], tup[2]))
330 331 332 333

    def run_serial(self):
        """Run the checkers in serial."""
        for checker in self.checkers:
334
            checker.run_checks(self.results_queue, self.statistics_queue)
335

336
    def run(self):
337
        """Run all the checkers.
338

339 340 341 342 343 344
        This will intelligently decide whether to run the checks in parallel
        or whether to run them in serial.

        If running the checks in parallel causes a problem (e.g.,
        https://gitlab.com/pycqa/flake8/issues/74) this also implements
        fallback to serial processing.
345
        """
346 347 348 349 350 351 352 353 354 355 356
        try:
            if self.using_multiprocessing:
                self.run_parallel()
            else:
                self.run_serial()
        except OSError as oserr:
            if oserr.errno not in SERIAL_RETRY_ERRNOS:
                LOG.exception(oserr)
                raise
            LOG.warning('Running in serial after OS exception, %r', oserr)
            self.run_serial()
357 358 359
        except KeyboardInterrupt:
            LOG.warning('Flake8 was interrupted by the user')
            raise exceptions.EarlyQuit('Early quit while running checks')
360 361
        finally:
            self._force_cleanup()
362

363 364 365 366 367 368 369
    def start(self, paths=None):
        """Start checking files.

        :param list paths:
            Path names to check. This is passed directly to
            :meth:`~Manager.make_checkers`.
        """
370
        LOG.info('Making checkers')
371
        self.make_checkers(paths)
372 373
        if not self.using_multiprocessing:
            return
374

375 376 377
        LOG.info('Populating process queue')
        for checker in self.checkers:
            self.process_queue.put(checker)
378

379
        for i in range(self.jobs):
380 381
            self.process_queue.put('DONE')

382 383
    def stop(self):
        """Stop checking files."""
384
        self._process_statistics()
385
        for proc in self.processes:
386
            LOG.info('Joining %s to the main process', proc.name)
387
            proc.join()
388 389 390 391 392


class FileChecker(object):
    """Manage running checks for a file and aggregate the results."""

393
    def __init__(self, filename, checks, options):
394 395 396 397 398 399 400
        """Initialize our file checker.

        :param str filename:
            Name of the file to check.
        :param checks:
            The plugins registered to check the file.
        :type checks:
401
            dict
402 403 404 405
        :param options:
            Parsed option values from config and command-line.
        :type options:
            optparse.Values
406
        """
407 408
        self.options = options
        self.filename = filename
409 410
        self.checks = checks
        self.results = []
411 412
        self.processor = self._make_processor()
        self.display_name = self.processor.filename
413 414 415 416 417
        self.statistics = {
            'tokens': 0,
            'logical lines': 0,
            'physical lines': len(self.processor.lines),
        }
418

419
    def _make_processor(self):
420
        try:
421
            return processor.FileProcessor(self.filename, self.options)
422 423 424 425 426 427 428 429 430 431 432
        except IOError:
            # If we can not read the file due to an IOError (e.g., the file
            # does not exist or we do not have the permissions to open it)
            # then we need to format that exception for the user.
            # NOTE(sigmavirus24): Historically, pep8 has always reported this
            # as an E902. We probably *want* a better error code for this
            # going forward.
            (exc_type, exception) = sys.exc_info()[:2]
            message = '{0}: {1}'.format(exc_type.__name__, exception)
            self.report('E902', 0, 0, message)
            return None
433

434
    def report(self, error_code, line_number, column, text, line=None):
435
        # type: (str, int, int, str) -> str
436
        """Report an error by storing it in the results list."""
437 438
        if error_code is None:
            error_code, text = text.split(' ', 1)
439

440
        physical_line = line
441 442
        # If we're recovering from a problem in _make_processor, we will not
        # have this attribute.
443
        if not physical_line and getattr(self, 'processor', None):
444 445
            physical_line = self.processor.line_for(line_number)

446
        error = (error_code, line_number, column, text, physical_line)
447
        self.results.append(error)
448
        return error_code
449

450
    def run_check(self, plugin, **arguments):
451
        """Run the check in a single plugin."""
452
        LOG.debug('Running %r with %r', plugin, arguments)
453
        try:
454 455 456 457
            self.processor.keyword_arguments_for(
                plugin['parameters'],
                arguments,
            )
458 459 460 461 462 463
        except AttributeError as ae:
            LOG.error('Plugin requested unknown parameters.')
            raise exceptions.PluginRequestedUnknownParameters(
                plugin=plugin,
                exception=ae,
            )
464
        return plugin['plugin'](**arguments)
465

466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498
    @staticmethod
    def _extract_syntax_information(exception):
        token = ()
        if len(exception.args) > 1:
            token = exception.args[1]
            if len(token) > 2:
                row, column = token[1:3]
        else:
            row, column = (1, 0)

        if column > 0 and token and isinstance(exception, SyntaxError):
            # NOTE(sigmavirus24): SyntaxErrors report 1-indexed column
            # numbers. We need to decrement the column number by 1 at
            # least.
            offset = 1
            physical_line = token[-1]
            if len(physical_line) == column and physical_line[-1] == '\n':
                # NOTE(sigmavirus24): By default, we increment the column
                # value so that it's always 1-indexed. The SyntaxError that
                # we are trying to handle here will end up being 2 past
                # the end of the line. This happens because the
                # SyntaxError is technically the character after the
                # new-line. For example, if the code is ``foo(\n`` then
                # ``\n`` will be 4, the empty string will be 5 but most
                # tools want to report the at column 4, i.e., the opening
                # parenthesis. Semantically, having a column number of 6 is
                # correct but not useful for tooling (e.g., editors that
                # constantly run Flake8 for users).
                # See also: https://gitlab.com/pycqa/flake8/issues/237
                offset += 1
            column -= offset
        return row, column

499 500 501 502 503 504
    def run_ast_checks(self):
        """Run all checks expecting an abstract syntax tree."""
        try:
            ast = self.processor.build_ast()
        except (ValueError, SyntaxError, TypeError):
            (exc_type, exception) = sys.exc_info()[:2]
505 506
            row, column = self._extract_syntax_information(exception)
            self.report('E999', row, column, '%s: %s' %
507 508 509
                        (exc_type.__name__, exception.args[0]))
            return

510
        for plugin in self.checks['ast_plugins']:
511
            checker = self.run_check(plugin, tree=ast)
512 513 514 515 516 517 518
            # If the plugin uses a class, call the run method of it, otherwise
            # the call should return something iterable itself
            try:
                runner = checker.run()
            except AttributeError:
                runner = checker
            for (line_number, offset, text, check) in runner:
519 520 521 522 523 524 525
                self.report(
                    error_code=None,
                    line_number=line_number,
                    column=offset,
                    text=text,
                )

526 527
    def run_logical_checks(self):
        """Run all checks expecting a logical line."""
528 529 530 531 532 533 534
        comments, logical_line, mapping = self.processor.build_logical_line()
        if not mapping:
            return
        self.processor.update_state(mapping)

        LOG.debug('Logical line: "%s"', logical_line.rstrip())

535
        for plugin in self.checks['logical_line_plugins']:
536
            self.processor.update_checker_state_for(plugin)
537 538 539 540
            results = self.run_check(plugin, logical_line=logical_line) or ()
            for offset, text in results:
                offset = find_offset(offset, mapping)
                line_number, column_offset = offset
541 542
                self.report(
                    error_code=None,
543
                    line_number=line_number,
544 545 546 547
                    column=column_offset,
                    text=text,
                )

548 549
        self.processor.next_logical_line()

550
    def run_physical_checks(self, physical_line, override_error_line=None):
551
        """Run all checks for a given physical line."""
552
        for plugin in self.checks['physical_line_plugins']:
553
            self.processor.update_checker_state_for(plugin)
554 555 556
            result = self.run_check(plugin, physical_line=physical_line)
            if result is not None:
                column_offset, text = result
557 558
                error_code = self.report(
                    error_code=None,
559 560
                    line_number=self.processor.line_number,
                    column=column_offset,
561
                    text=text,
562
                    line=(override_error_line or physical_line),
563 564 565 566
                )

                self.processor.check_physical_error(error_code, physical_line)

567 568 569 570 571 572 573 574
    def process_tokens(self):
        """Process tokens and trigger checks.

        This can raise a :class:`flake8.exceptions.InvalidSyntax` exception.
        Instead of using this directly, you should use
        :meth:`flake8.checker.FileChecker.run_checks`.
        """
        parens = 0
575
        statistics = self.statistics
576 577
        file_processor = self.processor
        for token in file_processor.generate_tokens():
578
            statistics['tokens'] += 1
579 580
            self.check_physical_eol(token)
            token_type, text = token[0:2]
581
            processor.log_token(LOG, token)
582
            if token_type == tokenize.OP:
583
                parens = processor.count_parentheses(parens, text)
584
            elif parens == 0:
585
                if processor.token_is_newline(token):
586
                    self.handle_newline(token_type)
587 588
                elif (processor.token_is_comment(token) and
                        len(file_processor.tokens) == 1):
589 590
                    self.handle_comment(token, text)

591
        if file_processor.tokens:
592
            # If any tokens are left over, process them
593
            self.run_physical_checks(file_processor.lines[-1])
594 595
            self.run_logical_checks()

596
    def run_checks(self, results_queue, statistics_queue):
597
        """Run checks against the file."""
598 599 600
        if self.processor.should_ignore_file():
            return

601
        try:
602
            self.process_tokens()
603 604 605
        except exceptions.InvalidSyntax as exc:
            self.report(exc.error_code, exc.line_number, exc.column_number,
                        exc.error_message)
606

607 608
        self.run_ast_checks()

609
        if results_queue is not None:
610
            results_queue.put((self.filename, self.results))
611

612 613 614 615 616
        logical_lines = self.processor.statistics['logical lines']
        self.statistics['logical lines'] = logical_lines
        if statistics_queue is not None:
            statistics_queue.put(self.statistics)

617 618 619 620 621 622 623 624 625 626 627 628
    def handle_comment(self, token, token_text):
        """Handle the logic when encountering a comment token."""
        # The comment also ends a physical line
        token = list(token)
        token[1] = token_text.rstrip('\r\n')
        token[3] = (token[2][0], token[2][1] + len(token[1]))
        self.processor.tokens = [tuple(token)]
        self.run_logical_checks()

    def handle_newline(self, token_type):
        """Handle the logic when encountering a newline token."""
        if token_type == tokenize.NEWLINE:
629
            self.run_logical_checks()
630 631 632 633 634 635 636 637
            self.processor.reset_blank_before()
        elif len(self.processor.tokens) == 1:
            # The physical line contains only this token.
            self.processor.visited_new_blank_line()
            self.processor.delete_first_token()
        else:
            self.run_logical_checks()

638 639
    def check_physical_eol(self, token):
        """Run physical checks if and only if it is at the end of the line."""
640
        if processor.is_eol_token(token):
641 642
            # Obviously, a newline token ends a single physical line.
            self.run_physical_checks(token[4])
643
        elif processor.is_multiline_string(token):
644 645 646 647 648 649 650 651 652 653 654 655 656 657
            # Less obviously, a string that contains newlines is a
            # multiline string, either triple-quoted or with internal
            # newlines backslash-escaped. Check every physical line in the
            # string *except* for the last one: its newline is outside of
            # the multiline string, so we consider it a regular physical
            # line, and will check it like any other physical line.
            #
            # Subtleties:
            # - have to wind self.line_number back because initially it
            #   points to the last line of the string, and we want
            #   check_physical() to give accurate feedback
            line_no = token[2][0]
            with self.processor.inside_multiline(line_number=line_no):
                for line in self.processor.split_line(token):
658 659
                    self.run_physical_checks(line + '\n',
                                             override_error_line=token[4])
660 661


662 663 664 665 666 667 668 669 670 671 672 673 674 675 676
def _run_checks_from_queue(process_queue, results_queue, statistics_queue):
    LOG.info('Running checks in parallel')
    try:
        for checker in iter(process_queue.get, 'DONE'):
            LOG.info('Checking "%s"', checker.filename)
            checker.run_checks(results_queue, statistics_queue)
    except exceptions.PluginRequestedUnknownParameters as exc:
        print(str(exc))
    except Exception as exc:
        LOG.error('Unhandled exception occurred')
        raise
    finally:
        results_queue.put('DONE')


677 678 679 680 681 682 683 684 685
def find_offset(offset, mapping):
    """Find the offset tuple for a single offset."""
    if isinstance(offset, tuple):
        return offset

    for token_offset, position in mapping:
        if offset <= token_offset:
            break
    return (position[0], position[1] + offset - token_offset)