



  

Rain 0.3.0-pre documentation

 [Rain](index.html#document-index) 

 stable

 

* [Overview](index.html#document-overview)

	+ [What is in the box](index.html#what-is-in-the-box)

	+ [Future directions](index.html#future-directions)

	+ [What we do *NOT* want to do](index.html#what-we-do-not-want-to-do)

	+ [Comparison with similar tools](index.html#comparison-with-similar-tools)

	+ [Roadmap](index.html#roadmap)

* [Quickstart](index.html#document-quickstart)

	+ [Introducing Rain Applications](index.html#introducing-rain-applications)

	+ [Writing your first Rain Application](index.html#writing-your-first-rain-application)

* [User’s Guide](index.html#document-user)

	+ [Basic terms](index.html#basic-terms)

	+ [Task definition and submission](index.html#task-definition-and-submission)

	+ [Fetching data objects](index.html#fetching-data-objects)

	+ [Inter-task dependencies](index.html#inter-task-dependencies)

	+ [More outputs](index.html#more-outputs)

	+ [Object data types](index.html#object-data-types)

	+ [Object content types](index.html#object-content-types)

	+ [Constant data objects](index.html#constant-data-objects)

	+ [Built-in tasks](index.html#built-in-tasks)

	+ [Running external programs](index.html#running-external-programs)

	+ [Python tasks](index.html#python-tasks)

	+ [Resources](index.html#resources)

	+ [Attributes ‘spec’ and ‘info’](index.html#attributes-spec-and-info)

	+ [Waiting for object(s) and task(s)](index.html#waiting-for-object-s-and-task-s)

	+ [Directories](index.html#directories)

	+ [Mapping data objects onto filesystem](index.html#mapping-data-objects-onto-filesystem)

	+ [Sessions](index.html#sessions)

* [Writing Own Executors](index.html#document-executors)

	+ [Rust tasklib](index.html#rust-tasklib)

	+ [C++ tasklib](index.html#c-tasklib)

	+ [Registration in governor](index.html#registration-in-governor)

	+ [Client API](index.html#client-api)

* [Installation, Running & Deployment](index.html#document-install)

	+ [Binaries](index.html#binaries)

	+ [Build via cargo](index.html#build-via-cargo)

	+ [Build from sources](index.html#build-from-sources)

	+ [Starting infrastructure](index.html#starting-infrastructure)

	+ [Arguments for program *rain*](index.html#arguments-for-program-rain)

* [Examples](index.html#document-examples)

	+ [Distributed cross-validation with libsvm](index.html#distributed-cross-validation-with-libsvm)

* [Python API](index.html#document-python_api)

	+ [Client API](index.html#client-api)

	+ [Remote Python tasks](index.html#remote-python-tasks)

* [Contributors’s Guide](index.html#document-dev)

	+ [Scripts](index.html#scripts)

	+ [Testing](index.html#testing)

	+ [Dashboard](index.html#dashboard)

[Rain](index.html#document-index)

* [Docs](index.html#document-index) »

* Rain 0.3.0-pre documentation

* [Edit on GitHub](https://github.com/substantic/rain/blob/35c0a8b1687e33b2014a52ba39b59d5e6a020d45/docs/guide/index.rst)

---

Welcome to Rain’s documentation![¶](#welcome-to-rain-s-documentation "Permalink to this headline")

==================================================================================================

Overview[¶](#overview "Permalink to this headline")

---------------------------------------------------

**Rain** is an open-source distributed computational framework for large-scale

task-based pipelines.

Rain aims to lower the entry barrier to the world of distributed computing and

to do so efficiently and within any scale. Our intention is to develop a light

yet robust distributed framework that features an intuitive [Python](https://www.python.org/) API,

straightforward installation and deployment with insightful monitoring on top.

Note

Despite that this is an early release of Rain, it is a fully functional

project that can be used out-of-the box. Being aware that there is still

a lot that can be improved and added, we are looking for external

users and collaborators to drive our future work, both enthusiasts, from the

industry and the scientific community. Talk to us online at [Gitter](https://gitter.im/substantic/rain) or via email

and let us know what your project needs and use-cases, submit bugs or feature

requests at [GitHub](https://github.com/substantic/rain) or even contribute with pull requests.

* **Dataflow programming.** Computation in Rain is defined as a flow graph of

tasks. Tasks may be built-in functions, Python code, or an external

applications, short and light or long-running and heavy. The system is

designed to integrate any code into a pipeline, respecting its resource

requirements, and to handle very large task graphs (hundreds thousands tasks).

* **Easy to use.** Rain was designed to be easy to deployed anywhere, ranging

from a single node deployments to large-scale distributed systems and clouds

ranging thousands of cores.

* **Rust core, Python API.** Rain is written in [Rust](https://www.rust-lang.org/en-US/) for safety and efficiency

and has a high-level Python API to Rain core infrastructure, and even supports

Python tasks out-of-the-box. Nevertheless, Rain core infrastructure provides

a language-independent interface that does not prevent adding support for

other languages in the future.

* **Tasks in Python/C++/Rust** Rain provides a way to define user-defined tasks

in Python, C++, and Rust.

* **Monitoring** Rain is designed to support both online and postmortem

monitoring.

\*\* [Get started now.](index.html#document-quickstart) \*\*

### What is in the box[¶](#what-is-in-the-box "Permalink to this headline")

Rain infrastructure composes of a central **server** component and **governor**

components, that may run on different machines. A governor may spawn one or more

**executors** that are local processes that provides execution of an external

code. Rain is distributed with Python executor. Rain also provides libraries

for C++ and Rust for writing own specialized executors.

Users interacts with server via

**client** applications. Rain is distributed with Python client API.

![Connection between basic components of Rain](_images/arch.svg)

#### Python Client[¶](#python-client "Permalink to this headline")

> 

> * Task-based programming model.

> * High-level interface to Rain core infrastructure.

> * Easy definition of various types of tasks and their inter-dependencies.

> * Python3 module.

> 

> 

> 

#### Rain Core Infrastructure[¶](#rain-core-infrastructure "Permalink to this headline")

> 

> * Basic scheduling heuristic respecting inter-task dependencies.

> * Rust implementation enabling easy build, deployment, and reliable run.

> * Distributed as all-in-one binary.

> * Direct governor-to-governor communication.

> * Basic dashboard for execution monitoring.

> 

> 

> 

#### Executors[¶](#executors "Permalink to this headline")

> 

> * Possibility to define own tasks in Python, C++, and Rust

> 

> 

> 

### Future directions[¶](#future-directions "Permalink to this headline")

There are many things to improve and even more new things to add. To work

efficiently we need to prioritize, and for that we need your feedback and use

cases. Which features would you like to see and put to good use? What kind of

pipelines do you run?

#### Better dashboard[¶](#better-dashboard "Permalink to this headline")

Better interactive view on the current and past computation status, including

post-mortem analysis. Which stats and views give you the most insight?

#### Better scheduler[¶](#better-scheduler "Permalink to this headline")

While surprisingly efficient, the current scheduler is currently mostly based on

heuristics and rules. We plan to replace it with an incremental global scheduler

based on belief propagation.

#### Resiliency[¶](#resiliency "Permalink to this headline")

The current version supports and propagates some failures (remote python task

exceptions, external program errors) but other errors still cause server panic

(e.g. governor node failure). The near-term goal is to have better failure modes

for introspection and possibly recovery. The system is designed to allow

building resiliency against task or governor failures via checkpoints in the task

graph (keeping file copies). It is not clear how useful to our users this would

be but it is on our radar.

#### Resources[¶](#resources "Permalink to this headline")

Currently, the only resources supported are CPU cores. We are working on also

supporting memory requirements, but other resources (GPUs, TPUs, disk space,

…) should be possible with enough work and interest.

#### Stream objects support[¶](#stream-objects-support "Permalink to this headline")

Some tasks work in a streaming fashion and it would be inefficient to wait for

their entire output before starting a consumer task. We plan to include

streaming data objects but there are semantic and usage issues about resources,

scheduling, multiple consumers and resiliency.

#### REST client interface[¶](#rest-client-interface "Permalink to this headline")

The capnp API is a bit heavy-handed for a client API. We plan to create a REST

API for the client applications, simplifying API creation in new languages, and

to unify it with the dashboard/status query API. External REST apis are

convenient for many users and they do not seem to be a performance bottleneck.

#### Easier Deployment in cloud settings[¶](#easier-deployment-in-cloud-settings "Permalink to this headline")

The Rust binary is already one statically linked file and one python-only

library, making distribution easy and running on PBS is already supported. We

would like to add better support for cloud settings, e.g. AWS and Kubernetes.

### What we do *NOT* want to do[¶](#what-we-do-not-want-to-do "Permalink to this headline")

There are also some directions we do NOT intend to focus on in the scope of Rain.

#### Visual editor[¶](#visual-editor "Permalink to this headline")

We do not plan to support visual creation and editing of pipelines. The scale of

reasonably editable workflows is usually very small. We focus on clean and easy

client APIs and great visualization.

#### User isolation and task sandboxing[¶](#user-isolation-and-task-sandboxing "Permalink to this headline")

We do not plan to limit malicious users or tasks from doing any harm. Use

existing tools for task isolation. The system is lightweight enough to have one

instance per user if necessary.

#### Fair user scheduling, accounting and quotas[¶](#fair-user-scheduling-accounting-and-quotas "Permalink to this headline")

When running multiple sessions, there is no intention to fairly schedule or

prioritize them. The objective is only overall efficient resource usage.

### Comparison with similar tools[¶](#comparison-with-similar-tools "Permalink to this headline")

TODO

### Roadmap[¶](#roadmap "Permalink to this headline")

<https://github.com/substantic/rain/issues/26>

Quickstart[¶](#quickstart "Permalink to this headline")

-------------------------------------------------------

### Introducing Rain Applications[¶](#introducing-rain-applications "Permalink to this headline")

Rain Applications are programs **defined on client side** and **executed on

Rain infrastructure** using the Rain API. Rain automaticaly distributes

execution of the applications in a distributed environment.

Rain Applications follows the paradigm of task oriented programming. Basic

building blocks for every Rain App are tasks - generic abstraction units

representing various kinds of computations ranging from native Python tasks to

3rd party software.

Rain tasks may be arbitrarily\* chained together and so provide complex

high-level functionality.

### Writing your first Rain Application[¶](#writing-your-first-rain-application "Permalink to this headline")

This section demonstrate how to start Rain infrastructure locally and execute

a simple “Hello world” application.

* **Start Rain infrastructue** Although, the components of Rain (server and

governor(s)) can be started manually, in order to simplify this process, we

provide “rain start” command to do it for you automatically. The following

command starts server and one local governor. (Starting Rain infrastructure on

distributed systems is described in [Starting infrastructure](index.html#start-rain).):

```

$ rain start --simple

```

* **Running “Hello World” example.** The following Python program creates a task

that joins two strings (This example is more explained in Section

tasks-and-objs.):

```

from rain.client import Client, tasks, blob

# Connect to server

client = Client("localhost", 7210)

# Create a new session

with client.new\_session() as session:

    # Create task (and two data objects)

    task = tasks.Concat((blob("Hello "), blob("world!"),))

    # Mark that the output should be kept after submit

    task.output.keep()

    # Submit all crated tasks to server

    session.submit()

    # Wait for completion of task and fetch results and get it as bytes

    result = task.output.fetch().get\_bytes()

    # Prints 'Hello world!'

    print(result)

```

User’s Guide[¶](#user-s-guide "Permalink to this headline")

-----------------------------------------------------------

### Basic terms[¶](#basic-terms "Permalink to this headline")

**Task** is a basic unit of work in Rain, it reads inputs and produces outputs.

Tasks are executed on computational nodes (computers where Rain governors are

running). Tasks can be external programs, python functions, and basic built-in

operations.

**Data objects** are objects that are read and created by tasks. Data objects

are immutable, once they are created they cannot be modified. They are generic

data blobs or directories with accompanying metadata. It is upto tasks to

interpret the data object content.

### Task definition and submission[¶](#task-definition-and-submission "Permalink to this headline")

Rain represents your computation as a graph of tasks and data objects. Tasks are

not eagerly executed during the graph construction. Instead, the actual

execution is managed by Rain infrastructure after an explicit submission. This

leads to a programming model in which you first only **define** a graph and then

**execute** it.

Let us consider the following example, where two constant objects are created

and merged together:

```

from rain.client import Client, tasks, blob

client = Client("localhost", 7210)  # Create a connection to the server

                                    # running at localhost:7210

with client.new\_session() as session:  # Creates a session

    a = blob("Hello ")    # Create a definition of data object in the current session

    b = blob("world!")    # Create a definition of data object in the current session

    tasks.Concat([a, b])  # Create a task definition in the current session

                          # that concatenates input data objects

    session.submit()      # Send the created graph into the server, where the computation

                          # is performed.

    session.wait\_all()    # Wait until all submitted tasks are completed

```

The graph composed in the session looks as follows:

![Example of task graph](_images/helloworld.svg)

When the graph is constructed, all created objects and tasks are put into the

active session. In many cases, it is sufficient just to create one session for

whole program lifetime, with one submit at the end. However, it is possible to

create more sessions or built a graph gradually with more submits. More details

are covered in Section [Sessions](#sessions).

### Fetching data objects[¶](#fetching-data-objects "Permalink to this headline")

Data objects produced by tasks are not transferred back to the client

automatically. If needed, this can be done using the `fetch()` method. It

returns `rain.common.DataInstance` that wraps data together with some

additional information. To get raw bytes from `rain.common.DataInstance`

you can call method `get\_bytes()`.

In the following example, we download the result back to the Python client

code. Expression `t.output` refers to the data object that is the output

of task `t`:

```

from rain.client import Client, tasks, blob

client = Client("localhost", 7210)

with client.new\_session() as session:

    a = blob("Hello ")

    b = blob("world!")

    t = tasks.Concat((a, b))

    t.output.keep()            # Tell server to keep result of task

    session.submit()           # Submit task graph

    result = t.output.fetch()  # Download result from the server

    print(result.get\_bytes())  # Prints b'Hello world!'

```

By default, Rain automatically removes data objects that are no longer needed

for further computation. Method `keep()` sets a flag to a given data object

that instructs the server to keep the object until the client does not

explicitly frees it. An object can be freed when the session is closed or when

`unkeep()` method is called. Method `keep()` may be called only before the

submit. Method `unkeep()` may be called on any “kept” object any time.

If method `fetch()` is called and the object has not been finished yet, the

method blocks until the object is not finished. Note that this is the reason,

why we did not use `wait\_all()` in this example.

### Inter-task dependencies[¶](#inter-task-dependencies "Permalink to this headline")

Naturally, an output of a task may be used as an input for another task. This

is demonstrated by the following example. In the example, we use

`tasks.Sleep(O, T)` that creates a task taking an arbitrary data object `O`

and waits for `T` seconds and then returns `O` as its output. Being aware

that such task is not very useful in practice, we find it useful as an

intuitive example to demostrate the concept of task chaining:

```

from rain.client import Client, tasks, blob

client = Client("localhost", 7210)

with client.new\_session() as session:

    a = blob("Hello ")

    b = blob("world!")

    t1 = tasks.Sleep(b, 1.0)   # Wait for one second and then returns 'b'

    t2 = tasks.Concat((a, t1.output))

    t2.output.keep()

    session.submit()           # Submit task graph

    result = t2.output.fetch()  # It will wait around 1 second

                                # and then returns b'Hello world'

```

If a task produces only a single output, we can ommit `.output` and directly

use the task as an input for another task. In our example, we can define `t2`

as follows:

```

t2 = tasks.Concat((a, t1))

```

This shortened notation is used in the rest of the text.

### More outputs[¶](#more-outputs "Permalink to this headline")

A task may generally create zero, one, or more outputs. All outputs are

accessible via attribute `outputs`. That contains an instance of

`rain.common.LabeledList`. It is an extension of a standard list

(indexed from zero), that also allows to be accessed via string labels.

```

# The following task creates two outputs labeled "output1" and "output2" with

# an equivalent of 'cat data | tee output1 > output2'.

t = tasks.Execute(["tee", Output("output1")], stdout="output2", stdin=data)

t.outputs["output1"]  # Access to output "output1"

t.outputs["output2"]  # Access to output "output2"

# There is also some helper functions:

# Keep all outputs (equivalent to: for o in t.outputs: o.keep())

t.keep\_outputs()

# After submit

# Fetch all outputs (equivalent to: [o.fetch() for o in t.outputs])

t.fetch\_outputs()

```

If a task has more than one output or zero outputs, then accessing attribute

`.output` throws an exception. Attribute `.outputs` is always availble

independently on the number of outputs.

### Object data types[¶](#object-data-types "Permalink to this headline")

Every data object represents either a single binary data blob or a directory.

Since these two object *data types* behave very differently, they are

distinguished and checked already when constructing the computation graph.

The *data type* may be one of:

* ‘blob’ - Binary data block. May have a [Object content types](#content-type) specified.

* ‘dir’ - Directory structure, see section [Directories](#directories).

We consider developing other data object “modes”, e.g. streams.

### Object content types[¶](#object-content-types "Permalink to this headline")

Binary data objecs represent different type of data in different formats.

The Rain infrastructure treats all data objects as raw binary blobs,

and it is up to tasks to interpret them. Content type is a string identifier

of the format of the data in tasks and clients. Python code also recognize

some of content types and allows to deserialize them directly.

Currently recognized content types are:

> 

> * ‘’ - Raw binary data, unknown or unspecified content type

> * ‘pickle’ - Serialized Python object

> * ‘cloudpickle’ - Serialized Python object via Cloudpickle

> * ‘json’ - Object serialized into JSON

> * ‘cbor’ - Object serialized into CBOR

> * ‘arrow’ - Object serialized with Apache Arrow

> * ‘text’ - UTF-8 string.

> * ‘text-<ENCODING>’ - Text with specified encoding

> * ‘mime/<MIME>’ - Content type defined as MIME type

> * ‘user/<TYPE>’ - User defined type, <TYPE> may be arbitrary string

> 

> 

> 

An object may have two different content-types: First, a type is specified

when constructing the task graph. Second, the type may be set by the task

executor dynamically (e.g. depending on some input data).

If present, the latter is taken to be the actual content type and must

be a sub-type of the former.

Any type is considered a subtype of the unspecified type.

### Constant data objects[¶](#constant-data-objects "Permalink to this headline")

Function `rain.client.blob()` serves for a creation of a constant data

object. The content of the data object is uploaded to the server together with

the task graph.

```

from rain.client import blob, pickled

blob(b"Raw data")  # Creates a data object with a defined content

blob(b"Raw data", label="input data")  # Data with a non-default label

                                       # (Default label is 'const')

blob("String data")  # Creates a data object from a string, the content type

                     # is set to 'text'

blob("[1, 2, 3, 4]", content\_type="json")  # Data with a specified content type

blob([1, 2, 3, 4], encode="json")  # Serialize python object to JSON and set

                                   # content type to "json"

blob([1, 2, 3, 4], encode="pickle")  # Serialize python object by pickle

                                     # content type to "pickle"

pickled([1, 2, 3, 4])  # Short-cut for blob(..., encode="pickle")

```

### Built-in tasks[¶](#built-in-tasks "Permalink to this headline")

The following tasks are supported directly by the Rain governor:

*Concat* (`rain.client.tasks.Concat`)

Concatencates inputs into one resulting blob.

*Load*, *LoadDir* (`rain.client.tasks.Load`, `rain.client.tasks.LoadDir`)

Creates data object from an external file or direftory.

(Note: The current version does not support tracking external resources;

therefore, this operation “internalizes” the file, i.e. it makes a copy

of it into the working directory.)

*Store* (`rain.client.tasks.Store`)

Saves data object to a filesystem.

The data are saved into local file system of the governor on which the

task is executed. This task is usually used for saving files to

a distributed file system, hence it does not matter which governor

performs the task.

*Sleep* (`rain.client.tasks.Sleep`)

Task that forwards its input as its

output after a specified delay. Mostly for testing and benchmarking.

*Execute* (`rain.client.tasks.SliceDirectory`)

Run an external program with given inputs, parameters and resources.

See `rain.client.Program` if you execute a program repeatedly

with different data.

*MakeDirectory* (`rain.client.tasks.MakeDirectory`)

Tasks that creates a directory combining the inputs under given paths.

*SliceDirectory* (`rain.client.tasks.SliceDirectory`)

Tasks that extracts a file or subdirectory from a directory object.

```

# This example demonstrates usage of four built-in tasks

from rain.client import tasks, Client, blob

client = Client("localhost", 7210)

with client.new\_session() as session:

    # Create tasks opening an external file

    data1 = tasks.Load("/path/to/data")

    # Create a constant object

    data2 = blob("constant data")

    # Merge two objects

    merge = tasks.Concat((data1, data2))

    # Sleep for 1s

    result = tasks.Sleep(merge, 1.0)

    # Write result into file

    tasks.Store(result, "/path/to/result")

    session.submit()

    session.wait\_all()

```

(Examples for the directory-related tasks are in section [Directories](#directories))

### Running external programs[¶](#running-external-programs "Permalink to this headline")

#### Task `tasks.Execute`[¶](#task-tasks-execute "Permalink to this headline")

The whole functionality is built around built-in task

`rain.client.tasks.Execute`. When a program is executed through

`rain.client.tasks.Execute`, then a new temporary directory is created.

This directory will be removed at the end of program execution. The current

working directory of the program is set to this directory.

The idea is that this directory is program’s sandbox where input data objects

are mapped and files created in this directory may be moved out as new data

objects when computation completes. Therefore, in contrast with many other

workflow systems, programs in rain should not be called with absolute paths in

arguments but use relative paths (to stay in its working directory).

Governors try to avoid unnecessary data object replication in the cases when

a data object is used by multiple tasks that run on the same governor.

If the executed program terminates with a non-zero code, then tasks fails and

content of standard error output is written into the error message.

The simple example looks as follow:

```

tasks.Execute("sleep 1")

```

This creates a task with no inputs and no outputs executing program “sleep”

with argument “1”. Arguments are parsed in shell-like manner.

Arguments can be also specified explicitly as a list:

```

tasks.Execute(("sleep",  "1"))

```

Command may be also interpreted by shell, if the argument `shell=True` is

provided:

```

tasks.Execute("sleep 1 && sleep 1", shell=True)

```

#### Outputs[¶](#outputs "Permalink to this headline")

Files created during task execution or task standard output can be used as the

output of `rain.client.tasks.Execute`. The following example calls program

`wget` that downloads web page at <https://github.com/> and saves it as

`index.html`. The created file is forwarded as the output of the task.

```

from rain.client import Client, task, Output

client = Client("localhost", 7210)

with client.new\_session() as session:

    t = tasks.Execute("wget https://github.com/",

                       output\_paths=[Output("index", path="index.html")])

    t.output.keep()

    session.submit()

    result = t.output.fetch().get\_bytes()

```

The class `rain.client.Output` allows to configure the outputs.

The first argument is the label of the output. The argument `path` sets the

path to the file used as output.

It is a relative path w.r.t. the working directory of the

task. If the path is not defined, then label is used as path; e.g.

`Output("my\_output")` is equivalent to `Output("my\_output",

path="my\_output")`. The Output instance can be also used for specification of

additional attributes such content type or size hint. Please see the class

documentation for more details.

If we do not want to configure the output, it is possible to use just a string

instead of instance of `Output`. It creates the output with the same label

and path as the given string.

Therefore we can create the previous task as follows:

```

t = tasks.Execute("wget https://github.com/", output\_paths=["index.html"])

```

The only difference is that label of the output is now “index.html” (not

“index”).

Of course, more than one output may be specified. Program `wget` allows to

redirect its log to a file through `--output-file` option:

```

t = tasks.Execute("wget https://github.com/ --output-file log",

                  outputs\_paths=["index.html", "log"])

```

This creates a task with two outputs with labels “index.html” and “log”.

The outputs are available using standard syntax, e.g. `t.outputs["log"]`.

Outputs can be also passed directly as program arguments.

This is a shortcut for two actions: passing the output path as an argument

and putting output into `output\_paths`.

The example above can be written as follows:

```

t = tasks.Execute(["wget", "https://github.com/", "--output-file", Output("log")],

                  output\_paths=["index.html"])

```

The argument `stdout` allows to use program’s standard output:

```

# Creates output from stdout labelled "stdout"

tasks.Execute("ls /", stdout=True)

# Creates output from stdout with label "my\_label"

tasks.Execute("ls /", stdout="my\_label")

# Creates output through Output object, argument 'path' is not allowed

tasks.Execute("ls /", stdout=Output("my\_label"))

```

#### Inputs[¶](#inputs "Permalink to this headline")

Data objects can be mapped into the working directory of

`rain.client.tasks()`. The simplest case is to use a data object directly

as arguments for a program. In such case, the data object is mapped into

randomly named file and the name is placed into program arguments.

Note that files are by default mapped only for reading (and proctected by

setting file permissions). More options of mapping is described in

[Mapping data objects onto filesystem](#fs-mappings).

```

from rain.client import Client, task, blob

client = Client("localhost", 7210)

with client.new\_session() as session:

    data = blob(b"It is\nrainy day\n")

    # Maps 'data' into file XXX where is a random name and executes

    # "grep rain XXX"

    task = tasks.Execute(["grep", "rain", data], stdout=True)

    task.output.keep()

    session.submit()

    print(task.output.fetch().get\_bytes())  # Prints b"rainy day"

```

For additional settings and file name control, there is

`rain.client.Input`, that is a counter-part for

`rain.client.Output`. It can be used as follows:

```

from rain.client import Client, task, Input

...

# It executes a program "a-program" with arguments "argument1" and "myfile"

# and while it maps dataobject in variable 'data' into file 'myfile'

my\_data = ... # A data object

task = tasks.Execute(["a-program", "argument1",

                      Input("my\_label", path="myfile", dataobj=my\_data)])

```

The argument `input\_paths` of `rain.client.tasks.Execute` serves to map

a data object into file without putting its filename into the program

arguments:

```

# It executes a program "a-program" with arguments "argument1"

# and while it maps dataobject in variable 'data' into file 'myfile'

tasks.Execute(["a-program", "argument1"],

              input\_paths=[Input("my\_label", path="myfile", dataobj=my\_data)])

```

The argument `stdin` serves to map a data object on the standard input of the

program:

```

# Executes a program "a-program" with argument "argument1" while mapping

# a data object on the standard input

tasks.Execute(["a-program", "argument1"], stdin=my\_data)

```

#### Factory `Program`[¶](#factory-program "Permalink to this headline")

In many cases, we need to call the same program with the same argument set.

Class `rain.client.Program` serves as a factory for

`rain.client.tasks.Execute` for this purpose. An instance of `Program`

can be called as a function that takes data objects; the call creates a task in

the active session.

```

from rain.client import Client, blob, Program, Input

rain\_grep = Program(["grep", "rain", Input("my\_input", path="my\_file")], stdout=True)

client = Client("localhost", 7210)

with client.new\_session() as session:

    data = blob(b"It is\nrainy day\n")

    # Creates a task that executes "grep rain my\_file" where dataobject in variable

    # 'data' is mapped into <FILE>

    task = rain\_grep(my\_input=data)

```

`Program` accepts the same arguments as `execute`, including

`input\_paths`, `output\_paths`, `stdin`, and `stdout`. The only

difference is that in all places where data object could be used, `Input`

instance (without `dataobj` argument) has to be used, since `Program`

defines “pattern” indepedently on a particular session.

### Python tasks[¶](#python-tasks "Permalink to this headline")

In addition to built-in tasks, Rain allows to run additional types of tasks via

executors. Rain is shipped with Python executor, that allows to execute

arbitrary Python code.

#### Decorator `@remote`[¶](#decorator-remote "Permalink to this headline")

Decorator `rain.client.remote()` turns a python function into a

Rain task. Let us consider the following example:

```

from rain.client import Client, remote

@remote()

def hello(ctx):

    return "Hello world!"

client = Client("localhost", 7210)

with client.new\_session() as session:

    t = hello()                # Create a task

    t.output.keep()

    session.submit()

    result = t.output.fetch()

    print(result)              # Prints b'Hello world!'

```

The decorator changes the behavior of the decorated function in a way that

calling it no longer executes it in the client but creates a task that executes

the function in a python executor. Governor starts and manages executors as

necessary, there is no need of any action from the user.

The decorated function should accept at least one argument. As the first

argument, the context of the execution is passed to the function. Context

enables some actions within the task. It is a convention to name this argument

as `ctx`.

#### Inputs[¶](#id1 "Permalink to this headline")

Decorated function may take more parameters than `ctx`; these parameters

define inputs of the task. By default, they can be arbitrary Python objects and

they are serialized via `cloudpickle`. If the decorated function is called

with a data object, it is invokend with `rain.common.DataInstance` that

contains data defined by the object:

```

from rain.client import Client, remote, blob

@remote()

def hello(ctx, data1, data2):

    return data1 + data2.get\_bytes()

client = Client("localhost", 7210)

with client.new\_session() as s:

    # Create data object

    data = blob("Rain!")

    # Creates a task calling function 'hello' in governor

    t = hello(b"Hello ", data)

    t.output.keep()

    s.submit()

    s.wait\_all()

    # Prints b'Hello Rain!"

    print(t.output.fetch().get\_bytes())

```

In remotely executed Python code, Rain data objects are replaced with actual

data instances. All occurences of data objects are replaced, even those

encapsulated in own data structures:

```

class MyClass:

    def \_\_init\_\_(self, my\_data):

        self.my\_data = my\_data

 @remote()

 def my\_call(ctx, input):

     # If we assume a call of this function as below,

     # we obtain an instance of MyClass where attribute 'my\_data'

     # is list of instances of DataInstance

     return b""

 ...

 my\_instance = MyClass([blob(b"data1"), blob(b"data2"), blob(b"data3")])

 task = my\_call(my\_instance)

```

Note

It is possible to pass also generators as arguments to remote functions, and

it works as expected. However, Rain has to include all data objects occuring

in related expressions as task dependencies. Therefore, you may create more

dependencies then expected. To avoid this problems, we recommend to evaluate

generators before passing to remote functions, especiialy if it is a

filtering kind of generator.

All metadata of data objects (including content type) are passed to the data

instances occuring in remote functions. Therefore, it is possible to call

method `load()` on data instances to deserialize objects according to

their content types:

```

@remote()

def fn1(ctx, data):

    # Load according content type. Throws an error if content type is not provided

    loaded\_data = data.load()

    ...

# Automatically call load() on specific argument

@remote(inputs={"data": Input(load=True)})

def fn2(ctx, data):

    ....

# Automatically call load() on all arguments

@remote(auto\_load=True)

def fn3(ctx, data):

    ....

# Example of calling:

data = blob([1,2,3,4], encode="json")

fn1(data)

```

The second case uses `rain.client.Input` to configure individual

parameters. It can be also used for additional configurations, like data-object

size hints for Rain scheduler, or content type specification:

```

# The following function asks for a dataobject with content type "json" as

# its argument. If the function is called the following happens:

# 1) If the input dataobject has content type "json", it is passed as it is

# 2) If the input dataobject has no content type (None), then content type "json"

     is set as the object content type

# 3) If the input dataobject has content type different from "json", the task fails

@remote(inputs={"data": Input(content\_type="json")})

def fn1(ctx, data):

    pass

```

#### Outputs[¶](#id2 "Permalink to this headline")

By default, it is expected that a remote function returns one data object. It

may return an instance of `bytes` or `str` that will be used as content of

the resulting data object. If an instance of bytes is returned then the content

type of resulting object is `None`, if a string is returned then the content

type is set to “text”. A remote function may also return a data instance, when

you want to set additional attributes of data object. More outputs may be

configured via `outputs` attribute of remote:

```

@remote()

def fn1(ctx):

    return b"Returning bytes"

@remote()

def fn2(ctx):

    return "Returning string"

# Configuring more unlabaled outputs

@remote(outputs=3)

def fn3(ctx):

    (b"data1", b"data2", b"data3")

# No output

@remote(outputs=0)

def fn4(ctx):

     pass

# Configuring labeled outputs

@remote(outputs=("label1", "label2"))

def fn5(ctx):

     return {"label1": b"data1", "label2": b"data2"}

# Set content types of resulting objects

@remote(outputs=(Output(content\_type="json"), Output(content\_type="json"))

def fn6(ctx):

    return ("[1, 2, 3]", "{'x': 123}")

# Automatically encode resulting objects

@remote(outputs=(Output(encode="pickle"), Output(encode="json"))

def fn7(ctx):

    return ([1, 2, 3], {"x": 123})

```

#### Debug stream[¶](#debug-stream "Permalink to this headline")

Method `debug` on the context allows to write messages into debug stream that

can be found in task attribute “debug” and it is also part of an error message

when the task fails.

```

@remote()

def remote\_fn(ctx):

    a = 11

    b = a + 10

    ctx.debug("Variable a = {}", a)

    ctx.debug("Variable b = {}", b)

    raise Exception("Error occured!")

# When this task is executed, you get the following error message:

#

# Exception: Error occured!

#

# Debug:

# Variable a = 11

# Variable b = 21

```

#### Type hints[¶](#type-hints "Permalink to this headline")

If you are using sufficiently new Python (>=3.5), you can use type hints

to define outputs and inputs, e.g.:

```

@remote

def test1(ctx, a : Input(content\_type="json")) -> Output(encode='pickle', label='test\_pickle');

    pass

```

### Resources[¶](#resources "Permalink to this headline")

In the current version, the only resource that can be configured is the number

of cpus. This following example shows how to request a a specific number of

cpus for a task:

```

# Reserve 4 CPUs for execution of a program

tasks.Execute("a-parallel-program", cpus=4)

# Resere 4 CPUs for a Python task

@remote(cpus=4)

def myfunction(ctx):

    pass

```

### Attributes ‘spec’ and ‘info’[¶](#attributes-spec-and-info "Permalink to this headline")

Most of the information about the tasks and data objects falls into

two categories:

* The user-created specification data (*spec*).

* The information about the task execution and object computation (*info*).

These are stored and transmitted separately. Once the objects and tasks

are submitted, the spec is immutable. The info is initially empty

and is set by the governor (and in part by the task executor). When

a task or object is finished, info is also immutable.

The data is transmitted as JSON, attributes with values `None`,

empty strings and empty lists may be omitted when encoding.

A client may ask for info attributes of any task/object as long as session

is open; “keep” flag is not necessary. Attributes are not updated

automatically, `fetch()` or `update()` has to be called to update

attributes.

#### Error, debugn and user[¶](#error-debugn-and-user "Permalink to this headline")

The task info and object info share `error` attribute. When non-empty,

the task is assumed to have failed. You may specify `error`

of an object to indicate the error more precisely, but it usually

indicates a failure of the generating task.

Note that empty `error` is assumedto mean success even if explicitly present.

The `debug` attribute is intended for any log messages from Rain or the user,

especially for internal and external debugging. General node progress is

normally not logged here as it is contained in the Rain event log.

This is the only attribute that is not immutable once set and may be appended

to.

Both task and object info and spec have a `user` dictionary intended

for any JSON-serializable data for any purpose. The keys prefixed with `\_`

are used internally in testing and development.

#### Task spec and info[¶](#task-spec-and-info "Permalink to this headline")

Task spec ( ::`rain.common.attributes.TaskSpec` in Python)

has the following attributes:

* `id` - Task ID tuple, type `rain.common.ID`.

* `task\_type` - The task-type identificator (e.g. “executor/method”).

* `config` - Any task-type specific configuration data, JSON-serializable.

* `inputs` - A list of input object IDs and labels as

::`rain.common.attributes.TaskSpecInput`

\* `id` - Input object ID.

\* `label` - Optional label.

* `outputs` - List of output object IDs.

* `resources` - Dictionary with resource specification.

* `user` - Arbitrary user json-serializable attributes.

Task info (::`rain.common.attributes.TaskInfo` in Python)

has the following attributes:

* `error` - Error message. Non-empty error indicates failure.

* `start\_time` - Time the task was started.

* `duration` - Real-time duration in seconds (floating-point number).

* `governor` - The ID of the governor that executed this task.

* `debug` - Debugging log, usually empty.

* `user` - Arbitrary json-serializable objects.

#### Data object spec and info[¶](#data-object-spec-and-info "Permalink to this headline")

Data object spec (::`rain.common.attributes.ObjectSpec` in Python)

has the following attributes:

* `id` - Object ID tuple, type `rain.common.ID`.

* `label` - Label (role) of this output at the generating task.

* `content\_type` - Specified content type name, see [content type](#content-type).

* `data\_type` - Object data type, `"blob"` or `"dir"`.

* `user` - Arbitrary user json-serializable attributes.

Data object info (::`rain.common.attributes.ObjectInfo` in Python)

has the following attributes:

* `error` - Error message. Non-empty error indicates failure.

* `size` - Final size in bytes (approximate for directories).

* `content\_type` - Content type after execution. Note that this must

be a sub-type of `spec.content\_type`.

* `debug` - Debugging log, usually empty.

* `user` - Arbitrary json-serializable objects.

#### Python API[¶](#python-api "Permalink to this headline")

In the client, the attributes are available as `spec` and `info` on

`rain.client.Task` and `rain.client.DataObject`.

An example of fetching and querying the attributes at the client:

```

with client.new\_session() as s:

    task = tasks.Execute("sleep 1")

    s.submit()

    s.wait\_all()

    # Download recent attributes

    task.update()

    # Print name of governor where task was executed

    print(task.info.governor)

```

In the python executor and remote tasks, the object attributes are

available on the input `rain.common.DataInstance`, the

task attributes on the execution context (::`rain.executor.context.Context`).

An example of remote attribute manipulation:

```

@remote()

def attr\_demo(ctx):

   # read user defined attributes

   foo = ctx.spec.user["foo"]

   # setup new "user\_info" attribute

   ctx.info.user["bar"] = [1, 2, foo]

   # Write some debug log

   ctx.debug("Running at governor", ctx.info.governor)

   return b"Result"

with client.new\_session() as session:

    task = attr\_demo()

    task.spec.user["foo"] = 42

    session.submit()

    session.wait\_all()

    task.update()

    # prints: [1, 2, 42]

    print(task.info.user["bar"])

    # prints the debug log

    print(task.info.debug)

```

### Waiting for object(s) and task(s)[¶](#waiting-for-object-s-and-task-s "Permalink to this headline")

Waiting for a completion of a single task/object is done using the `wait()`

method directly on awaited task or data object. Multiple tasks/objects can be

awaited at once using the `wait` method with a set of tasks/obejcts on

session:

```

with client.new\_session() as session:

    a = blob("Hello world")

    t1 = tasks.Sleep(a, 1.0)

    t2 = tasks.Sleep(a, 2.0)

    session.submit()

    t1.wait()  # This blocks until t1 is finished, independently of t2

    t2.output.wait()  # Waits until a data object is not finished

    # The same as two lines above, but since we are doing it at once, it is

    # slightly more efficient

    session.wait([t1, t2.output])

```

Note

Note that in the case of `wait()` (in contrast with `fetch()`), object

does not have to be marked as “kept”.

### Directories[¶](#directories "Permalink to this headline")

Rain allows to use directories in the similar way to blobs. Rain allows to

create directory data objects that can be passed to `tasks.Execute()`, remote

python code, and other places without any differences. There are only two

specific features:

> 

> * If a directory dataobject is mapped to a file system it is mapped as directory

> (not as a file as in the case of blobs).

> * If a directory is viewed as raw bytes (e.g. method `get\_bytes` on data

> instance), tar file is returned.

> 

> 

> 

A data type of an object (blob/directory) is a part of the

task graph and has to be determinated during its construction. To specify it in

places where `Input` and `Output` classes are used, there are classes

`rain.client.InputDir` and `rain.client.OutputDir`.

```

from rain import

from rain.client import Client, tasks, blob, OutputDir, directory

client = Client("localhost", 7210)

with client.new\_session() as session:

    # Creates a directory object from client's local file system

    # Recursively collects all files and directories in /path/to/dir

    d = directory("/path/to/dir")

    # Create blob data objects

    data1 = blob(b"12345")

    data2 = blob(b"67890")

    # Task that creates a directory from data objects

    d2 = tasks.MakeDirectory(tasks.make\_directory([

         ("myfile.txt", data1),  # Map 'data1' as file 'myfile.txt' into directory

         ("adir", d),  # Map directory 'd' as subdir named 'adir'

         ("a/deep/path/x", data2),  # Map 'data2' as a file 'x'; all subdirs on path is created

    ])

    # Task taking a file from a directory data object

    d3 = tasks.SliceDirectory(d2, "a/deep/path/x")

    # Task taking a directory from a directory data object

    # This is indicated by '/' at the end of the path.

    d3 = tasks.SliceDirectory(d2, "a/deep/")

    # Taking directory as outpout of task.execute

    tasks.Execute("git clone https://github.com/substantic/rain",

                  output\_paths=[OutputDir("rain")])

```

### Mapping data objects onto filesystem[¶](#mapping-data-objects-onto-filesystem "Permalink to this headline")

Rain knows two methods of maping a data objects onto filesystem.

* **write** - creates a fresh copy of data objects is created on filesystem that

can be freely modified. Changes of the file is *not* propagated back to data

object.

* **link** - symlink to the internal storage of governor. The user can only read

this data. This method may silently fall back to ‘write’ when governor has no file

system representation of the object.

Task `rain.client.tasks.Execute()` maps files by **link** method.

It can be changed by `write` argument of `Input`:

```

# Let 'obj' contains a data object

# THIS IS INVALID! You cannot modified linked objects

tasks.Execute("echo 'New line' >> myfile", shell=True,

              input\_paths=[Input("myfile", dataobj=obj)])

# This is ok. Writable copy of 'obj' is created.

tasks.Execute("echo 'New line' >> myfile", shell=True,

              input\_paths=[Input("myfile", dataobj=obj, write=True)])

```

Data instance has methods `write(path)` and `link(path)` that performs the

mapping to a given path. They can be used on both in executor and in client.

Let us note that in the current version **link** in the client always falls back

to **write**. Example:

```

@remote()

def my\_remote\_function(ctx, input1):

    input1.write("myfile")  # Writes data into 'myfile' that can be edited

                            # without change of the original object

    input1.link("myfile2")  # Creates a read-only file system representation

                            # of data object

```

Warning

Read-only property in linking method is forced by setting up file rights.

Therefore, as far you do not change permissions of files/directories, you are

proctected against accidental modifications of data objects. If you change

permissions or content of linked data objects, the behavior is undefined. Let

us remind that Rain is designed only for execution of trusted codes.

Obviously this kind of isolation is **not** a protection against malicious

users.

### Sessions[¶](#sessions "Permalink to this headline")

#### Overview[¶](#overview "Permalink to this headline")

The client allows to create one or more sessions. Sessions are the environment

scopes where application create task graphs and submit them into the server.

Sessions follows the following rules:

> 

> * Each client may manage multiple sessions. Tasks and data object in different

> sessions are independent and they may be executed simultaneously.

> * If a client disconnects, all sessions created by the client are terminated,

> i.e. running tasks are stopped and data objects are removed.

> (Persistent sessions are not supported in the current version)

> * If any task in a session fails, the session is labeled as failed, and all

> running tasks in the session are stopped. Any access to tasks/objects in the

> session will throw an exception containing error that caused the problem.

> Destroying the session is the only operation that does not throw the exception.

> Other sessions are not affected.

> 

> 

> 

#### Active session[¶](#active-session "Permalink to this headline")

Rain client maintains a global stack of sessions and `with` block moves a

session on the top of the stack and removes it from the stack when the block

ends. The session on the top of the stack is called *active* session. The

following example demonstrates when a session is active:

```

from rain.client import Client, tasks, blob

client = Client("localhost", 7210)

# no session is active

with client.new\_session() as a:

    # 'a' is active

    with client.new\_session() as b:

        # 'b' is active

        pass

    # 'b' is closed and 'a' is active again

# 'a' is closed and no session is active

```

Tasks and data objects are always created within the scope of active session.

Note

Which session is active is always a local information that only influences

tasks and data objects creation. This information is not propagated to the

server. Submitted tasks are running regardless the session is active or not.

#### Closing session[¶](#closing-session "Permalink to this headline")

Session may be closed manually by calling method `close()`, dropping the

client connection or leaving `with` block. To suppress the last named

behavior you can use the `bind\_only()` method as follows:

```

session = client.new\_session()

with session.bind\_only():

    # 'session' is active

    pass

# 'session' is not active here; however it is NOT closed

```

Once a session is closed, it is pernamently removed from the session stack and

cannot be reused again.

Note

The server holds tasks’ and objects’ metadata (e.g. performance information) as

long as a session is alive. If you use a long living client with many sessions,

sessions should be closed as soon as they are not needed.

#### Multiple submits[¶](#multiple-submits "Permalink to this headline")

The task graph does not have to be submmited at once; multiple submmits may

occur during the lifetime of a session. Data objects from previous submits

may be used while constructing a new new submit, the only condition is that

they have to be marked as “kept” explicitly.

```

with client.new\_session() as session:

   a = blob("Hello world")

   t1 = tasks.Sleep(a, 1.0)

   t1.output.keep()

   session.submit()  # First submit

   t2 = tasks.Sleep(t1.output, 1.0)

   session.submit()  # Second submit

   session.wait\_all()  # Wait until everything is finished

   t3 = tasks.Sleep(t1.output, 1.0)

   session.submit()  # Third submit

   session.wait\_all()  # Wait again until everything is finished

```

Let us remind that method `wait\_all()` waits until all currently running task

are finished, regardless in which submit they arrived to the server.

Writing Own Executors[¶](#writing-own-executors "Permalink to this headline")

-----------------------------------------------------------------------------

![Connection between basic components of Rain](_images/arch.svg)

This section covers how to write a new executor, i.e. how to create a program

that introduces new tasks type to Rain. A governor spawns and stops executors

as needed according tasks that are assigned to it. Each tasks always specifies

what kind of executor it needs.

There are generally two types of executors: **Universal executors** and

**Specialized executors**. The universal one allows to execute an arbitrary code

and specialized offers a fix of tasks that they provide.

The current version of Rain supports universal executor for Python. This is how

`@remote()` decorator works. It serializes a decorated function into a data

object and creates a task that needs Python executor that executes it.

For languages where code cannot be simply transferred in a portable way, Rain

offers **tasklibs**, a libraries for writing specialized executors. The current

version provides tasklibs for C++ and Rust. A tasklib allows to create a

stand-alone program that know how to communicate with governor and provides a

set of functions.

This sections shows how to write new tasks using tasklibs for C++ and Rust and

how to create run this tasks from client.

Note: Governor itself also provides some of basic task types, that are provided

through a virtual executor called **buildin**. You may see this “executor” in

dashboard.

### Rust tasklib[¶](#rust-tasklib "Permalink to this headline")

The documentation for writing executor in Rust can be found at

<https://docs.rs/rain_task/>. Registration of an executor into a governor and

using client API are same for all executors ([Registration in governor](#register-exec) and

[Client API](#task-api)).

### C++ tasklib[¶](#c-tasklib "Permalink to this headline")

Note

C++ tasklib is not fully finished. It allows to write basic task types, but

some of more advanced features (e.g. working with attributes) are not

implemented yet.

#### Getting started[¶](#getting-started "Permalink to this headline")

The following code shows how to create an executor named “example1” that

provides one task type “hello”. This task takes one blob as the input,

and returns one blob as the output.

```

#include <tasklib/executor.h>

int main()

{

  // Create executor, the argument is the name of the executor

  tasklib::Executor executor("example1");

  // Register task "hello"

  executor.add\_task("hello", [](tasklib::Context &ctx, auto &inputs, auto &outputs) {

      // Check that we been called exactly with 1 argument.

      // If not, the error message is set to context

      if (!ctx.check\_n\_args(1)) {

          return;

      }

      // This is body of our task, in our case, it reads the input data object

      // inserts "Hello" before the input and appends "!"

      auto& input1 = inputs[0];

      std::string str = "Hello " + input1->read\_as\_string() + "!";

      // Create new data instance and set it as one (and only) result

      // of the task

      outputs.push\_back(std::make\_unique<tasklib::MemDataInstance>(str));

  });

  // Connect to governor and serve registered tasks

  // This function is never finished.

  executor.start();

}

```

#### Building[¶](#building "Permalink to this headline")

To compile the example we need to creating following file structure:

* myexecutor

	+ myexecutor.cpp – Source code of our example

	+ CMakeFile.txt – CMake configuration file. The content is below.

	+ tasklib – Copy of tasklib from Rain repository (located in `rain/cpp/tasklib`)

Content of `CMakeFile.txt` is following:

```

cmake_minimum_required(VERSION 3.1)

project(myexecutor)

add_subdirectory(tasklib)

add_executable(myexecutor

              myexecutor.cpp)

target_include_directories(myexecutor PUBLIC ${CBOR_INCLUDE_DIRS} ${CMAKE_CURRENT_SOURCE_DIR}/src)

target_link_libraries (myexecutor tasklib ${CBOR_LIBRARIES} pthread)

```

Now, we can build the executor as follows:

```

$ cd myexecutor

$ mkdir _build

$ cd _build

$ cmake ..

$ make

```

### Registration in governor[¶](#registration-in-governor "Permalink to this headline")

When you write your own executors, you have to registrate them in the governor.

For this purpose, you have to create a configuration file for governor.

As an example, let us assume that we want to register called “example1”.

```

[executors.example1]

    command = "/path/to/executor/binary"

```

The configuration is in TOML format. If we save it as `/path/to/config.toml`

we can provide the path to the governor by starting as follows:

```

rain governor <SERVER\_ADDRESS> --config=/path/to/config.toml

```

or if you are using “rain start”:

```

rain start --simple --governor-config=/path/to/config

```

More about starting Rain can be found at [Starting infrastructure](index.html#start-rain).

### Client API[¶](#client-api "Permalink to this headline")

This section describes how to call own tasks from Python API.

Each task contains a string value called `task\_type` that specifies executor

and function. It has format `<EXECUTOR>/<FUNCTION>`. So far we have created

(and registered) own executor called `example1` that provides task `hello`.

The task type is [``](#id1)example1/hello`.

The followig code creates a class `Hello` that serves for calling our task:

```

from rain.client import Task

class Hello(Task):

    """ Task takes one blob as input and puts b"Hello " before

 and "!" after the input. """

    TASK\_TYPE = "example1/hello"

    def \_\_init\_\_(self, obj):

         # Define task with one input and one output,

         # Outputs may be a (labelled) list of data objects or a number.

         # If a number is used than it creates the specified number of blob outputs

         super().\_\_init\_\_(inputs=(obj,), outputs=1)

```

This class can be used to create task in task graph in the same way as tasks

from module `rain.client.tasks`, e.g.:

```

with client.new\_session() as session:

    a = blob("Hello world")

    t = Hello(a)

    session.submit()

    print(t.output.fetch().get\_bytes())  # prints b"Hello WORLD!"

```

Installation, Running & Deployment[¶](#installation-running-deployment "Permalink to this headline")

----------------------------------------------------------------------------------------------------

Rain Distributed Execution Framework consists is an all-in-one binary.

Rain API is a pure Python package with a set of dependencies installable via pip.

### Binaries[¶](#binaries "Permalink to this headline")

Rain provides a binary distribution for Linux/x64. The binary is almost fully

statically linked. The only dynamic dependancies are libc and sqlite3 (for logging

purpose).

Latest release can be found at <https://github.com/substantic/rain/releases>.

It can be downloaded and unpacked as follows:

```

$ wget https://github.com/substantic/rain/releases/download/v0.3.0-pre/rain-v0.3.0-pre-linux-x64.tar.xz

$ tar xvf rain-v0.3.0-pre-linux-x64.tar.xz

```

Installation of Python API:

```

$ pip3 install rain-python

```

### Build via cargo[¶](#build-via-cargo "Permalink to this headline")

If you have installed Rust toolchain, you can use cargo to build

Rust binaries and skip manual download:

```

$ cargo install rain_server

```

Note that you still have to install Python API through pip:

```

$ pip3 install rain-python

```

### Build from sources[¶](#build-from-sources "Permalink to this headline")

For building from sources, you need Rust and SQLite3 (for logging) and Capnp

compiler (for compiling protocol files) installed on your system.

```

# Example for installation of dependencies on Ubuntu

# Installation of latest Rust

$ curl https://sh.rustup.rs -sSf | sh

# Other dependencies

$ sudo apt-get install capnproto libsqlite3-dev

```

For building Rain, run the following commands:

```

$ git clone https://github.com/substantic/rain

$ cd rain

$ cargo build --release

```

After the installation, the final binary can be found `rain/target/relase/rain`.

Installation of Python API:

```

$ cd python

$ python setup.py install

```

### Starting infrastructure[¶](#starting-infrastructure "Permalink to this headline")

#### Starting local governors[¶](#starting-local-governors "Permalink to this headline")

The most simple case is running starting server and one governor with all

resources of the local machine. The following command do all work for you:

```

$ rain start --simple

```

If you want to start more local governors, you can use the following command.

It starts two governors with 4 assigned cpus and one with 2 assigned cpus:

```

$ rain start --local-wokers=[4,4,2]

```

#### Starting remote governors[¶](#starting-remote-governors "Permalink to this headline")

If you have more machines that are reachable via SSH you can use the following

command. We assume that file `my\_hosts` contains addresses of hosts, one per

line:

```

$ rain start --governor-host-file=my_hosts

```

Let us note, that current version assumes that assumes for each host that Rain

is placed in the same directory as on machine from which command is invoked.

If you are running Rain inside PBS scheduler (probably if you are using an HPC

machine), then you can simple run:

```

$ rain start --autoconf=pbs

```

It executes governor on each node allocated by PBS scheduler.

Note

We recommended to reserve one CPU for server unless you have long runnig

tasks. This reservation can be done through cgroups, or CPU pinning.

Another option (with less isolation) is to use option `-S`:

```

$ rain start -S --governor-host-file=my_hosts

```

If a remote machine is actually localhost (and therefore runs Rain server)

then `--cpus=-1` argument is used for the governor on that machine, i.e. the

governor will consider one cpu less on that machine.

#### Starting governors manually[¶](#starting-governors-manually "Permalink to this headline")

If you need a special setup that is not covered by `rain start` you can

simply start server and governors manually:

```

$ rain server                    # Start server

$ rain governor <SERVER-ADDRESS>   # Start governor

```

### Arguments for program *rain*[¶](#arguments-for-program-rain "Permalink to this headline")

#### Synopsis[¶](#synopsis "Permalink to this headline")

```

rain start --simple [--listen=ADDRESS] [--http-listen=ADDRESS]

         [-S] [--runprefix=CMD] [--logdir=DIR] [--workdir=DIR]

         [--governor-config=PATH]

rain start --autoconf=CONF [--listen=ADDRESS] [--http-listen=ADDRESS]

         [-S] [--runprefix=CMD] [--logdir=DIR] [--workdir=DIR]

         [--governor-config=PATH] [--remote-init=COMMANDS]

rain start --local-governors [--listen=ADDRESS] [--http-listen=ADDRESS]

         [-S] [--runprefix=CMD] [--logdir=DIR] [--workdir=DIR]

         [--governor-config=PATH]

rain start --governor-host-file=FILE [-S] [--listen=ADDRESS]

         [--http-listen=ADDRESS]

         [-S] [--runprefix=CMD] [--logdir=DIR] [--workdir=DIR]

         [--governor-config=PATH] [--remote-init=COMMANDS]

rain server [--listen=LISTEN\_ADDRESS] [--http-listen=LISTEN\_ADDRESS]

            [--logdir=DIR] [--ready-file=<FILE>]

rain governor [--cpus=N] [--workdir=DIR] [--logdir=DIR]

            [--ready-file=FILE] [--config=PATH] SERVER\_ADDRESS[:PORT]

rain --version | -v

rain --help | -h

```

#### Command: start[¶](#command-start "Permalink to this headline")

Starts Rain infrastructure (server & governors), makes sure that everything is

ready and terminates.

**–simple**

Starts server and one local governor that gains all resources of the local

machine.

**–autoconf=CONF**

Automatic configuration from the environment. Possible options are:

* *pbs* - If executed in an PBS job, it starts server on current node and

governor on each node.

**–local-governors=RESOURCES**

Start local with a given number of cpus. E.g. –local-governors=[4,4,2]

starts three governors: two with 4 cpus and one with 2 cpus.

**–governor-config=PATH**

Path to governor config. It is passed as –config argument for all governors.

**–governor-host-file=FILE**

Starts local server and remote governors. FILE should be file containing

name of hosts, one per line.

The current version assumes the following of each host:

* SSH server is running.

* Rain is installed in the same directory as on the machine

from which that `rain start` is executed.

**-S**

Serves for reserving a CPU on server node. If remote governor

detects that it is running on the same machine as server then it

is executed with `--cpus=-1`.

The detection is based on checking if the server PID exists on the remote

machine and program name is “rain”.

**–listen=(PORT|ADDRESS|ADDRESS:PORT)**

Set listening address of server. Default is 0.0.0.0:7210.

**–http-listen=(PORT|ADDRESS|ADDRESS:PORT)**

Set listening address of server for HTTP (dashboard). Default is 0.0.0.0:8080.

**–runprefix**

Set a command before rain programs. It is designed to used to run

analytical tools (e.g. –runprefix=”valgrind –tool=callgrind”)

**–logdir=DIR**

The option is unchanged propagated into the server and governors.

**–workdir=DIR**

The option is unchanged propagated into governors.

**–remote-init=COMMAND**

Commands executed on each remote connection. For example:

`--remote-init="export PATH=$PATH:/path/bin"`.

#### Command: server[¶](#command-server "Permalink to this headline")

Runs Rain server.

**–listen=(PORT|ADDRESS|ADDRESS:PORT)**

Set listening address of server. Default is 0.0.0.0:7210.

**–logdir=DIR**

Set logging directory of server. Default is /tmp/rain/logs/server-<HOSTNAME>-PID.

**–ready-file=FILE**

Create file containing a single line “ready”, when the server is fully initialized

and ready to accept connections.

#### Command: governor[¶](#command-governor "Permalink to this headline")

Runs Rain governor.

**SERVER\_ADDRESS[:PORT]**

An address where a server listens. If the port is omitted than port 7210 is

used.

**–config=PATH**

Set a path for a governor config.

**–cpus=N**

Set a number of cpus available to the governor (default: ‘detect’)

* If ‘detect’ is used then the all cores in the machine is used.

* If a positive number is used then value is used as the number of available

cpus.

* If a negative number -X is used then the number of cores is detected and X

is subtracted from this number, the resulting number is used as the number

of available cpus.

**–listen=(PORT|ADDRESS|ADDRESS:PORT)**

Set listening address of governor for governor-to-governor connections. When port is

0 then a open random port is assigned. The default is 0.0.0.0:0.

**–logdir=DIR**

Set the logging directory for the governor. Default is

`/tmp/rain/logs/governor-<HOSTNAME>-<PID>/logs`.

**–ready-file=FILE**

Creates the file containing a single line “ready”, when the governor is

connected to server and ready to accept governor-to-governor connections.

**–workdir=DIR**

Set the working directory where the governor stores intermediate results.

The defautl is `/tmp/rain/work/governor-<HOSTNAME>-<PID>`

Warning

Rain assumes that working directory is placed on a fast device (ideally

ramdisk). Avoid placing workdir on a network file system.

Examples[¶](#examples "Permalink to this headline")

---------------------------------------------------

### Distributed cross-validation with libsvm[¶](#distributed-cross-validation-with-libsvm "Permalink to this headline")

```

# =======================================================

# This example creates a simple cross-validation pipeline

# for libsvm tools over IRIS data set

#

# Requirements:

# 1) Installed svm-train and svm-predict

# (libsvm-tools package on Debian)

# 2) IRIS data set in CSV format, e.g.:

# https://raw.githubusercontent.com/pandas-dev/pandas/master/pandas/tests/data/iris.csv

# =======================================================

import os

from rain.client import Client, tasks, Program, Input, Output, remote

THIS\_DIR = os.path.dirname(os.path.abspath(\_\_file\_\_))

DATA\_FILE = os.path.join(THIS\_DIR, "iris.csv")

CHUNKS = 3

# Convert .csv to libsvm format

@remote()

def convert\_to\_libsvm\_format(ctx, data):

    lines = [line.split(",") for line in data.get\_str().rstrip().split("\n")]

    lines = lines[1:]  # Skip header

    labels = sorted(set(line[-1] for line in lines))

    result = "\n".join("{} 1:{} 2:{} 3:{} 4:{}".format(

        labels.index(line[4]),

        line[0], line[1], line[2], line[3]) for line in lines)

    return result

def main():

    # Program: SVM train

    # svm-train has following usage: svm-train <trainset> <model>

    # It reads <trainset> and creates file <model> with trained model

    train = Program(("svm-train", Input("data"), Output("output")))

    # Porgram: SVM predict

    # svm-predict has following usage: svm-predict <testdata> <model> <prediction>

    # It reads files <testdata> and <model> and creates file with prediction and

    # prints accuracy on standard output

    predict = Program(("svm-predict", Input("testdata"), Input("model"), Output("prediction")),

                    stdout=Output("accuracy"))

    # Connect to rain server

    client = Client("localhost", 7210)

    with client.new\_session() as session:

        # Load data - this is already task, so load is performed on governor

        input\_data = tasks.Load(DATA\_FILE)

        # Convert data - note that the function is marked @remote

        # so it is not executed now, but on a governor

        converted\_data = convert\_to\_libsvm\_format(input\_data)

        # Using unix command "sort" to shuffle dataset

        randomized\_data = tasks.Execute(("sort", "--random-sort", converted\_data), stdout=True)

        # Create chunks via unix command "split"

        chunks = tasks.Execute(("split", "-d", "-n", "l/{}".format(CHUNKS), randomized\_data),

                            output\_files=["x{:02}".format(i) for i in range(CHUNKS)]).outputs

                            # Note that we are taking "outputs" of the task here ==> ^^^^^^^^

        # Make folds

        train\_sets = [tasks.Concat(chunks[:i] + chunks[i+1:]) for i, c in enumerate(chunks)]

        # Train models

        models = [train(data=train\_set) for train\_set in train\_sets]

        # Compute predictions

        predictions = [predict(model=model, testdata=data) for model, data in zip(models, chunks)]

        # Set "keep" flag for "accuracy" output on predictions

        for p in predictions:

            p.outputs["accuracy"].keep()

        # Submit and wait until everything is not completed

        session.submit()

        session.wait\_all()

        # Print predictions

        for p in predictions:

            print(p.outputs["accuracy"].fetch().get\_str())

if \_\_name\_\_ == "\_\_main\_\_":

    main()

```

Python API[¶](#python-api "Permalink to this headline")

-------------------------------------------------------

The Rain Pyton API consists of two domains that observe the

workflow graph differently, although the concepts are similar

and some classes are used in both contexts.

* Code run at *the client*, creating sessions and task graphs,

executing and querying sessions. There, the tasks are only

created and declared, never actually executed.

* Python code that runs inside remote Pyhton tasks on the governors.

This code has access to the actual input data, but only sees the adjacent

data objects (input and output).

### Client API[¶](#client-api "Permalink to this headline")

#### Client[¶](#client "Permalink to this headline")

One instance per connection to a server.

#### Session[¶](#session "Permalink to this headline")

One instance per a constructed graph (possibly with multiple submits).

Tied to one `Client`.

#### Data objects[¶](#data-objects "Permalink to this headline")

Tied to a `Session`.

#### Tasks[¶](#tasks "Permalink to this headline")

Tied to a `Session`.

#### Attributes[¶](#attributes "Permalink to this headline")

#### Input and Output[¶](#input-and-output "Permalink to this headline")

These are helper objects are used to specify task

input and output attributes. In particular, specifying

an `Output` is the preferred way to set properties of the

output `DataObject`.

#### Builtin tasks and external programs[¶](#builtin-tasks-and-external-programs "Permalink to this headline")

Native Rain tasks to be run at the governors.

#### Data instance objects[¶](#data-instance-objects "Permalink to this headline")

Tied to a session and a `DataObject`. Also used in [Remote Python tasks](#sec-remote).

#### Resources[¶](#resources "Permalink to this headline")

Note

TODO: Describe and document task resources.

#### Labeled list[¶](#labeled-list "Permalink to this headline")

### Remote Python tasks[¶](#remote-python-tasks "Permalink to this headline")

API for creating routines to be run at the governors.

Created by the decorating with `remote` (preferred) or

by `Remote`.

Whe **specifying** the remote task in the client code, the relevant classes are

`Remote`, `Input`, `Output`, `RainException`, `RainWarning`, `LabeledList`

and the decorateor `remote`.

**Inside** the running remote task, only

`RainException`, `RainWarning`, `LabeledList`, `DataInstance` and `Context`

are relevant.

The inputs of a `Remote` task are arbitrary python objects

containing a `DataInstance` in place of every `DataObject`,

or loaded data object if `autoload=True` or `load=True` is

set on the `Input`.

The remote should return a list, tuple or `LabeledList` of

`DataInstance` (created by `Context.blob()`), `bytes` or `string`.

Contributors’s Guide[¶](#contributors-s-guide "Permalink to this headline")

---------------------------------------------------------------------------

We welcome contributions of any kind. Do not hesitate to open an GitHub issue or

contant us via email; this part of documentation is quite sparse.

### Scripts[¶](#scripts "Permalink to this headline")

* `utils/checks/stylecheck.sh` – Runs code style checks (flake8 & dry cargo fmt)

* `utils/checks/fullcheck.sh` – Runs stylecheck.sh + all available tests

* `utils/dist/make\_release.py` – Compiles release binary of Rain, creates Python package and

publishes release on GitHub

### Testing[¶](#testing "Permalink to this headline")

Rain contains two sets of tests:

> 

> * Unittests (in Rust)

> * Integration tests (in Python)

> 

> 

> 

#### Python tests[¶](#python-tests "Permalink to this headline")

Python tests are placed in `/rain/tests/pytests`.

To execute them simply run `py.test-3` in the root directory of Rain. The logs

are stored in `rain/tests/pytests/work`.

Important notes:

> 

> * Make sure you are running Python 3 py.test.

> * Working directory `rain/tests/pytests/work` is fully cleaned before every

> test! Therefore, if you want to see logs, make sure that no other test is

> executed after the test you want to see. See options `-x` and `-k` of

> py.test-3

> * By default, Python tests run with rain binary located in

> `rain/target/debug/` directory. This path can be modified using

> RAIN\_TEST\_BIN environment variable.

> 

> 

> 

### Dashboard[¶](#dashboard "Permalink to this headline")

Rain Dashboard is implement in JavaScript over NodeJs. However, we do not want

to have Node.js as a hard dependency when Rain is built from sources. Therefore,

compiled form of Dashboard is included in Rain git repository. Neverthless, if

you want to work on Dashboard, you need to install Node.js.

#### Installation[¶](#installation "Permalink to this headline")

(We assume that you have already installed Node.js.)

```

cd dashboard

npm install

```

#### Development[¶](#development "Permalink to this headline")

For development, just run:

```

npm start

```

It starts on Rain dashboard on port 3000. Now you can just edit dashboard

sources, **without** recompiling Rain binary. Dashboard in the development mode

assumes, that http rain server is running at localhost:8080. If you need, you

can change the address in `dashboard/src/utils/fetch.js`, but do not commit

this change, please.

#### Deployment[¶](#deployment "Permalink to this headline")

All Dashboard resources (including JS source codes) are included into Rain

binary. When Rain is compiled, files in `dashboard/dist` are read. To generate

`dist` directory from actual dashboard sources, you need to run:

```

cd dashboard

sh make\_dist.sh

```

Then you need rebuild Rain (e.g. `cargo build`). When you finish work on

dashboard, do not forget to include files in `dist` into repository.

Indices and tables[¶](#indices-and-tables "Permalink to this headline")

=======================================================================

* [Index](genindex.html)

* [Module Index](py-modindex.html)

* [Search Page](search.html)

