gsh/stdin.py

   1 # This program is free software; you can redistribute it and/or modify
   2 # it under the terms of the GNU General Public License as published by
   3 # the Free Software Foundation; either version 2 of the License, or
   4 # (at your option) any later version.
   5 #
   6 # This program is distributed in the hope that it will be useful,
   7 # but WITHOUT ANY WARRANTY; without even the implied warranty of
   8 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   9 # GNU Library General Public License for more details.
  10 #
  11 # You should have received a copy of the GNU General Public License
  12 # along with this program; if not, write to the Free Software
  13 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  14 #
  15 # See the COPYING file for license information.
  16 #
  17 # Copyright (c) 2006, 2007, 2008 Guillaume Chazarain <guichaz@gmail.com>
  18
  19 import asyncore
  20 import errno
  21 import os
  22 import readline # Just to say we want to use it with raw_input
  23 import signal
  24 import socket
  25 import subprocess
  26 import sys
  27 import tempfile
  28 import termios
  29 from threading import Thread, Event, Lock
  30
  31 from gsh import dispatchers, remote_dispatcher
  32 from gsh.console import console_output, set_last_status_length
  33 from gsh import completion
  34
  35 class input_buffer(object):
  36     """The shared input buffer between the main thread and the stdin thread"""
  37     def __init__(self):
  38         self.lock = Lock()
  39         self.buf = ''
  40
  41     def add(self, data):
  42         """Add data to the buffer"""
  43         self.lock.acquire()
  44         try:
  45             self.buf += data
  46         finally:
  47             self.lock.release()
  48
  49     def get(self):
  50         """Get the content of the buffer"""
  51         self.lock.acquire()
  52         try:
  53             data = self.buf
  54             if data:
  55                 self.buf = ''
  56                 return data
  57         finally:
  58             self.lock.release()
  59
  60 def ignore_sigchld(ignore):
  61     """Typically we don't want to create zombie. But when executing a user
  62     command (!command) the subprocess module relies on zombies not being
  63     automatically reclaimed"""
  64     if ignore:
  65         signal.signal(signal.SIGCHLD, signal.SIG_IGN)
  66         # Reclaim previously created zombies
  67         try:
  68             while os.waitpid(-1, os.WNOHANG) != (0, 0):
  69                 pass
  70         except OSError, e:
  71             if e.errno != errno.ECHILD:
  72                 raise
  73     else:
  74         signal.signal(signal.SIGCHLD, signal.SIG_DFL)
  75
  76 def process_input_buffer():
  77     """Send the content of the input buffer to all remote processes, this must
  78     be called in the main thread"""
  79     from gsh.control_commands_helpers import handle_control_command
  80     data = the_stdin_thread.input_buffer.get()
  81     if not data:
  82         return
  83
  84     remote_dispatcher.log('> ' + data)
  85
  86     if data.startswith(':'):
  87         handle_control_command(data[1:-1])
  88         return
  89
  90     if data.startswith('!'):
  91         ignore_sigchld(False)
  92         retcode = subprocess.call(data[1:], shell=True)
  93         ignore_sigchld(True)
  94         if retcode > 0:
  95             console_output('Child returned %d\n' % retcode)
  96         elif retcode < 0:
  97             console_output('Child was terminated by signal %d\n' % -retcode)
  98         return
  99
 100     for r in dispatchers.all_instances():
 101         try:
 102             r.dispatch_command(data)
 103         except asyncore.ExitNow, e:
 104             raise e
 105         except Exception, msg:
 106             console_output('%s for %s, disconnecting\n' % (msg, r.display_name))
 107             r.disconnect()
 108         else:
 109             if r.enabled and r.state is remote_dispatcher.STATE_IDLE:
 110                 r.change_state(remote_dispatcher.STATE_RUNNING)
 111
 112 # The stdin thread uses a synchronous (with ACK) socket to communicate with the
 113 # main thread, which is most of the time waiting in the poll() loop.
 114 # Socket character protocol:
 115 # d: there is new data to send
 116 # A: ACK, same reply for every message, communications are synchronous, so the
 117 # stdin thread sends a character to the socket, the main thread processes it,
 118 # sends the ACK, and the stdin thread can go on.
 119
 120 class socket_notification_reader(asyncore.dispatcher):
 121     """The socket reader in the main thread"""
 122     def __init__(self):
 123         asyncore.dispatcher.__init__(self, the_stdin_thread.socket_read)
 124
 125     def _do(self, c):
 126         if c == 'd':
 127             process_input_buffer()
 128         else:
 129             raise Exception, 'Unknown code: %s' % (c)
 130
 131     def handle_read(self):
 132         """Handle all the available character commands in the socket"""
 133         while True:
 134             try:
 135                 c = self.recv(1)
 136             except socket.error, why:
 137                 assert why[0] == errno.EWOULDBLOCK
 138                 return
 139             else:
 140                 self._do(c)
 141                 self.socket.setblocking(True)
 142                 self.send('A')
 143                 self.socket.setblocking(False)
 144
 145     def writable(self):
 146         """Our writes are blocking"""
 147         return False
 148
 149 def write_main_socket(c):
 150     """Synchronous write to the main socket, wait for ACK"""
 151     the_stdin_thread.socket_write.send(c)
 152     while True:
 153         try:
 154             the_stdin_thread.socket_write.recv(1)
 155         except socket.error, e:
 156             assert e[0] == errno.EINTR
 157         else:
 158             break
 159
 160 #
 161 # This file descriptor is used to interrupt readline in raw_input().
 162 # /dev/null is not enough as it does not get out of a 'Ctrl-R' reverse-i-search.
 163 # A Ctrl-C seems to make raw_input() return in all cases, and avoids printing
 164 # a newline
 165 tempfile_fd, tempfile_name = tempfile.mkstemp()
 166 os.remove(tempfile_name)
 167 os.write(tempfile_fd, chr(3))
 168
 169 def get_stdin_pid(cached_result=None):
 170     '''Try to get the PID of the stdin thread, otherwise get the whole process
 171     ID'''
 172     if cached_result is None:
 173         try:
 174             tasks = os.listdir('/proc/self/task')
 175         except OSError, e:
 176             if e.errno != errno.ENOENT:
 177                 raise
 178             cached_result = os.getpid()
 179         else:
 180             tasks.remove(str(os.getpid()))
 181             assert len(tasks) == 1
 182             cached_result = int(tasks[0])
 183     return cached_result
 184
 185 def interrupt_stdin_thread():
 186     """The stdin thread may be in raw_input(), get out of it"""
 187     dupped_stdin = os.dup(0) # Backup the stdin fd
 188     assert not the_stdin_thread.interrupt_asked # Sanity check
 189     the_stdin_thread.interrupt_asked = True # Not user triggered
 190     os.lseek(tempfile_fd, 0, 0) # Rewind in the temp file
 191     os.dup2(tempfile_fd, 0) # This will make raw_input() return
 192     pid = get_stdin_pid()
 193     os.kill(pid, signal.SIGWINCH) # Try harder to wake up raw_input()
 194     the_stdin_thread.out_of_raw_input.wait() # Wait for this return
 195     the_stdin_thread.interrupt_asked = False # Restore sanity
 196     os.dup2(dupped_stdin, 0) # Restore stdin
 197     os.close(dupped_stdin) # Cleanup
 198
 199 echo_enabled = True
 200 def set_echo(echo):
 201     global echo_enabled
 202     if echo != echo_enabled:
 203         fd = sys.stdin.fileno()
 204         attr = termios.tcgetattr(fd)
 205         if echo:
 206             attr[3] |= termios.ECHO
 207         else:
 208             attr[3] &= ~termios.ECHO
 209         termios.tcsetattr(fd, termios.TCSANOW, attr)
 210         echo_enabled = echo
 211
 212 class stdin_thread(Thread):
 213     """The stdin thread, used to call raw_input()"""
 214     def __init__(self):
 215         Thread.__init__(self, name='stdin thread')
 216         completion.install_completion_handler()
 217
 218     @staticmethod
 219     def activate(interactive):
 220         """Activate the thread at initialization time"""
 221         the_stdin_thread.input_buffer = input_buffer()
 222         if interactive:
 223             the_stdin_thread.raw_input_wanted = Event()
 224             the_stdin_thread.in_raw_input = Event()
 225             the_stdin_thread.out_of_raw_input = Event()
 226             the_stdin_thread.out_of_raw_input.set()
 227             s1, s2 = socket.socketpair()
 228             the_stdin_thread.socket_read, the_stdin_thread.socket_write = s1, s2
 229             the_stdin_thread.interrupt_asked = False
 230             the_stdin_thread.setDaemon(True)
 231             the_stdin_thread.start()
 232             the_stdin_thread.socket_notification = socket_notification_reader()
 233
 234     def want_raw_input(self):
 235         nr, total = dispatchers.count_awaited_processes()
 236         if nr:
 237             prompt = 'waiting (%d/%d)> ' % (nr, total)
 238         else:
 239             prompt = 'ready (%d)> ' % total
 240         self.prompt = prompt
 241         set_last_status_length(len(prompt))
 242         self.raw_input_wanted.set()
 243         self.socket_notification.handle_read()
 244         self.in_raw_input.wait()
 245         self.raw_input_wanted.clear()
 246
 247     def no_raw_input(self):
 248         if not self.out_of_raw_input.isSet():
 249             interrupt_stdin_thread()
 250
 251     # Beware of races
 252     def run(self):
 253         while True:
 254             self.raw_input_wanted.wait()
 255             self.out_of_raw_input.set()
 256             self.in_raw_input.set()
 257             self.out_of_raw_input.clear()
 258             cmd = None
 259             try:
 260                 cmd = raw_input(self.prompt)
 261             except EOFError:
 262                 if not self.interrupt_asked:
 263                     cmd = ':quit'
 264             if self.interrupt_asked:
 265                 cmd = None
 266             self.in_raw_input.clear()
 267             self.out_of_raw_input.set()
 268             if cmd:
 269                 if echo_enabled:
 270                     completion.add_to_history(cmd)
 271                 else:
 272                     completion.remove_last_history_item()
 273             set_echo(True)
 274             if cmd is not None:
 275                 self.input_buffer.add(cmd + '\n')
 276                 write_main_socket('d')
 277
 278 the_stdin_thread = stdin_thread()