<commit_before>(in-package :cl-rabbit-async)

(declaim (optimize (safety 3) (debug 3) (speed 0)))

(defclass async-channel ()
  ((connection        :type async-connection
                      :initarg :connection
                      :reader async-channel/connection)
   (channel           :type integer
                      :initarg :channel
                      :reader async-channel/channel)
   (message-callbacks :type list
                      :initarg :message-callbacks
                      :initform nil
                      :accessor async-channel/message-callbacks)
   (close-callbacks   :type list
                      :initarg :close-callbacks
                      :initform nil
                      :accessor async-channel/close-callbacks)
   (close-p           :type t
                      :initform nil
                      :accessor async-channel/close-p)))

(defclass async-connection ()
  ((connection    :type cl-rabbit::connection
                  :initarg :connection
                  :reader async-connection/connection)
   (channels      :type vector
                  :initform (make-array (list 10)
                  :element-type '(or null async-channel)
                  :initial-element nil
                  :adjustable t
                  :fill-pointer nil)
                  :reader async-connection/channels)
   (cmd-fd        :type fixnum
                  :accessor async-connection/cmd-fd)
   (cmd-fd-reader :type fixnum
                  :accessor async-connection/cmd-fd-reader)
   (close-p       :type t
                  :initform nil
                  :accessor async-connection/close-p)
   (b-current     :type (or null integer)
                  :initform nil
                  :accessor async-connection/b-current)
   (b-index       :type integer
                  :initform 0
                  :accessor async-connection/b-index)
   (b-lock        :type t
                  :initform (bordeaux-threads:make-lock "Sync connection lock")
                  :reader async-connection/b-lock)
   (b-condvar     :type t
                  :initform (bordeaux-threads:make-condition-variable :name "Sync connection condvar")
                  :reader async-connection/b-condvar)))

(defun process-unexpected-frame (conn)
  (simple-wait-frame conn))

(defun find-free-channel-index (async-conn)
  (with-accessors ((channels async-connection/channels))
      async-conn
    (let ((pos (position nil channels)))
      (or pos
          (let ((old-size (length channels)))
            (adjust-array channels (* old-size 2) :initial-element nil)
            old-size)))))

(defun wait-for-frame (async-conn)
  (log:info "Waiting for frame on async connection: ~s" async-conn)
  (with-accessors ((channels async-connection/channels)
                   (conn async-connection/connection))
      async-conn
    ;;
    (handler-bind ((cl-rabbit:rabbitmq-library-error
                    (lambda (condition)
                      (when (eq (cl-rabbit:rabbitmq-library-error/error-code condition)
                                :amqp-unexpected-frame)
                        (log:info "Got unexpected frame")
                        (process-unexpected-frame conn)))))
      ;;
      (let ((envelope (cl-rabbit:consume-message conn)))
        (let* ((channel-id (cl-rabbit:envelope/channel envelope))
               (index (1- channel-id)))
          (labels ((warn-nonexistent-channel ()
                     (log:warn "Message received for closed channel: ~a" channel-id)))
            (if (< index (length channels))
                (let ((async-channel (aref channels index)))
                  (if async-channel
                      (dolist (callback (async-channel/message-callbacks async-channel))
                        (funcall callback envelope))
                      ;; ELSE: Unused channel
                      (warn-nonexistent-channel)))
                ;; ELSE: Message for unused channel
                (warn-nonexistent-channel))))))))

(defun load-integer-from-fd (in-fd)
  (let ((size (cffi:foreign-type-size :int)))
    (cffi:with-foreign-pointer (buf size)
      (let ((result (iolib.syscalls:read in-fd buf size)))
        (unless (= result size)
          (error "Unable to read from pipe"))
        (cffi:mem-ref buf :int)))))

(defun handle-command (async-conn)
  (with-accessors ((b-lock async-connection/b-lock)
                   (b-condvar async-connection/b-condvar)
                   (b-current async-connection/b-current))
      async-conn
    (let ((request-index (load-integer-from-fd (async-connection/cmd-fd-reader async-conn))))
      (bordeaux-threads:with-lock-held (b-lock)
        (when b-current
          (error "The current handler index is not NIL: ~s" b-current))
        (setf b-current request-index)
        (bordeaux-threads:condition-notify b-condvar)
        (loop
           while b-current
           do (bordeaux-threads:condition-wait b-condvar b-lock))))))

(defun run-sync-loop (async-conn)
  (let* ((conn (async-connection/connection async-conn))
         (socket-fd (cl-rabbit::get-sockfd conn))
         (in-fd (async-connection/cmd-fd-reader async-conn))
         (event-base (make-instance 'iolib:event-base)))

    (iolib:set-io-handler event-base socket-fd
                          :read (lambda (fd type c)
                                  (declare (ignore fd type c))
                                  (wait-for-frame async-conn)))
    (iolib:set-io-handler event-base in-fd
                          :read (lambda (fd type c)
                                  (declare (ignore fd type c))
                                  (handle-command async-conn)))

    (loop
       if (or (cl-rabbit::data-in-buffer conn)
              (cl-rabbit::frames-enqueued conn))
       do (wait-for-frame async-conn)
       else do (iolib:event-dispatch event-base :one-shot t))))

(defun get-next-b-index (async-conn)
  (bordeaux-threads:with-lock-held ((async-connection/b-lock async-conn))
    (incf (async-connection/b-index async-conn))))

(defun run-in-sync-thread (async-conn fn)
  (with-accessors ((b-lock async-connection/b-lock)
                   (b-condvar async-connection/b-condvar)
                   (b-current async-connection/b-current))
      async-conn
    (let ((request-index (get-next-b-index async-conn))
          (size (cffi:foreign-type-size :int)))
      (cffi:with-foreign-pointer (buf size)
        (setf (cffi:mem-ref buf :int) request-index)<commit_msg>Use long long when sending request ID's over the pipe<commit_after>(in-package :cl-rabbit-async)

(declaim (optimize (safety 3) (debug 3) (speed 0)))

(defclass async-channel ()
  ((connection        :type async-connection
                      :initarg :connection
                      :reader async-channel/connection)
   (channel           :type integer
                      :initarg :channel
                      :reader async-channel/channel)
   (message-callbacks :type list
                      :initarg :message-callbacks
                      :initform nil
                      :accessor async-channel/message-callbacks)
   (close-callbacks   :type list
                      :initarg :close-callbacks
                      :initform nil
                      :accessor async-channel/close-callbacks)
   (close-p           :type t
                      :initform nil
                      :accessor async-channel/close-p)))

(defclass async-connection ()
  ((connection    :type cl-rabbit::connection
                  :initarg :connection
                  :reader async-connection/connection)
   (channels      :type vector
                  :initform (make-array (list 10)
                  :element-type '(or null async-channel)
                  :initial-element nil
                  :adjustable t
                  :fill-pointer nil)
                  :reader async-connection/channels)
   (cmd-fd        :type fixnum
                  :accessor async-connection/cmd-fd)
   (cmd-fd-reader :type fixnum
                  :accessor async-connection/cmd-fd-reader)
   (close-p       :type t
                  :initform nil
                  :accessor async-connection/close-p)
   (b-current     :type (or null integer)
                  :initform nil
                  :accessor async-connection/b-current)
   (b-index       :type integer
                  :initform 0
                  :accessor async-connection/b-index)
   (b-lock        :type t
                  :initform (bordeaux-threads:make-lock "Sync connection lock")
                  :reader async-connection/b-lock)
   (b-condvar     :type t
                  :initform (bordeaux-threads:make-condition-variable :name "Sync connection condvar")
                  :reader async-connection/b-condvar)))

(defun process-unexpected-frame (conn)
  (simple-wait-frame conn))

(defun find-free-channel-index (async-conn)
  (with-accessors ((channels async-connection/channels))
      async-conn
    (let ((pos (position nil channels)))
      (or pos
          (let ((old-size (length channels)))
            (adjust-array channels (* old-size 2) :initial-element nil)
            old-size)))))

(defun wait-for-frame (async-conn)
  (log:info "Waiting for frame on async connection: ~s" async-conn)
  (with-accessors ((channels async-connection/channels)
                   (conn async-connection/connection))
      async-conn
    ;;
    (handler-bind ((cl-rabbit:rabbitmq-library-error
                    (lambda (condition)
                      (when (eq (cl-rabbit:rabbitmq-library-error/error-code condition)
                                :amqp-unexpected-frame)
                        (log:info "Got unexpected frame")
                        (process-unexpected-frame conn)))))
      ;;
      (let ((envelope (cl-rabbit:consume-message conn)))
        (let* ((channel-id (cl-rabbit:envelope/channel envelope))
               (index (1- channel-id)))
          (labels ((warn-nonexistent-channel ()
                     (log:warn "Message received for closed channel: ~a" channel-id)))
            (if (< index (length channels))
                (let ((async-channel (aref channels index)))
                  (if async-channel
                      (dolist (callback (async-channel/message-callbacks async-channel))
                        (funcall callback envelope))
                      ;; ELSE: Unused channel
                      (warn-nonexistent-channel)))
                ;; ELSE: Message for unused channel
                (warn-nonexistent-channel))))))))

(defun load-integer-from-fd (in-fd)
  (let ((size (cffi:foreign-type-size :long-long)))
    (cffi:with-foreign-pointer (buf size)
      (let ((result (iolib.syscalls:read in-fd buf size)))
        (unless (= result size)
          (error "Unable to read from pipe"))
        (cffi:mem-ref buf :long-long)))))

(defun handle-command (async-conn)
  (with-accessors ((b-lock async-connection/b-lock)
                   (b-condvar async-connection/b-condvar)
                   (b-current async-connection/b-current))
      async-conn
    (let ((request-index (load-integer-from-fd (async-connection/cmd-fd-reader async-conn))))
      (bordeaux-threads:with-lock-held (b-lock)
        (when b-current
          (error "The current handler index is not NIL: ~s" b-current))
        (setf b-current request-index)
        (bordeaux-threads:condition-notify b-condvar)
        (loop
           while b-current
           do (bordeaux-threads:condition-wait b-condvar b-lock))))))

(defun run-sync-loop (async-conn)
  (let* ((conn (async-connection/connection async-conn))
         (socket-fd (cl-rabbit::get-sockfd conn))
         (in-fd (async-connection/cmd-fd-reader async-conn))
         (event-base (make-instance 'iolib:event-base)))

    (iolib:set-io-handler event-base socket-fd
                          :read (lambda (fd type c)
                                  (declare (ignore fd type c))
                                  (wait-for-frame async-conn)))
    (iolib:set-io-handler event-base in-fd
                          :read (lambda (fd type c)
                                  (declare (ignore fd type c))
                                  (handle-command async-conn)))

    (loop
       if (or (cl-rabbit::data-in-buffer conn)
              (cl-rabbit::frames-enqueued conn))
       do (wait-for-frame async-conn)
       else do (iolib:event-dispatch event-base :one-shot t))))

(defun get-next-b-index (async-conn)
  (bordeaux-threads:with-lock-held ((async-connection/b-lock async-conn))
    (incf (async-connection/b-index async-conn))))

(defun run-in-sync-thread (async-conn fn)
  (with-accessors ((b-lock async-connection/b-lock)
                   (b-condvar async-connection/b-condvar)
                   (b-current async-connection/b-current))
      async-conn
    (let ((request-index (get-next-b-index async-conn))
          (size (cffi:foreign-type-size :long-long)))
      (cffi:with-foreign-pointer (buf size)
        (setf (cffi:mem-ref buf :long-long) request-index)
{'fraction_non_alphanumeric': 0.11870779619398404, 'fraction_numerical': 0.0012277470841006752, 'mean_word_length': 1.7231508566652738, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, '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': '4705779', 'n_tokens_mistral': 3766, 'n_tokens_neox': 3567, 'n_words': 834}