自定义调度程序具有带超时/终止开关的顺序+半连续脚本？
Ryf*_*lex 7 python subprocess scheduler execfile python-2.7
下面是我的代码的一个重要部分,基本上如果你向下滚动到该execute_subscripts()功能,你可以看到我有两个脚本运行,通过execfile哪个工作很漂亮,他们表明prints,他们将traceback错误保存到错误文件.
我试图将第二个脚本变成一个不等待自己完成的脚本,然后再转到下一个脚本.
正如你所看到的,我试图用subprocess它Popen来启动一个无声的隐藏窗口......但它似乎没有运行,我不知道如何p.communicate()正确使用该函数来检索tracebacks和/或prints.
我也...需要帮助创建某种超时/终止开关,所以如果一个下标要么通过Popen或execfile路由没有在5分钟内完成它跳过它为该循环或重试并跳过如果它立即再次失败.
我明白我可能不应该strftime在时间使用....但是这部分对我来说很好,所以我认为不需要改变它.
from datetime import date, timedelta
from sched import scheduler
from time import time, sleep, strftime
import random
import traceback
import subprocess
s = scheduler(time, sleep)
random.seed()
def periodically(runtime, intsmall, intlarge, function):
## Get current time
currenttime = strftime('%H:%M:%S')
## If currenttime is anywhere between 23:40 and 23:50 then...
if currenttime > '23:40:00' and currenttime < '23:50:00':
## Open the error logging file as the variable "errors"
errors = open('MISC/ERROR(S).txt', 'a')
## Try to...
try:
## Call the clear subscript.
execfile("SUBSCRIPTS/CLEAR.py", {})
## On exception (fail)...
except Exception:
## Write the entire traceback error to file...
errors.write(traceback.format_exc() + '\n')
errors.write("\n\n")
## Close and exit the error logging file.
errors.close()
## Update time
currenttime = strftime('%H:%M:%S')
## Idle time
while currenttime >= '23:40:00' and currenttime <= '23:59:59' or currenttime >= '00:00:00' and currenttime <= '11:30:00':
## Update time
currenttime = strftime('%H:%M:%S')
print currenttime, "Idling..."
sleep(10)
## Update time
currenttime = strftime('%H:%M:%S')
## Initiate the scheduler.
runtime += random.randrange(intsmall, intlarge)
s.enter(runtime, 1, function, ())
s.run()
def execute_subscripts():
st = time()
print "Running..."
errors = open('MISC/ERROR(S).txt', 'a')
try:
execfile("SUBSCRIPTS/TESTSCRIPT.py", {})
except Exception:
errors.write(traceback.format_exc() + '\n')
errors.write("\n\n")
try:
execfile("SUBSCRIPTS/TEST.py", {})
except Exception:
errors.write(traceback.format_exc() + '\n')
errors.write("\n\n")
## subprocess.Popen(["pythonw", "SUBSCRIPTS/TEST.py", "0"], shell=True)
try:
execfile("SUBSCRIPTS/TESTSCRIPTTest.py", {})
except Exception:
errors.write(traceback.format_exc() + '\n')
errors.write("\n\n")
try:
execfile("SUBSCRIPTS/TESTTESTTEST.py", {})
except Exception:
errors.write(traceback.format_exc() + '\n')
errors.write("\n\n")
errors.close()
print """The whole routine took %.3f seconds""" % (time() - st)
while True:
periodically(50, -25, +90, execute_subscripts)
任何想法将不胜感激
添加了赏金,希望有人知道如何实现这一目标.
在此先感谢
Hyflex
我希望脚本能够做什么的例子......
下标1 - 在后台运行,将打印和错误从subscript1.py发送到main.py,不要等待它完成,转到下标2,10 秒后超时(或尽可能接近10秒,或超时)在所有下标被调用之后.)
下标2 - 在后台运行,将打印和错误从subscript2.py发送到main.py,等待它完成,然后进入下标3,超时10秒后(或尽可能接近10秒,或者毕竟超时)下标已被调用.)
下标3 - 在后台运行,将打印和错误从subscript3.py发送到main.py,等待它完成,然后进入下标4,10 秒后超时(或尽可能接近10秒,或者毕竟超时)下标已被调用.)
下标4 - 在后台运行,将打印和错误从subscript4.py发送到main.py,不要等待它完成,转到下标 5,10秒后超时(或尽可能接近10秒,或者调用所有下标后超时.)
下标5 - 在后台运行,将打印和错误从subscript5.py发送到main.py,等待它完成后再进入下一个下标(或者在这种情况下,循环结束), 10秒后超时(或接近尽可能10秒,或者在调用所有下标后超时.)
shx2的打印和回溯
[pid=9940] main running command: C:\Python27\python.exe SUB/subscript1.py (is_bg=False)
[pid=9940] main running command: C:\Python27\python.exe SUB/subscript1.py (is_bg=True)
Traceback (most recent call last):
File "C:\Test\main.py", line 21, in <module>
bg_proc1 = run_subscript(cmd, is_bg = True)
File "C:\Test\main.py", line 10, in run_subscript
return (cmd > sys.stdout) & BG # run in background
File "C:\Python27\lib\site-packages\plumbum\commands\modifiers.py", line 81, in __rand__
return Future(cmd.popen(), self.retcode)
File "C:\Python27\lib\site-packages\plumbum\commands\base.py", line 317, in popen
return self.cmd.popen(args, **kwargs)
File "C:\Python27\lib\site-packages\plumbum\commands\base.py", line 233, in popen
return self.cmd.popen(self.args + list(args), **kwargs)
File "C:\Python27\lib\site-packages\plumbum\machines\local.py", line 104, in popen
**kwargs)
File "C:\Python27\lib\site-packages\plumbum\machines\local.py", line 253, in _popen
stderr = stderr, cwd = str(cwd), env = env, **kwargs) # bufsize = 4096
File "C:\Python27\lib\subprocess.py", line 703, in __init__
errread, errwrite) = self._get_handles(stdin, stdout, stderr)
File "C:\Python27\lib\subprocess.py", line 851, in _get_handles
c2pwrite = msvcrt.get_osfhandle(stdout.fileno())
UnsupportedOperation: fileno
编辑: 
| --> # Sub 1.py # --> Sequential with timeout --> Started: 11:30.00 --> Estimated Completion: 11:30.01 (1 Second) --> Timeout at 11:30:10 (10 Seconds) --> # Sub 2.py # --> Sequential with timeout --> Started: 11:30.02 (or after time Sub 1.py's timeout) --> Estimated Completion: 11:30.03 (1 Second) --> Timeout at 11:30:13 (10 Seconds) --> # Sub 3.py # --> Sequential with timeout --> Started: 11:30.04 (or after time Sub 2.py's timeout) --> Estimated Completion: 11:30.08 (3 Seconds) --> Timeout at 11:30:18 (10 Seconds)
| ^ ^
| | |
| -------------------------------------------------------------------------------------------------------------------------------------------------- |
| --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
Scheduler -->|
| --> Sub 4.py --> Nonsequential with timeout --> Started: 11:30.00 --> Estimated Completion: 11:30.05 (5 Seconds) --> Timeout at 11:30:10 (15 Seconds)
|
| --> Sub 5.py --> Nonsequential with timeout --> Started: 11:30.00 --> Estimated Completion: 11:30.02 (2 Seconds) --> Timeout at 11:30:10 (10 Seconds)
|
| --> Sub 6.py --> Nonsequential with timeout --> Started: 11:30.00 --> Estimated Completion: 11:30.10 (10 Seconds) --> Timeout at 11:30:10 (25 Seconds)
希望这有助于直观地表达我想要实现的目标
正如您在问题中已经指出的那样,您实际上是在问两个不同的问题(在后台运行,并执行超时).幸运的是,两者的简短回答是同一个:
使用Plumbum!
Plumbum极大地简化了python脚本中类似shell脚本的元素,除此之外,还为后台运行命令提供了干净的接口,并强制执行超时.
下面是使用plumbum的示例.
在此示例中,子进程将全部运行相同的脚本 - subscript1.py.它会进行一些打印,有些会睡觉,有时也会随机失败.
subscript1.py
import os, sys, time, random
print '[pid=%s] STARTING %s' % (os.getpid(), sys.argv[0])
for i in range(3):
t = random.randint(1,5)
print '[pid=%s] sleeping for %s seconds' % (os.getpid(), t)
time.sleep(t)
# fail randomly
if t == 5:
raise RuntimeError('random error...')
print '[pid=%s] DONE %s' % (os.getpid(), sys.argv[0])
现在,下面的主要脚本main.py演示了如何在前台和后台运行子进程,有或没有超时,等待后台进程完成,以及处理子进程错误和超时.
main.py
import os, sys, time
from plumbum import FG, BG, ProcessExecutionError, ProcessTimedOut
from plumbum.cmd import python
cmd = python['subscript1.py'] # create the command to run (several times)
def run_subscript(cmd, is_bg = False):
print '[pid=%s] main running command: %s (is_bg=%s)' % (os.getpid(), cmd, is_bg)
if is_bg:
return (cmd > sys.stdout) & BG # run in background
else:
try:
return cmd & FG # run in foreground
except ProcessExecutionError, e:
print >>sys.stderr, e
# run a process in the foreground
run_subscript(cmd, is_bg = False)
# run two processes in the background, and one in the foreground
bg_proc1 = run_subscript(cmd, is_bg = True)
time.sleep(1)
bg_proc2 = run_subscript(cmd, is_bg = True)
time.sleep(1)
run_subscript(cmd, is_bg = False)
# wait for the background processes to finish
for bg_proc in ( bg_proc1, bg_proc2 ):
try:
bg_proc.wait()
except ProcessExecutionError, e:
print >>sys.stderr, e
# run a foreground process, which will time out
print '[pid=%s] main running command: %s (will time out)' % (os.getpid(), cmd)
try:
cmd.run(timeout = 2)
except ProcessTimedOut, e:
# command timed out
print >>sys.stderr, e
except ProcessExecutionError, e:
# command failed (but did not time out)
print >>sys.stderr, e
输出:
% python main.py
[pid=77311] main running command: /usr/local/bin/python subscript1.py (is_bg=False)
[pid=77314] STARTING subscript1.py
[pid=77314] sleeping for 1 seconds
[pid=77314] sleeping for 5 seconds
[pid=77314] sleeping for 3 seconds
[pid=77314] DONE subscript1.py
[pid=77311] main running command: /usr/local/bin/python subscript1.py (is_bg=True)
[pid=77316] STARTING subscript1.py
[pid=77316] sleeping for 5 seconds
[pid=77311] main running command: /usr/local/bin/python subscript1.py (is_bg=True)
[pid=77317] STARTING subscript1.py
[pid=77317] sleeping for 1 seconds
[pid=77311] main running command: /usr/local/bin/python subscript1.py (is_bg=False)
[pid=77317] sleeping for 5 seconds
[pid=77318] STARTING subscript1.py
[pid=77318] sleeping for 5 seconds
[pid=77316] sleeping for 2 seconds
[pid=77316] sleeping for 4 seconds
[pid=77317] sleeping for 5 seconds
[pid=77318] sleeping for 2 seconds
[pid=77318] sleeping for 3 seconds
[pid=77316] DONE subscript1.py
[pid=77318] DONE subscript1.py
Command line: ['/usr/local/bin/python', 'subscript1.py']
Exit code: 1
Stderr: | Traceback (most recent call last):
| File "subscript1.py", line 13, in <module>
| raise RuntimeError('random error...')
| RuntimeError: random error...
[pid=77311] main running command: /usr/local/bin/python subscript1.py (will time out)
('Process did not terminate within 2 seconds', ['/usr/local/bin/python', 'subscript1.py'])
编辑:
我现在意识到我的示例代码没有演示在后台运行命令并对其执行超时.为此,只需使用cmd.bgrun(...)而不是cmd.run(...).
您获得的错误与重定向有关,并且必须与您在Windows上运行的事实相关.这可能是Windows上的plumbum的兼容性问题,或者我的代码可能不完美,即可能有另一种使用plumbum使其工作的方法.不幸的是,我没有一台Windows机器来测试它......
我希望这有帮助.
如果我明白你想要做什么， subprocess.Popen() 就是可行的方法。这是一个简单的类，我认为它可以提供您想要的所有功能：
from time import sleep
import subprocess
import datetime
import os
class Worker:
def __init__(self, cmd):
print datetime.datetime.now(), ":: starting subprocess :: %s"%cmd
self.cmd = cmd
self.log = "[running :: %s]\n"%cmd
self.subp = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
self.start_time = datetime.datetime.now()
def wait_to_finish(self, timeout_seconds = None):
while True:
retcode = self.subp.poll()
if retcode is not None:
self.get_process_output()
self.log += "\n[subprocess finished, return code: %d]\n"%retcode
print datetime.datetime.now(), ":: subprocess %s exited, retcode=%d"%(self.cmd, retcode)
return
else:
# process hasn't finished yet
sleep(1)
if timeout_seconds is not None:
cur_time = datetime.datetime.now()
if (cur_time - self.start_time).seconds > timeout_seconds:
print datetime.datetime.now(), ":: subprocess %s :: killing after %d seconds"%(self.cmd, timeout_seconds)
self.kill()
return
def still_running(self):
return (self.subp.poll() is None)
def kill(self):
self.subp.terminate()
self.get_process_output()
self.log += "\n[subprocess killed by explicit request]\n"
return
def get_process_output(self):
out, err = self.subp.communicate()
self.log += out
self.log += err
您发出命令，班级就会在后台启动它。然后，您可以等待它完成，并可选择超时（从进程启动的时间开始计算）。您可以获取进程输出，如果需要，还可以显式终止该进程。
这只是一个展示其功能的简单示例：
# Start two subprocesses in the background
worker1 = Worker([r'c:\python26\python.exe', 'sub1.py'])
worker2 = Worker([r'c:\python26\python.exe', 'sub2.py'])
# Wait for both to finish, kill after 10 seconds timeout
worker1.wait_to_finish(timeout_seconds = 10)
worker2.wait_to_finish(timeout_seconds = 10)
# Start another subprocess giving it 5 seconds to finish
worker3 = Worker([r'c:\python26\python.exe', 'sub3.py'])
worker3.wait_to_finish(timeout_seconds = 5)
print "----LOG1----\n" + worker1.log
print "----LOG2----\n" + worker2.log
print "----LOG3----\n" + worker3.log
子1.py:
from time import sleep
print "sub1 output: start"
sleep(5)
print "sub1 output: finish"
子2.py:
print "sub2 output: start"
erroneous_command()
子3.py:
from time import sleep
import sys
print "sub3 output: start, sleeping 15 sec"
sys.stdout.flush()
sleep(15)
print "sub3 output: finish"
这是输出：
2013-11-06 15:31:17.296000 :: starting subprocess :: ['c:\\python26\\python.exe', 'sub1.py']
2013-11-06 15:31:17.300000 :: starting subprocess :: ['c:\\python26\\python.exe', 'sub2.py']
2013-11-06 15:31:23.306000 :: subprocess ['c:\\python26\\python.exe', 'sub1.py'] exited, retcode=0
2013-11-06 15:31:23.309000 :: subprocess ['c:\\python26\\python.exe', 'sub2.py'] exited, retcode=1
2013-11-06 15:31:23.310000 :: starting subprocess :: ['c:\\python26\\python.exe', 'sub3.py']
2013-11-06 15:31:29.314000 :: subprocess ['c:\\python26\\python.exe', 'sub3.py'] :: killing after 5 seconds
----LOG1----
[running :: ['c:\\python26\\python.exe', 'sub1.py']]
sub1 output: start
sub1 output: finish
[subprocess finished, return code: 0]
----LOG2----
[running :: ['c:\\python26\\python.exe', 'sub2.py']]
sub2 output: start
Traceback (most recent call last):
File "sub2.py", line 2, in <module>
erroneous_command()
NameError: name 'erroneous_command' is not defined
[subprocess finished, return code: 1]
----LOG3----
[running :: ['c:\\python26\\python.exe', 'sub3.py']]
sub3 output: start, sleeping 15 sec
[subprocess killed by explicit request]
就实施调度而言，我可以建议几个选项，但选择实际上取决于您的任务是什么：
1）如果可以指定任意时间点的精确调度，那么就可以实现完全同步的调度器：
while True:
# check time
# check currently running processes :: workerX.still_running()
# -> if some are past their timeout, kill them workerX.kill()
# start new subprocesses according to your scheduling logic
sleep(1)
2) 如果您有几个定义明确的脚本序列，您希望每 10 秒“即发即忘”，则将每个序列放入其自己的 .py 脚本中（使用“import Worker”），并每隔10 秒，还定期检查哪些序列已退出以收集其日志。
3）如果您的序列是动态定义的并且您更喜欢“即发即忘”方法，那么线程将是最好的方法。