packages/vaex-core/vaex/misc/parallelize.py
def parallelize(cores=None, fork=True, flatten=False, info=False, infoclass=InfoThreadProgressBar, init=None, *args, **kwargs):
	"""Function decorater that executes the function in parallel
	
	Usage::

		@parallelize(cores=10, info=True)
		def f(x):
			return x**2
		
		x = numpy.arange(0, 100, 0.1)
		y = f(x) # this gets executed parallel

	:param cores: number of cpus/cores to use (if None, it counts the cores using /proc/cpuinfo)
	:param fork: fork a process (should always be true since of the GIT, but can be used with c modules that release the GIT)
	:param flatten: if False and each return value is a list, final result will be a list of lists, if True, all lists are combined to one big list
	:param info: show progress bar (see infoclass)
	:param infoclass: class to instantiate that shows the progress (default shows progressbar)
	:param init: function to be called in each forked process before executing, can be used to set the seed, takes a integer as parameter (number that identifies the process)
	:param args: extra arguments passed to function
	:param kwargs: extra keyword arguments passed to function
			
	Example::
			
		@parallelize(cores=10, info=True, n=2)
		def f(x, n):
			return x**n
		
		x = numpy.arange(0, 100, 0.1)
		y = f(x) # this gets executed parallel
			
	
	
	"""
	if cores == None:
		cores = multiprocessing.cpu_count()
	def wrapper(f):
		def execute(*multiargs):
			results = []
			len(list(zip(*multiargs)))
			N = len(multiargs[0])
			if info:
				print("running %i jobs on %i cores" % (N, cores))
			taskQueue = queue.Queue(len(multiargs[0]))
			#for timenr in range(times):
			#	taskQueue.put(timenr)
			for tasknr, _args in enumerate(zip(*multiargs)):
				taskQueue.put((tasknr, list(_args)))
			#for timenr in range(times):
			#	result = f(*args, **kwargs)
			#	results.append(result)
			executions = [Execution(taskQueue, fork, f, init, corenr, args, kwargs) for corenr in range(cores)]
			if info:
				infoobj = infoclass(len(multiargs[0]), executions)
				infoobj.start()
			for i, execution in enumerate(executions):
				execution.setName("T-%d" % i)
				execution.start()
			#if 1:
			#	watchdog = Watchdog(executions)
			#	watchdog.start()
			error = False
			for execution in executions:
				log("joining:",execution.getName())
				try:
					execution.join()
				except BaseException:
					error = True
				results.extend(execution.results)
				if execution.error:
					error = True 
			if info:
				infoobj.join()
			if error:
				print("error", file=sys.stderr)
				results = None
				raise Exception("error in one or more of the executors")
			else:
				results.sort(cmp=lambda a, b: cmp(a[0], b[0]))
				results = [k[1] for k in results]
				#print "bla", results
				if flatten:
					flatresults = []
					for result in results:
						flatresults.extend(result)
					results = flatresults
			return results
		return execute
	return wrapper
def parallelize(cores=None, fork=True, flatten=False, info=False, infoclass=InfoThreadProgressBar, init=None, *args, **kwargs):
	"""Function decorater that executes the function in parallel
	
	Usage::

		@parallelize(cores=10, info=True)
		def f(x):
			return x**2
		
		x = numpy.arange(0, 100, 0.1)
		y = f(x) # this gets executed parallel

	:param cores: number of cpus/cores to use (if None, it counts the cores using /proc/cpuinfo)
	:param fork: fork a process (should always be true since of the GIT, but can be used with c modules that release the GIT)
	:param flatten: if False and each return value is a list, final result will be a list of lists, if True, all lists are combined to one big list
	:param info: show progress bar (see infoclass)
	:param infoclass: class to instantiate that shows the progress (default shows progressbar)
	:param init: function to be called in each forked process before executing, can be used to set the seed, takes a integer as parameter (number that identifies the process)
	:param args: extra arguments passed to function
	:param kwargs: extra keyword arguments passed to function
			
	Example::
			
		@parallelize(cores=10, info=True, n=2)
		def f(x, n):
			return x**n
		
		x = numpy.arange(0, 100, 0.1)
		y = f(x) # this gets executed parallel
			
	
	
	"""
	if cores == None:
		cores = multiprocessing.cpu_count()
	def wrapper(f):
		def execute(*multiargs):
			results = []
			len(list(zip(*multiargs)))
			N = len(multiargs[0])
			if info:
				print("running %i jobs on %i cores" % (N, cores))
			taskQueue = queue.Queue(len(multiargs[0]))
			#for timenr in range(times):
			#	taskQueue.put(timenr)
			for tasknr, _args in enumerate(zip(*multiargs)):
				taskQueue.put((tasknr, list(_args)))
			#for timenr in range(times):
			#	result = f(*args, **kwargs)
			#	results.append(result)
			executions = [Execution(taskQueue, fork, f, init, corenr, args, kwargs) for corenr in range(cores)]
			if info:
				infoobj = infoclass(len(multiargs[0]), executions)
				infoobj.start()
			for i, execution in enumerate(executions):
				execution.setName("T-%d" % i)
				execution.start()
			#if 1:
			#	watchdog = Watchdog(executions)
			#	watchdog.start()
			error = False
			for execution in executions:
				log("joining:",execution.getName())
				try:
					execution.join()
				except BaseException:
					error = True
				results.extend(execution.results)
				if execution.error:
					error = True 
			if info:
				infoobj.join()
			if error:
				print("error", file=sys.stderr)
				results = None
				raise Exception("error in one or more of the executors")
			else:
				results.sort(cmp=lambda a, b: cmp(a[0], b[0]))
				results = [k[1] for k in results]
				#print "bla", results
				if flatten:
					flatresults = []
					for result in results:
						flatresults.extend(result)
					results = flatresults
			return results
		return execute
	return wrapper
def parallelize ( cores = None , fork = True , flatten = False , info = False , infoclass = InfoThreadProgressBar , init = None , * args , * * kwargs ) : if cores == None : cores = multiprocessing . cpu_count ( ) def wrapper ( f ) : def execute ( * multiargs ) : results = [ ] len ( list ( zip ( * multiargs ) ) ) N = len ( multiargs [ 0 ] ) if info : print ( "running %i jobs on %i cores" % ( N , cores ) ) taskQueue = queue . Queue ( len ( multiargs [ 0 ] ) ) #for timenr in range(times): #	taskQueue.put(timenr) for tasknr , _args in enumerate ( zip ( * multiargs ) ) : taskQueue . put ( ( tasknr , list ( _args ) ) ) #for timenr in range(times): #	result = f(*args, **kwargs) #	results.append(result) executions = [ Execution ( taskQueue , fork , f , init , corenr , args , kwargs ) for corenr in range ( cores ) ] if info : infoobj = infoclass ( len ( multiargs [ 0 ] ) , executions ) infoobj . start ( ) for i , execution in enumerate ( executions ) : execution . setName ( "T-%d" % i ) execution . start ( ) #if 1: #	watchdog = Watchdog(executions) #	watchdog.start() error = False for execution in executions : log ( "joining:" , execution . getName ( ) ) try : execution . join ( ) except BaseException : error = True results . extend ( execution . results ) if execution . error : error = True if info : infoobj . join ( ) if error : print ( "error" , file = sys . stderr ) results = None raise Exception ( "error in one or more of the executors" ) else : results . sort ( cmp = lambda a , b : cmp ( a [ 0 ] , b [ 0 ] ) ) results = [ k [ 1 ] for k in results ] #print "bla", results if flatten : flatresults = [ ] for result in results : flatresults . extend ( result ) results = flatresults return results return execute return wrapper
Function decorater that executes the function in parallel
	
	Usage::

		@parallelize(cores=10, info=True)
		def f(x):
			return x**2
		
		x = numpy.arange(0, 100, 0.1)
		y = f(x) # this gets executed parallel

	:param cores: number of cpus/cores to use (if None, it counts the cores using /proc/cpuinfo)
	:param fork: fork a process (should always be true since of the GIT, but can be used with c modules that release the GIT)
	:param flatten: if False and each return value is a list, final result will be a list of lists, if True, all lists are combined to one big list
	:param info: show progress bar (see infoclass)
	:param infoclass: class to instantiate that shows the progress (default shows progressbar)
	:param init: function to be called in each forked process before executing, can be used to set the seed, takes a integer as parameter (number that identifies the process)
	:param args: extra arguments passed to function
	:param kwargs: extra keyword arguments passed to function
			
	Example::
			
		@parallelize(cores=10, info=True, n=2)
		def f(x, n):
			return x**n
		
		x = numpy.arange(0, 100, 0.1)
		y = f(x) # this gets executed parallel
Function decorater that executes the function in parallel Usage ::