1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
#!/usr/bin/env python
# vim:fileencoding=utf-8
from __future__ import (unicode_literals, division, absolute_import,
print_function)
__license__ = 'GPL v3'
__copyright__ = '2014, Kovid Goyal <kovid at kovidgoyal.net>'
import pdb, socket, inspect, sys, select, os, atexit, time
from calibre import prints
from calibre.utils.ipc import eintr_retry_call
from calibre.constants import cache_dir
PROMPT = b'(debug) '
QUESTION = b'\x00\x01\x02'
class RemotePdb(pdb.Pdb):
def __init__(self, addr="127.0.0.1", port=4444, skip=None):
try:
prints("pdb is running on %s:%d" % (addr, port), file=sys.stderr)
except IOError:
pass
# Open a reusable socket to allow for reloads
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
self.sock.bind((addr, port))
self.sock.listen(1)
clientsocket, address = self.sock.accept()
self.handle = clientsocket.makefile('rw')
pdb.Pdb.__init__(self, completekey='tab', stdin=self.handle, stdout=self.handle, skip=skip)
self.prompt = PROMPT
def prints(self, *args, **kwargs):
kwargs['file'] = self.handle
prints(*args, **kwargs)
def ask_question(self, query):
self.handle.write(QUESTION)
self.prints(query, end='')
self.handle.write(PROMPT)
self.handle.flush()
return self.handle.readline()
def end_session(self, *args):
self.clear_all_breaks()
self.reset()
del self.handle
try:
self.sock.shutdown(socket.SHUT_RDWR)
self.sock.close()
except socket.error:
pass
return pdb.Pdb.do_continue(self, None)
def do_clear(self, arg):
if not arg:
ans = self.ask_question("Clear all breaks? [y/n]: ")
if ans.strip().lower() in {b'y', b'yes'}:
self.clear_all_breaks()
self.prints('All breaks cleared')
return
return pdb.Pdb.do_clear(self, arg)
do_cl = do_clear
def do_continue(self, arg):
if not self.breaks:
ans = self.ask_question(
'There are no breakpoints set. Continuing will terminate this debug session. Are you sure? [y/n]: ')
if ans.strip().lower() in {b'y', b'yes'}:
return self.end_session()
return
return pdb.Pdb.do_continue(self, arg)
do_c = do_cont = do_continue
do_EOF = do_quit = do_exit = do_q = end_session
def set_trace(port=4444, skip=None):
frame = inspect.currentframe().f_back
try:
debugger = RemotePdb(port=port, skip=skip)
debugger.set_trace(frame)
except KeyboardInterrupt:
prints('Debugging aborted by keyboard interrupt')
except Exception:
prints('Failed to run debugger')
import traceback
traceback.print_exc()
def cli(port=4444):
prints('Connecting to remote debugger on port %d...' % port)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
for i in xrange(20):
try:
sock.connect(('127.0.0.1', port))
break
except socket.error:
pass
time.sleep(0.1)
else:
try:
sock.connect(('127.0.0.1', port))
except socket.error as err:
prints('Failed to connect to remote debugger:', err, file=sys.stderr)
raise SystemExit(1)
prints('Connected to remote process')
import readline
histfile = os.path.join(cache_dir(), 'rpdb.history')
try:
readline.read_history_file(histfile)
except IOError:
pass
atexit.register(readline.write_history_file, histfile)
p = pdb.Pdb()
readline.set_completer(p.complete)
readline.parse_and_bind("tab: complete")
try:
while True:
recvd = b''
while not recvd.endswith(PROMPT) or select.select([sock], [], [], 0) == ([sock], [], []):
buf = eintr_retry_call(sock.recv, 16 * 1024)
if not buf:
return
recvd += buf
recvd = recvd[:-len(PROMPT)]
if recvd.startswith(QUESTION):
recvd = recvd[len(QUESTION):]
sys.stdout.write(recvd)
raw = sys.stdin.readline() or b'n'
else:
sys.stdout.write(recvd)
raw = b''
try:
raw = raw_input(PROMPT) + b'\n'
except (EOFError, KeyboardInterrupt):
pass
if not raw:
raw = b'quit\n'
eintr_retry_call(sock.send, raw)
except KeyboardInterrupt:
pass
if __name__ == '__main__':
cli()
|