Closures and functional programming

How to use the @closure decorator for stream processing

TinyChain lets developers write code that mostly looks like regular Python, but executes in a distributed concurrent runtime. One important difference comes up when iterating over a Map, Tuple, or Stream (analogous to a Python dict, tuple, or generator). TinyChain handles these cases using functional programming with the filter, fold, for_each, and map methods. An easy example is using map to create a new Tuple based on an existing Tuple:

import os
import tinychain as tc

HOST = tc.host.Host(os.getenv("TC_HOST", "http://127.0.0.1:8702"))
ENDPOINT = "/transact/hypothetical"

# initialize a new execution context
cxt = tc.Context()

# instantiate a Tuple
cxt.tuple = tc.Tuple([1, 2, 3])

@tc.get_op
def pow(x: tc.Number):
    return x**2

# create a new Tuple by squaring the elements in `cxt.Tuple`
cxt.raised = cxt.tuple.map(pow)

if __name__ == "__main__":
    # check that the implementation works as expected
    assert HOST.post(ENDPOINT, cxt) == [1, 4, 9]

Often in these cases it's necessary to reference some state in the calling context in the function applied to the stream. You can do this using a closure:

# ...
cxt.tuple = tc.Tuple([1, 2, 3])
cxt.exponent = 2

# capture `cxt.exponent` from the outer context
@tc.closure(cxt.exponent)
@tc.get_op
def pow(x: tc.Number):
    return x**cxt.exponent

cxt.raised = cxt.tuple.map(pow)
# ...

Collection types like Table and Tensor all support copying from and into a Stream. For example, you can create a Tensor by reading fields from a Table:

# ...
# initialize a new execution context
cxt = tc.Context()

# initialize a new table
key = [tc.Column("order_id", tc.U64)]
values = [tc.Column("price", tc.U64)]
schema = tc.table.Schema(key, values)
cxt.table = tc.table.Table(schema)

# add a row
cxt.place_order = cxt.table.insert([1], [499])

# convert to a tensor
schema = [[1], tc.U64]
cxt.prices = tc.tensor.Dense.copy_from(schema, cxt.table.select(["price"]).rows())
cxt.result = tc.After(cxt.place_order, cxt.prices)

if __name__ == "__main__":
    # check that the implementation works as expected
    print(HOST.post(ENDPOINT, cxt))

Another common use-case for Stream is in place of a for loop. For example:

cxt = tc.Context()
cxt.tensor = tc.tensor.Dense.constant([3], 1)

@tc.closure(cxt.tensor)
@tc.get_op
def pow(i: tc.UInt):
    return cxt.tensor[i].write(i**2)

cxt.update = tc.After(tc.Stream.range(cxt.tensor.size).for_each(pow), cxt.tensor)

Examples

You can find more complex examples of functional programming in the codebase:

• the Tensor client tests

• Table