Apr 25
5 min read
CoreDump: The Repl Pattern
Welcome to CoreDump, the one stop for techies, in this dump, the Trio introduces the Repl pattern.
Dump 04.25.2022.18.39
Gecko: The Repl pattern does have security concerns, maybe it is a new era of literate programming, imagine being able to just add the repl pattern, and load and link/compile any repl? Maybe the repl should be interactive?
Trio: With the repl pattern, amazing things are possible, it goes a step ahead from the wall pattern, bringing better collaboration and media to coding!, the generator is similar to the iterable, so generatable code, not native, can be reversed from persistence.
Maybe some examples are in place here, the best part is auto completion and reverse indexed code, say from Google or Black Duck, so the auto completion would say class repl(), with what ever constructor and a menu would popup in the IDE, with functions, maybe FaaS, maybe just source code snippets, and a static analysis tool in the repl pattern, it inherits? You can trace the snippets and repl them, clean verified and interactive.
Say you need a O(1) algorithm for the nth term of some sequence, like the Fibonacci, so you find the snippet in RxPy, and here comes the repl pattern, the code is inherited from the repl class, linked and a window analyzes the code, verified and you commit it! That simple, Dr Bheemaiah calls it green coding …
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
https://github.com/glenjamin/node-fib
"""
import decimal
import functools
import itertools
import math
import os
import sysfrom collections import OrderedDictfrom twisted.internet import reactor, defer, task, utils, threads
from twisted.python import log
from twisted.web import http, resource, server, statictry: import psutil
except ImportError:
psutil = Noneclass FibResource(resource.Resource):
"""Serve nth fibonacci number on /n request.Cancel computations on disconnect.
"""
isLeaf = True
def __init__(self, fib):
resource.Resource.__init__(self)
self.fib = fibdef render_GET(self, request):
def report_timings(request, start=reactor.seconds()):
print >> request, "\n%.0f ms" % (1000*(reactor.seconds() - start),)# parse URL
try: n = int(request.postpath[0])
except (IndexError, ValueError), e:
request.setResponseCode(http.BAD_REQUEST)
return str(e)# find nth fibonacci number
d = defer.maybeDeferred(self.fib, n)
@d.addCallback
def finish_request(nth_fibonacci):
request.write(str(nth_fibonacci))
report_timings(request)
report_process_info(request)
request.finish()
@d.addErrback
def fail_request(reason):
if reason and not reason.check(defer.CancelledError):
# show page with stacktrace
request.processingFailed(reason)
request.notifyFinish().addErrback(lambda _: d.cancel())
return server.NOT_DONE_YETif psutil is not None:
def report_process_info(request, p = psutil.Process(os.getpid())):
"""Report CPU/memory usage of the process."""
for f in [p.get_memory_info, p.get_threads]:
print >> request, f()
else:
report_process_info = lambda *args,**kwargs: Nonedef gcd(a, b):
"""Return GCD (greatest common divisor) for positive integers a,b.>>> gcd(12, 15) == gcd(15, 12) == 3
True
>>> gcd(1, 100)
1
>>> gcd(3, 100)
1
>>> gcd(5, 100)
5
>>> gcd(10, 0)
10
"""
while b:
a, b = b, a % b
return adef cooperator(iterable, n, yield_interval, callback):
"""
- call next(iterable) `n` times
- yield None every `yield_interval` iterations or more times
- call callback with the result of the nth (last) iteration or None>>> import pprint
>>> list(cooperator(range(1, 10), 5, 3, pprint.pprint))
5
[None, None, None, None, None]
>>> list(cooperator(range(1, 10), 7, 3, pprint.pprint))
7
[None, None, None]
"""
f = None
yield_interval = gcd(max(n-1, 1), yield_interval)
for f in itertools.islice(iterable, 0, n, yield_interval):
yield None
callback(f)def gen2deferred(yield_interval=1):
"""Decorator that converts generator to deferred."""
def gen_decorator(func):
@functools.wraps(func)
def wrapper(n):
def stop_task(unused_deferred):
try: t.stop()
except task.TaskFailed: pass
d = defer.Deferred(canceller=stop_task)
t = task.cooperate(cooperator(func(n), n+1,
yield_interval, d.callback))
return d
return wrapper
return gen_decorator@gen2deferred(yield_interval=1000)
def iterfib(n=None, a=0, b=1):
"""Lazely generate fibonacci sequence.O(n) steps, O(n) in memory (to hold the result)NOTE: each step involves bignumbers so it is actually O(n) in time
"""
while 1:
yield a
a, b = b, a+b@gen2deferred(yield_interval=1)
def sicpfib(n):
"""Compute nth fibonacci number. Yielding None during computations.O(log(n)) steps, O(n) in memory (to hold the result)
Function Fib(count)
a ← 1
b ← 0
p ← 0
q ← 1While count > 0 Do
If Even(count) Then
p ← p² + q²
q ← 2pq + q²
count ← count ÷ 2
Else
a ← bq + aq + ap
b ← bp + aq
count ← count - 1
End If
End WhileReturn b
End FunctionSee http://stackoverflow.com/questions/1525521/nth-fibonacci-number-in-sublinear-time/1526036#1526036
"""
a, b, p, q = 1, 0, 0, 1
while n > 0:
yield None
if n % 2 == 0: # even
oldp = p
p = p*p + q*q
q = 2*oldp*q + q*q
n //= 2
else:
olda = a
a = b*q + a*q + a*p
b = b*p + olda*q
n -= 1
yield bdef _recfib(n):
"""
>>> _recfib(10)
55
"""
if n == 0: return 0
if n == 1: return 1
return _recfib(n-2) + _recfib(n-1)def recfib(n):
"""Unmodified blocking alg.NOTE: uninterruptable
O(a**n) steps, O(a**n) memory in a thread
"""
return threads.deferToThread(_recfib, n)def memoize_deferred(threshold):
"""fifo cache limited to threshold items for a function that
returns deferred."""
def setcache(value, n):
if len(cache) == threshold: # limit number of items to threshold
cache.popitem(last=False) # fifo
cache[n] = value
return valuecache = OrderedDict()
def decorator(func):
@functools.wraps(func)
def wrapper(n):
try:
value = cache[n]
return defer.succeed(value)
except KeyError:
d = func(n)
d.addCallback(setcache, n)
return d
return wrapper
return decoratoron_next_tick = lambda f,*a, **kw: task.deferLater(reactor, 0, f, *a, **kw)@memoize_deferred(threshold=3) # remember a few last values
@defer.inlineCallbacks
def memfib(n):
"""Return deferred nth fibonacci number."""
if n == 0: f = 0
elif n == 1: f = 1
else:
a = yield on_next_tick(memfib, n-2)
b = yield on_next_tick(memfib, n-1)
f = a + b
defer.returnValue(f)def binet_decimal(n, precision=None):
"""Calculate nth fibonacci number using Binet's formula.O(1) steps, O(1) in memoryNOTE: uninterruptable
>>> map(binet_decimal, range(10))
['0', '1', '1', '2', '3', '5', '8', '13', '21', '34']
"""
with decimal.localcontext() as cxt:
if precision is not None:
cxt.prec = precision
with decimal.localcontext(cxt) as nested_cxt:
nested_cxt.prec += 2 # increase prec. for intermediate results
sqrt5 = decimal.Decimal(5).sqrt()
f = ((1 + sqrt5) / 2)**n / sqrt5
s = str(+f.to_integral()) # round to required precision
return s
binetfib = binet_decimaldef ndigits_fibn(n):
"""Find number of decimal digits in fib(n)."""
phi = (1 + math.sqrt(5)) / 2
return int(n*math.log10(phi)-math.log10(5)/2)+1def binetfib_exact(n):
"""Call binetfib() with calculated precision.O(1) *bigdecimal steps*, O(n) in memory
"""
return utils.getProcessOutput(sys.executable,
['-c', """import fibonacci as f, sys
sys.stdout.write(f.binet_decimal({n}, f.ndigits_fibn({n})))""".format(n=n)])# from twisted/internet/utils.py
# modified to kill child process on deferred.cancel()
def _callProtocolWithDeferred(protocol, executable, args, env, path, reactor=None):
if reactor is None:
from twisted.internet import reactord = defer.Deferred(canceller=lambda d: p.transport.signalProcess('KILL'))
p = protocol(d)
reactor.spawnProcess(p, executable, (executable,)+tuple(args), env, path)
return d
#HACK: patch t.i.utils
utils._callProtocolWithDeferred = _callProtocolWithDeferred
del _callProtocolWithDeferreddef getFibFactory():
root = resource.Resource()html = '<!doctype html><html><body><ul>'
for f in [iterfib, sicpfib, binetfib, binetfib_exact,
memfib, recfib,
]:
html += '<li><a href="/{f}/17">{f}</a>\n'.format(f=f.__name__)
root.putChild(f.__name__, FibResource(f))
root.putChild('', static.Data(html, 'text/html'))
return server.Site(root)def shutdown(reason, reactor, stopping=[]):
"""Stop the reactor."""
if stopping: return
stopping.append(True)
if reason:
log.msg(reason.value)
reactor.callWhenRunning(reactor.stop)portstr = "tcp:1597"if __name__ == '__main__':
import doctest; doctest.testmod()from twisted.internet import endpoints
log.startLogging(sys.stdout)
endpoint = endpoints.serverFromString(reactor, portstr)
d = endpoint.listen(getFibFactory())
d.addErrback(shutdown, reactor)
reactor.run()
else: # twistd -ny
from twisted.application import strports
from twisted.application.service import Application
application = Application("fibonacci")service = strports.service(portstr, getFibFactory())
service.setServiceParent(application)
Ref: GitHub — zed/txfib: Calculate nth fibonacci number in twisted
We have quite a few functions…..
we have, binetfib O(1) steps, O(1) memory in the reactor thread the same but with fixed precision.
So class Repl, has a class Repl->O.binetfib(), with Repl->O.static_analysis(_functions__[]) for a HOF on the inherited function list , array or any other HOF data structure.
