これがこの問題に対する私の見解です。ユーザースクリプトをバニラCPython内で実行する必要があるということは、ミニ言語用のインタープリターを作成するか、Pythonバイトコードにコンパイルして（またはPythonをソース言語として使用して）実行する前にバイトコードを「サニタイズ」する必要があることを意味します。
ユーザーがPythonでスクリプトを記述でき、解析ツリーから安全でない構文をフィルタリングしたり、安全でないオペコードを削除したりすることで、ソースとバイトコードを十分にサニタイズできるという前提に基づいて、簡単な例を示しました。バイトコード。
ソリューションの2番目の部分では、ユーザースクリプトのバイトコードがウォッチドッグタスクによって定期的に中断される必要があります。これにより、ユーザースクリプトがオペコードの制限を超えないようにし、これらすべてをバニラCPythonで実行します。
私の試みの要約。これは主に問題の2番目の部分に焦点を当てています。
- ユーザースクリプトはPythonで書かれています。
- バイトプレイを使用して、バイトコードをフィルタリングおよび変更します。
- ユーザーのバイトコードをインストルメント化してオペコードカウンターを挿入し、コンテキストがウォッチドッグタスクに切り替わる関数を呼び出します。
- グリーンレットを使用してユーザーのバイトコードを実行し、ユーザーのスクリプトとウォッチドッグコルーチンを切り替えます。
- ウォッチドッグは、エラーを発生させる前に実行できるオペコードの数に事前設定された制限を適用します。
うまくいけば、これは少なくとも正しい方向に進むでしょう。あなたがそれに到達したとき、私はあなたの解決策についてもっと聞きたいです。
のソースコードlowperf.py：
# std
import ast
import dis
import sys
from pprint import pprint
# vendor
import byteplay
import greenlet
# bytecode snippet to increment our global opcode counter
INCREMENT = [
(byteplay.LOAD_GLOBAL, '__op_counter'),
(byteplay.LOAD_CONST, 1),
(byteplay.INPLACE_ADD, None),
(byteplay.STORE_GLOBAL, '__op_counter')
]
# bytecode snippet to perform a yield to our watchdog tasklet.
YIELD = [
(byteplay.LOAD_GLOBAL, '__yield'),
(byteplay.LOAD_GLOBAL, '__op_counter'),
(byteplay.CALL_FUNCTION, 1),
(byteplay.POP_TOP, None)
]
def instrument(orig):
"""
Instrument bytecode. We place a call to our yield function before
jumps and returns. You could choose alternate places depending on
your use case.
"""
line_count = 0
res = []
for op, arg in orig.code:
line_count += 1
# NOTE: you could put an advanced bytecode filter here.
# whenever a code block is loaded we must instrument it
if op == byteplay.LOAD_CONST and isinstance(arg, byteplay.Code):
code = instrument(arg)
res.append((op, code))
continue
# 'setlineno' opcode is a safe place to increment our global
# opcode counter.
if op == byteplay.SetLineno:
res += INCREMENT
line_count += 1
# append the opcode and its argument
res.append((op, arg))
# if we're at a jump or return, or we've processed 10 lines of
# source code, insert a call to our yield function. you could
# choose other places to yield more appropriate for your app.
if op in (byteplay.JUMP_ABSOLUTE, byteplay.RETURN_VALUE) \
or line_count > 10:
res += YIELD
line_count = 0
# finally, build and return new code object
return byteplay.Code(res, orig.freevars, orig.args, orig.varargs,
orig.varkwargs, orig.newlocals, orig.name, orig.filename,
orig.firstlineno, orig.docstring)
def transform(path):
"""
Transform the Python source into a form safe to execute and return
the bytecode.
"""
# NOTE: you could call ast.parse(data, path) here to get an
# abstract syntax tree, then filter that tree down before compiling
# it into bytecode. i've skipped that step as it is pretty verbose.
data = open(path, 'rb').read()
suite = compile(data, path, 'exec')
orig = byteplay.Code.from_code(suite)
return instrument(orig)
def execute(path, limit = 40):
"""
This transforms the user's source code into bytecode, instrumenting
it, then kicks off the watchdog and user script tasklets.
"""
code = transform(path)
target = greenlet.greenlet(run_task)
def watcher_task(op_count):
"""
Task which is yielded to by the user script, making sure it doesn't
use too many resources.
"""
while 1:
if op_count > limit:
raise RuntimeError("script used too many resources")
op_count = target.switch()
watcher = greenlet.greenlet(watcher_task)
target.switch(code, watcher.switch)
def run_task(code, yield_func):
"This is the greenlet task which runs our user's script."
globals_ = {'__yield': yield_func, '__op_counter': 0}
eval(code.to_code(), globals_, globals_)
execute(sys.argv[1])
これがサンプルのユーザースクリプトuser.pyです：
def otherfunc(b):
return b * 7
def myfunc(a):
for i in range(0, 20):
print i, otherfunc(i + a + 3)
myfunc(2)
実行例は次のとおりです。
% python lowperf.py user.py
0 35
1 42
2 49
3 56
4 63
5 70
6 77
7 84
8 91
9 98
10 105
11 112
Traceback (most recent call last):
File "lowperf.py", line 114, in <module>
execute(sys.argv[1])
File "lowperf.py", line 105, in execute
target.switch(code, watcher.switch)
File "lowperf.py", line 101, in watcher_task
raise RuntimeError("script used too many resources")
RuntimeError: script used too many resources