Python Cookbook 13.11. Detecting Inactive Computers

Credit: Nicola Larosa

Problem

问题

You need to monitor the working state of a number of computers connected to a TCP/IP network.

你需要监控连接在一个TCP/IP网络里的一些计算机的工作状态.

Solution

解决

The key idea in this recipe is to have every computer periodically send a heartbeat UDP packet to a computer acting as the server for this heartbeat-monitoring service. The server keeps track of how much time has passed since each computer last sent a heartbeat and reports on computers that have been silent for too long.

本配方的重点就是让每台计算机定期发送一个UDP心跳包给一台运行着心跳监控服务的服务器.这台服务器记录每台机器最后一次发送心跳包至今的时间,并报告沉默太久的计算机.

Here is the "client" program, HeartbeatClient.py, which must run on every computer we need to monitor:

这是客户端程序HeartbeatClient.py,我们需要监控的每台计算机上都将运行这个程序:

""" Heartbeat client, sends out a UDP packet periodically """
import socket, time
SERVER_IP = '192.168.0.15'; SERVER_PORT = 43278; BEAT_PERIOD = 5
print 'Sending heartbeat to IP %s , port %d' % (SERVER_IP, SERVER_PORT)
print 'press Ctrl-C to stop'
while True:
    hbSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    hbSocket.sendto('PyHB', (SERVER_IP, SERVER_PORT))
    if __debug__:
        print 'Time: %s' % time.ctime( )
    time.sleep(BEAT_PERIOD)

The server program, which receives and keeps track of these "heartbeats", must run on the machine whose address is given as SERVER_IP in the "client" program. The server must support concurrency, since many heartbeats from different computers might arrive simultaneously. A server program has essentially two ways to support concurrency: multithreading, or asynchronous operation. Here is a multithreaded ThreadedBeatServer.py, using only modules from the Python Standard Library:
用来接收和记录心跳的服务器端程序必须运行的机器的地址必须是客户端程序中SERVER_IP变量所指定的值.服务器必须支持并发,因为不同计算机发出的心跳包可能同时到达.一个服务器程序有两种方法来支持并发:多线程或者异步操作,这是一个使用Python标准库中够的模块构建的多线程程序

#该程序在我的windows xp上无法运行,由于多个套接字不能绑定同一个端口
""" Threaded heartbeat server """
import socket, threading, time
UDP_PORT = 43278; CHECK_PERIOD = 20; CHECK_TIMEOUT = 15
class Heartbeats(dict):
    """ Manage shared heartbeats dictionary with thread locking """
    def __init__(self):
        super(Heartbeats, self).__init__( )
        self._lock = threading.Lock( )
    def __setitem__(self, key, value):
        """ Create or update the dictionary entry for a client """
        self._lock.acquire( )
        try:
            super(Heartbeats, self).__setitem__(key, value)
        finally:
            self._lock.release( )
    def getSilent(self):
        """ Return a list of clients with heartbeat older than CHECK_TIMEOUT """
        limit = time.time( ) - CHECK_TIMEOUT
        self._lock.acquire( )
        try:
            silent = [ip for (ip, ipTime) in self.items( ) if ipTime < limit]
        finally:
            self._lock.release( )
        return silent
class Receiver(threading.Thread):
    """ Receive UDP packets and log them in the heartbeats dictionary """
    def __init__(self, goOnEvent, heartbeats):
        super(Receiver, self).__init__( )
        self.goOnEvent = goOnEvent
        self.heartbeats = heartbeats
        self.recSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.recSocket.settimeout(CHECK_TIMEOUT)
        self.recSocket.bind(('', UDP_PORT))
    def run(self):
        while self.goOnEvent.isSet( ):
            try:
                data, addr = self.recSocket.recvfrom(5)
                if data == 'PyHB':
                    self.heartbeats[addr[0]] = time.time( )
            except socket.timeout:
                pass
def main(num_receivers=3):
    receiverEvent = threading.Event( )
    receiverEvent.set( )
    heartbeats = Heartbeats( )
    receivers = [  ]
    for i in range(num_receivers):
        receiver = Receiver(goOnEvent=receiverEvent, heartbeats=heartbeats)
        receiver.start( )
        receivers.append(receiver)
    print 'Threaded heartbeat server listening on port %d' % UDP_PORT
    print 'press Ctrl-C to stop'
    try:
        while True:
            silent = heartbeats.getSilent( )
            print 'Silent clients: %s' % silent
            time.sleep(CHECK_PERIOD)
    except KeyboardInterrupt:
        print 'Exiting, please wait'
        receiverEvent.clear( )
        for receiver in receivers:
            receiver.join( )
        print 'Finished.'
if __name__ == '__main__':
    main( )

.
As an alternative, here is an asynchronous AsyncBeatServer.py program based on the powerful Twisted framework:
这里提供了一个基于Twisted框架的异步程序 AsyncBeatServer.py作为另外可选的程序.

import time
from twisted.application import internet, service
from twisted.internet import protocol
from twisted.python import log
UDP_PORT = 43278; CHECK_PERIOD = 20; CHECK_TIMEOUT = 15
class Receiver(protocol.DatagramProtocol):
    """ Receive UDP packets and log them in the "client"s dictionary """
    def datagramReceived(self, data, (ip, port)):
        if data == 'PyHB':
            self.callback(ip)
class DetectorService(internet.TimerService):
    """ Detect clients not sending heartbeats for too long """
    def __init__(self):
        internet.TimerService.__init__(self, CHECK_PERIOD, self.detect)
        self.beats = {  }
    def update(self, ip):
        self.beats[ip] = time.time( )
    def detect(self):
        """ Log a list of clients with heartbeat older than CHECK_TIMEOUT """
        limit = time.time( ) - CHECK_TIMEOUT
        silent = [ip for (ip, ipTime) in self.beats.items( ) if ipTime < limit]
        log.msg('Silent clients: %s' % silent)
application = service.Application('Heartbeat')
# define and link the silent clients' detector service
detectorSvc = DetectorService( )
detectorSvc.setServiceParent(application)
# create an instance of the Receiver protocol, and give it the callback
receiver = Receiver( )
receiver.callback = detectorSvc.update
# define and link the UDP server service, passing the receiver in
udpServer = internet.UDPServer(UDP_PORT, receiver)
udpServer.setServiceParent(application)
# each service is started automatically by Twisted at launch time
log.msg('Asynchronous heartbeat server listening on port %d\n'
    'press Ctrl-C to stop\n' % UDP_PORT)

Discussion

讨论

When a number of computers are connected by a TCP/IP network, we are often interested in monitoring their working state. The client and server programs presented in this recipe help you detect when a computer stops working, while having minimal impact on network traffic and requiring very little setup. Note that this recipe does not monitor the working state of single, specific services running on a machine, just that of the TCP/IP stack and the underlying operating system and hardware components.

当一些计算机连接在一个TCP/IP网络上时,我们经常需要监控他们的工作状态.这个配方中给出的客户端和服务器端程序可以帮助你检测一个计算着在何时停止工作,而且它只需要极少的设置和很小的网络流量就可以工作.注意这个配方并没有在一台机器上运行一个单独的特别的服务,只是使用了TCP/IP协议栈和基本操作系统及计算机组件.

This PyHeartBeat approach is made up of two files: a client program, HeartbeatClient.py, sends UDP packets to the server, while a server program, either ThreadedBeatServer.py (using only modules from the Python Standard Library to implement a multithreaded approach) or AsyncBeatServer.py (implementing an asynchronous approach based on the powerful Twisted framework), runs on a central computer to listen for such packets and detect inactive clients. Client programs, running on any number of computers, periodically send UDP packets to the server program that runs on the central computer. The server program, in either version, dynamically builds a dictionary that stores the IP addresses of the "client" computers and the timestamp of the last packet received from each one. At the same time, the server program periodically checks the dictionary, checking whether any of the timestamps are older than a defined timeout, to identify clients that have been silent too long.

这个PyHeartBeat使用了两个文件:一个HeartbeatClient.py客户端程序来发送UDP包给服务器,同时一个 叫ThreadedBeatServer.py(仅仅使用了Python标准库的模块来实现多线程)或者 AsyncBeatServer.py(基于Twisted框架实现的异步方法)的服务器程序在一个中心计算机上运行来监听这些包并发现不活跃的客户端.运行在任意数量的计算机上的客户端程序定期地发思念过UDP包给运行在中心计算机上的服务器程序.服务器程序则动态地构建一个字典来存储客户端的ip地址以及它最后发送的心跳包的时间戳.同时,服务器端程序将定期地检测字典来检查是否有时间戳以及超时了,并判定客户端是否沉默太久.

In this kind of application, there is no need to use reliable TCP connections since the loss of a packet now and then does not produce false alarms, as long as the server-checking timeout is kept suitably larger than the "client"-sending period. Since we may have hundreds of computers to monitor, it is best to keep the bandwidth used and the load on the server at a minimum: we do this by periodically sending a small UDP packet, instead of setting up a relatively expensive TCP connection per client.

这种程序没有必要使用TCP连接,因为丢失一些包并不会城市一个错误警告,知道服务器检测到客户端发送的时间间隔超时了.我们通过间歇性地发送小的UDP包来保证即使有数以百计的计算机要监控,还是能保证带宽和服务器的负载很小,而我们要和每个客户端保持一个TCP链接这个代价将会很昂贵.

The packets are sent from each client every 5 seconds, while the server checks the dictionary every 20 seconds, and the server's timeout defaults to 15 seconds. These parameters, along with the server IP number and port used, can be adapted to one's needs.

每个客户端将会每隔5秒发送一次这些包,而服务器将每隔20秒检查一次字典,同时服务器的超时时间会默认设置成15秒.这些参数,和服务器ip以及端口一样,可以因需求而调整.

Threaded server

线程服务器

In the threaded server, a small number of worker threads listen to the UDP packets coming from the "client"s, while the main thread periodically checks the recorded heartbeats. The shared data structure, a dictionary, must be locked and released at each access, both while writing and reading, to avoid data corruption on concurrent access. Such data corruption would typically manifest itself as intermittent, time-dependent bugs that are difficult to reproduce, investigate, and correct.

在这个线程服务器中,当主线程在定时检查记录的心跳包时,其他的工作线程在监听客户端发过来的UDP包.这个共享数据结构必须要在每一次读写访问时加锁和释放锁,以防止同步访问时数据损坏.这种数据损坏主要表现为随机性,不定时出现的问题,因此难以再现,检查和修复.

A very sound alternative to such meticulous use of locking around access to a resource is to dedicate a specialized thread to be the only one interacting with the resource (in this case, the dictionary), while all other threads send work requests to the specialized thread with a Queue.Queue instance. A Queue-based approach is more scalable when per-resource locking gets too complicated to manage easily: Queue is less bug-prone and, in particular, avoids worries about deadlocks. See Recipe 9.3, Recipe 9.5, Recipe 9.4, and Recipe 11.9 for more information about Queue and examples of using Queue to structure the architecture of a multithreaded program.

一个很好用于这种资源访问粒度的锁定情况的可选方案是专门制定一个线程来使用资源(在这个例子中的字典),其他线程通过一个Queue.Queue实例发送工作请求给这个特殊的线程.一个基于队列的方法在难以控制每个资源的锁定时更具有扩展性.队列会出现更少的bug而且实际能避免死锁.参考食谱9.3,食谱9.5,食谱9.4已经食谱11.9来获取更多关于队列的信息以及使用队列来结构化构建一个多线程程序的例子.

Asynchronous server

The Twisted server employs an asynchronous, event-driven model based on the Twisted framework (http://www.twistedmatrix.com/). The framework is built around a central "reactor" that dispatches events from a queue in a single thread, and monitors network and host resources. The user program is composed of short code fragments invoked by the reactor when dispatching the matching events. Such a working model guarantees that only one user code fragment is executing at any given time, eliminating at the root all problems of concurrent access to shared data structures. Asynchronous servers can provide excellent performance and scalability under very heavy loads, by avoiding the threading and locking overheads of multithreader servers.

Twisted服务器使用了一个基于Twisted框架 (http://www.twistedmatrix.com/)的异步事件驱动模型.这个框架是围绕着一个中心"反应堆"来搭建的,该"反应堆"从一个队列里分发事件给单个线程,并监控网络和主机资源.这个用户程序包括了"反应堆"分发相应事件的代码块.这个工作方式保证任何时刻都只有一个用户代码段在执行,从根本上消除了并发访问共享数据结构的所有问题.异步服务器避免了线程和锁定多线程服务器的费用,因此能在非常大的负载下提供良好的性能和可扩展性 .

The asynchronous server program presented in this recipe is composed of one application and two services, the UDPServer and the DetectorService, respectively. It is invoked at any command shell by means of the twistd command, with the following options:

 本配方中给出的异步服务器模型包括了一个应用和两个服务,分别是UDP服务器和观察者服务.可以再热河命令行脚本上通过twistd命令并松狮如下参数来调用它.

$ twistd -ony AsyncBeatServer.py

The twistd command controls the reactor, and many other delicate facets of a server's operation, leaving the script it loads the sole responsibility of defining a global variable named application, implementing the needed services, and connecting the service objects to the application object.

 twistd命令控制反应堆,以及其他服务器运行的微妙方面,让它所加载的脚本来定义一个叫做application的全局变量,实现所需服务,并且连接服务对象和应用对象.

Normally, twistd runs as a daemon and logs to a file (or to other logging facilities, depending on configuration options), but in this case, with the -ony flags, we're specifically asking twistd to run in the foreground and with logging to standard output, so we can better see what's going on. Note that the most popular file extension for scripts to be loaded by twistd is .tac, although in this recipe I have used the more generally familiar extension .py. The choice of file extension is just a convention, in this case: twistd can work with Python source files with any file extension, since you pass the full filename, extension included, as an explicit command-line argument anyway.

通常,twisted作为一个守护进程运行,并且往一个文件(或者其他的日志设备,取决于配置选项)里打日志,但是在这个例子中,由于使用了-ony 标志,我们指定twisted在前台运行并将日志记录到标准输出,这样我们能更好地看到它如何运行.注意twisted所加载的脚本的场景扩展名是.tac,虽然我在这个配方中使用的感觉通用的扩展名.py.这个文件扩展名的选择仅仅是一种转换,在这个例子中,twisted可以使用任意的文件扩展名和python源文件一起运行,因为你给出了完整的文件名和扩展名, 就像一个明确的命令行参数一样.

See Also

参考

Documentation for the standard library modules socket, tHReading, Queue and time in the Library Reference and Python in a Nutshell; twisted is at http://www.twistedmatrix.com; Jeff Bauer has a related program, known as Mr. Creosote (http://starship.python.net/crew/jbauer/creosote/), using UDP for logging information; UDP is described in depth in W. Richard Stevens, UNIX Network Programming, Volume 1: Networking APIs-Sockets and XTI, 2d ed. (Prentice-Hall); for the truly curious, the UDP protocol is defined in the two-page RFC 768 (http://www.ietf.org/rfc/rfc768.txt), which, when compared with current RFCs, shows how much the Internet infrastructure has evolved in 20 years.