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
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