* parallel-remote.el (parallel-remote-send): Renamed from `parallel-send'.

* parallel.el (parallel-send): New function to send data to the remote
process.
* test-parallel.el: Rename `parallel-send' to `parallel-remote-send'.
(parallel-continue-when-executed): New test.

Author: daimrod

parallel-remote.el

@@ -22,13 +22,15 @@
 
 ;;; Code:
 
+(require 'cl)
+
 (defvar parallel-service nil)
 (defvar parallel-task-id nil)
 (defvar parallel-client nil)
 (defvar parallel--executed nil)
 (defvar parallel-continue-when-executed nil)
 
-(defun parallel-send (data)
+(defun parallel-remote-send (data)
   (process-send-string parallel-client
                        (format "%S " (cons parallel-task-id data))))
 
@@ -40,7 +42,7 @@
                                               :host "localhost"
                                               :family 'ipv4))
   (set-process-filter parallel-client #'parallel-remote--filter)
-  (parallel-send 'code)
+  (parallel-remote-send 'code)
   (when noninteractive                  ; Batch Mode
     ;; The evaluation is done in the `parallel--filter' but in Batch
     ;; Mode, Emacs doesn't wait for the input, it stops as soon as
@@ -49,17 +51,31 @@
       (sleep-for 10))))                 ; arbitrary chosen
 
 (defun parallel-remote--filter (_proc output)
-  (parallel-send
-   (if (or noninteractive
-           (not debug-on-error))
-       (condition-case err
-           (eval (read output))
-         (error err))
-     (eval (read output))))
+  (dolist (code (parallel--read-output output))
+    (parallel-remote-send
+     (if (or noninteractive
+             (not debug-on-error))
+         (condition-case err
+             (eval code)
+           (error err))
+       (eval code))))
   (unless parallel-continue-when-executed
     (setq parallel--executed t)
     (kill-emacs)))
 
+(defun parallel--read-output (output)
+  "Read lisp forms from output and return them as a list."
+  (loop with output = (replace-regexp-in-string
+                       "\\`[ \t\n]*" ""
+                       (replace-regexp-in-string "[ \t\n]*\\'" "" output)) ; trim string
+        with start = 0
+        with end = (length output)
+        for ret = (read-from-string output start end)
+        for data = (first ret)
+        do (setq start (rest ret))
+        collect data
+        until (= start end)))
+
 (provide 'parallel-remote)
 
 ;;; parallel-remote.el ends here

parallel.el

@@ -23,7 +23,7 @@
 ;;; Code:
 
 (require 'cl)
-(require 'find-func)
+(require 'parallel-remote)
 
 (defgroup parallel nil
   "Execute stuff in parallel"
@@ -114,7 +114,7 @@
         library-path (or library-path
                          (plist-get config :library-path)
                          (plist-get parallel-config :library-path)
-                         (find-library-name "parallel-remote")))
+                         (locate-library "parallel-remote")))
 
   (let ((task (parallel--new-task))
         proc tunnel ssh-args)
@@ -128,6 +128,7 @@
       (put task 'on-event on-event))
     (put task 'results nil)
     (put task 'status 'run)
+    (put task 'queue nil)
 
     ;; We need to get the tunnel if it exists so we can send the right
     ;; `service' to the remote.
@@ -241,23 +242,19 @@ to `funcall' FUN with ENV as arguments."
 (defun parallel--filter (connection output)
   "Server filter used to retrieve the results send by the remote
 process and send the code to be executed by it."
-  (loop with output = (replace-regexp-in-string
-                       "\\`[ \t\n]*" ""
-                       (replace-regexp-in-string "[ \t\n]*\\'" "" output)) ; trim string
-        with start = 0
-        with end = (length output)
-        for ret = (read-from-string output start end)
-        for data = (first ret)
-        do (setq start (rest ret))
-        do (parallel--process-output connection (first data) (rest data))
-        until (= start end)))
+  (dolist (data (parallel--read-output output))
+    (parallel--process-output connection (first data) (rest data))))
 
 (defun parallel--process-output (connection task result)
+  (put task 'connection connection)
   (cond ((and (not (get task 'initialized))
               (eq result 'code))
-         (process-send-string connection
-                              (parallel--call-with-env (get task 'exec-fun)
-                                                       (get task 'env)))
+         (parallel-send task
+                        (get task 'exec-fun)
+                        (get task 'env))
+         (let ((code nil))
+           (while (setq code (pop (get task 'queue)))
+             (parallel-send task (car code) (cdr code))))
          (put task 'initialized t))
         (t
          (push result (get task 'results))
@@ -298,6 +295,15 @@ result returned by exec-fun."
   "Stop TASK."
   (delete-process (get task 'proc)))
 
+(defun parallel-send (task fun &optional env)
+  "Send FUN to be evaluated by TASK in ENV."
+  (let ((connection (get task 'connection)))
+    (if connection
+        (process-send-string
+         connection
+         (parallel--call-with-env fun env))
+      (push (cons fun env) (get task 'queue)))))
+
 (provide 'parallel)
 
 ;;; parallel.el ends here

test-parallel.el

@@ -28,9 +28,9 @@
 (ert-deftest parallel-basic ()
   (should (= 42 (parallel-get-result (parallel-start (lambda () 42))))))
 
-(ert-deftest parallel-send ()
+(ert-deftest parallel-remote-send ()
   (should (equal (parallel-get-results (parallel-start (lambda ()
-                                                         (parallel-send 12)
+                                                         (parallel-remote-send 12)
                                                          42)))
                  (list 42 12))))
 
@@ -61,8 +61,8 @@
              (apply #'+
                     (parallel-get-results
                      (parallel-start (lambda ()
-                                       (parallel-send 12)
-                                       (parallel-send 42)
+                                       (parallel-remote-send 12)
+                                       (parallel-remote-send 42)
                                        0)
                                      :on-event
                                      (lambda (data)
@@ -84,7 +84,7 @@
 (ert-deftest parallel-other-library ()
   (let ((library (make-temp-file "parallel-")))
     (with-temp-file library
-      (insert-file-contents (find-library-name "parallel-remote"))
+      (insert-file-contents (locate-library "parallel-remote"))
       (insert "(defun parallel--test () 42)"))
     (should (equal 42
                    (parallel-get-result
@@ -109,6 +109,12 @@
   ;; HACKY ERT run tests in alphabetical order, hence the name.
   (should (eq nil parallel--tasks)))
 
+(ert-deftest parallel-continue-when-executed ()
+  (let ((task (parallel-start (lambda () 42) :continue-when-executed t :timeout 5)))
+    (parallel-send task (lambda () (setq parallel-continue-when-executed nil)))
+    (should (equal (list nil 42)
+                   (parallel-get-results task)))))
+
 (provide 'test-parallel)
 
 ;;; test-parallel.el ends here