Some improvements on burn script

# Introduction

This notebook is a compilation of commands to perform operations with HTCondor through the command line.

This documentation is a compilation of commands to perform operations with HTCondor through the command line.

## Documentation

    condor_submit -interactive requirements='regexp("td\d", Name)'

# Slot querying

    condor_status gpu01 -af Name Gpus cpus memory/1024

# Test program

The `burn.py` program will fill up the memory and the cpu for an specific amount of time and will log the memory and cpu consumption.

    $ python burn.py --help

    usage: burn.py [-h] [--memory MEMORY] [--threads THREADS] [--walltime WALLTIME]

                [--loglevel {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}]

    options:

    -h, --help            show this help message and exit

    --memory MEMORY, -m MEMORY

                            Number of max GB of RSS to use

    --threads THREADS, -t THREADS

                            Number of CPUs to stress, will default to the number of threads available

    --walltime WALLTIME, -w WALLTIME

                            Walltime in seconds

    --loglevel {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}, -l {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}

import argparse

from concurrent.futures import ThreadPoolExecutor

from datetime import datetime

import logging

import os

import psutil

import threading

import time

import numpy as np

# Set up logging

log_levels = list(logging.getLevelNamesMapping().keys())

# create logger with default DEBUG level

logger = logging.getLogger('test_HTCondor')

logger.setLevel(logging.DEBUG)

# create console handler and set level to debug

ch = logging.StreamHandler()

ch.setLevel(logging.DEBUG)

# create formatter

formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# add formatter to ch

ch.setFormatter(formatter)

# add ch to logger

logger.addHandler(ch)

MATRIX_SIZE_GB = 0.1

N_MATRICES = 10

def create_parser():

    parser = argparse.ArgumentParser()

    parser.add_argument('--memory', '-m', type=int,

        default=2,

        help='Number of max GB of RSS to use')

    parser.add_argument('--threads', '-t', type=int,

        help='Number of CPUs to stress, will default to the number of threads available')

    parser.add_argument('--walltime', '-w', type=int,

        default=60,

        help='Walltime in seconds'

    )

    parser.add_argument('--loglevel', '-l', default='INFO',

        choices=log_levels)

    return parser

class Burner:

    def __init__(self, memory_gb, walltime_s, n_threads=None):

        self.memory_gb = memory_gb

        self.walltime_s = walltime_s

        self.set_n_threads(n_threads)

    def set_n_threads(self, n_threads):

        if not n_threads is None:

            self.n_threads = n_threads

        elif 'OMP_NUM_THREADS' in os.environ:

            self.n_threads = int(os.environ['OMP_NUM_THREADS'])

        else:

            self.n_threads = os.cpu_count()

    def start(self):

        logger.info(f'Starting test')

        myhost = os.uname()[1]

        logger.info(f'Running at {myhost}')

        # initial monitoring

        process = psutil.Process(os.getpid())

        curr_mem = process.memory_info().rss/(1024**3)

        logger.info(f'Starting process. RSS={curr_mem:.2f} GB')

        # compute

        # Build matrix

        # 3 times the size of the matrix is needed during execution

        matrix_size_gb = self.memory_gb / self.n_threads / 3

        n_bits = matrix_size_gb*1024*1024*1024*8

        n_floats = n_bits/64

        len_matrix = np.ceil(np.sqrt(n_floats)).astype('int')

        init_time = pivot_time = datetime.now()

        stop_thread = False

        thread_pool = []

        # Creating a pool of threads

        # the cpu usage does not correspond to the number of threads

        # since numpy will try to use all cpus available

        # we use multiple threads to avoid locking the cpu

        # when writing to memory

        executor = ThreadPoolExecutor(max_workers=self.n_threads)

        for i in range(self.n_threads):

            executor.submit(self.burn, i, len_matrix, lambda: stop_thread)

        while abs((pivot_time - init_time).total_seconds()) < self.walltime_s:

            pivot_time = datetime.now()

            curr_mem = process.memory_info().rss/(1024**3)

            curr_proc = process.cpu_percent()

            logger.info(f'CPU burning: RSS={curr_mem:.2f} GB, CPU={curr_proc:.1f} %')

            time.sleep(1)

        stop_thread = True

        logger.info('Shutting down, this may take a while ...')

        executor.shutdown()

    def burn(self, thread_ind, len_matrix, stop_function):

        '''Fill the memory and burn the CPU by doing

        random matrices multiplications.

        This function is intended to run in a thread so that

        it can be monitored and stopped.

        Arguments:

            thread_ind: int, index of the thread, just for logging

            matrix: np.array, matrix to perform the operations

            stop_function: function, when True, burning will stop

        '''

        x = np.random.rand(len_matrix, len_matrix)

        # Burn!

        i = 1

        while True:

            logger.debug(f'Thread {thread_ind} Running iteration {i}')

            y = x**2 @ x

            if stop_function():

                logger.debug(f'Thread {thread_ind} Stopping during matrix multiplication')

                return

            i += 1

def main(args):

    ch.setLevel(args.loglevel)

    # Burn!

    burner = Burner(args.memory, args.walltime, args.threads)

    burner.start()

if __name__ == '__main__':

    parser = create_parser()

    args = parser.parse_args()

    main(args)