Skip to content

Commit caa9a27

Browse files
DonJayamanneDonJayamanne
committed
running code in kernel
1 parent d0bccd6 commit caa9a27

3 files changed

Lines changed: 179 additions & 26 deletions

File tree

‎pythonFiles/PythonTools/ipythonServer.py‎

Lines changed: 119 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -45,9 +45,9 @@
4545
BaseException = Exception
4646

4747
try:
48-
from Queue import Empty # Python 2
48+
from Queue import Empty, Queue # Python 2
4949
except ImportError:
50-
from queue import Empty # Python 3
50+
from queue import Empty, Queue # Python 3
5151

5252
DEBUG = os.environ.get('DEBUG_DJAYAMANNE_IPYTHON') is not None
5353
TEST = os.environ.get('PYTHON_DONJAYAMANNE_TEST') is not None
@@ -128,12 +128,18 @@ def release(self):
128128
- Not the best, but that's how it will have to be done
129129
- http://www.xavierdupre.fr/app/pyquickhelper/helpsphinx/_modules/pyquickhelper/ipythonhelper/notebook_runner.html
130130
"""
131+
132+
131133
class iPythonKernelResponseMonitor(object):
132-
def __init__(self, kernelUUID, socketConnection):
134+
135+
def __init__(self, kernelUUID, socketConnection, send_lock, shell_channel, iopub_channel):
133136
import threading
134137
self.kernel = multiKernelManager.get_kernel(kernelUUID)
135138
self.conn = socketConnection
136139
self.is_stop_requested = False
140+
self.shell_channel = shell_channel
141+
self.iopub_channel = iopub_channel
142+
self.send_lock = send_lock
137143
thread.start_new_thread(self.start_processing, ())
138144

139145
def stop(self):
@@ -142,6 +148,14 @@ def stop(self):
142148
def check_for_exit_socket_loop(self):
143149
return self.is_stop_requested
144150

151+
def _populateErrorContents(self, sourceContents, targetContents):
152+
try:
153+
targetContents.ename = sourceContents.ename
154+
targetContents.evalue = sourceContents.evalue
155+
targetContents.traceback = sourceContents.traceback
156+
except AttributeError:
157+
pass
158+
145159
def start_processing(self):
146160
"""loop to read the io ports/messages"""
147161

@@ -151,6 +165,31 @@ def start_processing(self):
151165
if self.check_for_exit_socket_loop():
152166
break
153167

168+
try:
169+
# We can ignore msgtype=execute_request
170+
171+
exe_result = self.shell_channel.get_shell_msg(timeout=1)
172+
# message can be JSON, but not always
173+
# (http://jupyter-client.readthedocs.io/en/latest/messaging.html)
174+
# assume for now that dates are the only crappy (non JSONable stuff sent)
175+
json_to_send = json.dumps(exe_result, default=str)
176+
with self.send_lock:
177+
_debug_write('shell_result')
178+
write_bytes(self.conn, iPythonSocketServer._SHEL)
179+
write_string(self.conn, json_to_send)
180+
except Empty:
181+
pass
182+
183+
try:
184+
msg = self.iopub_channel.get_iopub_msg(timeout=10)
185+
json_to_send = json.dumps(msg, default=str)
186+
with self.send_lock:
187+
_debug_write('iopub_msg')
188+
write_bytes(self.conn, iPythonSocketServer._IOPB)
189+
write_string(self.conn, json_to_send)
190+
except Empty:
191+
pass
192+
154193
except IPythonExitException:
155194
_debug_write('IPythonExitException')
156195
_debug_write(traceback.format_exc())
@@ -185,6 +224,9 @@ class iPythonSocketServer(object):
185224
_STPK = to_bytes('STPK')
186225
_RSTK = to_bytes('RSTK')
187226
_ITPK = to_bytes('ITPK')
227+
_RUN = to_bytes('RUN ')
228+
_SHEL = to_bytes('SHEL')
229+
_IOPB = to_bytes('IOPB')
188230

189231
def __init__(self):
190232
import threading
@@ -199,6 +241,8 @@ def __init__(self):
199241
self.execute_item_lock = threading.Lock()
200242
# lock starts acquired (we use it like manual reset event)
201243
self.execute_item_lock.acquire()
244+
self.kernelMonitor = None
245+
self.shell_channel = None
202246

203247
def connect(self, port):
204248
# start a new thread for communicating w/ the remote process
@@ -332,14 +376,15 @@ def _cmd_strk(self, id):
332376
except socket.timeout:
333377
pass
334378
kernelUUID = multiKernelManager.start_kernel(kernel_name=kernelName)
379+
self._postStartKernel(kernelUUID)
380+
335381
# get the config and the connection FileExistsError
336-
kernel = multiKernelManager.get_kernel(kernelUUID)
337382
try:
338-
config = kernel.config
383+
config = kernel_manager.config
339384
except:
340385
config = {}
341386
try:
342-
connection_file = kernel.connection_file
387+
connection_file = kernel_manager.connection_file
343388
except:
344389
connection_file = ""
345390

@@ -351,6 +396,28 @@ def _cmd_strk(self, id):
351396
write_string(self.conn, json.dumps(config))
352397
write_string(self.conn, connection_file)
353398

399+
def _postStartKernel(self, kernelUUID):
400+
kernel_manager = multiKernelManager.get_kernel(kernelUUID)
401+
kernel_client = kernel_manager.client()
402+
kernel_client.start_channels()
403+
404+
try:
405+
# IPython 3.x
406+
kernel_client.wait_for_ready()
407+
iopub = kernel_client
408+
shell = kernel_client
409+
except AttributeError:
410+
# Ipython 2.x
411+
# Based on https://github.com/paulgb/runipy/pull/49/files
412+
iopub = kernel_client.iopub_channel
413+
shell = kernel_client.shell_channel
414+
shell.get_shell_msg = shell.get_msg
415+
iopub.get_iopub_msg = iopub.get_msg
416+
417+
self.shell_channel = shell
418+
self.kernelMonitor = iPythonKernelResponseMonitor(
419+
kernelUUID, self.conn, self.send_lock, shell, iopub)
420+
354421
def _cmd_stpk(self, id):
355422
"""Shutdown a kernel by UUID"""
356423
while True:
@@ -359,12 +426,28 @@ def _cmd_stpk(self, id):
359426
break
360427
except socket.timeout:
361428
pass
429+
362430
try:
363-
kernel = multiKernelManager.get_kernel(kernelUUID)
364-
kernel.shutdown_kernel()
431+
if self.kernelMonitor is not None:
432+
self.kernelMonitor.stop()
433+
finally:
434+
pass
435+
436+
try:
437+
kernel_manager = multiKernelManager.get_kernel(kernelUUID)
438+
kernel_client = kernel_manager.client()
439+
kernel_client.stop_channels()
440+
finally:
441+
pass
442+
443+
try:
444+
kernel_manager = multiKernelManager.get_kernel(kernelUUID)
445+
kernel_manager.shutdown_kernel()
365446
except:
366447
pass
367448
finally:
449+
self.shell_channel = None
450+
self.kernelMonitor = None
368451
with self.send_lock:
369452
write_bytes(self.conn, iPythonSocketServer._STPK)
370453
write_string(self.conn, id)
@@ -377,8 +460,18 @@ def _cmd_rstk(self, id):
377460
break
378461
except socket.timeout:
379462
pass
380-
kernel = multiKernelManager.get_kernel(kernelUUID)
381-
kernel.restart_kernel(now=True)
463+
464+
try:
465+
if self.kernelMonitor is not None:
466+
self.kernelMonitor.stop()
467+
finally:
468+
self.kernelMonitor = None
469+
470+
kernel_manager = multiKernelManager.get_kernel(kernelUUID)
471+
kernel_manager.restart_kernel(now=True)
472+
473+
self._postStartKernel(kernelUUID)
474+
382475
with self.send_lock:
383476
write_bytes(self.conn, iPythonSocketServer._RSTK)
384477
write_string(self.conn, id)
@@ -391,16 +484,25 @@ def _cmd_itpk(self, id):
391484
break
392485
except socket.timeout:
393486
pass
394-
kernel = multiKernelManager.get_kernel(kernelUUID)
395-
kernel.interrupt_kernel()
487+
kernel_manager = multiKernelManager.get_kernel(kernelUUID)
488+
kernel_manager.interrupt_kernel()
396489
with self.send_lock:
397490
write_bytes(self.conn, iPythonSocketServer._ITPK)
398491
write_string(self.conn, id)
399492

400-
def _cmd_run(self):
401-
"""runs the received snippet of code"""
402-
# self.run_command(read_string(self.conn))
403-
pass
493+
def _cmd_run(self, id):
494+
"""runs the received snippet of code (kernel is expected to have been started)"""
495+
while True:
496+
try:
497+
code = read_string(self.conn)
498+
break
499+
except socket.timeout:
500+
pass
501+
msg_id = self.shell_channel.execute(code)
502+
with self.send_lock:
503+
write_bytes(self.conn, iPythonSocketServer._RUN)
504+
write_string(self.conn, id)
505+
write_string(self.conn, msg_id)
404506

405507
def _cmd_abrt(self):
406508
"""aborts the current running command"""
@@ -501,6 +603,7 @@ def flush(self):
501603
to_bytes('stpk'): True,
502604
to_bytes('rstk'): True,
503605
to_bytes('itpk'): True,
606+
to_bytes('run '): True,
504607
}
505608

506609

‎src/client/jupyter/jupyter_client-Kernel.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,11 @@ export class JupyterClientKernel extends Kernel {
4040
};
4141

4242
public execute(code: string, onResults: Function) {
43+
this.jupyterClient.runCode(code).then(() => {
44+
const y = '';
45+
}).catch(reason => {
46+
const x = '';
47+
})
4348
};
4449

4550
public executeWatch(code: string, onResults: Function) {

‎src/client/jupyter/jupyter_client/ipythonAdapter.ts‎

Lines changed: 55 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,11 @@
11
"use strict";
22

3-
import * as net from "net";
43
import { SocketCallbackHandler } from "../../common/comms/socketCallbackHandler";
54
import { Commands, ResponseCommands } from "./commands";
6-
import { SocketStream } from "../../Common/comms/SocketStream";
75
import { SocketServer } from '../../common/comms/socketServer';
86
import { IdDispenser } from '../../common/idDispenser';
97
import { createDeferred, Deferred } from '../../common/helpers';
10-
import {KernelCommand} from './contracts';
8+
import { KernelCommand } from './contracts';
119

1210
export class iPythonAdapter extends SocketCallbackHandler {
1311
private idDispenser: IdDispenser;
@@ -20,6 +18,9 @@ export class iPythonAdapter extends SocketCallbackHandler {
2018
this.registerCommandHandler(ResponseCommands.KernelInterrupted, this.onKernelCommandComplete.bind(this));
2119
this.registerCommandHandler(ResponseCommands.KernelRestarted, this.onKernelCommandComplete.bind(this));
2220
this.registerCommandHandler(ResponseCommands.KernelShutdown, this.onKernelCommandComplete.bind(this));
21+
this.registerCommandHandler(ResponseCommands.RunCode, this.onCodeSentForExecution.bind(this));
22+
this.registerCommandHandler(ResponseCommands.ShellResult, this.onShellResult.bind(this));
23+
this.registerCommandHandler(ResponseCommands.IOPUBMessage, this.onIOPUBMessage.bind(this));
2324
this.idDispenser = new IdDispenser();
2425
}
2526

@@ -114,23 +115,23 @@ export class iPythonAdapter extends SocketCallbackHandler {
114115
this.releaseId(id);
115116
def.resolve([kernelUUID, config, connectionFile]);
116117
}
117-
public sendKernelCommand(kernelUUID: string, command:KernelCommand): Promise<any> {
118+
public sendKernelCommand(kernelUUID: string, command: KernelCommand): Promise<any> {
118119
const [def, id] = this.createId<any>();
119-
let commandBytes:Buffer;
120-
switch(command){
121-
case KernelCommand.interrupt:{
120+
let commandBytes: Buffer;
121+
switch (command) {
122+
case KernelCommand.interrupt: {
122123
commandBytes = Commands.InterruptKernelBytes;
123124
break;
124125
}
125-
case KernelCommand.restart:{
126+
case KernelCommand.restart: {
126127
commandBytes = Commands.RestartKernelBytes;
127128
break;
128129
}
129-
case KernelCommand.shutdown:{
130+
case KernelCommand.shutdown: {
130131
commandBytes = Commands.ShutdownKernelBytes;
131132
break;
132133
}
133-
default:{
134+
default: {
134135
throw new Error('Unrecognized Kernel Command');
135136
}
136137
}
@@ -167,6 +168,50 @@ export class iPythonAdapter extends SocketCallbackHandler {
167168
def.resolve(message);
168169
}
169170

171+
runCode(code): Promise<any> {
172+
const [def, id] = this.createId<string[]>();
173+
this.SendRawCommand(Commands.RunCodeBytes);
174+
this.stream.WriteString(id);
175+
this.stream.WriteString(code)
176+
return def.promise;
177+
}
178+
private onCodeSentForExecution() {
179+
const id = this.stream.readStringInTransaction();
180+
const msg_id = this.stream.readStringInTransaction();
181+
if (msg_id == undefined) {
182+
return;
183+
}
184+
const def = this.pendingCommands.get(id);
185+
this.releaseId(id);
186+
def.resolve(msg_id);
187+
}
188+
189+
private onShellResult() {
190+
const jsonResult = this.stream.readStringInTransaction();
191+
if (jsonResult == undefined) {
192+
return;
193+
}
194+
try {
195+
const y = JSON.parse(jsonResult);
196+
}
197+
catch (ex) {
198+
const x = '';
199+
}
200+
}
201+
202+
private onIOPUBMessage() {
203+
const jsonResult = this.stream.readStringInTransaction();
204+
if (jsonResult == undefined) {
205+
return;
206+
}
207+
try {
208+
const y = JSON.parse(jsonResult);
209+
}
210+
catch (ex) {
211+
const x = '';
212+
}
213+
}
214+
170215
private onError() {
171216
const cmd = this.stream.readStringInTransaction();
172217
const id = this.stream.readStringInTransaction();

0 commit comments

Comments
 (0)