def adjust_workers(num_flows, num_cpus, worker_sockets, log_fh=None):
 qiime_config = load_qiime_config()
 min_per_core = int(qiime_config['denoiser_min_per_core'])
 if (num_flows < ((num_cpus - 1) * min_per_core)):
  if log_fh:
   log_fh.write('Adjusting number of workers:\n')
   log_fh.write(('flows: %d   cpus:%d\n' % (num_flows, num_cpus)))
  per_core = max(min_per_core, ((num_flows / num_cpus) + 1))
  for i in range(num_cpus):
   if ((i * per_core) > num_flows):
    worker_sock = worker_sockets.pop()
    worker_sock.close()
    num_cpus = (num_cpus - 1)
    if log_fh:
     log_fh.write(('released worker %d\n' % i))
  if log_fh:
   log_fh.write(('New number of cpus:%d\n' % num_cpus))
 if ((num_cpus == 0) or (num_cpus != len(worker_sockets))):
  raise ValueError('Adjust_workers screwed up!')
 return num_cpus

def adjust_workers ( num_flows , num_cpus , worker_sockets , log_fh = None ) : qiime_config = load_qiime_config ( ) min_per_core = int ( qiime_config [ 'denoiser_min_per_core' ] ) if ( num_flows < ( ( num_cpus - 1 ) * min_per_core ) ) : if log_fh : log_fh . write ( 'Adjusting number of workers:\n' ) log_fh . write ( ( 'flows: %d   cpus:%d\n' % ( num_flows , num_cpus ) ) ) per_core = max ( min_per_core , ( ( num_flows / num_cpus ) + 1 ) ) for i in range ( num_cpus ) : if ( ( i * per_core ) > num_flows ) : worker_sock = worker_sockets . pop ( ) worker_sock . close ( ) num_cpus = ( num_cpus - 1 ) if log_fh : log_fh . write ( ( 'released worker %d\n' % i ) ) if log_fh : log_fh . write ( ( 'New number of cpus:%d\n' % num_cpus ) ) if ( ( num_cpus == 0 ) or ( num_cpus != len ( worker_sockets ) ) ) : raise ValueError ( 'Adjust_workers screwed up!' ) return num_cpus
stop workers no longer needed .