<commit_before>;; -*- mode: scheme; coding: utf-8 -*-
;; "Supervisor" or super server for kahua
;;
;;  Copyright (c) 2003-2006 Scheme Arts, L.L.C., All rights reserved.
;;  Copyright (c) 2003-2006 Time Intermedia Corporation, All rights reserved.
;;  See COPYING for terms and conditions of using this software
;;

;; For clients, this server works as a receptionist of kahua system.
;; It opens a socket where initial clients will connect.
;; This server doesn't know about internals of kahua server, but
;; just dispatches it to the worker servers.
;; Eventually, this server will manage multiple worker servers for
;; load balancing or hot restarting.
;; For now, we have only one worker server, so this is just an outline
;; of what we will ultimately do.

(use gauche.charconv)
(use gauche.net)
(use gauche.threads)
(use gauche.process)
(use gauche.selector)
(use gauche.listener)
(use gauche.parseopt)
(use gauche.parameter)
(use gauche.collection)
(use gauche.mop.singleton)
(use srfi-1)
(use srfi-2)
(use srfi-11)
(use srfi-13)
(use file.util)
(use util.queue)
(use util.list)
(use util.match)
(use kahua.config)
(use kahua.gsid)
(use kahua.developer)
(use kahua.util)
(use kahua.thread-pool)
(use kahua.protocol.worker)

(define *spvr* #f) ;; bound to supervisor object for convenience

(define *default-sigmask* #f)

(define *worker-types* '()) ;; for multi-thread version

(define *default-worker-type* #f)

(define-constant *TERMINATION-SIGNALS* (sys-sigset-add! (make <sys-sigset>) SIGTERM SIGINT SIGHUP))

;; Supervisor protocol
;;
;; [Session initiation]
;;
;;  A client first connect to a well known socket of the supervisor.
;;  This first request is called session-initiating request.
;;  At this point, a client wouldn't have complete GSID.  It may have
;;  state ID from the last session, but it certainly doesn't have
;;  continuation ID.
;;
;;  When the supervisor observes it, and the request is not for direct
;;  administrative request for the supervisor itself, it assigns a
;;  worker and forwards the session-initiating request to the worker.
;;  The worker will return a reply, usually accompanied by GSID.
;;  The supervisor forwards the reply to the client.
;;
;;  Afterwards, the client can figure out the worker ID encoded in GSID,
;;  and directly connects to the worker.
;;
;;  Client can optionally pass the worker type it wants to talk to,
;;  using "x-kahua-worker" header.   If such header is absent, the
;;  supervisor selects the default worker.
;;
;;  As a special case, "spvr" is given to "x-kahua-worker", the body
;;  is interpreted by the spvr process as a command.  See "supervisor
;;  commands" below.
;;
;; [Message format]
;;
;;  Request and reply both consist of two S-expressions, a header and
;;  a body.  A header is a list of two-element lists, resembles to
;;  what rfc822-header->list returns.   A body can be any valid sexpr.
;;  It is arguable whether this format is adequat or not.  Let's see.
;;
;;  In a request header, "x-kahua-sgsid" elemnt carries state GSID,
;;  and "x-kahua-cgsid" carries continuation GSID.  Other header can
;;  be freely used by the worker.  A client may send additional information
;;  in the header, and the worker should ignore the header element that
;;  it doesn't understand.
;;
;;  In a reply header, "x-kahua-sgsid" and "x-kahua-cgsid" are also
;;  used to carry GSID.  It also contains "x-kahua-status", whose value
;;  is either "OK", "ERROR", or "SPVR-ERROR".   "ERROR" indicates
;;  an error occurred in the worker, and "SPVR-ERROR" indicates an
;;  error occurred in the supervisor.  The message body of error replies
;;  contains a list of a error message string (for now).
;;

;;;=================================================================
;;; Global structure
;;;

(define-class <kahua-spvr> ()
  ((sockbase     :init-form (kahua-sockbase))
   (wtype-table  :init-form (make-hash-table 'eq?)      :getter wtype-table-of)
   (wid-table    :init-form (make-hash-table 'string=?) :getter wid-table-of)
   (wno-table    :init-form (make-hash-table 'eq?)      :getter wno-table-of)
   (selector     :init-form (make <selector>)           :getter selector-of)
   (mutex        :init-form (make-mutex)                :getter mutex-of)
   (gosh-path    :init-keyword :gosh-path) ; Absolute path of gosh, passed
                                           ; by wrapper script.
   (lib-path     :init-keyword :lib-path)  ; Path where kahua library files
                                           ; are installed.
   (keyserv      :init-value #f)           ; keyserver process
   (httpd        :init-value #f)	   ; httpd process
   ))

(define-class <kahua-worker-type> ()
  ((spvr   :init-keyword :spvr :getter spvr-of) ; back ptr to spvr
   (name   :init-keyword :name :getter name-of)	; worker type name (symbol)
   (count  :init-keyword :count
	   :getter count-of :init-value 0)	; count of running worker
   (workers :init-value #f :getter workers-of)  ; circular list of kahua-worker instances.
   (mutex  :init-form (make-mutex) :getter mutex-of)))

(define-class <kahua-worker> ()
  ((type   :getter type-of	      ; back ptr to kahua-worker-type
	   :init-keyword :type)
   (wid    :getter wid-of)	      ; worker id (string)
   (wno    :getter wno-of)	      ; Worker No.
   (process :getter process-of)	      ; worker process
   (sockaddr :getter sockaddr-of)     ; socket address
   (logger :getter logger-of)	      ; log output procedure has one arguments
   (start-time :getter start-time-of  ; timestamp
               :init-form (sys-time))
   (zombee :getter zombee?	      ; it's going to shutdown?
	   :init-form #f)
   ;; internal
   (next-wno :allocation :class :init-value 0)
   ))

(define-class <kahua-keyserv> ()
  ((process :init-keyword :process) ;; <process>
   (id      :init-keyword :id)      ;; keyserver id
   ))

;;;=================================================================
;;; Error handling
;;;

(define (spvr-errorf class fmt . args)
  (apply kahua:log-format fmt args)
  (apply errorf class fmt args))

(define (spvr-error-header err)
  `(("x-kahua-status" "SPVR-ERROR" ,(class-name (class-of err)))))

;; If errors occur before spvr service starts, we should terminate
;; spvr with appropriate error message.
(define (app-error msg . args)
  (apply format #t msg args)
  (newline)
  (exit 1))

;;;=================================================================
;;; Miscellaneous utilities
;;;

(define (send-message out header body)
  (write header out) (newline out)
  (write body out)   (newline out)
  (flush out))

(define (receive-message in)
  (let* ((header (read in))
         (body   (read in)))
    (values header body)))

(define (log-worker-action action worker)
  ((logger-of worker) action))

(define (get-worker-type header)
  (cond ((assoc "x-kahua-worker" header)
         => (lambda (p) (string->symbol (cadr p))))
        (else #f)))

;; returns a list suitable to pass run-process.  option-list<commit_msg> r3170@raven:  bizenn | 2007-09-05 17:54:09 +0900  Add log output.<commit_after>;; -*- mode: scheme; coding: utf-8 -*-
;; "Supervisor" or super server for kahua
;;
;;  Copyright (c) 2003-2006 Scheme Arts, L.L.C., All rights reserved.
;;  Copyright (c) 2003-2006 Time Intermedia Corporation, All rights reserved.
;;  See COPYING for terms and conditions of using this software
;;

;; For clients, this server works as a receptionist of kahua system.
;; It opens a socket where initial clients will connect.
;; This server doesn't know about internals of kahua server, but
;; just dispatches it to the worker servers.
;; Eventually, this server will manage multiple worker servers for
;; load balancing or hot restarting.
;; For now, we have only one worker server, so this is just an outline
;; of what we will ultimately do.

(use gauche.charconv)
(use gauche.net)
(use gauche.threads)
(use gauche.process)
(use gauche.selector)
(use gauche.listener)
(use gauche.parseopt)
(use gauche.parameter)
(use gauche.collection)
(use gauche.mop.singleton)
(use srfi-1)
(use srfi-2)
(use srfi-11)
(use srfi-13)
(use file.util)
(use util.queue)
(use util.list)
(use util.match)
(use kahua.config)
(use kahua.gsid)
(use kahua.developer)
(use kahua.util)
(use kahua.thread-pool)
(use kahua.protocol.worker)

(define *spvr* #f) ;; bound to supervisor object for convenience

(define *default-sigmask* #f)

(define *worker-types* '()) ;; for multi-thread version

(define *default-worker-type* #f)

(define-constant *TERMINATION-SIGNALS* (sys-sigset-add! (make <sys-sigset>) SIGTERM SIGINT SIGHUP))

;; Supervisor protocol
;;
;; [Session initiation]
;;
;;  A client first connect to a well known socket of the supervisor.
;;  This first request is called session-initiating request.
;;  At this point, a client wouldn't have complete GSID.  It may have
;;  state ID from the last session, but it certainly doesn't have
;;  continuation ID.
;;
;;  When the supervisor observes it, and the request is not for direct
;;  administrative request for the supervisor itself, it assigns a
;;  worker and forwards the session-initiating request to the worker.
;;  The worker will return a reply, usually accompanied by GSID.
;;  The supervisor forwards the reply to the client.
;;
;;  Afterwards, the client can figure out the worker ID encoded in GSID,
;;  and directly connects to the worker.
;;
;;  Client can optionally pass the worker type it wants to talk to,
;;  using "x-kahua-worker" header.   If such header is absent, the
;;  supervisor selects the default worker.
;;
;;  As a special case, "spvr" is given to "x-kahua-worker", the body
;;  is interpreted by the spvr process as a command.  See "supervisor
;;  commands" below.
;;
;; [Message format]
;;
;;  Request and reply both consist of two S-expressions, a header and
;;  a body.  A header is a list of two-element lists, resembles to
;;  what rfc822-header->list returns.   A body can be any valid sexpr.
;;  It is arguable whether this format is adequat or not.  Let's see.
;;
;;  In a request header, "x-kahua-sgsid" elemnt carries state GSID,
;;  and "x-kahua-cgsid" carries continuation GSID.  Other header can
;;  be freely used by the worker.  A client may send additional information
;;  in the header, and the worker should ignore the header element that
;;  it doesn't understand.
;;
;;  In a reply header, "x-kahua-sgsid" and "x-kahua-cgsid" are also
;;  used to carry GSID.  It also contains "x-kahua-status", whose value
;;  is either "OK", "ERROR", or "SPVR-ERROR".   "ERROR" indicates
;;  an error occurred in the worker, and "SPVR-ERROR" indicates an
;;  error occurred in the supervisor.  The message body of error replies
;;  contains a list of a error message string (for now).
;;

;;;=================================================================
;;; Global structure
;;;

(define-class <kahua-spvr> ()
  ((sockbase     :init-form (kahua-sockbase))
   (wtype-table  :init-form (make-hash-table 'eq?)      :getter wtype-table-of)
   (wid-table    :init-form (make-hash-table 'string=?) :getter wid-table-of)
   (wno-table    :init-form (make-hash-table 'eq?)      :getter wno-table-of)
   (selector     :init-form (make <selector>)           :getter selector-of)
   (mutex        :init-form (make-mutex)                :getter mutex-of)
   (gosh-path    :init-keyword :gosh-path) ; Absolute path of gosh, passed
                                           ; by wrapper script.
   (lib-path     :init-keyword :lib-path)  ; Path where kahua library files
                                           ; are installed.
   (keyserv      :init-value #f)           ; keyserver process
   (httpd        :init-value #f)	   ; httpd process
   ))

(define-class <kahua-worker-type> ()
  ((spvr   :init-keyword :spvr :getter spvr-of) ; back ptr to spvr
   (name   :init-keyword :name :getter name-of)	; worker type name (symbol)
   (count  :init-keyword :count
	   :getter count-of :init-value 0)	; count of running worker
   (workers :init-value #f :getter workers-of)  ; circular list of kahua-worker instances.
   (mutex  :init-form (make-mutex) :getter mutex-of)))

(define-class <kahua-worker> ()
  ((type   :getter type-of	      ; back ptr to kahua-worker-type
	   :init-keyword :type)
   (wid    :getter wid-of)	      ; worker id (string)
   (wno    :getter wno-of)	      ; Worker No.
   (process :getter process-of)	      ; worker process
   (sockaddr :getter sockaddr-of)     ; socket address
   (logger :getter logger-of)	      ; log output procedure has one arguments
   (start-time :getter start-time-of  ; timestamp
               :init-form (sys-time))
   (zombee :getter zombee?	      ; it's going to shutdown?
	   :init-form #f)
   ;; internal
   (next-wno :allocation :class :init-value 0)
   ))

(define-class <kahua-keyserv> ()
  ((process :init-keyword :process) ;; <process>
   (id      :init-keyword :id)      ;; keyserver id
   ))

;;;=================================================================
;;; Error handling
;;;

(define (spvr-errorf class fmt . args)
  (apply kahua:log-format fmt args)
  (apply errorf class fmt args))

(define (spvr-error-header err)
  (let1 err-name (class-name (class-of err))
    (kahua:log-format "SPVR-ERROR: ~a: ~a"
		      err-name (if (slot-exists? err 'message) (slot-ref err 'message) ""))
    `(("x-kahua-status" "SPVR-ERROR" ,err-name))))

;; If errors occur before spvr service starts, we should terminate
;; spvr with appropriate error message.
(define (app-error msg . args)
  (apply format #t msg args)
  (newline)
  (exit 1))

;;;=================================================================
;;; Miscellaneous utilities
;;;

(define (send-message out header body)
  (write header out) (newline out)
  (write body out)   (newline out)
  (flush out))

(define (receive-message in)
  (let* ((header (read in))
         (body   (read in)))
    (values header body)))

(define (log-worker-action action worker)
  ((logger-of worker) action))

(define (get-worker-type header)
  (cond ((assoc "x-kahua-worker" header)
         => (lambda (p) (string->symbol (cadr p))))
        (else #f)))

;; returns a list suitable to pass run-process.  option-list
{'fraction_non_alphanumeric': 0.14795954452224344, 'fraction_numerical': 0.005728835136855506, 'mean_word_length': 3.3561306223043745, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2713092', 'n_tokens_mistral': 4644, 'n_tokens_neox': 4480, 'n_words': 1616}