Decorate interface for parallel computation

class sage.parallel.decorate.Fork(timeout=0, verbose=False)[source]

Bases: object

A fork decorator class.

class sage.parallel.decorate.Parallel(p_iter='fork', ncpus=None, **kwds)[source]

Bases: object

Create a parallel-decorated function. This is the object created by parallel().

class sage.parallel.decorate.ParallelFunction(parallel, func)[source]

Bases: object

Class which parallelizes a function or class method. This is typically accessed indirectly through Parallel.__call__().

sage.parallel.decorate.fork(f=None, timeout=0, verbose=False)[source]

Decorate a function so that when called it runs in a forked subprocess.

This means that it will not have any in-memory side effects on the parent Sage process. The pexpect interfaces are all reset.

INPUT:

  • f – a function

  • timeout – (default: 0) if positive, kill the subprocess after this many seconds (wall time)

  • verbose – boolean (default: False); whether to print anything about what the decorator does (e.g., killing the subprocess)

Warning

The forked subprocess will not have access to data created in pexpect interfaces. This behavior with respect to pexpect interfaces is very important to keep in mind when setting up certain computations. It’s the one big limitation of this decorator.

EXAMPLES:

We create a function and run it with the fork decorator. Note that it does not have a side effect. Despite trying to change the global variable a below in g, the variable a does not get changed:

sage: a = 5
sage: @fork
....: def g(n, m):
....:     global a
....:     a = 10
....:     return factorial(n).ndigits() + m
sage: g(5, m=5)
8
sage: a
5

>>> from sage.all import *
>>> a = Integer(5)
>>> @fork
... def g(n, m):
...     global a
...     a = Integer(10)
...     return factorial(n).ndigits() + m
>>> g(Integer(5), m=Integer(5))
8
>>> a
5

We use fork to make sure that the function terminates after 100 ms, no matter what:

sage: @fork(timeout=0.1, verbose=True)
....: def g(n, m): return factorial(n).ndigits() + m
sage: g(10^7, m=5)
Killing subprocess ... with input ((10000000,), {'m': 5}) which took too long
'NO DATA (timed out)'

>>> from sage.all import *
>>> @fork(timeout=RealNumber('0.1'), verbose=True)
... def g(n, m): return factorial(n).ndigits() + m
>>> g(Integer(10)**Integer(7), m=Integer(5))
Killing subprocess ... with input ((10000000,), {'m': 5}) which took too long
'NO DATA (timed out)'

We illustrate that the state of the pexpect interface is not altered by forked functions (they get their own new pexpect interfaces!):

sage: # needs sage.libs.pari
sage: gp.eval('a = 5')
'5'
sage: @fork()
....: def g():
....:     gp.eval('a = 10')
....:     return gp.eval('a')
sage: g()
'10'
sage: gp.eval('a')
'5'

>>> from sage.all import *
>>> # needs sage.libs.pari
>>> gp.eval('a = 5')
'5'
>>> @fork()
... def g():
...     gp.eval('a = 10')
...     return gp.eval('a')
>>> g()
'10'
>>> gp.eval('a')
'5'

We illustrate that the forked function has its own pexpect interface:

sage: gp.eval('a = 15')                                                         # needs sage.libs.pari
'15'
sage: @fork()
....: def g(): return gp.eval('a')
sage: g()                                                                       # needs sage.libs.pari
'a'

>>> from sage.all import *
>>> gp.eval('a = 15')                                                         # needs sage.libs.pari
'15'
>>> @fork()
... def g(): return gp.eval('a')
>>> g()                                                                       # needs sage.libs.pari
'a'

We illustrate that segfaulting subprocesses are no trouble at all:

sage: cython('def f(): print(<char*>0)')                                        # needs sage.misc.cython
sage: @fork
....: def g():
....:     os.environ["CYSIGNALS_CRASH_NDEBUG"]="yes" # skip enhanced backtrace (it is slow)
....:     f()
sage: print("this works"); g()                                                  # needs sage.misc.cython
this works...

------------------------------------------------------------------------
Unhandled SIG...
------------------------------------------------------------------------
'NO DATA'

>>> from sage.all import *
>>> cython('def f(): print(<char*>0)')                                        # needs sage.misc.cython
>>> @fork
... def g():
...     os.environ["CYSIGNALS_CRASH_NDEBUG"]="yes" # skip enhanced backtrace (it is slow)
...     f()
>>> print("this works"); g()                                                  # needs sage.misc.cython
this works...
<BLANKLINE>
------------------------------------------------------------------------
Unhandled SIG...
------------------------------------------------------------------------
'NO DATA'
sage.parallel.decorate.normalize_input(a)[source]

Convert a to a pair (args, kwds) using some rules:

  • if already of that form, leave that way.

  • if a is a tuple make (a,{})

  • if a is a dict make (tuple(),a)

  • otherwise make ((a,),{})

INPUT:

  • a – object

OUTPUT:

  • args – tuple

  • kwds – dictionary

EXAMPLES:

sage: sage.parallel.decorate.normalize_input( (2, {3:4}) )
((2, {3: 4}), {})
sage: sage.parallel.decorate.normalize_input( (2,3) )
((2, 3), {})
sage: sage.parallel.decorate.normalize_input( {3:4} )
((), {3: 4})
sage: sage.parallel.decorate.normalize_input( 5 )
((5,), {})

>>> from sage.all import *
>>> sage.parallel.decorate.normalize_input( (Integer(2), {Integer(3):Integer(4)}) )
((2, {3: 4}), {})
>>> sage.parallel.decorate.normalize_input( (Integer(2),Integer(3)) )
((2, 3), {})
>>> sage.parallel.decorate.normalize_input( {Integer(3):Integer(4)} )
((), {3: 4})
>>> sage.parallel.decorate.normalize_input( Integer(5) )
((5,), {})
sage.parallel.decorate.parallel(p_iter='fork', ncpus=None, **kwds)[source]

This is a decorator that gives a function a parallel interface, allowing it to be called with a list of inputs, whose values will be computed in parallel.

Warning

The parallel subprocesses will not have access to data created in pexpect interfaces. This behavior with respect to pexpect interfaces is very important to keep in mind when setting up certain computations. It’s the one big limitation of this decorator.

INPUT:

  • p_iter – parallel iterator function or string: - 'fork' – (default) use a new forked subprocess for each input - 'multiprocessing' – use multiprocessing library - 'reference' – use a fake serial reference implementation

  • ncpus – integer; maximal number of subprocesses to use at the same time

  • timeout – number of seconds until each subprocess is killed (only supported by 'fork'; zero means not at all)

  • reseed_rng: reseed the rng (random number generator) in each subprocess

Warning

If you use anything but 'fork' above, then a whole new subprocess is spawned, so none of your local state (variables, certain functions, etc.) is available.

EXAMPLES:

We create a simple decoration for a simple function. The number of cpus (or cores, or hardware threads) is automatically detected:

sage: @parallel
....: def f(n): return n*n
sage: f(10)
100
sage: sorted(list(f([1,2,3])))
[(((1,), {}), 1), (((2,), {}), 4), (((3,), {}), 9)]

>>> from sage.all import *
>>> @parallel
... def f(n): return n*n
>>> f(Integer(10))
100
>>> sorted(list(f([Integer(1),Integer(2),Integer(3)])))
[(((1,), {}), 1), (((2,), {}), 4), (((3,), {}), 9)]

We use exactly two cpus:

sage: @parallel(2)
....: def f(n): return n*n

>>> from sage.all import *
>>> @parallel(Integer(2))
... def f(n): return n*n

We create a decorator that uses three subprocesses, and times out individual processes after 10 seconds:

sage: @parallel(ncpus=3, timeout=10)
....: def fac(n): return factor(2^n-1)
sage: for X, Y in sorted(list(fac([101,119,151,197,209]))): print((X,Y))        # needs sage.libs.pari
(((101,), {}), 7432339208719 * 341117531003194129)
(((119,), {}), 127 * 239 * 20231 * 131071 * 62983048367 * 131105292137)
(((151,), {}), 18121 * 55871 * 165799 * 2332951 * 7289088383388253664437433)
(((197,), {}), 7487 * 26828803997912886929710867041891989490486893845712448833)
(((209,), {}), 23 * 89 * 524287 * 94803416684681 * 1512348937147247 * 5346950541323960232319657)

sage: @parallel('multiprocessing')
....: def f(N): return N^2
sage: v = list(f([1,2,4])); v.sort(); v
[(((1,), {}), 1), (((2,), {}), 4), (((4,), {}), 16)]
sage: @parallel('reference')
....: def f(N): return N^2
sage: v = list(f([1,2,4])); v.sort(); v
[(((1,), {}), 1), (((2,), {}), 4), (((4,), {}), 16)]

>>> from sage.all import *
>>> @parallel(ncpus=Integer(3), timeout=Integer(10))
... def fac(n): return factor(Integer(2)**n-Integer(1))
>>> for X, Y in sorted(list(fac([Integer(101),Integer(119),Integer(151),Integer(197),Integer(209)]))): print((X,Y))        # needs sage.libs.pari
(((101,), {}), 7432339208719 * 341117531003194129)
(((119,), {}), 127 * 239 * 20231 * 131071 * 62983048367 * 131105292137)
(((151,), {}), 18121 * 55871 * 165799 * 2332951 * 7289088383388253664437433)
(((197,), {}), 7487 * 26828803997912886929710867041891989490486893845712448833)
(((209,), {}), 23 * 89 * 524287 * 94803416684681 * 1512348937147247 * 5346950541323960232319657)

>>> @parallel('multiprocessing')
... def f(N): return N**Integer(2)
>>> v = list(f([Integer(1),Integer(2),Integer(4)])); v.sort(); v
[(((1,), {}), 1), (((2,), {}), 4), (((4,), {}), 16)]
>>> @parallel('reference')
... def f(N): return N**Integer(2)
>>> v = list(f([Integer(1),Integer(2),Integer(4)])); v.sort(); v
[(((1,), {}), 1), (((2,), {}), 4), (((4,), {}), 16)]

For functions that take multiple arguments, enclose the arguments in tuples when calling the parallel function:

sage: @parallel
....: def f(a, b): return a*b
sage: for X, Y in sorted(list(f([(2,3),(3,5),(5,7)]))): print((X, Y))
(((2, 3), {}), 6)
(((3, 5), {}), 15)
(((5, 7), {}), 35)

>>> from sage.all import *
>>> @parallel
... def f(a, b): return a*b
>>> for X, Y in sorted(list(f([(Integer(2),Integer(3)),(Integer(3),Integer(5)),(Integer(5),Integer(7))]))): print((X, Y))
(((2, 3), {}), 6)
(((3, 5), {}), 15)
(((5, 7), {}), 35)

For functions that take a single tuple as an argument, enclose it in an additional tuple at call time, to distinguish it as the first argument, as opposed to a tuple of arguments:

sage: @parallel
....: def firstEntry(aTuple): return aTuple[0]
sage: for X, Y in sorted(list(firstEntry([((1,2,3,4),),((5,6,7,8),)]))): print((X, Y))
((((1, 2, 3, 4),), {}), 1)
((((5, 6, 7, 8),), {}), 5)

>>> from sage.all import *
>>> @parallel
... def firstEntry(aTuple): return aTuple[Integer(0)]
>>> for X, Y in sorted(list(firstEntry([((Integer(1),Integer(2),Integer(3),Integer(4)),),((Integer(5),Integer(6),Integer(7),Integer(8)),)]))): print((X, Y))
((((1, 2, 3, 4),), {}), 1)
((((5, 6, 7, 8),), {}), 5)

The parallel decorator also works with methods, classmethods, and staticmethods. Be sure to apply the parallel decorator after (“above”) either the classmethod or staticmethod decorators:

sage: class Foo():
....:     @parallel(2)
....:     def square(self, n):
....:         return n*n
....:     @parallel(2)
....:     @classmethod
....:     def square_classmethod(cls, n):
....:         return n*n
sage: a = Foo()
sage: a.square(3)
9
sage: sorted(a.square([2,3]))
[(((2,), {}), 4), (((3,), {}), 9)]
sage: Foo.square_classmethod(3)
9
sage: sorted(Foo.square_classmethod([2,3]))
[(((2,), {}), 4), (((3,), {}), 9)]
sage: Foo.square_classmethod(3)
9

>>> from sage.all import *
>>> class Foo():
...     @parallel(Integer(2))
...     def square(self, n):
...         return n*n
...     @parallel(Integer(2))
...     @classmethod
...     def square_classmethod(cls, n):
...         return n*n
>>> a = Foo()
>>> a.square(Integer(3))
9
>>> sorted(a.square([Integer(2),Integer(3)]))
[(((2,), {}), 4), (((3,), {}), 9)]
>>> Foo.square_classmethod(Integer(3))
9
>>> sorted(Foo.square_classmethod([Integer(2),Integer(3)]))
[(((2,), {}), 4), (((3,), {}), 9)]
>>> Foo.square_classmethod(Integer(3))
9

By default, all subprocesses use the same random seed and therefore the same deterministic randomness. For functions that should be randomized, we can reseed the random seed in each subprocess:

sage: @parallel(reseed_rng=True)
....: def unif(n): return ZZ.random_element(x=0, y=n)
sage: set_random_seed(42)
sage: sorted(unif([1000]*3)) # random
[(((1000,), {}), 444), (((1000,), {}), 597), (((1000,), {}), 640)]

>>> from sage.all import *
>>> @parallel(reseed_rng=True)
... def unif(n): return ZZ.random_element(x=Integer(0), y=n)
>>> set_random_seed(Integer(42))
>>> sorted(unif([Integer(1000)]*Integer(3))) # random
[(((1000,), {}), 444), (((1000,), {}), 597), (((1000,), {}), 640)]

Warning

Currently, parallel methods do not work with the multiprocessing implementation.