arrdem/source
1
0
Fork 0
Code Issues 11 Pull requests Projects Releases Packages Wiki Activity

[NO TESTS] WIP

Browse source
This commit is contained in:
Reid 'arrdem' McKenzie 2021-08-03 08:42:19 -06:00
parent 642219d84c
commit 4a46ea97a1
3 changed files with 62 additions and 310 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
projects/flowmetal/NOTES.org
View file

@ -1,4 +0,0 @@
#+TITLE: Notes
https://github.com/Pyrlang/Pyrlang
https://en.wikipedia.org/wiki/Single_system_image

260
projects/flowmetal/README.md
View file

@ -39,10 +39,11 @@ I previously [blogged a bit](https://www.arrdem.com/2019/04/01/the_silver_tower/
I'm convinced that a programming environment based around [virtual resiliency](https://www.microsoft.com/en-us/research/publication/a-m-b-r-o-s-i-a-providing-performant-virtual-resiliency-for-distributed-applications/) is a worthwhile goal (having independently invented it) and worth trying to bring to a mainstream general purpose platform like Python.
Flowmetal is an interpreted language backed by a durable event store.
The execution history of a program is persisted to the durable store as execution precedes.
If an interpretation step fails to persist, it can't have external effects and can be retried or recovered.
The execution history of a program persists to the durable store as execution precedes.
If an interpretation step fails to persist, it can't have external effects.
This is the fundamental insight behind Microsoft AMBROSIA.
The event store also provides Flowmetal's only interface for communicating with external systems.
Other systems can attach to Flowmetal's datastore and send events to and receive them from Flowmetal.
Other systems can attach to Flowmetal's data store and send events to and receive them from Flowmetal.
For instance Flowmetal contains a reference implementation of a HTTP callback connector and of a HTTP request connector.
This allows Flowmetal programs to request that HTTP requests be sent on their behalf, consume the result, and wait for callbacks.
@ -57,9 +58,9 @@ A Flowmetal setup could look something like this -
^ ^
| |
v v
+-----------------------+ +------------------------+
| HTTP server connector | | HTTP request connector |
+-----------------------+ +------------------------+
+-----------------------+ +------------------------+
| HTTP server connector | | HTTP request connector |
+-----------------------+ +------------------------+
^ ^
| |
v v
@ -74,253 +75,6 @@ A Flowmetal setup could look something like this -
+--------------------------+
```
## Example - Await
A common pattern working in distributed environments is to want to request another system perform a job and wait for its results.
There are lots of parallels here to making a function or RPC call, except that it's a distributed system with complex failure modes.
In a perfect world we'd want to just write something like this -
```python
#!/usr/bin/env python3.10
from service.client import Client
CLIENT = Client("http://service.local", api_key="...")
job = client.create_job(...)
result = await job
# Do something with the result
```
There's some room for variance here around API design taste, but this snippet is probably familiar to many Python readers.
Let's think about its failure modes.
First, that `await` is doing a lot of heavy lifting.
Presumably it's wrapping up a polling loop of some sort.
That may be acceptable in some circumstances, but it really leaves to the client library implementer the question of what an acceptable retry policy is.
Second, this snippet assumes that `create_job` will succeed.
There won't be an authorization error, or a network transit error, or a remote server error or anything like that.
Third, there's no other record of whatever `job` is.
If the Python interpreter running this program dies, or the user gets bored and `C-c`'s it or the computer encounters a problem, the job will be lost.
Maybe that's OK, maybe it isn't.
But it's a risk.
Now, let's think about taking on some of the complexity needed to solve these problems ourselves.
### Retrying challenges
We can manually write the retry loop polling a remote API.
``` python
#!/usr/bin/env python3.10
from datetime import datetime, timedelta
from service.client import Client
CLIENT = Client("http://service.local", api_key="...")
AWAIT_TIMEOUT = timedelta(minutes=30)
POLL_TIME = timedelta(seconds=10)
def sleep(duration=POLL_TIME):
"""A slightly more useful sleep. Has our default and does coercion."""
from time import sleep
if isinstance(duration, timedelta):
duration = duration.total_seconds()
sleep(duration)
# Create a job, assuming idempotence
while True:
try:
job = client.create_job(...)
start_time = datetime.now()
break
except:
sleep()
# Waiting for the job
while True:
# Time-based timeout
if datetime.now() - start_time > AWAIT_TIMEOUT:
raise TimeoutError
# Checking the job status, no backoff linear polling
try:
if not job.complete():
continue
except:
sleep()
continue
# Trying to read the job result, re-using the retry loop & total timeout machinery
try:
result = job.get()
break
except:
sleep()
continue
# Do something with the result
```
We could pull [retrying](https://pypi.org/project/retrying/) off the shelf and get some real mileage here.
`retrying` is a super handy little library that provides the `@retry` decorator, which implements a variety of common retrying concerns such as retrying N times with linear or exponential back-off, and such.
It's really just the `while/try/except` state machine we just wrote a couple times as a decorator.
``` python
#!/usr/bin/env python3.10
from datetime import datetime, timedelta
from retrying import retry
from service.client import Client
CLIENT = Client("http://service.local", api_key="...")
AWAIT_TIMEOUT = timedelta(minutes=30)
POLL_TIME = timedelta(seconds=10)
class StillWaitingException(Exception):
"""Something we can throw to signal we're still waiting on an external event."""
@retry(wait_fixed=POLL_TIME.total_milliseconds())
def r_create_job(client):
"""R[eliable] create job. Retries over exceptions forever with a delay. No jitter."""
return client.create_job()
@retry(stop_max_delay=AWAIT_TIMEOUT.total_milliseconds(),
wait_fixed=POLL_TIME.total_milliseconds())
def r_get_job(job):
"""R[eliable] get job. Retries over exceptions up to a total time with a delay. No jitter."""
if not job.complete():
raise StillWaitingException
return job.get()
job = r_create_job(client)
result = r_get_job(job)
# Do something with the result
```
That's pretty good!
We've preserved most of our direct control over the mechanical retrying behavior, we can tweak it or choose a different provider.
And we've managed to get the syntactic density of the original `await` example back ... almost.
This is where Python's lack of an anonymous function block syntax and other lexical structures becomes a sharp limiter.
In another language like Javascript or LUA, you could probably get this down to something like -
``` lua
-- retry is a function of retrying options to a function of a callable to retry
-- which returns a zero-argument callable which will execute the callable with
-- the retrying behavior as specified.
client = Client("http://service.local", api_key="...")
retry_config = {} -- Fake, obviously
with_retry = retry(retry_config)
job = with_retry(
funtion ()
return client.start_plan(...)
end)()
result = with_retry(
function()
if job.complete() then
return job.get()
end
end)()
```
The insight here is that the "callback" function we're defining in the Python example as `r_get_job` and soforth has no intrinsic need to be named.
In fact choosing the arbitrary names `r_get_job` and `r_create_job` puts more load on the programmer and the reader.
Python's lack of block anonymous procedures precludes us from cramming the `if complete then get` operation or anything more complex into a `lambda` without some serious syntax crimes.
Using [PEP-0342](https://www.python.org/dev/peps/pep-0342/#new-generator-method-send-value), it's possible to implement arbitrary coroutines in Python by `.send()`ing values to generators which may treat `yield` statements as rvalues for receiving remotely sent inputs.
This makes it possible to explicitly yield control to a remote interpreter, which will return or resume the couroutine with a result value.
Microsoft's [Durable Functions](https://docs.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-overview?tabs=python) use exactly this behavor to implement durable functions.
The "functions" provided by the API return sentinels which can be yielded to an external interpreter, which triggers processing and returns control when there are results.
This is [interpreter effect conversion pattern (Extensible Effects)](http://okmij.org/ftp/Haskell/extensible/exteff.pdf) as seen in Haskell and other tools; applied.
``` python
import azure.functions as func
import azure.durable_functions as df
def orchestrator_function(context: df.DurableOrchestrationContext):
x = yield context.call_activity("F1", None)
y = yield context.call_activity("F2", x)
z = yield context.call_activity("F3", y)
result = yield context.call_activity("F4", z)
return result
main = df.Orchestrator.create(orchestrator_function)
```
Now it would seem that you could "just" automate doing rewriting that to something like this -
``` python
@df.Durable
def main(ctx):
x = context.call_activity("F1", None)
y = context.call_activity("F2", x)
z = context.call_activity("F3", y)
return context.call_activity("F4", z)
```
There's some prior art for doing this (https://eigenfoo.xyz/manipulating-python-asts/, https://greentreesnakes.readthedocs.io/en/latest/manipulating.html#modifying-the-tree) but it's a lot of legwork for not much.
There are also some pretty gaping correctness holes in taking the decorator based rewriting approach;
how do you deal with rewriting imported code, or code that's in classes/behind `@property` and other such tricks?
Just not worth it.
Now, what we _can_ do is try to hijack the entire Python interpreter to implement the properties/tracing/history recording we want there.
The default cpython lacks hooks for doing this, but we can write a python-in-python interpreter and "lift" the user's program into an interpreter we control, which ultimately gets most of its behavior "for free" from the underlying cpython interpreter.
There's [an example](https://github.com/pfalcon/pyastinterp) of doing this as part of the pycopy project; although there it's more of a Scheme-style proof of metacircular self-hosting.
There's a modified copy of the astinterp in `scratch/` which is capable of running a considerable subset of py2/3.9 to the point of being able to source-import many libraries including `requests` and run PyPi sourced library code along with user code under hoisted interpretation.
It doesn't support coroutines/generators yet, and there's some machinery required to make it "safe" (meaningfully single-stepable; "fix"/support eval, enable user-defined import/`__import__` through the lifted python VM) but as a proof of concept of a lifted VM I'm genuinely shocked how well this works.
Next questions here revolve around how to "snapshot" the state of the interpreter meaningfully, and how to build a replayable interpreter log.
There are some specific challenges around how Python code interacts with native C code that could limit the viability of this approach, but at the absolute least this fully sandboxed Python interpreter could be used to implement whatever underlying magic could be desired and restricted to some language subset as desired.
The goal is to make something like this work -
``` python
from df import Activity
f1 = Activity("F1")
f2 = Activity("F2")
f3 = Activity("F3")
f4 = Activity("F4")
def main():
return f4(f3(f2(f1(None))))
```
Which may offer a possible solution to the interpreter checkpointing problem - only checkpoint "supported" operations.
Here the `Activity().__call__` operation would have special support, as with `datetime.datetime.now()` and controlling `time.sleep()`, threading and possibly `random.Random` seeding which cannot trivially be made repeatable.
### Durability challenges
FIXME - manually implementing snapshotting and recovery is hard
### Leverage with language support
FIXME - What does a DSL that helps with all this look like?
## License
Mirrored from https://git.arrdem.com/arrdem/flowmetal

68
projects/flowmetal/scratch/astinterp.py
View file

@ -152,15 +152,12 @@ class InterpFuncWrap:
return self.interp.call_func(self.node, self, *args, **kwargs)
# Python don't fully treat objects, even those defining __call__() special
# method, as a true callable. For example, such objects aren't automatically
# converted to bound methods if looked up as another object's attributes.
# As we want our "interpreted functions" to behave as close as possible to
# real functions, we just wrap function object with a real function. An
# alternative might have been to perform needed checks and explicitly
# bind a method using types.MethodType() in visit_Attribute (but then maybe
# there would be still other cases of "callable object" vs "function"
# discrepancies).
# Python don't fully treat objects, even those defining __call__() special method, as a true callable. For example, such
# objects aren't automatically converted to bound methods if looked up as another object's attributes. As we want our
# "interpreted functions" to behave as close as possible to real functions, we just wrap function object with a real
# function. An alternative might have been to perform needed checks and explicitly bind a method using
# types.MethodType() in visit_Attribute (but then maybe there would be still other cases of "callable object" vs
# "function" discrepancies).
def InterpFunc(fun):
def func(*args, **kwargs):
return fun.__call__(*args, **kwargs)
@ -203,21 +200,19 @@ class ModuleInterpreter(StrictNodeVisitor):
self.system = system
self.fname = fname
self.ns = self.module_ns = ModuleNS(node)
# Call stack (in terms of function AST nodes).
self.call_stack = []
# To implement "store" operation, we need to arguments: location and
# value to store. The operation itself is handled by a node visitor
# (e.g. visit_Name), and location is represented by AST node, but
# there's no support to pass additional arguments to a visitor
# (likely, because it would be a burden to explicit pass such
# additional arguments thru the chain of visitors). So instead, we
# store this value as field. As interpretation happens sequentially,
# there's no risk that it will be overwritten "concurrently".
# To implement "store" operation, we need to arguments: location and value to store. The operation itself is
# handled by a node visitor (e.g. visit_Name), and location is represented by AST node, but there's no support
# to pass additional arguments to a visitor (likely, because it would be a burden to explicit pass such
# additional arguments thru the chain of visitors). So instead, we store this value as field. As interpretation
# happens sequentially, there's no risk that it will be overwritten "concurrently".
self.store_val = None
# Current active exception, for bare "raise", which doesn't work
# across function boundaries (and that's how we have it - exception
# would be caught in visit_Try, while re-rasing would happen in
# visit_Raise).
# Current active exception, for bare "raise", which doesn't work across function boundaries (and that's how we
# have it - exception would be caught in visit_Try, while re-rasing would happen in visit_Raise).
self.cur_exc = []
def push_ns(self, new_ns):
@ -685,13 +680,10 @@ class ModuleInterpreter(StrictNodeVisitor):
if func is builtins.super and not args:
if not self.ns.parent or not isinstance(self.ns.parent, ClassNS):
raise RuntimeError("super(): no arguments")
# As we're creating methods dynamically outside of class, super()
# without argument won't work, as that requires __class__ cell.
# Creating that would be cumbersome (Pycopy definitely lacks
# enough introspection for that), so we substitute 2 implied
# args (which argumentless super() would take from cell and
# 1st arg to func). In our case, we take them from prepared
# bookkeeping info.
# As we're creating methods dynamically outside of class, super() without argument won't work, as that
# requires __class__ cell. Creating that would be cumbersome (Pycopy definitely lacks enough introspection
# for that), so we substitute 2 implied args (which argumentless super() would take from cell and 1st arg to
# func). In our case, we take them from prepared bookkeeping info.
args = (self.ns.parent.cls, self.ns["self"])
return func(*args, **kwargs)
@ -826,11 +818,9 @@ class ModuleInterpreter(StrictNodeVisitor):
if isinstance(node.ctx, ast.Load):
res = NO_VAR
ns = self.ns
# We always lookup in the current namespace (on the first
# iteration), but afterwards we always skip class namespaces.
# Or put it another way, class code can look up in its own
# namespace, but that's the only case when the class namespace
# is consulted.
# We always lookup in the current namespace (on the first iteration), but afterwards we always skip class
# namespaces. Or put it another way, class code can look up in its own namespace, but that's the only case
# when the class namespace is consulted.
skip_classes = False
while ns:
if not (skip_classes and isinstance(ns, ClassNS)):
@ -843,8 +833,10 @@ class ModuleInterpreter(StrictNodeVisitor):
if res is NONLOCAL:
ns = self.resolve_nonlocal(node.id, ns.parent)
return ns[node.id]
if res is GLOBAL:
res = self.module_ns.get(node.id, NO_VAR)
if res is not NO_VAR:
return res
@ -852,26 +844,34 @@ class ModuleInterpreter(StrictNodeVisitor):
return getattr(builtins, node.id)
except AttributeError:
raise NameError("name '{}' is not defined".format(node.id))
elif isinstance(node.ctx, ast.Store):
res = self.ns.get(node.id, NO_VAR)
if res is GLOBAL:
self.module_ns[node.id] = self.store_val
elif res is NONLOCAL:
ns = self.resolve_nonlocal(node.id, self.ns.parent)
ns[node.id] = self.store_val
else:
self.ns[node.id] = self.store_val
elif isinstance(node.ctx, ast.Del):
res = self.ns.get(node.id, NO_VAR)
if res is NO_VAR:
raise NameError("name '{}' is not defined".format(node.id))
elif res is GLOBAL:
del self.module_ns[node.id]
elif res is NONLOCAL:
ns = self.resolve_nonlocal(node.id, self.ns.parent)
del ns[node.id]
else:
del self.ns[node.id]
else:
raise NotImplementedError
@ -937,9 +937,11 @@ class InterpreterSystem(object):
mod = self.load(f)
self.modules[name] = mod.ns
break
elif os.path.isfile(e):
# FIXME (arrdem 2021-05-31)
raise RuntimeError("Import from .zip/.whl/.egg archives aren't supported yet")
else:
self.modules[name] = __import__(name, globals, locals, fromlist, level)