我有一个Python测试程序用于测试另一个软件组件的功能,让我们称之为被测组件(COT).Python测试程序通过持久TCP连接连接到COT.Python程序正在使用Python套接字API.现在为了模拟物理链路的故障,我想让Python程序关闭套接字,但不要正确断开连接.即我不想再在TCP通道上发送任何东西,包括任何TCP SYN/ ACK/ FIN.我只想让套接字保持沉默.它不能再响应远程数据包了.

这并不像看起来那么容易,因为调用close套接字会将TCP FIN数据包发送到远程端.(优雅的断开连接).那么如何在不发送任何数据包的情况下杀死套接字呢？

我无法关闭Python程序本身,因为它需要维护与其他组件的其他连接.有关信息,套接字在单独的线程中运行.所以我想到了突然杀死线程,但这也不是那么容易.(有没有办法在Python中杀死一个线程？)

有任何想法吗？

我的程序是按以下方式设计的:

1. 程序的第一部分从传感器获取实时值,并使用Matplotlib绘制它.这必须持续很长时间.而且,它将信息记录到数据库中.
2. 第二部分是IP摄像头.我必须从IP摄像头获取输入并显示它.为了显示我正在使用OpenCV的imshow方法.另外,我正在存储来自IP Camera的视频.

问题:我已经有了算法,问题是我需要在while循环中运行这两个算法.条件是我无法退出任何一个.现在线程是一个很好的选择,但我已经阅读了关于GIL的内容,那么如何运行两个无限循环呢？

from multiprocessing import Process

def methodA():
    while TRUE:
        do something

def methodB():
    while TRUE:
        do something

p=Process(target=methodA())
p.start()
p1=Process(target=methodB())
p1.start()

现在,当我开始处理p它开始执行,之后如何开始p1同时运行？

假设我有一个长期运行的python函数,看起来像这样？

import random
import time
from rx import Observable
def intns(x):
    y = random.randint(5,10)
    print(y)
    print('begin')
    time.sleep(y)
    print('end')
    return x

我希望能够设置超时1000ms.

所以我就像这样,通过上面的强烈计算创建一个可观察的并映射它.

a = Observable.repeat(1).map(lambda x: intns(x))

现在,每个值发出的,如果超过1000毫秒我就越想尽快结束观察到,当我到达1000ms使用on_error或on_completed

a.timeout(1000).subscribe(lambda x: print(x), lambda x: print(x))

上面的语句确实得到超时和调用on_error,但它继续完成计算强烈的计算,然后才返回到下一个语句.有没有更好的方法呢？

最后一个语句打印以下内容

8 # no of seconds to sleep
begin # begins sleeping, trying to emit the first value
Timeout # operation times out, and calls on_error
end # thread waits till the function ends

这个想法是,如果一个特定的函数超时,我希望能够继续我的程序,并忽略结果.

我想知道 …

我事先就问题的长度道歉,我不想遗漏任何东西.

一些背景资料

我正在尝试通过编写使用Windows API来模拟击键,鼠标移动和窗口/控制操作的Python应用程序来自动化数据输入过程.我不得不求助于这种方法,因为我没有(还)有直接访问数据存储/数据库所需的安全许可(例如,使用SQL)或间接通过更适合的API.官僚主义,这是一种痛苦;-)

数据输入过程涉及由于物品可用性的变化而更正销售订单.不可用的文章要么从订单中删除,要么被其他合适的文章替换.

最初我希望人类能够监控自动数据输入过程,以确保一切正常.为了实现这一点,我一方面减慢了操作的速度,另一方面也通过固定窗口告知用户当前正在进行的操作.

实际的问题

为了允许用户暂停自动化过程,我将Pause/Break键注册为热键,在处理程序中我想暂停自动化功能.但是,我目前正在努力找到一种方法来正确暂停自动化功能的执行.当调用pause函数时,无论它在做什么,我都希望自动化过程停止在其轨道上.我不希望它甚至执行另一次击键.

更新[23/01]:我实际上想要做的不仅仅是暂停,我希望能够在自动化过程运行时与自动化流程进行通信并请求它暂停,跳过当前的销售订单,完全放弃甚至可能更多.

任何人都可以告诉我The Right Way(TM)来实现我想要的吗？

更多信息

以下是自动化工作原理的示例(我正在使用pywinauto库):

from pywinauto import application
app = application.Application()
app.start_("notepad")
app.Notepad.TypeKeys("abcdef")

更新[25/01]:在我的应用程序工作几天后,我注意到我并没有真正使用pywinauto那么多,现在我只是用它来查找窗口然后我直接SendKeysCtypes.SendKeys用来模拟键盘输入和win32api模拟鼠标输入的功能.

到目前为止我发现了什么

以下是我在寻找答案时遇到的一些方法:

1. 我可以在两个单独的进程中分离自动化功能和接口+热键监听器.我们将前者称为"automator",将后者称为"manager".然后,管理器可以通过向进程发送SIGSTOP信号并使用SIGCONT信号(或Windows等效的SuspendThread/ResumeThread)取消暂停来暂停执行automator .

   为了能够更新用户界面,自动机将需要通过某种IPC机制通知管理器其进展.

   缺点:

   • 使用SIGSTOP会不会有点苛刻？它甚至可以正常工作吗？很多人似乎都在反对它,甚至称之为"危险".

   • 我担心实施IPC机制会有点复杂.另一方面,我与DBus合作过,实施起来并不难.

2. 第二种方法和许多人似乎建议的方法涉及使用线程,并且基本归结为以下(简化):

   while True:
       if self.pause: # pause
       # Do the work...
   

   Run Code Online (Sandbox Code Playgroud)

   但是,这样做似乎只会在没有更多工作要做之后暂停.我认为这种方法可行的唯一方法是将工作(整个自动化过程)划分为较小的工作段(即任务).在开始新任务之前,工作线程将检查它是否应该暂停并等待.

   缺点:

   • 似乎是将工作划分为较小部分的实现,例如上面的部分,将是非常丑陋的代码(美学上).

     我想象的方式,所有语句都将转换为类似于:( queue.put((function, args))例如queue.put((app.Notepad.TypeKeys, "abcdef")))并且您将使自动化过程线程贯穿任务并在开始任务之前不断检查暂停状态.那不可能是正确的......

   • 该程序实际上不会停止在其轨道上,但在实际暂停之前首先完成一项任务(无论多么小).

取得了进展

更新[23/01]:我已经通过提到的SuspendThread/ResumeThread功能使用第一种方法实现了我的应用程序的一个版本.到目前为止,这似乎工作非常好,也允许我编写自动化的东西,就像你写任何其他脚本一样.我遇到的唯一怪癖是键盘修饰符(CTRL,ALT,SHIFT)在暂停时"卡住".我可以轻松解决的问题.

我还使用第二种方法(线程和信号/消息传递)编写了测试并实现了暂停功能.然而,它看起来真的很丑陋(检查暂停标志和与"做工作"相关的一切).所以如果有人能给我一个类似于第二种方法的正确例子,我会很感激.

相关问题

• 暂停一个过程？

• 使用线程类暂停线程

  Alex Martelli发布了一个答案:

  没有其他线程强制暂停线程的方法(除了其他线程杀死该线程之外) - 目标线程必须通过偶尔检查相应的"标志"来配合(线程.条件可能适合于暂停/取消暂停案例).

  然后他提到了多处理模块和SIGSTOP/SIGCONT. …

是否可以使用相同的子进程命令按顺序执行任意数量的命令？

我需要每个命令在执行之前等待前一个命令完成,我需要它们全部在同一个会话/ shell中执行.我还需要它在Python 2.6,Python 3.5中工作.我还需要subprocess命令才能在Linux,Windows和macOS中工作(这就是为什么我echo这里只使用命令作为示例).

请参阅下面的非工作代码:

import sys
import subprocess

cmds = ['echo start', 'echo mid', 'echo end']

p = subprocess.Popen(cmd=tuple([item for item in cmds]),
                     stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

for line in iter(p.stdout.readline, b''):
    sys.stdout.flush()
    print(">>> " + line.rstrip())

如果这是不可能的,我应该采取哪种方法,以便在同一会话/ shell中以同步顺序执行我的命令？

我有一个使用paramiko的自动化过程并出现此错误:

Exception in thread Thread-1 (most likely raised during interpreter 
shutdown)

....
....
<type 'exceptions.AttributeError'>: 'NoneType' object has no attribute 
'error' 

我知道这是清理/线程中的问题,但我不知道如何解决它.

我有最新版本(1.7.6)并根据此线程,它已解决,所以我直接下载代码但仍然得到错误.

在winxp/win2003下的Python 2.5/2.6上发生了故障.

我在__del__析构函数中关闭连接,然后在脚本结束之前关闭它,这些都不起作用.还有更多,使用这个错误发生在前面,所以也许与解释器关闭无关？

当用python下载大文件时,我想为时间限制设置一个时间限制,不仅适用于连接过程,还适用于下载.

我正在尝试使用以下python代码:

import requests

r = requests.get('http://ipv4.download.thinkbroadband.com/1GB.zip', timeout = 0.5, prefetch = False)

print r.headers['content-length']

print len(r.raw.read())

这不起作用(下载没有时间限制),正如文档中正确指出的那样:https://requests.readthedocs.org/en/latest/user/quickstart/#timeouts

如果有可能,这将是伟大的:

r.raw.read(timeout = 10)

问题是,如何为下载设置时间限制？

如果我将Python解释器嵌入到C或C++程序中,如本例所示,有没有办法限制解释器运行的时间？有没有什么可以阻止Python代码进入无限循环,从而阻止PyObject_CallObject(或等效)返回？

类似地,如果Python代码创建了一个新线程,那么有什么东西可以阻止这个线程进入无限循环并永远运行吗？

目前还不清楚如何Parallel在python中正确地删除joblib的工作者.其他人在这里,这里,这里和这里有类似的问题.

在我的例子中,我正在使用一个由50 joblib名工人组成的 threading后端池.

并行调用(线程):

output = Parallel(n_jobs=50, backend  = 'threading')
    (delayed(get_output)(INPUT) 
        for INPUT in list)

在这里,Parallel挂起没有错误,len(list) <= n_jobs但只有在n_jobs => -1.

为了克服这个问题,人们给予 说明如何创建一个超时装饰的Parallel功能(get_output(INPUT)使用)在上面的例子)multiprocessing:

主要功能(装饰):

@with_timeout(10)    # multiprocessing
def get_output(INPUT):     # threading
    output = do_stuff(INPUT)
    return output

多处理装饰器:

def with_timeout(timeout):
    def decorator(decorated):
        @functools.wraps(decorated)
        def inner(*args, **kwargs):
            pool = multiprocessing.pool.ThreadPool(1)
            async_result = pool.apply_async(decorated, args, kwargs)
            try:
                return …

我有一个 python3 程序，它启动第二个线程（除了主线程）来异步处理一些事件。理想情况下，我的程序可以正常运行，并且永远不会有未处理的异常。但事情发生了。当/如果有异常，我希望整个解释器以错误代码退出，就好像它是单个线程一样。那可能吗？

现在，如果在生成的线程上发生异常，它会打印出通常的错误信息，但不会退出。主线程一直在运行。

例子

import threading
import time

def countdown(initial):
    while True:
        print(initial[0])
        initial = initial[1:]
        time.sleep(1)

if __name__ == '__main__':
    helper = threading.Thread(target=countdown, args=['failsoon'])
    helper.start()
    time.sleep(0.5)
    #countdown('THISWILLTAKELONGERTOFAILBECAUSEITSMOREDATA')
    countdown('FAST')

该countdown最终将无法访问[0]，因为它已经清空引起从字符串IndexError: string index out of range错误。目标是无论主程序还是辅助程序先死，整个程序都会死，但堆栈跟踪信息仍然输出。

尝试的解决方案

经过一番挖掘，我的想法是使用sys.excepthook. 我添加了以下内容：

def killAll(etype, value, tb):
    print('KILL ALL')
    traceback.print_exception(etype, value, tb)
    os.kill(os.getpid(), signal.SIGKILL)

sys.excepthook = killAll

如果主线程是第一个死亡的线程，则此方法有效。但在另一种情况下则不然。这似乎是一个已知问题（https://bugs.python.org/issue1230540）。我会在那里尝试一些解决方法。

虽然该示例显示了我创建的一个主线程和一个辅助线程，但我对我可能正在运行启动线程的其他人的库的一般情况感兴趣。