(code_str, timeout=600, MAX_LEN=1000)
| 304 | output_queue.put(output_capture.getvalue()) |
| 305 | |
| 306 | def execute_code(code_str, timeout=600, MAX_LEN=1000): |
| 307 | #code_str = code_str.replace("\\n", "\n") |
| 308 | code_str = "from utils import *\n" + code_str |
| 309 | if "load_dataset('pubmed" in code_str: |
| 310 | return "[CODE EXECUTION ERROR] pubmed Download took way too long. Program terminated" |
| 311 | if "exit(" in code_str: |
| 312 | return "[CODE EXECUTION ERROR] The exit() command is not allowed you must remove this." |
| 313 | output_queue = multiprocessing.Queue() |
| 314 | proc = multiprocessing.Process(target=worker_run_code, args=(code_str, output_queue)) |
| 315 | proc.start() |
| 316 | proc.join(timeout) |
| 317 | if proc.is_alive(): |
| 318 | proc.terminate() # Forcefully kill the process |
| 319 | proc.join() |
| 320 | return (f"[CODE EXECUTION ERROR]: Code execution exceeded the timeout limit of {timeout} seconds. " |
| 321 | "You must reduce the time complexity of your code.") |
| 322 | else: |
| 323 | if not output_queue.empty(): output = output_queue.get() |
| 324 | else: output = "" |
| 325 | return output |
no outgoing calls
no test coverage detected