Source Code for Module x2go.gevent_subprocess

  1  # -*- coding: utf-8 -*- 
  2   
  3  # Copyright (C) 2010-2013 by Mike Gabriel <mike.gabriel@das-netzwerkteam.de> 
  4  # 
  5  # Python X2Go is free software; you can redistribute it and/or modify 
  6  # it under the terms of the GNU Affero General Public License as published by 
  7  # the Free Software Foundation; either version 3 of the License, or 
  8  # (at your option) any later version. 
  9  # 
 10  # Python X2Go is distributed in the hope that it will be useful, 
 11  # but WITHOUT ANY WARRANTY; without even the implied warranty of 
 12  # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
 13  # GNU Affero General Public License for more details. 
 14  # 
 15  # You should have received a copy of the GNU Affero General Public License 
 16  # along with this program; if not, write to the 
 17  # Free Software Foundation, Inc., 
 18  # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA. 
 19   
 20  # gevent_subprocess was found here: http://groups.google.com/group/gevent/browse_thread/thread/dba1a5d29e0a60ff 
 21  # Mark Visser <mjmvisser@gmail.com> 
 22   
 23  ### 
 24  ### Mark Visser, Sat, 20 Nov 2010 13:30:16 -0500 
 25  ### 
 26  ### <quote> 
 27  ### Hi Mike, 
 28  ### 
 29  ### I hereby place that code snippet in the public domain, feel free to apply any license that is appropriate! 
 30  ### 
 31  ### cheers, 
 32  ### -Mark 
 33  ### </quote> 
 34  ### 
 35  ### Thus, I place myself as the copyright holder for code in this file 
 36  ### for cases in that it is used in context of the X2Go project. 
 37  ### 
 38   
 39   
 40  """Implementation of the standard :mod:`subprocess` module that spawns greenlets""" 
 41  import errno 
 42  import sys 
 43  import fcntl, os 
 44   
 45  _subprocess = __import__('subprocess') 
 46   
 47  from gevent import socket, select, hub 
 48   
 49  # identical to original 
 50  CalledProcessError = _subprocess.CalledProcessError 
 51  MAXFD = _subprocess.MAXFD 
 52  PIPE = _subprocess.PIPE 
 53  STDOUT = _subprocess.STDOUT 
 54  call = _subprocess.call 
 55  check_call = _subprocess.check_call 
 56  list2cmdline = _subprocess.list2cmdline 
 57   
 58   
 59 -class Popen(object): 
 60 -    def __init__(self, *args, **kwargs): 
 61          # delegate to an actual Popen object 
 62          self.__p = _subprocess.Popen(*args, **kwargs) 
 63          # make the file handles nonblocking 
 64          if self.stdin is not None: 
 65              fcntl.fcntl(self.stdin, fcntl.F_SETFL, os.O_NONBLOCK) 
 66          if self.stdout is not None: 
 67              fcntl.fcntl(self.stdout, fcntl.F_SETFL, os.O_NONBLOCK) 
 68          if self.stderr is not None: 
 69              fcntl.fcntl(self.stderr, fcntl.F_SETFL, os.O_NONBLOCK) 
 70   
 71 -    def __getattr__(self, name): 
 72          # delegate attribute lookup to the real Popen object 
 73          return getattr(self.__p, name) 
 74   
 75 -    def _write_pipe(self, f, input): 
 76          # writes the given input to f without blocking 
 77          if input: 
 78              bytes_total = len(input) 
 79              bytes_written = 0 
 80              while bytes_written < bytes_total: 
 81                  try: 
 82                      # f.write() doesn't return anything, so use os.write. 
 83                      bytes_written += os.write(f.fileno(), input[bytes_written:]) 
 84                  except IOError, ex: 
 85                      if ex[0] != errno.EAGAIN: 
 86                          raise 
 87                      sys.exc_clear() 
 88                  socket.wait_write(f.fileno()) 
 89          f.close() 
 90   
 91 -    def _read_pipe(self, f): 
 92          # reads output from f without blocking 
 93          # returns output 
 94          chunks = [] 
 95          while True: 
 96              try: 
 97                  chunk = f.read(4096) 
 98                  if not chunk: 
 99                      break 
100                  chunks.append(chunk) 
101              except IOError, ex: 
102                  if ex[0] != errno.EAGAIN: 
103                      raise 
104                  sys.exc_clear() 
105              socket.wait_read(f.fileno()) 
106          f.close() 
107          return ''.join(chunks) 
108   
109 -    def communicate(self, input=None): 
110          # Optimization: If we are only using one pipe, or no pipe at 
111          # all, using select() is unnecessary. 
112          if [self.stdin, self.stdout, self.stderr].count(None) >= 2: 
113              stdout = None 
114              stderr = None 
115              if self.stdin: 
116                  self._write_pipe(self.stdin, input) 
117              elif self.stdout: 
118                  stdout = self._read_pipe(self.stdout) 
119              elif self.stderr: 
120                  stderr = self._read_pipe(self.stderr) 
121              self.wait() 
122              return (stdout, stderr) 
123          else: 
124              return self._communicate(input) 
125   
126 -    def _communicate(self, input): 
127          # identical to original... all the heavy lifting is done 
128          # in gevent.select.select 
129          read_set = [] 
130          write_set = [] 
131          stdout = None # Return 
132          stderr = None # Return 
133   
134          if self.stdin: 
135              # Flush stdin buffer. 
136              self.stdin.flush() 
137              if input: 
138                  write_set.append(self.stdin) 
139              else: 
140                  self.stdin.close() 
141          if self.stdout: 
142              read_set.append(self.stdout) 
143              stdout = [] 
144          if self.stderr: 
145              read_set.append(self.stderr) 
146              stderr = [] 
147   
148          input_offset = 0 
149          while read_set or write_set: 
150              try: 
151                  rlist, wlist, xlist = select.select(read_set, write_set, []) 
152              except select.error, e: 
153                  if e.args[0] == errno.EINTR: 
154                      continue 
155                  raise 
156   
157              if self.stdin in wlist: 
158                  # When select has indicated that the file is writable, 
159                  # we can write up to PIPE_BUF bytes without risk 
160                  # blocking.  POSIX defines PIPE_BUF >= 512 
161                  bytes_written = os.write(self.stdin.fileno(), buffer(input, input_offset, 512)) 
162                  input_offset += bytes_written 
163                  if input_offset >= len(input): 
164                      self.stdin.close() 
165                      write_set.remove(self.stdin) 
166   
167              if self.stdout in rlist: 
168                  data = os.read(self.stdout.fileno(), 1024) 
169                  if data == "": 
170                      self.stdout.close() 
171                      read_set.remove(self.stdout) 
172                  stdout.append(data) 
173   
174              if self.stderr in rlist: 
175                  data = os.read(self.stderr.fileno(), 1024) 
176                  if data == "": 
177                      self.stderr.close() 
178                      read_set.remove(self.stderr) 
179                  stderr.append(data) 
180   
181          # All data exchanged.  Translate lists into strings. 
182          if stdout is not None: 
183              stdout = ''.join(stdout) 
184          if stderr is not None: 
185              stderr = ''.join(stderr) 
186   
187          # Translate newlines, if requested.  We cannot let the file 
188          # object do the translation: It is based on stdio, which is 
189          # impossible to combine with select (unless forcing no 
190          # buffering). 
191          if self.universal_newlines and hasattr(file, 'newlines'): 
192              if stdout: 
193                  stdout = self._translate_newlines(stdout) 
194              if stderr: 
195                  stderr = self._translate_newlines(stderr) 
196   
197          self.wait() 
198          return (stdout, stderr) 
199   
200 -    def wait(self, check_interval=0.01): 
201          # non-blocking, use hub.sleep 
202          try: 
203              while True: 
204                  status = self.poll() 
205                  if status >= 0: 
206                      return status 
207                  hub.sleep(check_interval) 
208          except OSError, e: 
209              if e.errno == errno.ECHILD: 
210                  # no child process, this happens if the child process 
211                  # already died and has been cleaned up 
212                  return -1 
213              else: 
214                  raise 
215