Hijacking Function Calls for Durability
By Miki Tebeka
In the previous post we discussed how to intercept function calls in the user Python code and transform them. For example, the following code:
event = ml.enrich(event)is transformed into:
event = ml.enrich(event)event = _ak_call(ml.enrich, event)
In this blog post, we’re going to look at how _ak_call works.
Durable Execution
AutoKitteh transforms function calls into temporal activities.
_ak_call(ml.enrich, event) causes four RPC calls, as shown in the diagram below.
That’s a lot of overhead for a function call, but it allows AutoKitteh to replay a function that were already executed in a failing workflow. The inner blue part in the diagram, is not called on replay, AutoKitteh returns the stored value from the initial call to ml.enrich(event).
Converting a function call into an RPC call poses several challenges, in the blog post I’ll cover serialization and error handling.
When you call a function, the arguments and the return values are in memory and can be any Python object. When you make an RPC call, you need to serialize the arguments to the function and its return (or error) value.
There are many serialization formats, from JSON to protocol buffers and many, many more. Since we’re using the serialization only in Python, we decided to use pickle which is a built-in serialization format that supports a wide variety of Python types.
But, there are still a lot of types that pickle can’t handle. For example threading.Lock, open files, lambdas and more. AutoKitteh solves this in two ways: Avoiding serialization altogether and an autokitteh.activity decorator.
When AutoKitteh’s Python runner starts an _ak_call flow, it knows what the function’s arguments are. AutoKitteh stores these values in memory and when it receives the Execute call, it gets these arguments from the memory. This way, we avoid serializing the function call. We cannot do that same for the return value, since AutoKitteh stores it for a replay.
Let’s say the user wrote this code:
from urllib.request
def on_event(event):
with urlopen(event.url) as fp:
data = fp.read()
print(data)
The call to urlopen becomes an activity, but the return value of urlopen cannot be pickled, and this code will fail. The solution is to use the autokitteh.activity decorator.
from urllib.request
import autokitteh
def on_event(event):
data = read_url(event.url)
print(data)
@autokitteh.activity
def read_url(url):
with urlopen(url) as fp:
return fp.read()
Even though read_url is a local function, AutoKitteh will run it as an activity since it’s marked with the autokitteh.activity decorator. The return value from read_url is bytes which can be pickled.
Warning:
Pickle is simple and powerful, but can be tricky at times and is also a security risk. We’re OK with the security risk since we’re sending messages internally. And, we still encounter some fun issues such as exceptions who can be pickled but not unpickled from time to time.
Error Handling
Error handling is hard in regular code, more so in RPC.
Consider the following workflow:
def on_event(event):
try:
event = enrich(event)
except KeyError as err:
pass
print(event)
@autokitteh.activity
def enrich(event):
event = event.copy()
event["name"] = "{first} {last}".format(**event)
return eventIn the regular run of the code, if the event does not have first or last keys, enrich will raise KeyError. on_event will catch the exception and will print the event.
However, AutoKitteh changes the flow of execution to several RPC calls and the exception from enrich happens when AutoKitteh calls Execute. When AutoKitteh calls ActivityReply, we want to raise the exception in the user code.
To do that, we use concurrent.fututres.Future, it allows us to block the execution of on_event until there’s a call to ActivityReply.
Here’s a simplified version of the code:
import pickle
from collections import namedtuple
from concurrent.futures import Future
Call = namedtuple('Call', 'fn args kw fut')
Result = namedtuple('Result', 'value error')
class RunnerError(Exception):
pass
class Runner:
def __init__(self, ak_client):
self.ak_client = ak_client
self.call = None
def activity(self, fn, args, kw):
if self.call is not None:
raise RunnerError('nested activity')
fut = Future()
self.call = Call(fn, args, kw, fut)
self.ak_client.Activity() # Async RPC call
return fut.result() # Block until we get the answer or error
def execute(self):
"""Called from AutoKitteh"""
if self.call is None:
raise RunnerError('no active call')
fn, args, kw, _ = self.call
value = error = None
try:
value = fn(*args, **kw)
except Exception as err:
error = err
result = Result(value, error)
return pickle.dumps(result)
def activity_reply(self, data):
"""Called from AutoKitteh"""
if self.call is None:
raise RunnerError('no active call')
call = self.call
self.call = None
result = pickle.loads(data)
if result.error:
call.fut.set_exception(result.error)
else:
call.fut.set_result(result.value)
On a call to activity, we create a Future, do an async call to AutoKitteh Activity and the block on fut.result(). On call to execute from AutoKitteh, the runner returns a Result object which contains both the return value and a possible exception. Then AutoKitteh calls activity_reply with this result. If there’s an error – it sets it in fut.set_exception, otherwise we set the result using fut.set_result.
The current challenge we’re facing is that pickle does not store the exception traceback which is found in err.__traceback__, and we’d like to restore it. But traceback objects are not pickleable, we might need to write our own. If you’re thinking: “Why not save the exception in memory like you do to the Call?” The answer is that activity_reply might come from a replay, and then the exception won’t be in memory.
Conclusion
Running modified user code while trying to keep the original semantics is hard. Using RPC adds another level of complexity. In this blog post we’ve shown some of the challenges we face and how we solved them. Things might change in the future as we encounter more real-life user code. If you want to help us – head over to https://autokitteh.com and run some workflows.
