<commit_before>"""Collection of low level, parallel, functions and VecOperations class."""

import sys  
import copy
import time as T
import numpy as N
import util
import parallel as parallel_mod
import vectors as V

   
class VecOperations(object):
    """Responsible for low level operations on vecs.

    Kwargs:
        inner_product: inner product function
        
        max_vecs_per_node: max number of vecs that can be in memory
        simultaneously per node.
        
        verbose: print non-essential statuses and warnings, boolean.
        
        print_interval: max of how frequently progress is printed, in seconds.

    The class is a collection of non-trivial parallel methods used in the 
    high-level modred classes like POD, BPOD, and DMD. 
    
    Note: It is generally best to use all available processors, however this
    depends on the computer and the nature of the functions
    supplied. In some cases, loading from file is slower with more workers.
    """
    
    def __init__(self, inner_product=None, 
        max_vecs_per_node=None, verbose=True, print_interval=10):
        """Constructor. """
        self.inner_product = inner_product
        self.verbose = verbose 
        self.print_interval = print_interval
        self.prev_print_time = 0.
        self.parallel = parallel_mod.default_instance
        
        if max_vecs_per_node is None:
            self.max_vecs_per_node = int(1e6) #N.inf? it's a float not an int...
            self.print_msg('Warning: max_vecs_per_node was not specified. '
                'Assuming infinte vecs can be in memory per node. Decrease '
                'max_vecs_per_node if get memory errors.')
        else:
            self.max_vecs_per_node = max_vecs_per_node
        
        if self.max_vecs_per_node < \
            2 * self.parallel.get_num_procs() / self.parallel.get_num_nodes(): 
            self.max_vecs_per_proc = 2
            self.print_msg('Warning: max_vecs_per_node too small for given '
                'number of nodes and procs. Assuming 2 vecs can be '
                'in memory per processor. If possible, increase ' 
                'max_vecs_per_node for a speedup.')
        else:
            self.max_vecs_per_proc = self.max_vecs_per_node * \
                self.parallel.get_num_nodes()/self.parallel.get_num_procs()
                
    def _check_inner_product(self):
        """Check that inner_product is defined"""
        if self.inner_product is None:
            raise RuntimeError('No inner product function/callable defined')
        
    
    def print_msg(self, msg, output_channel=sys.stdout):
        """Print a message from rank 0 if verbose"""
        if self.verbose and self.parallel.is_rank_zero():
            print >> output_channel, msg


    def sanity_check(self, test_vec_handle):
        """Check user-supplied vec handle and vec objects.
        
        Args:
            test_vec_handle: a vector handle.
        
        The add and mult functions are tested for the generic object.  
        This is not a complete testing, but catches some common mistakes.
        Raises error if a check fails.
        
        TODO: Other things which could be tested:
            get/put doesn't effect other vecs (memory problems)
        """
        self._check_inner_product()
        tol = 1e-10

        test_vec = test_vec_handle.get()
        vec_copy = copy.deepcopy(test_vec)
        vec_copy_mag2 = self.inner_product(vec_copy, vec_copy)
        
        factor = 2.
        vec_mult = test_vec * factor
        
        if abs(self.inner_product(vec_mult, vec_mult) -
                vec_copy_mag2 * factor**2) > tol:
            raise ValueError('Multiplication of vec/mode failed')
        
        if abs(self.inner_product(test_vec, test_vec) - 
                vec_copy_mag2) > tol:  
            raise ValueError('Original vec modified by multiplication!') 
        vec_add = test_vec + test_vec
        if abs(self.inner_product(vec_add, vec_add) - vec_copy_mag2 * 4) > tol:
            raise ValueError('Addition does not give correct result')
        
        if abs(self.inner_product(test_vec, test_vec) - vec_copy_mag2) > tol:  
            raise ValueError('Original vec modified by addition!')       
        
        vec_add_mult = test_vec * factor + test_vec
        if abs(self.inner_product(vec_add_mult, vec_add_mult) - vec_copy_mag2 *
                (factor + 1) ** 2) > tol:
            raise ValueError('Multiplication and addition of vec/mode are '+\
                'inconsistent')
        
        if abs(self.inner_product(test_vec, test_vec) - vec_copy_mag2) > tol:  
            raise ValueError('Original vec modified by combo of mult/add!') 
        
        #vecSub = 3.5*test_vec - test_vec
        #N.testing.assert_array_almost_equal(vecSub,2.5*test_vec)
        #N.testing.assert_array_almost_equal(test_vec,vec_copy)
        self.print_msg('Passed the sanity check')


    def compute_inner_product_mat(self, row_vec_handles, col_vec_handles):
        """Computes a matrix of inner products and returns it.
        
        Args:
            row_vec_handles: list of row vec handles (e.g. BPOD adjoints, "Y")
          
            col_vec_handles: list of column vec handles (e.g. BPOD directs, "X")

        Within this method, the vecs are retrieved in a memory-efficient
        chunks so that they are not all in memory at once.
        The row vecs and col vecs are assumed to be different.        
        When they are the same (POD), a different method is used.
        
        Each processor is responsible for retrieving a subset of the rows and
        columns. The processor which retrieves a particular column vec then sends
        it to each successive processor so it can be used to compute all IPs
        for the current row chunk on each processor. This is repeated until all
        processors are done with all of their row chunks. If there are 2
        processors::
           
                | x o |
          rank0 | x o |
                | x o |
            -
                | o x |
          rank1 | o x |
                | o x |
        
        In the next step, rank 0 sends column 0 to rank 1 and rank 1
        sends column 1 to rank 0. The remaining IPs are filled in::
        
                | x x |
          rank0 | x x |
                | x x |
            -
                | x x |
          rank1 | x x |
                | x x |
          
        When the number of cols and rows is
        not divisible by the number of processors, the processors are assigned
        unequal numbers of tasks. However, all processors are always
        part of the passing circle.
        
        This is also generalized to allow the columns to be read in chunks, 
        rather than only 1 at a time.
        If we change the implementation to use hybrid distributed-shared 
        memory where it is best to work in
        operation-units (load/gets, IPs, etc) of multiples of the number
        of processors sharing memory (procs/node).
        
        The scaling is:
        
        - num gets / processor ~ (n_r*n_c/((max-1)*n_p*n_p)) + n_r/n_p
        - num MPI sends / processor ~ (n_p-1)*(n_r/((max-1)*n_p))*n_c/n_p
        - num inner products / processor ~ n_r*n_c/n_p
            
        where n_r is number of rows, n_c number of columns, max is
        max_vecs_per_proc = max_vecs_per_node/num_procs_per_node, and n_p is
        number of processors.
        
        It is enforced that there are more columns than rows by doing an
        internal transpose and un-transpose. This improves efficiency by placing
        the larger of n_c and n_r on the quadratically scaled portion.
        
        It is good to use all available processors, even if it lowers max.
        However, sometimes simultaneous loads actually makes each load slow.     
        """
        self._check_inner_product()
        row_vec_handles = util.make_list(row_vec_handles)
        col_vec_handles = util.make_list(col_vec_handles)
            
        num_cols = len(col_vec_handles)<commit_msg>minor changes to docs only.<commit_after>"""Collection of parallelized functions of vectors and vector handles."""

import sys  
import copy
import time as T
import numpy as N
import util
import parallel as parallel_mod
import vectors as V

   
class VecOperations(object):
    """Responsible for low level operations on vecs.

    Kwargs:
        inner_product: inner product function
        
        max_vecs_per_node: max number of vecs that can be in memory
        simultaneously per node.
        
        verbose: print non-essential statuses and warnings, boolean.
        
        print_interval: max of how frequently progress is printed, in seconds.

    The class is a collection of parallelized methods used in the 
    high-level modred classes like POD, BPOD, and DMD. 

    It is good to use all available processors, even if it lowers max.

    Note: It is generally best to use all available processors, even if this
    lowers ``max_vecs_per_node`` proportionally. 
    However this depends on the computer and the nature of the functions
    supplied, and sometimes loading from file is slower with more processors.
    """
    def __init__(self, inner_product=None, 
        max_vecs_per_node=None, verbose=True, print_interval=10):
        """Constructor. """
        self.inner_product = inner_product
        self.verbose = verbose 
        self.print_interval = print_interval
        self.prev_print_time = 0.
        self.parallel = parallel_mod.default_instance
        
        if max_vecs_per_node is None:
            self.max_vecs_per_node = 10000 # different default?
            self.print_msg('Warning: max_vecs_per_node was not specified. '
                'Assuming 10000 vecs can be in memory per node. Decrease '
                'max_vecs_per_node if memory errors.')
        else:
            self.max_vecs_per_node = max_vecs_per_node
        
        if self.max_vecs_per_node < \
            2 * self.parallel.get_num_procs() / self.parallel.get_num_nodes(): 
            self.max_vecs_per_proc = 2
            self.print_msg('Warning: max_vecs_per_node too small for given '
                'number of nodes and procs. Assuming 2 vecs can be '
                'in memory per processor. If possible, increase ' 
                'max_vecs_per_node for a speedup.')
        else:
            self.max_vecs_per_proc = self.max_vecs_per_node * \
                self.parallel.get_num_nodes()/self.parallel.get_num_procs()
                
    def _check_inner_product(self):
        """Check that inner_product is defined"""
        if self.inner_product is None:
            raise RuntimeError('No inner product function is defined')
        
    
    def print_msg(self, msg, output_channel=sys.stdout):
        """Print a message from rank 0 if verbose"""
        if self.verbose and self.parallel.is_rank_zero():
            print >> output_channel, msg

    def sanity_check_in_memory(self, test_vec):
        """Same as ``sanity_check`` but takes a vec instead of handle"""
        self.sanity_check(V.InMemoryVecHandle(test_vec)

    def sanity_check(self, test_vec_handle):
        """Check user-supplied vec handle and vec objects.
        
        Args:
            test_vec_handle: a vector handle.
        
        The add and mult functions are tested for the vector object.  
        This is not a complete testing, but catches some common mistakes.
        Raises an error if a check fails.
        
        TODO: Other things which could be tested:
            get/put doesn't effect other vecs (memory problems)
        """
        self._check_inner_product()
        tol = 1e-10

        test_vec = test_vec_handle.get()
        vec_copy = copy.deepcopy(test_vec)
        vec_copy_mag2 = self.inner_product(vec_copy, vec_copy)
        
        factor = 2.
        vec_mult = test_vec * factor
        
        if abs(self.inner_product(vec_mult, vec_mult) -
                vec_copy_mag2 * factor**2) > tol:
            raise ValueError('Multiplication of vec/mode failed')
        
        if abs(self.inner_product(test_vec, test_vec) - 
                vec_copy_mag2) > tol:  
            raise ValueError('Original vec modified by multiplication!') 
        vec_add = test_vec + test_vec
        if abs(self.inner_product(vec_add, vec_add) - vec_copy_mag2 * 4) > tol:
            raise ValueError('Addition does not give correct result')
        
        if abs(self.inner_product(test_vec, test_vec) - vec_copy_mag2) > tol:  
            raise ValueError('Original vec modified by addition!')       
        
        vec_add_mult = test_vec * factor + test_vec
        if abs(self.inner_product(vec_add_mult, vec_add_mult) - vec_copy_mag2 *
                (factor + 1) ** 2) > tol:
            raise ValueError('Multiplication and addition of vec/mode are '+\
                'inconsistent')
        
        if abs(self.inner_product(test_vec, test_vec) - vec_copy_mag2) > tol:  
            raise ValueError('Original vec modified by combo of mult/add!') 
        
        #vecSub = 3.5*test_vec - test_vec
        #N.testing.assert_array_almost_equal(vecSub,2.5*test_vec)
        #N.testing.assert_array_almost_equal(test_vec,vec_copy)
        self.print_msg('Passed the sanity check')


    def compute_inner_product_mat(self, row_vec_handles, col_vec_handles):
        """Computes a matrix of inner products and returns it.
        
        Args:
            row_vec_handles: list of row vec handles (e.g. BPOD adjoints, "Y")
          
            col_vec_handles: list of column vec handles (e.g. BPOD directs, "X")

        The vecs are retrieved in memory-efficient
        chunks and are not all in memory at once.
        The row vecs and col vecs are assumed to be different.        
        When they are the same, use ``compute_symmetric_inner_product`` for a 
        2x speedup.
        
        Each processor is responsible for retrieving a subset of the rows and
        columns. The processors then exchange columns via MPI so it can be 
        used to compute all IPs for the rows on each processor. This is 
        repeated until all processors are done with all of their row chunks.
        If there are 2 processors::
           
                | x o |
          rank0 | x o |
                | x o |
            -
                | o x |
          rank1 | o x |
                | o x |
        
        In the next step, rank 0 sends column 0 to rank 1 and rank 1
        sends column 1 to rank 0. The remaining IPs are filled in::
        
                | x x |
          rank0 | x x |
                | x x |
            -
                | x x |
          rank1 | x x |
                | x x |
          
        When the number of cols and rows is
        not divisible by the number of processors, the processors are assigned
        unequal numbers of tasks. However, all processors are always
        part of the passing cycle.
        
        The scaling is:
        
        - num gets / processor ~ (n_r*n_c/((max-1)*n_p*n_p)) + n_r/n_p
        - num MPI sends / processor ~ (n_p-1)*(n_r/((max-1)*n_p))*n_c/n_p
        - num inner products / processor ~ n_r*n_c/n_p
            
        where n_r is number of rows, n_c number of columns, max is
        max_vecs_per_proc = max_vecs_per_node/num_procs_per_node, and n_p is
        number of processors.
        
        If there are more rows than columns, then an internal transpose
        and un-transpose is performed to improve efficiency (since n_c only
        appears in the scaling in the quadratic term).
        """
        self._check_inner_product()
        row_vec_handles = util.make_list(row_vec_handles)
        col_vec_handles = util.make_list(col_vec_handles)
            
        num_cols = len(col_vec_handles)
{'fraction_non_alphanumeric': 0.06491547791114183, 'fraction_numerical': 0.005907119964808647, 'mean_word_length': 1.9846211552888222, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 19, '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': '2931373', 'n_tokens_mistral': 4627, 'n_tokens_neox': 4430, 'n_words': 1535}