Source Code for Module flumotion.twisted.fdserver

  1  # -*- Mode: Python -*- 
  2  # vi:si:et:sw=4:sts=4:ts=4 
  3  # 
  4  # Flumotion - a streaming media server 
  5  # Copyright (C) 2004,2005,2006,2007 Fluendo, S.L. (www.fluendo.com). 
  6  # All rights reserved. 
  7   
  8  # This file may be distributed and/or modified under the terms of 
  9  # the GNU General Public License version 2 as published by 
 10  # the Free Software Foundation. 
 11  # This file is distributed without any warranty; without even the implied 
 12  # warranty of merchantability or fitness for a particular purpose. 
 13  # See "LICENSE.GPL" in the source distribution for more information. 
 14   
 15  # Licensees having purchased or holding a valid Flumotion Advanced 
 16  # Streaming Server license may use this file in accordance with the 
 17  # Flumotion Advanced Streaming Server Commercial License Agreement. 
 18  # See "LICENSE.Flumotion" in the source distribution for more information. 
 19   
 20  # Headers in this file shall remain intact. 
 21   
 22  from flumotion.common import log 
 23  from flumotion.extern.fdpass import fdpass 
 24   
 25  from twisted.internet import unix, main, address, tcp 
 26  from twisted.spread import pb 
 27   
 28  import errno 
 29  import os 
 30  import socket 
 31  import struct 
 32   
 33  __version__ = "$Rev: 6125 $" 
 34   
 35   
 36  # Heavily based on 
 37  # http://twistedmatrix.com/trac/browser/sandbox/exarkun/copyover/server.py 
 38  # and client.py 
 39  # Thanks for the inspiration! 
 40   
 41  # Since we're doing this over a stream socket, our file descriptor messages 
 42  # aren't guaranteed to be received alone; they could arrive along with some 
 43  # unrelated data. 
 44  # So, we prefix the message with a 16 byte magic signature, and a length, 
 45  # and if we receive file descriptors decode based on this. 
 46  # 
 47  # map() instead of a string to workaround gettext encoding problems. 
 48  # 
 49  MAGIC_SIGNATURE = ''.join(map(chr, [253, 252, 142, 127, 7, 71, 185, 234, 
 50                                      161, 117, 238, 216, 220, 54, 200, 163])) 
 51   
 52 -class FDServer(unix.Server): 
 53 -    def sendFileDescriptor(self, fileno, data=""): 
 54          message = struct.pack("@16sI", MAGIC_SIGNATURE, len(data)) + data 
 55          return fdpass.writefds(self.fileno(), [fileno], message) 
 56   
 57 -class FDPort(unix.Port): 
 58      transport = FDServer 
 59   
 60 -class FDClient(unix.Client): #, log.Loggable): 
 61   
 62 -    def doRead(self): 
 63          if not self.connected: 
 64              return 
 65          try: 
 66              (fds, message) = fdpass.readfds(self.fileno(), 64 * 1024) 
 67          except socket.error, se: 
 68              if se.args[0] == errno.EWOULDBLOCK: 
 69                  return 
 70              else: 
 71                  return main.CONNECTION_LOST 
 72          else: 
 73              if not message: 
 74                  return main.CONNECTION_DONE 
 75   
 76              if len(fds) > 0: 
 77                  # Look for our magic cookie in (possibly) the midst of other 
 78                  # data. Pass surrounding chunks, if any, onto dataReceived(), 
 79                  # which (undocumentedly) must return None unless a failure 
 80                  # occurred. 
 81                  # Pass the actual FDs and their message to 
 82                  # fileDescriptorsReceived() 
 83                  offset = message.find(MAGIC_SIGNATURE) 
 84                  if offset < 0: 
 85                      # Old servers did not send this; be hopeful that this 
 86                      # doesn't have bits of other protocol (i.e. PB) mixed up 
 87                      # in it. 
 88                      return self.protocol.fileDescriptorsReceived(fds, message) 
 89                  elif offset > 0: 
 90                      ret = self.protocol.dataReceived(message[0:offset]) 
 91                      if ret: 
 92                          return ret 
 93   
 94                  msglen = struct.unpack("@I", message[offset+16:offset+20])[0] 
 95                  offset += 20 
 96                  ret = self.protocol.fileDescriptorsReceived(fds, 
 97                      message[offset:offset+msglen]) 
 98                  if ret: 
 99                      return ret 
100   
101                  if offset+msglen < len(message): 
102                      return self.protocol.dataReceived(message[offset+msglen:]) 
103                  return ret 
104              else: 
105                #  self.debug("No FDs, passing to dataReceived") 
106                  return self.protocol.dataReceived(message) 
107   
108 -class FDConnector(unix.Connector): 
109 -    def _makeTransport(self): 
110          return FDClient(self.address, self, self.reactor) 
111   
112 -class FDPassingBroker(pb.Broker, log.Loggable): 
113      """ 
114      A pb.Broker subclass that handles FDs being passed to it (with associated 
115      data) over the same connection as the normal PB data stream. 
116      When an FD is seen, it creates new protocol objects for them from the 
117      childFactory attribute. 
118      """ 
119      # FIXME: looks like we can only use our own subclasses that take 
120      # three __init__ args 
121 -    def __init__(self, childFactory, connectionClass, **kwargs): 
122          """ 
123          @param connectionClass: a subclass of L{twisted.internet.tcp.Connection} 
124          """ 
125          pb.Broker.__init__(self, **kwargs) 
126   
127          self.childFactory = childFactory 
128          self._connectionClass = connectionClass 
129   
130      # This is the complex bit. If our underlying transport receives a file 
131      # descriptor, this gets called - along with the data we got with the FD. 
132      # We create an appropriate protocol object, and attach it to the reactor. 
133 -    def fileDescriptorsReceived(self, fds, message): 
134          if len(fds) == 1: 
135              fd = fds[0] 
136   
137              # Note that we hardcode IPv4 here! 
138              sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM) 
139   
140              self.debug("Received FD %d->%d" % (fd, sock.fileno())) 
141   
142              # Undocumentedly (other than a comment in 
143              # Python/Modules/socketmodule.c), socket.fromfd() calls dup() on 
144              # the passed FD before it actually wraps it in a socket object. 
145              # So, we need to close the FD that we originally had... 
146              os.close(fd) 
147   
148              try: 
149                  peeraddr = sock.getpeername() 
150              except socket.error: 
151                  self.info("Socket disconnected before being passed to client") 
152                  sock.close() 
153                  return 
154   
155              # Based on bits in tcp.Port.doRead() 
156              addr = address._ServerFactoryIPv4Address('TCP', 
157                  peeraddr[0], peeraddr[1]) 
158              protocol = self.childFactory.buildProtocol(addr) 
159   
160              self._connectionClass(sock, protocol, peeraddr, message) 
161          else: 
162              self.warning("Unexpected: FD-passing message with len(fds) != 1") 
163   
164 -class _SocketMaybeCloser(tcp._SocketCloser): 
165      keepSocketAlive = False 
166   
167 -    def _closeSocket(self): 
168          # We override this (from tcp._SocketCloser) so that we can close sockets 
169          # properly in the normal case, but once we've passed our socket on via 
170          # the FD-channel, we just close() it (not calling shutdown() which will 
171          # close the TCP channel without closing the FD itself) 
172          if self.keepSocketAlive: 
173              try: 
174                  self.socket.close() 
175              except socket.error: 
176                  pass 
177          else: 
178              tcp.Server._closeSocket(self) 
179   
180 -class PassableServerConnection(_SocketMaybeCloser, tcp.Server): 
181      """ 
182      A subclass of tcp.Server that permits passing the FDs used to other 
183      processes (by just calling close(2) rather than shutdown(2) on them) 
184      """ 
185      pass 
186   
187 -class PassableServerPort(tcp.Port): 
188      transport = PassableServerConnection 
189