hiro/emacs-jupyter
1
0
Fork 0
mirror of https://github.com/vale981/emacs-jupyter synced 2025-03-05 07:41:37 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

Heartbeat channels are implemented with timers instead of a subprocess

Browse source
This commit is contained in:
Nathaniel Nicandro 2017-12-30 23:30:53 -06:00
parent b57040e96c
commit c7c200d571
1 changed files with 51 additions and 128 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

179
jupyter-channels.el
View file

@ -11,15 +11,6 @@
:control zmq-DEALER) :control zmq-DEALER)
"The socket types for the various channels used by `jupyter'.") "The socket types for the various channels used by `jupyter'.")
;; TODO: Each channel has its own process like the heartbeat channel and does
;; the majority of encoding and decoding messages there. The parent emacs
;; process then is only responsible to carry out the actions in the messages
;; and construct replies which are then sent to the process.
;;
;; The current implementation still creates a process, but only for polling the
;; file descriptor of the socket to check for incoming messages. What would be
;; simpler is to create one polling process and handle messages through a
;; single filter function instead of per channel.
(defclass jupyter-channel () (defclass jupyter-channel ()
((type ((type
:type keyword :type keyword
@ -124,166 +115,98 @@ A channel is alive if its socket property is bound to a
((type ((type
:type keyword :type keyword
:initform :hb :initform :hb
:documentation "The type of this channel. Should be one of :documentation "The type of this channel is `:hb'.")
the keys in `jupyter-channel-socket-types', excluding `:hb'
which corresponds to the heartbeat channel.")
(endpoint (endpoint
:type string :type string
:initarg :endpoint :initarg :endpoint
:documentation "The endpoint this channel is connected to. :documentation "The endpoint this channel is connected to.
Typical endpoints look like \"tcp://127.0.0.1:5555\".") Typical endpoints look like \"tcp://127.0.0.1:5555\".")
;; channel must be restarted for this to be updated (socket
:type (or null zmq-socket)
:initform nil
:documentation "The socket used for communicating with the kernel.")
(time-to-dead (time-to-dead
:type integer :type integer
:initform 1 :initform 1
:documentation "The time in seconds to wait for a response :documentation "The time in seconds to wait for a response
from the kernel until the connection is assumed to be dead.") from the kernel until the connection is assumed to be dead. Note
that this slot only takes effect when starting the channel.")
(beating (beating
:type (or boolean symbol) :type (or boolean symbol)
:initform t :initform t
:documentation "A flag variable indicating that the heartbeat :documentation "A flag variable indicating that the heartbeat
channel is sending and receiving messages with the kernel.") channel is communicating with the kernel.")
(paused (paused
:type boolean :type boolean
:initform nil :initform t
:documentation "A flag variable indicating that the heartbeat :documentation "A flag variable indicating that the heartbeat
channel is paused and not communicating with the kernel. To channel is paused and not communicating with the kernel. To
pause the heartbeat channel use `jupyter-hb-pause', to unpause pause the heartbeat channel use `jupyter-hb-pause', to unpause
use `jupyter-hb-unpause'.") use `jupyter-hb-unpause'.")
(process (timer
:type (or null process) :type (or null timer)
:initform nil :initform nil
:documentation "The underlying process which runs the :documentation "The timer which sends and receives heartbeat
heartbeat channel and communicates with the kernel.")) messages from the kernel."))
:documentation "A base class for heartbeat channels.") :documentation "A base class for heartbeat channels.")
(cl-defmethod jupyter-channel-alive-p ((channel jupyter-hb-channel)) (cl-defmethod jupyter-channel-alive-p ((channel jupyter-hb-channel))
"Return non-nil if CHANNEL is alive." "Return non-nil if CHANNEL is alive."
(process-live-p (oref channel process))) (and (oref channel timer) (memq (oref channel timer) timer-list)))
(cl-defmethod jupyter-hb-beating-p ((channel jupyter-hb-channel)) (cl-defmethod jupyter-hb-beating-p ((channel jupyter-hb-channel))
"Return non-nil if the kernel associated with CHANNEL is still "Return non-nil if the kernel associated with CHANNEL is still
connected." connected."
(unless (jupyter-channel-alive-p channel) (unless (jupyter-channel-alive-p channel)
(error "Heartbeat process not started")) (error "Heartbeat channel not alive"))
(process-send-string (oref channel process) "beating\n")
(accept-process-output (oref channel process) nil nil 1)
(oref channel beating)) (oref channel beating))
(cl-defmethod jupyter-hb-pause ((channel jupyter-hb-channel)) (cl-defmethod jupyter-hb-pause ((channel jupyter-hb-channel))
"Pause checking for heartbeat events on CHANNEL." "Pause checking for heartbeat events on CHANNEL."
(unless (jupyter-channel-alive-p channel) (unless (jupyter-channel-alive-p channel)
(error "Heartbeat process not started")) (error "Heartbeat channel not alive"))
(process-send-string (oref channel process) "pause\n") (oset channel paused t))
(accept-process-output (oref channel process) nil nil 1))
(cl-defmethod jupyter-hb-unpause ((channel jupyter-hb-channel)) (cl-defmethod jupyter-hb-unpause ((channel jupyter-hb-channel))
"Unpause checking for heatbeat events on CHANNEL." "Unpause checking for heatbeat events on CHANNEL."
(unless (jupyter-channel-alive-p channel) (unless (jupyter-channel-alive-p channel)
(error "Heartbeat process not started")) (error "Heartbeat channel not alive"))
(process-send-string (oref channel process) "unpause\n") (oset channel paused nil))
(accept-process-output (oref channel process) nil nil 1))
(cl-defmethod jupyter-stop-channel ((channel jupyter-hb-channel)) (cl-defmethod jupyter-stop-channel ((channel jupyter-hb-channel))
"Stop a CHANNEL." "Stop the heartbeat CHANNEL."
(let ((proc (oref channel process))) (cancel-timer (oref channel timer))
(when proc (zmq-socket-set (oref channel socket) zmq-LINGER 0)
(delete-process proc) (zmq-close (oref channel socket))
(kill-buffer (process-buffer proc)) (oset channel socket nil)
(oset channel process nil)))) (oset channel timer nil))
;; TODO: Convert the heartbeat to a timer function that runs every second
;; instead. I can just check zmq-EVENTS every second to see if the channel is
;; beating
(cl-defmethod jupyter-start-channel ((channel jupyter-hb-channel) &key identity) (cl-defmethod jupyter-start-channel ((channel jupyter-hb-channel) &key identity)
"Start a CHANNEL." (unless (jupyter-channel-alive-p channel)
(declare (indent 1)) (oset channel socket (jupyter-connect-channel
(jupyter-stop-channel channel) :hb (oref channel endpoint) identity))
;; https://github.com/jupyter/jupyter_client/blob/master/jupyter_client/channels.py (oset channel timer
(let* (run-with-timer
((time-to-dead (oref channel time-to-dead)) 0 (oref channel time-to-dead)
(proc (lexical-let ((identity identity)
(zmq-start-process (sent nil))
`(lambda (ctx) (lambda (channel)
(let ((beating t) (let ((sock (oref channel socket)))
(paused nil) (when sent
(request-time nil) (setq sent nil)
(wait-time nil) (if (condition-case nil
(last-success nil)) (zmq-recv sock zmq-NOBLOCK)
(while t (zmq-EAGAIN nil))
(catch 'restart (oset channel beating t)
(with-zmq-socket sock ,(plist-get jupyter-channel-socket-types (oset channel beating nil)
(oref channel type)) (zmq-socket-set sock zmq-LINGER 0)
((zmq-LINGER 1000)) (zmq-close sock)
,(when identity (setq sock (jupyter-connect-channel
`(zmq-socket-set sock zmq-ROUTING_ID ,identity)) :hb (oref channel endpoint) identity))
(zmq-connect sock ,(oref channel endpoint)) (oset channel socket sock)))
(with-zmq-poller (unless (oref channel paused)
;; Poll STDIN to avoid blocking (zmq-send sock "ping")
(zmq-poller-register (current-zmq-poller) 0 zmq-POLLIN) (setq sent t)))))
(zmq-poller-register (current-zmq-poller) sock zmq-POLLIN) channel))))
(while t
;; Send a ping request to the heartbeat channel and poll
;; for the reply. If any commands from stdin arrive
;; while polling, handle those and continue waiting.
;; Once the reply is received, keep polling for stdin
;; for the remaining time-to-dead period. After waiting
;; send another ping.
(if request-time
(setq wait-time (* (ceiling
(- ,time-to-dead
(float-time
(time-subtract
(current-time)
request-time))))
1000))
(unless paused
(zmq-send sock "ping"))
(setq request-time (current-time)
wait-time ,(* (ceiling time-to-dead) 1000)))
(let ((event
(condition-case err
(zmq-poller-wait (current-zmq-poller)
(if (> wait-time 0) wait-time 0))
(zmq-EINTR nil)
(error (signal (car err) (cdr err))))))
(cond
((and event (integerp (car event)))
(cl-case (read-minibuffer "")
(beating
(zmq-prin1 (cons 'beating beating)))
(pause
(setq paused t)
(zmq-prin1 '(pause . t)))
(unpause
(setq paused nil)
(zmq-prin1 '(unpause . t)))))
(event
(zmq-recv sock)
(setq beating t
last-success t))
;; When no events have arrived after the poll, its an
;; indication that a reply has been received and we
;; should send another one so set request-time to nil
;; to force another send, note that the send will not
;; happen if we are paused.
((or paused last-success)
(setq request-time nil
last-success nil))
(t
(setq beating nil
request-time nil
last-success nil)
(throw 'restart t)))))))))))
(lexical-let ((channel channel))
(lambda (event)
(cl-case (car event)
(pause (oset channel paused (cdr event)))
(unpause (oset channel paused (not (cdr event))))
(beating (oset channel beating (cdr event)))
(otherwise (error "Invalid event from heartbeat channel."))))))))
;; Don't query when exiting
(set-process-query-on-exit-flag proc nil)
(oset channel process proc)))
(provide 'jupyter-channels) (provide 'jupyter-channels)