Megatest

;;     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
;;     GNU General Public License for more details.
;; 
;;     You should have received a copy of the GNU General Public License
;;     along with Megatest.  If not, see <http://www.gnu.org/licenses/>.
;;

(declare (unit portlogger))

(module
 portlogger
 (portlogger:set-configdat!
  portlogger:set-printers!
  portlogger:set-default-log-port!
  portlogger:open-db
  portlogger:open-run-close
  portlogger:take-port
  portlogger:get-prev-used-port
  portlogger:find-port
  portlogger:set-port
  portlogger:release-port
  portlogger:set-failed
  portlogger:is-port-in-use
  portlogger:main
)
 
(import scheme posix chicken data-structures ports)

(require-extension (srfi 18) extras tcp s11n)
(use srfi-1 posix srfi-69 hostinfo dot-locking z3 regex)
(use (prefix sqlite3 sqlite3:))
(use (prefix mtconfigf configf:))

;; lsof -i

(define *configdat* #f)
(define (portlogger:set-configdat! cfgdat)
  (set! *configdat* cfgdat))

(define (debug:print level port . params)
  (with-output-to-port
      port
    (lambda ()(apply print params))))
(define debug:print-error debug:print)
(define *default-log-port* (current-error-port))

(define (portlogger:set-printers! pdebug pdebugerr)
  (set! debug:print pdebug)
  (set! debug:print-error pdebugerr))
(define (portlogger:set-default-log-port! port)
  (set! *default-log-port* port))

(define (portlogger:open-db fname)
  (let* ((avail    #t) ;; for now - assume wait on journal not needed (tasks:wait-on-journal fname 5 remove: #t)) ;; wait up to about 10 seconds for the journal to go away
	 (exists   (file-exists? fname))
	 (db       (if avail 
		       (sqlite3:open-database fname)
		       (begin
			 (system (conc "rm -f " fname))
			 (sqlite3:open-database fname))))
	 (handler  (sqlite3:make-busy-timeout 136000))
	 (canwrite (file-write-access? fname)))








    (sqlite3:set-busy-handler! db handler)
    (sqlite3:execute db "PRAGMA synchronous = 0;")

    (sqlite3:execute 
     db
     "CREATE TABLE IF NOT EXISTS ports (
            port INTEGER PRIMARY KEY,
            state TEXT DEFAULT 'not-used',
            fail_count INTEGER DEFAULT 0,
            update_time TIMESTAMP DEFAULT (strftime('%s','now')) );")
    db))

(define (portlogger:open-run-close proc . params)
  (let* ((fname  (conc "/tmp/." (current-user-name) "-portlogger.db"))
	 (avail  #t)) ;; (tasks:wait-on-journal fname 10))) ;; wait up to about 10 seconds for the journal to go away
    (handle-exceptions
     exn
     (begin
       ;; (release-dot-lock fname)
       (debug:print-error 0 *default-log-port* "portlogger:open-run-close failed. " proc " " params)
       (debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
       (debug:print 5 *default-log-port* "exn=" (condition->list exn))
       (if (file-exists? fname)(delete-file fname)) ;; brutally get rid of it
       (print-call-chain (current-error-port)))
     (let* (;; (lock   (obtain-dot-lock fname 2 9 10))
	    (db     (portlogger:open-db fname))
	    (res    (apply proc db params)))
       (sqlite3:finalize! db)
       ;; (release-dot-lock fname)
       res))))

;; (fold-row PROC INIT DATABASE SQL . PARAMETERS) 
(define (portlogger:take-port db portnum)
  (let* ((qry1 (sqlite3:prepare db "INSERT INTO ports (port,state) VALUES (?,?);"))
	 (qry2 (sqlite3:prepare db "UPDATE ports SET state=?,update_time=strftime('%s','now') WHERE port=?;"))
	 (qry3 (sqlite3:prepare db "SELECT state FROM ports WHERE port=?;"))
	 (res  ;; (sqlite3:with-transaction ;; move the transaction up to the find-port call
		;; db
		;; (lambda ()
		  ;; (fold-row (lambda (var curr) (or var curr)) #f db "SELECT var FROM foo WHERE id=100;")
		  (let* ((curr #f)
			 (res  #f))
		    (set! curr (sqlite3:fold-row
				(lambda (var curr)
				  (or curr var curr))
				"not-tried"
				qry3
				portnum))
		    ;; (print "curr=" curr)
		    (set! res (case (string->symbol curr)
				((released)  (sqlite3:execute qry2 "taken" portnum) 'taken)
				((not-tried) (sqlite3:execute qry1 portnum "taken") 'taken)
				((taken)                                            'already-taken)
				((failed)                                           'failed)
				(else                                               'error)))
		    ;; (print "res=" res)
		    res))) ;; ))
    (sqlite3:finalize! qry1)
    (sqlite3:finalize! qry2)
    (sqlite3:finalize! qry3)
    res))

(define (portlogger:get-prev-used-port db)
  (handle-exceptions

     (lambda (var curr)
       (or curr var curr))
     #f
     db
     "SELECT (port) FROM ports WHERE state='released' LIMIT 1;")))

(define (portlogger:find-port db)
  (let ((lowport (let ((val (configf:lookup *configdat* "server" "lowport")))
		   (if (and val 
			    (string->number val))
		       (string->number val)
		       32768))))
    (sqlite3:with-transaction
     db
     (lambda ()
       (let loop ((numtries 0))
	 (let* ((portnum (or (portlogger:get-prev-used-port db)
			     (+ lowport ;; top of registered ports is 49152 but let's use ports in the registered range
				(random (- 64000 lowport))))))
	   (handle-exceptions
	    exn
	    (begin
	      (debug:print 0 *default-log-port* "EXCEPTION: portlogger database probably overloaded or unreadable. If you see this message again remove /tmp/.$USER-portlogger.db")
	      (debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
	      (debug:print 5 *default-log-port* "exn=" (condition->list exn))
	      (print-call-chain (current-error-port))
	      (debug:print 0 *default-log-port* "Continuing anyway."))
	    (portlogger:take-port db portnum) ;; always "take the port"
	    (if (portlogger:is-port-in-use portnum)
		portnum
		(loop (add1 numtries))))))))))

;; set port to "released", "failed" etc.
;; 
(define (portlogger:set-port db portnum value)
  (sqlite3:execute db "UPDATE ports SET state=?,update_time=strftime('%s','now') WHERE port=?;" value portnum))

;; release port
(define (portlogger:release-port db portnum)
  (sqlite3:execute db "UPDATE ports SET state=?,update_time=strftime('%s','now') WHERE port=?;" "released" portnum))

;; set port to failed (attempted to take but got error)
;;
(define (portlogger:set-failed db portnum)
  (sqlite3:execute db "UPDATE ports SET state='failed',fail_count=fail_count+1,update_time=strftime('%s','now') WHERE port=?;" portnum))

;; pulled from mtut - TODO: remove from mtut
;;
(define (portlogger:is-port-in-use port-num)
  (let-values (((inp oup pid)
                (process "netstat" (list  "-tulpn" ))))
    (let loop ((inl (read-line inp)))
      (if (not (eof-object? inl))
	  (begin 
	    (if (string-search (regexp (conc ":" port-num "\\s+")) inl)
		#t
		(loop (read-line inp))))))))

;;======================================================================
;; MAIN
;;======================================================================

(define (portlogger:main . args)
  (let* ((dbfname (conc "/tmp/." (current-user-name) "-portlogger.db"))

	     ((find)(portlogger:find-port db))
	     ((set) (let ((port  (cadr  args))
			  (state (caddr args)))
		      (portlogger:set-port db 
					   (if (number? port) port (string->number port))
					   state)
		      state))
	     ((failed)(portlogger:set-failed db (string->number (cadr args))) 'failed)
	     (else "nosuchcommand")))))
    (sqlite3:finalize! db)
    result))
     
;; (print (apply portlogger:main (cdr (argv))))
)