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 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 process_input_buffer():
  61     """Send the content of the input buffer to all remote processes, this must
  62     be called in the main thread"""
  63     from gsh.control_commands_helpers import handle_control_command
  64     data = the_stdin_thread.input_buffer.get()
  65     remote_dispatcher.log('> ' + data)
  66
  67     if data.startswith(':'):
  68         handle_control_command(data[1:-1])
  69         return
  70
  71     if data.startswith('!'):
  72         try:
  73             retcode = subprocess.call(data[1:], shell=True)
  74         except OSError, e:
  75             if e.errno == errno.EINTR:
  76                 console_output('Child was interrupted\n')
  77                 retcode = 0
  78             else:
  79                 raise
  80         if retcode > 128 and retcode <= 192:
  81             retcode = 128 - retcode
  82         if retcode > 0:
  83             console_output('Child returned %d\n' % retcode)
  84         elif retcode < 0:
  85             console_output('Child was terminated by signal %d\n' % -retcode)
  86         return
  87
  88     for r in dispatchers.all_instances():
  89         try:
  90             r.dispatch_command(data)
  91         except asyncore.ExitNow, e:
  92             raise e
  93         except Exception, msg:
  94             console_output('%s for %s, disconnecting\n' % (msg, r.display_name))
  95             r.disconnect()
  96         else:
  97             if r.enabled and r.state is remote_dispatcher.STATE_IDLE:
  98                 r.change_state(remote_dispatcher.STATE_RUNNING)
  99
 100 # The stdin thread uses a synchronous (with ACK) socket to communicate with the
 101 # main thread, which is most of the time waiting in the poll() loop.
 102 # Socket character protocol:
 103 # d: there is new data to send
 104 # A: ACK, same reply for every message, communications are synchronous, so the
 105 # stdin thread sends a character to the socket, the main thread processes it,
 106 # sends the ACK, and the stdin thread can go on.
 107
 108 class socket_notification_reader(asyncore.dispatcher):
 109     """The socket reader in the main thread"""
 110     def __init__(self):
 111         asyncore.dispatcher.__init__(self, the_stdin_thread.socket_read)
 112
 113     def _do(self, c):
 114         if c == 'd':
 115             process_input_buffer()
 116         else:
 117             raise Exception, 'Unknown code: %s' % (c)
 118
 119     def handle_read(self):
 120         """Handle all the available character commands in the socket"""
 121         while True:
 122             try:
 123                 c = self.recv(1)
 124             except socket.error, why:
 125                 if why[0] == errno.EWOULDBLOCK:
 126                     return
 127                 else:
 128                     raise
 129             else:
 130                 self._do(c)
 131                 self.socket.setblocking(True)
 132                 self.send('A')
 133                 self.socket.setblocking(False)
 134
 135     def writable(self):
 136         """Our writes are blocking"""
 137         return False
 138
 139 def write_main_socket(c):
 140     """Synchronous write to the main socket, wait for ACK"""
 141     the_stdin_thread.socket_write.send(c)
 142     while True:
 143         try:
 144             the_stdin_thread.socket_write.recv(1)
 145         except socket.error, e:
 146             if e[0] != errno.EINTR:
 147                 raise
 148         else:
 149             break
 150
 151 #
 152 # This file descriptor is used to interrupt readline in raw_input().
 153 # /dev/null is not enough as it does not get out of a 'Ctrl-R' reverse-i-search.
 154 # A Ctrl-C seems to make raw_input() return in all cases, and avoids printing
 155 # a newline
 156 tempfile_fd, tempfile_name = tempfile.mkstemp()
 157 os.remove(tempfile_name)
 158 os.write(tempfile_fd, chr(3))
 159
 160 def get_stdin_pid(cached_result=None):
 161     """Try to get the PID of the stdin thread, otherwise get the whole process
 162     ID"""
 163     if cached_result is None:
 164         try:
 165             tasks = os.listdir('/proc/self/task')
 166         except OSError, e:
 167             if e.errno != errno.ENOENT:
 168                 raise
 169             cached_result = os.getpid()
 170         else:
 171             tasks.remove(str(os.getpid()))
 172             assert len(tasks) == 1
 173             cached_result = int(tasks[0])
 174     return cached_result
 175
 176 def interrupt_stdin_thread():
 177     """The stdin thread may be in raw_input(), get out of it"""
 178     dupped_stdin = os.dup(0) # Backup the stdin fd
 179     assert not the_stdin_thread.interrupt_asked # Sanity check
 180     the_stdin_thread.interrupt_asked = True # Not user triggered
 181     os.lseek(tempfile_fd, 0, 0) # Rewind in the temp file
 182     os.dup2(tempfile_fd, 0) # This will make raw_input() return
 183     pid = get_stdin_pid()
 184     os.kill(pid, signal.SIGWINCH) # Try harder to wake up raw_input()
 185     the_stdin_thread.out_of_raw_input.wait() # Wait for this return
 186     the_stdin_thread.interrupt_asked = False # Restore sanity
 187     os.dup2(dupped_stdin, 0) # Restore stdin
 188     os.close(dupped_stdin) # Cleanup
 189
 190 echo_enabled = True
 191 def set_echo(echo):
 192     global echo_enabled
 193     if echo != echo_enabled:
 194         fd = sys.stdin.fileno()
 195         attr = termios.tcgetattr(fd)
 196         if echo:
 197             attr[3] |= termios.ECHO
 198         else:
 199             attr[3] &= ~termios.ECHO
 200         termios.tcsetattr(fd, termios.TCSANOW, attr)
 201         echo_enabled = echo
 202
 203 class stdin_thread(Thread):
 204     """The stdin thread, used to call raw_input()"""
 205     def __init__(self):
 206         Thread.__init__(self, name='stdin thread')
 207         completion.install_completion_handler()
 208
 209     @staticmethod
 210     def activate(interactive):
 211         """Activate the thread at initialization time"""
 212         the_stdin_thread.input_buffer = input_buffer()
 213         if interactive:
 214             the_stdin_thread.raw_input_wanted = Event()
 215             the_stdin_thread.in_raw_input = Event()
 216             the_stdin_thread.out_of_raw_input = Event()
 217             the_stdin_thread.out_of_raw_input.set()
 218             s1, s2 = socket.socketpair()
 219             the_stdin_thread.socket_read, the_stdin_thread.socket_write = s1, s2
 220             the_stdin_thread.interrupt_asked = False
 221             the_stdin_thread.setDaemon(True)
 222             the_stdin_thread.start()
 223             the_stdin_thread.socket_notification = socket_notification_reader()
 224             the_stdin_thread.prepend_text = None
 225             readline.set_pre_input_hook(the_stdin_thread.prepend_previous_text)
 226
 227     def prepend_previous_text(self):
 228         if self.prepend_text:
 229             readline.insert_text(self.prepend_text)
 230             readline.redisplay()
 231             self.prepend_text = None
 232
 233     def want_raw_input(self):
 234         nr, total = dispatchers.count_awaited_processes()
 235         if nr:
 236             prompt = 'waiting (%d/%d)> ' % (nr, total)
 237         else:
 238             prompt = 'ready (%d)> ' % total
 239         self.prompt = prompt
 240         set_last_status_length(len(prompt))
 241         self.raw_input_wanted.set()
 242         while not self.in_raw_input.isSet():
 243             self.socket_notification.handle_read()
 244             self.in_raw_input.wait(0.1)
 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 self.interrupt_asked:
 263                     cmd = readline.get_line_buffer()
 264                 else:
 265                     cmd = chr(4) # Ctrl-D
 266             if self.interrupt_asked:
 267                 self.prepend_text = cmd
 268                 cmd = None
 269             self.in_raw_input.clear()
 270             self.out_of_raw_input.set()
 271             if cmd:
 272                 if echo_enabled:
 273                     completion.add_to_history(cmd)
 274                 else:
 275                     completion.remove_last_history_item()
 276             set_echo(True)
 277             if cmd is not None:
 278                 self.input_buffer.add(cmd + '\n')
 279                 write_main_socket('d')
 280
 281 the_stdin_thread = stdin_thread()