<commit_before>from collections import OrderedDict, namedtuple
from contextlib import contextmanager
from functools import reduce
from operator import mul
from pathlib import Path
from subprocess import DEVNULL, PIPE, run
from time import time as seq_time
import os

import cgen as c
from sympy import S

from devito.ir.iet import (ExpressionBundle, List, TimedList, Section,
                           Iteration, FindNodes, Transformer)
from devito.ir.support import IntervalGroup
from devito.logger import warning, error
from devito.mpi import MPI
from devito.mpi.routines import MPICall, MPIList, RemainderCall
from devito.parameters import configuration
from devito.passes.iet import BusyWait
from devito.symbolics import subs_op_args
from devito.tools import DefaultOrderedDict, flatten

__all__ = ['create_profile']


SectionData = namedtuple('SectionData', 'ops sops points traffic itermaps')
PerfKey = namedtuple('PerfKey', 'name rank')
PerfInput = namedtuple('PerfInput', 'time ops points traffic sops itershapes')
PerfEntry = namedtuple('PerfEntry', 'time gflopss gpointss oi ops itershapes')


class Profiler(object):

    _default_includes = []
    _default_libs = []
    _ext_calls = []

    """Metadata for a profiled code section."""

    def __init__(self, name):
        self.name = name

        # Operation reductions observed in sections
        self._ops = []

        # C-level code sections
        self._sections = OrderedDict()
        self._subsections = OrderedDict()

        # Python-level timers
        self.py_timers = OrderedDict()

        self.initialized = True

    def analyze(self, iet):
        """
        Analyze the Sections in the given IET. This populates `self._sections`.
        """
        sections = FindNodes(Section).visit(iet)
        for s in sections:
            if s.name in self._sections:
                continue

            bundles = FindNodes(ExpressionBundle).visit(s)

            # Total operation count
            ops = sum(i.ops*i.ispace.size for i in bundles)

            # Operation count at each section iteration
            sops = sum(i.ops for i in bundles)

            # Total memory traffic
            mapper = {}
            for i in bundles:
                for k, v in i.traffic.items():
                    mapper.setdefault(k, []).append(v)
            traffic = 0
            for i in mapper.values():
                try:
                    traffic += IntervalGroup.generate('union', *i).size
                except ValueError:
                    # Over different iteration spaces
                    traffic += sum(j.size for j in i)

            # Each ExpressionBundle lives in its own iteration space
            itermaps = [i.ispace.dimension_map for i in bundles]

            # Track how many grid points are written within `s`
            points = []
            for i in bundles:
                writes = {e.write for e in i.exprs
                          if e.is_tensor and e.write.is_TimeFunction}
                points.append(i.size*len(writes))
            points = sum(points)

            self._sections[s.name] = SectionData(ops, sops, points, traffic, itermaps)

    def track_subsection(self, sname, name):
        v = self._subsections.setdefault(sname, OrderedDict())
        v[name] = SectionData(S.Zero, S.Zero, S.Zero, S.Zero, [])

    def instrument(self, iet, timer):
        """
        Instrument the given IET for C-level performance profiling.
        """
        sections = FindNodes(Section).visit(iet)
        if sections:
            mapper = {}
            for i in sections:
                n = i.name
                assert n in timer.fields
                mapper[i] = i._rebuild(body=TimedList(timer=timer, lname=n, body=i.body))
            return Transformer(mapper, nested=True).visit(iet)
        else:
            return iet

    @contextmanager
    def timer_on(self, name, comm=None):
        """
        Measure the execution time of a Python-level code region.

        Parameters
        ----------
        name : str
            A representative string for the timed region.
        comm : MPI communicator, optional
            If provided, the global execution time is derived by a single MPI
            rank, with timers started and stopped right after an MPI barrier.
        """
        if comm and comm is not MPI.COMM_NULL:
            comm.Barrier()
            tic = MPI.Wtime()
            yield
            comm.Barrier()
            toc = MPI.Wtime()
        else:
            tic = seq_time()
            yield
            toc = seq_time()
        self.py_timers[name] = toc - tic

    def record_ops_variation(self, initial, final):
        """
        Record the variation in operation count experienced by a section due to
        a flop-reducing transformation.
        """
        self._ops.append((initial, final))

    @property
    def all_sections(self):
        return list(self._sections) + flatten(self._subsections.values())

    @property
    def trackable_subsections(self):
        return ()

    def summary(self, args, dtype, reduce_over=None):
        """
        Return a PerformanceSummary of the profiled sections.

        Parameters
        ----------
        args : dict
            A mapper from argument names to run-time values from which the Profiler
            infers iteration space and execution times of a run.
        dtype : data-type
            The data type of the objects in the profiled sections. Used to compute
            the operational intensity.
        """
        comm = args.comm

        summary = PerformanceSummary()
        for name, data in self._sections.items():
            # Time to run the section
            time = max(getattr(args[self.name]._obj, name), 10e-7)

            # Add performance data
            if comm is not MPI.COMM_NULL:
                # With MPI enabled, we add one entry per section per rank
                times = comm.allgather(time)
                assert comm.size == len(times)
                for rank in range(comm.size):
                    summary.add(name, rank, times[rank])
            else:
                summary.add(name, None, time)

        return summary


class ProfilerVerbose1(Profiler):

    @property
    def trackable_subsections(self):
        return (MPIList, RemainderCall, BusyWait)


class ProfilerVerbose2(Profiler):

    @property
    def trackable_subsections(self):
        return (MPICall, BusyWait)


class AdvancedProfiler(Profiler):

    # Override basic summary so that arguments other than runtime are computed.
    def summary(self, args, dtype, reduce_over=None):
        grid = args.grid
        comm = args.comm

        # Produce sections summary
        summary = PerformanceSummary()
        for name, data in self._sections.items():
            # Time to run the section
            time = max(getattr(args[self.name]._obj, name), 10e-7)

            # Number of FLOPs performed
            ops = int(subs_op_args(data.ops, args))

            # Number of grid points computed
            points = int(subs_op_args(data.points, args))

            # Compulsory traffic
            traffic = float(subs_op_args(data.traffic, args)*dtype().itemsize)

            # Runtime itermaps/itershapes
            itermaps = [OrderedDict([(k, int(subs_op_args(v, args)))
                                     for k, v in i.items()])
                        for i in data.itermaps]
            itershapes = tuple(tuple(i.values()) for i in itermaps)

            # Add local performance data
            if comm is not MPI.COMM_NULL:
                # With MPI enabled, we add one entry per section per rank
                times = comm.allgather(time)
                assert comm.size == len(times)
                opss = comm.allgather(ops)
                pointss = comm.allgather(points)
                traffics = comm.allgather(traffic)
                sops = [data.sops]*comm.size
                itershapess = comm.allgather(itershapes)
                items = list(zip(times, opss, pointss, traffics, sops, itershapess))
                for rank in range(comm.size):
                    summary.add(name, rank, *items[rank])
            else:
                summary.add(name, None, time, ops, points, traffic, data.sops, itershapes)

        # Enrich summary with subsections data
        for sname, v in self._subsections.items():
            for name, data in v.items():<commit_msg>operator: Use FD-Gpts/s instead of Gpts/s<commit_after>from collections import OrderedDict, namedtuple
from contextlib import contextmanager
from functools import reduce
from operator import mul
from pathlib import Path
from subprocess import DEVNULL, PIPE, run
from time import time as seq_time
import os

import cgen as c
from sympy import S

from devito.ir.iet import (ExpressionBundle, List, TimedList, Section,
                           Iteration, FindNodes, Transformer)
from devito.ir.support import IntervalGroup
from devito.logger import warning, error
from devito.mpi import MPI
from devito.mpi.routines import MPICall, MPIList, RemainderCall
from devito.parameters import configuration
from devito.passes.iet import BusyWait
from devito.symbolics import subs_op_args
from devito.tools import DefaultOrderedDict, flatten

__all__ = ['create_profile']


SectionData = namedtuple('SectionData', 'ops sops points traffic itermaps')
PerfKey = namedtuple('PerfKey', 'name rank')
PerfInput = namedtuple('PerfInput', 'time ops points traffic sops itershapes')
PerfEntry = namedtuple('PerfEntry', 'time gflopss gpointss oi ops itershapes')


class Profiler(object):

    _default_includes = []
    _default_libs = []
    _ext_calls = []

    """Metadata for a profiled code section."""

    def __init__(self, name):
        self.name = name

        # Operation reductions observed in sections
        self._ops = []

        # C-level code sections
        self._sections = OrderedDict()
        self._subsections = OrderedDict()

        # Python-level timers
        self.py_timers = OrderedDict()

        self.initialized = True

    def analyze(self, iet):
        """
        Analyze the Sections in the given IET. This populates `self._sections`.
        """
        sections = FindNodes(Section).visit(iet)
        for s in sections:
            if s.name in self._sections:
                continue

            bundles = FindNodes(ExpressionBundle).visit(s)

            # Total operation count
            ops = sum(i.ops*i.ispace.size for i in bundles)

            # Operation count at each section iteration
            sops = sum(i.ops for i in bundles)

            # Total memory traffic
            mapper = {}
            for i in bundles:
                for k, v in i.traffic.items():
                    mapper.setdefault(k, []).append(v)
            traffic = 0
            for i in mapper.values():
                try:
                    traffic += IntervalGroup.generate('union', *i).size
                except ValueError:
                    # Over different iteration spaces
                    traffic += sum(j.size for j in i)

            # Each ExpressionBundle lives in its own iteration space
            itermaps = [i.ispace.dimension_map for i in bundles]

            # Track how many fd-points are written within `s`
            points = []
            total_writes = 0
            for i in bundles:
                writes = {e.write for e in i.exprs
                          if e.is_tensor and e.write.is_TimeFunction}
                total_writes += len(writes)
                # If we have writes only consider one iteration space size
                # per bundle otherwise append 0
                if len(writes):
                    points.append(i.size)
                else:
                    points.append(i.size*len(writes))

            # list(set()) will return a list with unique elements
            # discarding duplicate grid sizes
            points = list(set(points))

            fdpoints = sum(points)

            self._sections[s.name] = SectionData(ops, sops, fdpoints, traffic, itermaps)

    def track_subsection(self, sname, name):
        v = self._subsections.setdefault(sname, OrderedDict())
        v[name] = SectionData(S.Zero, S.Zero, S.Zero, S.Zero, [])

    def instrument(self, iet, timer):
        """
        Instrument the given IET for C-level performance profiling.
        """
        sections = FindNodes(Section).visit(iet)
        if sections:
            mapper = {}
            for i in sections:
                n = i.name
                assert n in timer.fields
                mapper[i] = i._rebuild(body=TimedList(timer=timer, lname=n, body=i.body))
            return Transformer(mapper, nested=True).visit(iet)
        else:
            return iet

    @contextmanager
    def timer_on(self, name, comm=None):
        """
        Measure the execution time of a Python-level code region.

        Parameters
        ----------
        name : str
            A representative string for the timed region.
        comm : MPI communicator, optional
            If provided, the global execution time is derived by a single MPI
            rank, with timers started and stopped right after an MPI barrier.
        """
        if comm and comm is not MPI.COMM_NULL:
            comm.Barrier()
            tic = MPI.Wtime()
            yield
            comm.Barrier()
            toc = MPI.Wtime()
        else:
            tic = seq_time()
            yield
            toc = seq_time()
        self.py_timers[name] = toc - tic

    def record_ops_variation(self, initial, final):
        """
        Record the variation in operation count experienced by a section due to
        a flop-reducing transformation.
        """
        self._ops.append((initial, final))

    @property
    def all_sections(self):
        return list(self._sections) + flatten(self._subsections.values())

    @property
    def trackable_subsections(self):
        return ()

    def summary(self, args, dtype, reduce_over=None):
        """
        Return a PerformanceSummary of the profiled sections.

        Parameters
        ----------
        args : dict
            A mapper from argument names to run-time values from which the Profiler
            infers iteration space and execution times of a run.
        dtype : data-type
            The data type of the objects in the profiled sections. Used to compute
            the operational intensity.
        """
        comm = args.comm

        summary = PerformanceSummary()
        for name, data in self._sections.items():
            # Time to run the section
            time = max(getattr(args[self.name]._obj, name), 10e-7)

            # Add performance data
            if comm is not MPI.COMM_NULL:
                # With MPI enabled, we add one entry per section per rank
                times = comm.allgather(time)
                assert comm.size == len(times)
                for rank in range(comm.size):
                    summary.add(name, rank, times[rank])
            else:
                summary.add(name, None, time)

        return summary


class ProfilerVerbose1(Profiler):

    @property
    def trackable_subsections(self):
        return (MPIList, RemainderCall, BusyWait)


class ProfilerVerbose2(Profiler):

    @property
    def trackable_subsections(self):
        return (MPICall, BusyWait)


class AdvancedProfiler(Profiler):

    # Override basic summary so that arguments other than runtime are computed.
    def summary(self, args, dtype, reduce_over=None):
        grid = args.grid
        comm = args.comm

        # Produce sections summary
        summary = PerformanceSummary()
        for name, data in self._sections.items():
            # Time to run the section
            time = max(getattr(args[self.name]._obj, name), 10e-7)

            # Number of FLOPs performed
            ops = int(subs_op_args(data.ops, args))

            points = int(subs_op_args(data.points, args))

            # Compulsory traffic
            traffic = float(subs_op_args(data.traffic, args)*dtype().itemsize)

            # Runtime itermaps/itershapes
            itermaps = [OrderedDict([(k, int(subs_op_args(v, args)))
                                     for k, v in i.items()])
                        for i in data.itermaps]
            itershapes = tuple(tuple(i.values()) for i in itermaps)

            # Add local performance data
            if comm is not MPI.COMM_NULL:
                # With MPI enabled, we add one entry per section per rank
                times = comm.allgather(time)
                assert comm.size == len(times)
                opss = comm.allgather(ops)
                pointss = comm.allgather(points)
                traffics = comm.allgather(traffic)
                sops = [data.sops]*comm.size
                itershapess = comm.allgather(itershapes)
                items = list(zip(times, opss, pointss, traffics, sops, itershapess))
                for rank in range(comm.size):
                    summary.add(name, rank, *items[rank])
            else:
                summary.add(name, None, time, ops, points, traffic, data.sops, itershapes)

        # Enrich summary with subsections data
        for sname, v in self._subsections.items():
            for name, data in v.items():
{'fraction_non_alphanumeric': 0.08335252129864149, 'fraction_numerical': 0.0011512779184895234, 'mean_word_length': 1.891163255117324, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4525772', 'n_tokens_mistral': 4784, 'n_tokens_neox': 4598, 'n_words': 1473}