Как безопасно запустить ненадежный кусок кода?

Question

Предположим, вы работаете с каким-то сложным фрагментом кода, которому вы не можете доверять. Есть ли способ безопасно выполнить его, не потеряв контроль над сценарием?

Примером может служить функция, которая работает только иногда и может случайно / эффектно завершаться с ошибкой. Как вы можете повторить попытку, пока она не заработает? Я попытался взломать с использованием модуля потоков, но не смог аккуратно уничтожить зависший поток.

#!/usr/bin/env python

import os
import sys
import random

def unreliable_code():

  def ok():
    return "it worked!!"

  def fail():
    return "it didn't work"

  def crash():
    1/0

  def hang():
    while True: 
      pass

  def bye():
    os._exit(0)

  return random.choice([ok, fail, crash, hang, bye])()


result = None
while result != "it worked!!":
  # ???

Answer 1

Чтобы быть в безопасности от исключений, используйте try / exc (но, я думаю, вы это знаете).

Чтобы быть в безопасности от зависания кода (бесконечный цикл), единственный способ, которым я знаю, - запустить код в другом процессе.,Этот дочерний процесс, который вы можете убить из родительского процесса, если он не завершится достаточно быстро.

Чтобы быть в безопасности от неприятного кода (делать то, чего он не должен), посмотрите на http://pypi.python.org/pypi/RestrictedPython.

Answer 2

В вашем реальном приложении вы можете переключиться на многопроцессорность?Кажется, что то, что вы просите, может быть сделано с multiprocessing + threading.Timer + try/except.

Взгляните на это:

class SafeProcess(Process):
    def __init__(self, queue, *args, **kwargs):
        self.queue = queue
        super().__init__(*args, **kwargs)
    def run(self):
        print('Running')
        try:
            result = self._target(*self._args, **self._kwargs)
            self.queue.put_nowait(result)
        except:
            print('Exception')

result = None
while result != 'it worked!!':
    q = Queue()
    p = SafeProcess(q, target=unreliable_code)
    p.start()
    t = Timer(1, p.terminate)   # in case it should hang
    t.start()
    p.join()
    t.cancel()
    try:
        result = q.get_nowait()
    except queues.Empty:
        print('Empty')
    print(result)

Это в одном (счастливом) случае дало мне:

Running
Empty
None
Running
it worked!!

В ваших примерах кода у вас есть 4 из 5 шансов получить ошибку, поэтому вы также можете создать пулили что-то, что повысит ваши шансы на получение правильного результата.

Answer 3

Вы можете попробовать запустить его в песочнице .