threads-jss.lisp 4.69 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11
;;;; Copyright (C) 2011 Mark Evenson

(in-package #:threads)

(require 'abcl-contrib)
(eval-when (:compile-toplevel)
           (require 'jss))

(defparameter *server-running-p* nil)

;;; XXX possibly need multiple thread pools
12 13
(defparameter *thread-pool* nil
  "The current JVM class implementing the ScheduledThreadPool abstraction.")
14 15 16 17 18 19 20 21 22 23 24 25 26
(defparameter *scheduled-futures* nil)
(defparameter *incoming-scheduled-future* nil)
(defparameter *watch-queue-future* nil)


;;;; Configure the directories for a threadpool from these defaults.
(defparameter *root* #p"/var/tmp/abcl-threads/")

(defparameter *logs* (merge-pathnames "logs/" *root*))

(defparameter *incoming* (merge-pathnames "incoming/" *root*))
(defparameter *dirs* (list *incoming*))

27 28 29 30 31
(defparameter *queue* (merge-pathnames "queue/" *root*))

(defparameter *processed* (merge-pathnames "processed/" *root*))


32 33 34 35
;;;; A simple logging abstraction.

(defconstant +month-names+ '("Jan" "Feb" "Mar" "Apr" "May" "Jun" 
                             "Jul" "Aug" "Sep" "Oct" "Nov" "Dec"))
36
(defconstant +seconds+ (java:jfield "java.util.concurrent.TimeUnit" "SECONDS"))
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
(defparameter *log* *standard-output*)

(defun format-time (universal-time)
    (multiple-value-bind 
          (second minute hour date month year day-of-week dst-p tz)
        (decode-universal-time universal-time)
      (format nil "~&~A ~A ~2,'0D:~2,'0D:~2,'0D" 
              (nth (1- month) +month-names+) date hour minute second)))

(defmacro log (message &rest parameters)
  `(when *log*
     (format *log* "~A " (format-time (get-universal-time)))
     (format *log* ,message ,@parameters)
     (format *log* "~&")))

;;; Start a pool of hungry philosophers.
(defun start-server () 
  (when *server-running-p*
    (error "Server not recorded as stopped."))
  (unless 
      (mapcar #'ensure-directories-exist *dirs*)
    (error "Failed to create directories under '~A'." *root*))
  (let ((logfile (merge-pathnames "abcl-threads.log" *logs*)))
    (setf *log* 
          (open logfile :direction :output :if-exists :append :if-does-not-exist :create))
    (format *standard-output* "Logging to ~A." logfile))
  (log "Starts.")
  (schedule-threads)
  (setf *server-running-p* t))

(defun stop-server (&key (force nil))
  (unless force
    (unless *server-running-p*
      (error "Server not recorded as running.")))
  (log "Stopping the server.")
  (dolist (future `(,*incoming-scheduled-future* ,*watch-queue-future* ,@*scheduled-futures*))
    (when (not (or (#"isCancelled" future)
                   (#"isDone" future)))
      (#"cancel" future t)))
  (#"shutdown" *thread-pool*)
  (close *log*)
  (setf *server-running-p* nil))

(defun schedule-threads ()
  (log "Starting thread pool.")
  (when *thread-pool*
    (log "Removing existing incoming thread pool."))
  (setf *thread-pool*
        (#"newScheduledThreadPool" 'java.util.concurrent.Executors 1))
  (#"setExecuteExistingDelayedTasksAfterShutdownPolicy" *thread-pool* nil)
  (initialize-queue)
  (log "Scheduling queue watcher.")
  (setf *watch-queue-future* 
        (#"scheduleWithFixedDelay" 
         *thread-pool*
         (make-watch-queue) 10 1 +seconds+))
  (log "Scheduling incoming watcher.")
  (setf *incoming-scheduled-future*
        (#"scheduleWithFixedDelay" 
         *thread-pool*
         (make-process-incoming) 1 1 +seconds+)))

99 100 101 102 103
(defun make-process-incoming ()
  (java:jinterface-implementation "java.lang.Runnable" "run" #'process-incoming))

(defun process-incoming ()
  (flet ((reject-input (file invalid) 
104 105 106 107
           (warn (format nil "~A is ~A" file invalid)))
         (process (file)
           nil))

108 109 110 111 112 113 114 115 116 117
  (let ((incoming (directory (merge-pathnames *incoming* "*"))))
    (unless incoming
      (return-from process-incoming))
    (log "Processing ~A incoming items." (length incoming))
    (let (table error)
      (dolist (file incoming)
        (setf error nil)
        (log "Analyzing ~A." file)
        (setf table
              (handler-case 
118
                  (process file)
119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136
                (t (e) 
                  (log "Failed to process ~A because ~A" file e)
                  (setf error e))))
        (if error 
            (reject-input file (if (listp error) error (list error)))
            (multiple-value-bind (valid invalid)
                (validate table)
              (if invalid 
                  (progn 
                    (log "Rejecting ~A because of invalid rows." file)
                    (reject-input file invalid))
                  (let ((incoming 
                         (make-pathname :defaults *queue* 
                                        :name (pathname-name file)
                                        :type (pathname-type file))))
                    (log "Inserting ~A." incoming)
                    (rename-file file incoming))))))))))