Source code for pwem.protocols.protocol_tests

# **************************************************************************
# *
# * Authors:     R. Marabini (roberto@cnb.csic.es)
# *              Pablo Conesa (pconesa@cnb.csic.es)
# *
# * Unidad de  Bioinformatica of Centro Nacional de Biotecnologia , CSIC
# *
# * This program is free software; you can redistribute it and/or modify
# * it under the terms of the GNU General Public License as published by
# * the Free Software Foundation; either version 3 of the License, or
# * (at your option) any later version.
# *
# * This program is distributed in the hope that it will be useful,
# * but WITHOUT ANY WARRANTY; without even the implied warranty of
# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# * GNU General Public License for more details.
# *
# * You should have received a copy of the GNU General Public License
# * along with this program; if not, write to the Free Software
# * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
# * 02111-1307  USA
# *
# *  All comments concerning this program package may be sent to the
# *  e-mail address 'scipion@cnb.csic.es'
# *
# **************************************************************************
""" This module is to add protocols that need to be there for testing purposes
but nos available for regular users. This will hide them and prevent the their
 appearance in any GUI search or tree."""
import time

import pyworkflow.protocol as pwprot
import pyworkflow.protocol.params as params
from pyworkflow import VERSION_1_1


STRESS_NG = 'stress-ng'


[docs]class ProtTests(pwprot.Protocol):
[docs] @classmethod def isDisabled(cls): """ Return True for all test protocols. Disabled protocols will not be offered in the available protocols.""" return True
[docs]class ProtStress(ProtTests): """ stress will stress test a computer system in various selectable ways. Several options require the program stress-ng. """ _label = 'stress' _program = STRESS_NG _lastUpdateVersion = VERSION_1_1 def __init__(self, **kwargs): ProtTests.__init__(self, **kwargs) self.xmippMic = {} # --------------------------- DEFINE param functions ---------------------- def _defineParams(self, form): form.addSection(label='Input') form.addParam('noCpu', params.IntParam, default=0, label="No. CPU stressors", help="start N workers spinning on sqrt(rand())") form.addParam('noMem', params.IntParam, default=0, label="No. memory stressors", help="start N workers continuously calling mmap(2)/munmap" "(2) and writing to the allocated memory. It will " "use 100% per worker ") form.addParam('amountMem', params.IntParam, default=256, label="Memory per stressors (Mb)", help="allocate N bytes per vm worker. each stressors will" " use 100% of a CPU") form.addParam('noIO', params.IntParam, default=0, label="No IO stressors", help="start N workers spinning on sync() (Disk io)") form.addParam('timeout', params.IntParam, default=60, label="TimeOut (sec)", help="timeout after N seconds. Total execution time is " "timeout plus delay") form.addParam('delay', params.IntParam, default=0, label="delay (sec)", help="wait this seconds before stressing the system " "seconds") form.addParam('extraParams', params.StringParam, label='Additional parameters', default=" ", help='Additional parameters for stress-ng ' '(http://kernel.ubuntu.com/~cking/stress-ng/)') # ---------------------- INSERT steps functions ---------------------------- def _insertAllSteps(self): if self.delay != 0: self._insertFunctionStep('delayStep') self._insertFunctionStep('stressStep') # ------------------------ STEPS functions ---------------------------------
[docs] def delayStep(self): time.sleep(self.delay)
[docs] def stressStep(self): if self.noCpu == 0 and self.noIO == 0 and self.noMem == 0: time.sleep(self.timeout) else: args = " " if self.noCpu != 0: args += " --cpu %d" % self.noCpu if self.noIO != 0: args += " --io %d" % self.noIO if self.noMem != 0: args += " --mmap %d" % self.noMem if self.amountMem != 0: args += " --mmap-bytes %dM" % self.amountMem args += " --timeout %ds" % self.timeout args += " --metrics-brief" if self.extraParams is not None: args += " %s " % self.extraParams self.runJob(self._program, args)
# ------------------------ INFO functions ---------------------------------- def _validate(self): message = [] from distutils.spawn import find_executable if self.noCpu == 0 and self.noIO == 0 and self.noMem == 0: pass else: if find_executable(self._program) is None: message = ["Cannot find executable %s" % self._program] return message def _summary(self): message = "%d CPU, %d Mem, %d IO stressor" % (self.cpu, self.mem, self.io) return [message] def _methods(self): return []
# class ProtOutputTest(ProtTests): # """ Protocol to test scalar output and input linking""" # _label = 'test output' # # def __init__(self, **args): # Protocol.__init__(self, **args) # self.name = params.String(args.get('name', None)) # # def _defineParams(self, form): # # section = form.addSection("Input") # section.addParam('iBoxSize', params.IntParam, allowsPointers=True, # default=10, # label='Input box size as Integer', # validators=[params.Positive]) # # section.addParam('nullableInteger', params.IntParam, allowsPointers=True, # label='Nullable Integer', allowsNull=True) # # def _createOutputStep(self): # # New Output would be an Integer # boxSize = Integer(10) # # if self.iBoxSize.hasValue(): # boxSize.set(2*int(self.iBoxSize.get())) # # self._defineOutputs(oBoxSize=boxSize) # # def _insertAllSteps(self): # self._insertFunctionStep('_createOutputStep') # # # class ProtMultiPointerTest(ProtTests): # """ Class to test how multipointer params are exported to json""" # def _defineParams(self, form): # # # This should cover Multipointer params that points to attributes... # # therefore extended attribute of pointers should be used # form.addParam('mpToAttr', params.MultiPointerParam, # label="Multipointer to attribute", # pointerClass='String', # help="Should point to String inside another protocol") # # # This should cover Multipointer params that points to protocols... # # therefore extended attribute of pointers should NOT be used # form.addParam('mpToProts', params.MultiPointerParam, # label="Multipointer to sets", # pointerClass='Protocol', # help="Should point to another protocol")