1.7. PyCOMPSs: Using objects, lists, and synchronization. Managing fault-tolerance.

In this example we will see how classes and objects can be used from PyCOMPSs, and that class methods can become tasks. The example also illustrates the current fault-tolerance management provided by the runtime.

Import the PyCOMPSs library

import pycompss.interactive as ipycompss

Start the runtime

Initialize COMPSs runtime Parameters indicates if the execution will generate task graph, tracefile, monitor interval and debug information.

import os
if 'BINDER_SERVICE_HOST' in os.environ:
    ipycompss.start(graph=True, debug=False,
                    project_xml='../xml/project.xml',
                    resources_xml='../xml/resources.xml')
else:
    ipycompss.start(graph=True, monitor=1000, trace=False, debug=False)

******************************************************
*************** PyCOMPSs Interactive *****************
******************************************************
*          .-~~-.--.           _____       _______   *
*         :         )         |____ \     /  ___  \  *
*   .~ ~ -.\       /.- ~~ .     ___) |    | (___) |  *
*   >       `.   .'       <    / ___/     \____   /  *
*  (         .- -.         )  | |___   _      /  /   *
*   `- -.-~  `- -'  ~-.- -'   |_____| |_|    /__/    *
*     (        :        )           _ _ .-:          *
*      ~--.    :    .--~        .-~  .-~  }          *
*          ~-.-^-.-~ \_      .~  .-~   .~            *
*                   \ \ '     \ '_ _ -~              *
*                    \`.\`.    //                    *
*           . - ~ ~-.__\`.\`-.//                     *
*       .-~   . - ~  }~ ~ ~-.~-.                     *
*     .' .-~      .-~       :/~-.~-./:               *
*    /_~_ _ . - ~                 ~-.~-._            *
*                                     ~-.<           *
******************************************************
* - Starting COMPSs runtime...                       *
* - Log path : /home/javier/.COMPSs/InteractiveMode_07/
* - PyCOMPSs Runtime started... Have fun!            *
******************************************************

Importing task and arguments directionality modules

Import task module before annotating functions or methods

from pycompss.api.api import compss_barrier
from pycompss.api.api import compss_wait_on
from pycompss.api.task import task
from pycompss.api.parameter import *

Declaring a class

%%writefile my_shaper.py

from pycompss.api.task import task
from pycompss.api.parameter import IN
import sys

class Shape(object):
    def __init__(self,x,y):
        self.x = x
        self.y = y
        description = "This shape has not been described yet"

    @task(returns=int, target_direction=IN)
    def area(self):
        return self.x * self.y

    @task()
    def scaleSize(self,scale):
        self.x = self.x * scale
        self.y = self.y * scale

    # on_failure= 'IGNORE', on_failure= 'RETRY', on_failure= 'FAIL', 'CANCEL_SUCCESSORS'
    @task(on_failure= 'CANCEL_SUCCESSORS')
    def downScale(self,scale):
        if (scale <= 0):
            sys.exit(1)
        else:
            self.x = self.x/scale
            self.y = self.y/scale

    @task(returns=int, target_direction=IN)
    def perimeter(self):
        return 2 * self.x + 2 * self.y

    def describe(self,text):
        self.description = text

    @task(target_direction=IN)
    def infoShape(self):
        print('Shape x=', self.x, 'y= ', self.y)

Overwriting my_shaper.py

Invoking tasks

from my_shaper import Shape

my_shapes = []
my_shapes.append(Shape(100,45))
my_shapes.append(Shape(50,50))
my_shapes.append(Shape(10,100))
my_shapes.append(Shape(20,30))
my_shapes.append(Shape(200,25))

all_perimeters = []

i=4
for this_shape in my_shapes:
    this_shape.scaleSize(2)
    this_shape.area()
    i = i - 1
    this_shape.downScale(i)
    all_perimeters.append(this_shape.perimeter())

Synchronizing results from tasks

# all_perimeters = compss_wait_on(all_perimeters)
# print all_perimeters

Stop the runtime

ipycompss.stop(sync=False)

******************************************************
*************** STOPPING PyCOMPSs ******************
******************************************************
Checking if any issue happened.

[ERRMGR]  -  WARNING: Job 15, running Task 15 on worker localhost, has failed.
[ERRMGR]  -  WARNING: Notifying task 15 failure
[ERRMGR]  -  WARNING: Task 'my_shaper.Shape.downScale' TOTALLY FAILED.
[ERRMGR]  -  WARNING: Task 16(Action: 16) with name my_shaper.Shape.perimeter has been cancelled.
[ERRMGR]  -  WARNING: Task failed: [[Task id: 15], [Status: FAILED], [Core id: 2], [Priority: false], [NumNodes: 1], [MustReplicate: false], [MustDistribute: false], [my_shaper.Shape.downScale(INT_T)]]
[ERRMGR]  -  WARNING: Task canceled: [[Task id: 16], [Status: CANCELED], [Core id: 3], [Priority: false], [NumNodes: 1], [MustReplicate: false], [MustDistribute: false], [my_shaper.Shape.perimeter()]]
[ERRMGR]  -  WARNING: Job 18, running Task 19 on worker localhost, has failed.
[ERRMGR]  -  WARNING: Notifying task 19 failure
[ERRMGR]  -  WARNING: Task 'my_shaper.Shape.downScale' TOTALLY FAILED.
[ERRMGR]  -  WARNING: Task 20(Action: 20) with name my_shaper.Shape.perimeter has been cancelled.
[ERRMGR]  -  WARNING: Task failed: [[Task id: 19], [Status: FAILED], [Core id: 2], [Priority: false], [NumNodes: 1], [MustReplicate: false], [MustDistribute: false], [my_shaper.Shape.downScale(INT_T)]]
[ERRMGR]  -  WARNING: Task canceled: [[Task id: 20], [Status: CANCELED], [Core id: 3], [Priority: false], [NumNodes: 1], [MustReplicate: false], [MustDistribute: false], [my_shaper.Shape.perimeter()]]

Warning: some of the variables used with PyCOMPSs may
         have not been brought to the master.
******************************************************