(in-package "PVM")

;;;
;;; [incr tsdb()] --- Competence and Performance Profiling Environment
;;; Copyright (c) 1996 -- 2005 Stephan Oepen (oe@csli.stanford.edu)
;;;               2007 Ben Waldron
;;;
;;; This program is free software; you can redistribute it and/or modify it
;;; under the terms of the GNU Lesser General Public License as published by
;;; the Free Software Foundation; either version 2.1 of the License, or (at
;;; your option) any later version.
;;;
;;; This program is distributed in the hope that it will be useful, but WITHOUT
;;; ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
;;; FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
;;; License for more details.
;;; 

(defconstant %pvm_no_parent% -23)
(defconstant %pvm_task_fail% 42)
(defconstant %pvm_lisp_message% 50)
(defconstant %pvm_c_message% 51)

(defparameter 
  *pvm*
  (format
   nil "exec ~a"
   (namestring (make-pathname :directory (pathname-directory make::bin-dir)
                              :name "pvm"))))
(defparameter 
  *pvmd*
  (format
   nil "exec ~a"
   (namestring (make-pathname :directory (pathname-directory make::bin-dir)
                              :name "pvmd3"))))

(defparameter *pvm-encoding* #+:logon :utf-8 #-:logon nil)

(defparameter *pvm-debug-p* nil)

(defparameter *pvm-pending-events* nil)

(defun initialize-pvm ()
  (setf *pvm*
    (format
     nil "exec ~a"
     (namestring (make-pathname :directory (pathname-directory make::bin-dir)
                                :name "pvm"))))
  (setf *pvmd*
    (format
     nil "exec ~a"
     (namestring (make-pathname :directory (pathname-directory make::bin-dir)
                                :name "pvmd3")))))

(defstruct cpu 
  host spawn options encoding
  architecture class task threshold name grammar template
  edges analyses results wait quantum
  preprocessor tagger reader
  create complete)

(defstruct client tid cpu task protocol form status load host)

(defun current-user ()
  (or #+(and :allegro-version>= (version>= 5 0)) 
      (sys:user-name)
      #+(and :allegro (not (and :allegro-version>= (version>= 5 0))))
      (system:getenv "USER")
      #+(and :mcl :powerpc) 
      (ccl:process-name ccl:*current-process*)
      #+:lucid 
      (lcl:environment-variable "USER")
      #+:sbcl (sb-ext:posix-getenv "USER")
      "nobody"))

(define-alien-routine
    ("getpid" current-pid) signed)

(define-alien-routine pvm_mytid signed)

(let ((user (current-user))
      tid)
  (defun make-tmp-file (prefix)
    (unless tid (setf tid (pvm_mytid)))
    (let ((file (format 
                 nil 
                 "/tmp/~a.~a.~a.~a" 
                 prefix user tid (gensym ""))))
    (when (probe-file file) (delete-file file))
    file)))

(define-alien-routine load_average double-float)

(define-alien-routine
    ("pvm_register" _pvm_register) signed
  (file c-string)
  (debugp signed))

(defun pvm_register (&optional (file t) debugp)
  (let ((file (cond
               ((or (eq file t) (equal file "")) "")
               ((eq file nil) "/dev/null")
               ((stringp file) file)
               (t "")))
        (debugp (if debugp 1 0)))
    (_pvm_register file debugp)))

(define-alien-routine pvm_quit signed)

(define-alien-routine pvmendtask signed)

(define-alien-routine pvm_kill signed
  (tid signed))

(define-alien-routine pvm_parent signed)

(define-alien-routine
    ("pvm_create" _pvm_create) signed
  (task c-string)
  (argv (* c-string))
  (host c-string)
  (architecture c-string))

(defun pvm_create (task arguments &key host architecture)
  (let* ((arguments
          (when (consp arguments)
            (loop
                for argument in arguments
                for string = (if (stringp argument) 
                               argument
                               (format nil "~(~s~)" argument))
                for native = #+:ics (excl:string-to-native string)
                             #-:ics string
                collect native)))
         (argv
          #+:ics
          (let* ((size (+ (length arguments) 1))
                 (vector (allocate-fobject (list :array '(* :char) size))))
            (loop
                for argument in arguments
                for i from 0
                do
                  (setf (fslot-value-typed '(:array (* :char)) nil vector i)
                    argument)
                finally 
                  (incf i)
                  (setf (fslot-value-typed '(:array (* :char)) nil vector i)
                    0))
            vector)
          #-:ics
          (make-array (+ (length arguments) 1)
                                :initial-contents (append arguments '(0))))
         (host (if host (string host) ""))
         (architecture 
          (if (or host (null architecture)) "" (string architecture)))
         (result (_pvm_create task argv host architecture)))
    #+(and :ics :allegro)
    (loop
        for argument in arguments
        when (integerp argument) do (excl:aclfree argument))
    result))

(define-alien-routine pvm_flush signed)

(define-alien-routine
    ("pvm_poll" _pvm_poll) signed
  (tid signed)
  (tag signed)
  (block signed)
  #-:ics
  (output c-string)
  #+:ics
  (output signed)
  (size signed))

(define-alien-routine
    pvm_collect signed
  #-:ics
  (output c-string)
  #+:ics
  (output signed)
  (size signed))

(let* ((limit (or #+(and :allegro-version>= (version>= 7 0) (not :64bit))
                  (expt 2 24)
                  #+(and :allegro-version>= (version>= 7 0) :64bit)
                  (expt 2 26)
                  array-total-size-limit))
       (size 4096)
       (output (make-array 
                (+ size 1) :element-type 'character 
		#+:allegro :allocation 
		#+:allegro :static))
       #+:ics
       (output (excl:lispval-other-to-address output)))
  (defun pvm_poll (tid tag block)
    (let* ((block (cond ((null block) 0)
                        ((numberp block) (round block))
                        (t -1)))
           (encoding (or *pvm-encoding*
                         #+:allegro (excl:locale-external-format excl:*locale*)
			 #+:sbcl (SB-IMPL::DEFAULT-EXTERNAL-FORMAT)
			 ))
           (status (_pvm_poll tid tag block output size)))
      (cond
       ((zerop status) nil)
       ((< status 0) :error)
       ((> status size)
        (if (>= (+ status 4096) (- limit 2))
          (when *pvm-debug-p*
            (format 
             t 
             "pvm-poll(): rejecting over-sized message (~d)~%"
             (+ status 4096)))
          (let ((new (make-array 
                      (+ status 4096) :element-type 'character 
                      #+:allegro :allocation 
		      #+:allegro :static)))

            ;;
            ;; _fix_me_
            ;; we would want to free the old .output. array here; apparently, 
            ;; the way recommended in the 5.0.1 documentation---viz. aclfree()
            ;; plus lispval-other-to-address()---is unavailable because the
            ;; latter function does not exist :-{.             (16-feb-00; oe)
            ;;
            ;; it seems, ACL 6.0 now has lispval-other-to-address(). 
            ;;                                                 (13-sep-01; oe)
            ;;
            #+:allegro
	    (when (integerp output) (excl:aclfree output))
            (setf output 
              #-:ics new #+:ics (excl:lispval-other-to-address new))
            (setf size (- (array-total-size new) 1))))
        (let ((status (pvm_collect output size)))
          (if (< status 0)
            :error
            (multiple-value-bind (result condition)
                (ignore-errors 
                 (let ((string 
                        #-:ics output 
                        #+:ics (excl:native-to-string 
                                output :external-format encoding)))
                   (when *pvm-debug-p*
                     (format t "pvm_poll(): read `~a'.~%" string))
                   (read-from-string string t nil :end status)))
              (when (and (null result) condition)
                (format t "~&pvm_poll(): error `~a'.~%" condition))
              (or result :error)))))
       (t
        (multiple-value-bind (result condition)
            (ignore-errors 
             (let ((string 
                    #-:ics output 
                    #+:ics (excl:native-to-string
                            output :external-format encoding)))
               (when *pvm-debug-p*
                 (format t "pvm_poll(): read `~a'.~%" string))
               (read-from-string string t nil :end status)))
          (when (and (null result) condition)
            (format
             t
             "~&pvm_poll(): error `~a'.~%" condition))
          (or result :error)))))))

(define-alien-routine
    ("pvm_transmit" _pvm_transmit) signed
  (tid signed)
  (tag signed)
  (string c-string))

(defun pvm_transmit (tid tag form)
  ;;
  ;; _fix_me_
  ;; why did we, several years ago, not want *print-readably* on?  i can hardly
  ;; see how that can be good?                                 (10-mar-04; oe)
  ;;
  (let* (#+:null
         (*print-readably* nil)
         #+:null
         (string (format nil "~s" form))
         (string (with-standard-io-syntax (write-to-string form)))
         (encoding (or *pvm-encoding*
                       #+:allegro (excl:locale-external-format excl:*locale*)
		       #+:sbcl (SB-IMPL::DEFAULT-EXTERNAL-FORMAT)
		       ))
         (status 
          ;(excl:with-native-string (address string :external-format encoding)
          ;  (_pvm_transmit tid tag address))
	  (_pvm_transmit tid tag string) ;;bmw
	  
	  ))
    status))

(define-alien-routine
    pvm_tidtohost signed
  (tid signed))

(define-alien-routine
    ("pvm_vm_info" _pvm_vm_info) signed
  (file c-string))

(defun pvm_vm_info ()
  (let ((file (make-tmp-file ".pvm.io")))
    (let ((status (_pvm_vm_info file)))
      (cond
       ((zerop status) nil)
       ((or (< status 0) (not (probe-file file))) :error)
       (t
        (let* ((buffer (make-string status))
               (length (with-open-file (stream file :direction :input)
                         (read-sequence buffer stream))))
          (unless *pvm-debug-p* (delete-file file))
          (when (eq status length)
            (multiple-value-bind (result condition)
                (ignore-errors (read-from-string buffer))
              (when (and (null result) condition *pvm-debug-p*)
                (format
                 t
                 "~&pvm_vm_info(): error `~a'.~%" condition))
              (or result :error)))))))))

(define-alien-routine
    ("pvm_announce" pvm_announce) signed
  (class c-string)
  (version c-string)
  (user c-string))

(define-alien-routine
    ("pvm_delinfo" pvm_delinfo) signed
  (class c-string)
  (index signed)
  (flags signed))

(defun pvm_retract (class)
  (pvm_delinfo class 0 0))

(define-alien-routine
    ("pvm_task_info" _pvm_task_info) signed
  (tid signed)
  (file c-string))

(defun pvm_task_info (&optional (tid 0))
  (let ((file (make-tmp-file ".pvm.io" )))
    (let ((status (_pvm_task_info tid file)))
      (cond
       ((zerop status) nil)
       ((or (< status 0) (not (probe-file file))) :error)
       (t
        (let* ((buffer (make-string status))
               (length (with-open-file (stream file :direction :input)
                         (read-sequence buffer stream))))
          (unless *pvm-debug-p* (delete-file file))
          (when (eq status length)
            (multiple-value-bind (result condition)
                (ignore-errors (read-from-string buffer))
              (when (and (null result) condition *pvm-debug-p*)
                (format
                 t
                 "~&pvm_task_info(): error `~a'.~%" condition))
              (or result :error)))))))))

(defun pvm_halt (&key (user 
		       (sb-ext:posix-getenv "USER")
		       ))
  (pvm_quit)
  (let ((file (format nil "/tmp/.pvm.halt.~a.~a" user (gensym ""))))
    (with-open-file (stream file :direction :output :if-exists :supersede)
      (format stream "halt~%"))
    (run-process 
     *pvm* :wait t 
     :output "/dev/null" :if-output-exists :append
     :error-output "/dev/null" :if-error-output-exists :append
     :input file)
    (when (probe-file file) (delete-file file))))

(defun pvm_start (&key reset (user (sb-ext:posix-getenv "USER")))
  (pvm_quit)
  (when reset (pvm_halt :user user))
  (let ((status (run-process 
                 *pvmd* :wait t 
                 :output "/dev/null" :if-output-exists :append
                 :error-output "/dev/null" :if-error-output-exists :append)))
    (if (zerop status) :ok :error)))

(defmacro message-p (message)
  `(and (consp ,message)
        (find :tag ,message :key #'first)
        (find :remote ,message :key #'first)))

(defmacro message-tag (message)
  `(rest (assoc :tag ,message)))
(defmacro message-load (message)
  `(rest (assoc :load ,message)))
(defmacro message-remote (message)
  `(rest (assoc :remote ,message)))
(defmacro message-content (message)
  `(rest (assoc :content ,message)))
(defmacro message-corpse (message)
  `(rest (assoc :corpse ,message)))

(defun shutdown (tid)
  (pvm_transmit tid %pvm_lisp_message% '(:shutdown)))

(defun tid-status (tid)
  (find tid (rest (assoc :tasks (pvm_task_info tid))) 
        :key #'(lambda (task) (rest (assoc :tid task)))))
         
(defun revaluate (tid form 
                  &optional (block t) 
                  &key (verbose t) (key 0) interrupt)
  (when (and block (null (tid-status tid)))
    (when verbose
      (format
       t
       "revaluate(): invalid task id ~d.~%" tid))
    (return-from revaluate :error))
  (if (< (pvm_transmit 
          tid %pvm_lisp_message% (list :eval key form))
         0)
    :error
    (if block
      (loop
          for message = (pvm_poll tid %pvm_lisp_message% block)
          when (eq message :error)
          return :error
          when (and (message-p message)
                    (eql (message-tag message) %pvm_lisp_message%)
                    (eq (first (message-content message)) :return)
                    (eql (second (message-content message)) key))
          return (third (message-content message))
          else when (interrupt-p interrupt)
          do
            (return-from revaluate :interrupt))
      :ok)))

(defun interrupt-p (interrupt)
  (when (and interrupt (probe-file interrupt))
    (delete-file interrupt)
    t))