| 21 | |
| 22 | |
| 23 | class CodeExecutor: |
| 24 | def __init__(self, code, timeout, use_process: bool): |
| 25 | self.code = format_code(code) |
| 26 | self.timeout = timeout |
| 27 | self.error = '' |
| 28 | self.use_process = use_process |
| 29 | |
| 30 | def execute_code(self, return_val): |
| 31 | try: |
| 32 | f = StringIO() |
| 33 | with redirect_stdout(f): |
| 34 | exec(self.code, globals(), locals()) |
| 35 | s = f.getvalue() |
| 36 | s = s.strip('\n') |
| 37 | return_val['result'] = s |
| 38 | except Exception: |
| 39 | pass |
| 40 | |
| 41 | @staticmethod |
| 42 | def execute_code_with_string(code, index, return_val): |
| 43 | code = format_code(code) |
| 44 | try: |
| 45 | f = StringIO() |
| 46 | with redirect_stdout(f): |
| 47 | exec(code, globals(), locals()) |
| 48 | s = f.getvalue() |
| 49 | s = s.strip('\n') |
| 50 | return_val[index] = s |
| 51 | except Exception as e: |
| 52 | # print(e) |
| 53 | pass |
| 54 | |
| 55 | def run(self): |
| 56 | if self.use_process: |
| 57 | manager = multiprocessing.Manager() |
| 58 | return_dict = manager.dict() |
| 59 | process = multiprocessing.Process( |
| 60 | target=self.execute_code, args=(return_dict,)) |
| 61 | process.start() |
| 62 | process.join(timeout=self.timeout) |
| 63 | process.terminate() |
| 64 | else: |
| 65 | return_dict = {} |
| 66 | thread = threading.Thread( |
| 67 | target=self.execute_code, args=(return_dict,)) |
| 68 | thread.start() |
| 69 | thread.join(timeout=self.timeout) |
| 70 | if thread.is_alive(): |
| 71 | thread.join() # Ensures the thread is terminated before continuing |
| 72 | print('time out!') |
| 73 | self.error = 'Execution timed out' |
| 74 | |
| 75 | if 'result' in return_dict: |
| 76 | return return_dict['result'] |
| 77 | else: |
| 78 | return '' |
| 79 | |
| 80 |
no outgoing calls
no test coverage detected