Source Code for Module src.pyflowConfig

  1  # 
  2  # pyFlow - a lightweight parallel task engine 
  3  # 
  4  # Copyright (c) 2012-2017 Illumina, Inc. 
  5  # All rights reserved. 
  6  # 
  7  # Redistribution and use in source and binary forms, with or without 
  8  # modification, are permitted provided that the following conditions 
  9  # are met: 
 10  # 
 11  # 1. Redistributions of source code must retain the above copyright 
 12  #    notice, this list of conditions and the following disclaimer. 
 13  # 
 14  # 2. Redistributions in binary form must reproduce the above copyright 
 15  #    notice, this list of conditions and the following disclaimer in 
 16  #    the documentation and/or other materials provided with the 
 17  #    distribution. 
 18  # 
 19  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
 20  # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
 21  # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS 
 22  # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE 
 23  # COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 
 24  # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, 
 25  # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; 
 26  # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 
 27  # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 
 28  # LIABILITY, OR TORT INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY 
 29  # WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 30  # POSSIBILITY OF SUCH DAMAGE. 
 31  # 
 32  # 
 33   
 34  """ 
 35  pyflowConfig 
 36   
 37  This file defines a class instance 'siteConfig' containing pyflow components 
 38  which are the most likely to need site-specific configuration. 
 39  """ 
 40   
 41  import os 
 42   
 43   
 44  # this is the only object pyflow needs to import, it 
 45  # is defined at the end of this module: 
 46  # 
 47  siteConfig = None 
 48   
 49   
 50  # depending on network setup getfqdn() can be slow, so cache calls to this function here: 
 51  # 
 52 -def _getHostName() : 
 53      import socket 
 54      return socket.getfqdn() 
 55   
 56  cachedHostName = None 
 57   
 58   
 59 -def getHostName() : 
 60      global cachedHostName 
 61      if cachedHostName is None : 
 62          cachedHostName = _getHostName() 
 63      return cachedHostName 
 64   
 65   
 66 -def getDomainName() : 
 67      hn = getHostName().split(".") 
 68      if len(hn) > 1 : hn = hn[1:] 
 69      return ".".join(hn) 
 70   
 71   
 72   
 73 -class DefaultSiteConfig(object) : 
 74      """ 
 75      Default configuration settings are designed to work with as 
 76      many sites as technically feasible 
 77      """ 
 78   
 79      # All email notifications come from the following e-mail address 
 80      # 
 81      mailFrom = "pyflow-bot@" + getDomainName() 
 82   
 83      # Default memory (in megabytes) requested by each command task: 
 84      # 
 85      defaultTaskMemMb = 2048 
 86   
 87      # In local run mode, this is the defalt memory per thread that we 
 88      # assume is available: 
 89      # 
 90      defaultHostMemMbPerCore = 2048 
 91   
 92      # maximum number of jobs which can be submitted to sge at once: 
 93      # 
 94      # allowed values are "unlimited" or None for unlimited jobs, or 
 95      # a positive integer describing the max number of jobs 
 96      # 
 97      maxSGEJobs = 128 
 98   
 99      # both getHostName and getDomainName are used in the 
100      # siteConfig factory, so these are not designed to be 
101      # overridden at present: 
102      getHostName = staticmethod(getHostName) 
103      getDomainName = staticmethod(getDomainName) 
104   
105      @classmethod 
106 -    def qsubResourceArg(cls, nCores, memMb) : 
107          """ 
108          When a task is launched using qsub in sge mode, it will call this 
109          function to specify the requested number of threads and megabytes 
110          of memory. The returned argument list will be appended to the qsub 
111          arguments. 
112   
113          nCores -- number of threads requested 
114          memMb -- memory requested (in megabytes) 
115          """ 
116          nCores = int(nCores) 
117          memMb = int(memMb) 
118          return cls._qsubResourceArgConfig(nCores, memMb) 
119   
120      @classmethod 
121 -    def _qsubResourceArgConfig(cls, nCores, memMb) : 
122          """ 
123          The default function is designed for maximum 
124          portability  -- it just provides more memory 
125          via more threads. 
126          """ 
127   
128          # this is the memory we assume is available per 
129          # thread on the cluster: 
130          # 
131          class Constants(object) : megsPerCore = 4096 
132   
133          memCores = 1 + ((memMb - 1) / Constants.megsPerCore) 
134   
135          qsubCores = max(nCores, memCores) 
136   
137          if qsubCores <= 1 : return [] 
138          return ["-pe", "threaded", str(qsubCores)] 
139   
140   
141      @classmethod 
142 -    def getSgeMakePrefix(cls, nCores, memMb, schedulerArgList) : 
143          """ 
144          This prefix will be added to ' -C directory', and run from 
145          a local process to handle sge make jobs. 
146   
147          Note that memMb hasn't been well defined for make jobs yet, 
148          is it the per task memory limit? The first application to 
149          accually make use of this will have to setup the convention, 
150          it is ignored right now... 
151          """ 
152          nCores = int(nCores) 
153          memMb = int(memMb) 
154   
155          retval = ["qmake", 
156                  "-V", 
157                  "-now", "n", 
158                  "-cwd", 
159                  "-N", "pyflowMakeTask"] 
160   
161          # user arguments to run() (usually q specification: 
162          retval.extend(schedulerArgList) 
163   
164          #### use qmake parallel environment: 
165          # retval.extend(["-pe","make",str(nCores),"--"]) 
166   
167          #### ...OR use 'dynamic' sge make environment: 
168          retval.extend(["--", "-j", str(nCores)]) 
169   
170          return retval 
171   
172   
173   
174 -def getEnvVar(key) : 
175      if key in os.environ : return os.environ[key] 
176      return None 
177   
178   
179   
180 -class hvmemSGEConfig(DefaultSiteConfig) : 
181      """ 
182      This config assumes 'h_vmem' is defined on the SGE instance 
183   
184      """ 
185   
186      @classmethod 
187 -    def _qsubResourceArgConfig(cls, nCores, memMb) : 
188   
189          retval = [] 
190   
191          # specify memory requirements 
192          memGb = 1 + ((memMb - 1) / 1024) 
193          reqArg = "h_vmem=%iG" % (memGb) 
194          retval.extend(["-l", reqArg]) 
195   
196          if nCores > 1 : 
197              retval.extend(["-pe", "threaded", str(nCores)]) 
198   
199          return retval 
200   
201   
202   
203  # 
204  # final step is the selection of this run's siteConfig object: 
205  # 
206   
207 -def siteConfigFactory() : 
208      # TODO: add an import time test to determine if h_vmem is valid 
209      return  hvmemSGEConfig 
210   
211   
212  siteConfig = siteConfigFactory() 
213