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.
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.
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.
15 # See the COPYING file for license information.
17 # Copyright (c) 2006, 2007 Guillaume Chazarain <guichaz@yahoo.fr>
24 import readline
# Just to say we want to use it with raw_input
27 from threading
import Thread
, Event
, Lock
29 from gsh
import remote_dispatcher
30 from gsh
.console
import console_output
32 # Handling of stdin is certainly the most complex part of gsh
34 stdin_fcntl_flags
= fcntl
.fcntl(0, fcntl
.F_GETFL
, 0)
36 def set_stdin_blocking(blocking
):
37 """We set O_NONBLOCK on stdin only when we read from it"""
39 flags
= stdin_fcntl_flags
& ~os
.O_NONBLOCK
41 flags
= stdin_fcntl_flags | os
.O_NONBLOCK
42 fcntl
.fcntl(0, fcntl
.F_SETFL
, flags
)
44 class stdin_dispatcher(asyncore
.file_dispatcher
):
45 """The stdin reader in the main thread => no fancy editing"""
47 asyncore
.file_dispatcher
.__init
__(self
, 0)
48 self
.is_readable
= True
49 atexit
.register(fcntl
.fcntl
, 0, fcntl
.F_SETFL
, stdin_fcntl_flags
)
50 set_stdin_blocking(True)
53 """We set it to be readable only when the stdin thread is not in
55 return self
.is_readable
57 def handle_expt(self
):
58 # Emulate the select with poll as in: asyncore.loop(use_poll=True)
62 """We don't write to stdin"""
65 def handle_close(self
):
66 """Ctrl-D was received but the remote processes were not ready"""
67 remote_dispatcher
.dispatch_termination_to_all()
69 def handle_read(self
):
70 """Some data can be read on stdin"""
73 set_stdin_blocking(False)
75 data
= self
.recv(4096)
77 set_stdin_blocking(True)
79 if e
.errno
== errno
.EAGAIN
:
80 # End of available data
86 # Handle the just read data
87 the_stdin_thread
.input_buffer
.add(data
)
88 process_input_buffer()
91 self
.is_readable
= False
94 class input_buffer(object):
95 """The shared input buffer between the main thread and the stdin thread"""
101 """Add data to the buffer"""
109 """Get the content of the buffer"""
119 def process_input_buffer():
120 """Send the content of the input buffer to all remote processes, this must
121 be called in the main thread"""
122 data
= the_stdin_thread
.input_buffer
.get()
125 for r
in remote_dispatcher
.all_instances():
127 r
.dispatch_write(data
)
128 except Exception, msg
:
129 console_output('%s for %s, disconnecting\n' % (msg
, r
.display_name
),
135 if r
.enabled
and r
.state
is remote_dispatcher
.STATE_IDLE
:
136 r
.change_state(remote_dispatcher
.STATE_EXPECTING_NEXT_LINE
)
138 # The stdin thread uses a synchronous (with ACK) socket to communicate with the
139 # main thread, which is most of the time waiting in the poll() loop.
140 # Socket character protocol:
141 # s: entering in raw_input, the main loop should not read stdin
142 # e: leaving raw_input, the main loop can read stdin
143 # q: Ctrl-D was pressed, exiting
144 # d: there is new data to send
145 # A: ACK, same reply for every message, communications are synchronous, so the
146 # stdin thread sends a character to the socket, the main thread processes it,
147 # sends the ACK, and the stdin thread can go on.
149 class socket_notification_reader(asyncore
.dispatcher
):
150 """The socket reader in the main thread"""
152 asyncore
.dispatcher
.__init
__(self
, the_stdin_thread
.socket_read
)
156 the_stdin_dispatcher
.is_readable
= c
== 'e'
159 remote_dispatcher
.dispatch_termination_to_all()
161 process_input_buffer()
163 raise Exception, 'Unknown code: %s' % (c
)
165 def handle_read(self
):
166 """Handle all the available character commands in the socket"""
170 except socket
.error
, why
:
171 assert why
[0] == errno
.EWOULDBLOCK
175 self
.socket
.setblocking(True)
177 self
.socket
.setblocking(False)
180 """Our writes are blocking"""
183 # All the words that have been typed in gsh. Used by the completion mechanism.
184 history_words
= set()
186 def complete(text
, state
):
187 """On tab press, return the next possible completion"""
189 matches
= [w
for w
in history_words
if len(w
) > l
and w
.startswith(text
)]
190 if state
<= len(matches
):
191 return matches
[state
]
193 def write_main_socket(c
):
194 """Synchronous write to the main socket, wait for ACK"""
195 the_stdin_thread
.socket_write
.send(c
)
197 the_stdin_thread
.socket_write
.recv(1)
199 class stdin_thread(Thread
):
200 """The stdin thread, used to call raw_input()"""
202 Thread
.__init
__(self
, name
='stdin thread')
205 def activate(interactive
):
206 """Activate the thread at initialization time"""
207 the_stdin_thread
.ready_event
= Event()
208 the_stdin_thread
.input_buffer
= input_buffer()
210 the_stdin_thread
.interrupted_event
= Event()
211 s1
, s2
= socket
.socketpair()
212 the_stdin_thread
.socket_read
, the_stdin_thread
.socket_write
= s1
, s2
213 the_stdin_thread
.wants_control_shell
= False
214 the_stdin_thread
.setDaemon(True)
215 the_stdin_thread
.start()
216 socket_notification_reader()
218 the_stdin_thread
.ready_event
.set()
223 self
.ready_event
.wait()
224 nr
, total
= remote_dispatcher
.count_completed_processes()
227 # The remote processes are ready, the thread can call raw_input
228 self
.interrupted_event
.clear()
231 write_main_socket('s')
232 readline
.set_completer(complete
)
233 readline
.parse_and_bind('tab: complete')
234 readline
.set_completer_delims(' \t\n')
235 cmd
= raw_input('ready (%d)> ' % (nr
))
236 if self
.wants_control_shell
:
237 # This seems to be needed if Ctrl-C is hit when some
238 # text is in the line buffer
240 if len(history_words
) < 10000:
241 words
= [w
+ ' ' for w
in cmd
.split() if len(w
) > 1]
242 history_words
.update(words
)
244 if not self
.wants_control_shell
:
245 write_main_socket('e')
247 if self
.wants_control_shell
:
248 self
.ready_event
.clear()
249 # Ok, we are no more in raw_input(), tell it to the
251 self
.interrupted_event
.set()
253 write_main_socket('q')
256 self
.ready_event
.clear()
257 self
.input_buffer
.add(cmd
+ '\n')
258 write_main_socket('d')
260 the_stdin_thread
= stdin_thread()
261 the_stdin_dispatcher
= stdin_dispatcher()