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

(define-condition socket-closed (connection-error) ()
  (:report (lambda (c s) (format s "Socket closed: ~a: ~a" (conn-errcode c) (conn-errmsg c))))
  (:documentation "Thrown when a closed socket is being operated on."))

(defclass socket ()
  ((c :accessor socket-c :initarg :c :initform (cffi:null-pointer))
   (closed :accessor socket-closed :initarg :closed :initform nil)
   (direction :accessor socket-direction :initarg :direction :initform nil))
  (:documentation "Wraps around a libevent bufferevent socket."))

(defun get-last-tcp-err ()
  "Since libevent provides a macro but not a function for getting the last error
   code, we have to essentially rebuild that macro here.

   NOTE: As libevent says, this function cannot be considered idempotent. In
   other words, calling it more than once may result in different values
   returned. So, call once and save the result until the next time an error
   happens."
  #+(or :windows :win32 :win64)
    (let ((code (cffi:foreign-funcall "WSAGetLastError" :int)))
      (values code
              (cffi:foreign-funcall "evutil_socket_error_to_string" :int code :string)))
  #-(or :windows :win32 :win64)
    (let ((code (cffi:mem-aref (cffi:foreign-symbol-pointer "errno") :int)))
      (values code
              (cffi:foreign-funcall "strerror" :int code :string))))

(defun close-socket (socket)
  "Free a socket (bufferevent) and clear out all associated data."
  ;; grab the data pointer associated with the bufferevent and free it. see
  ;; comment tcp-send about data-pointer for a better explanation.
  (check-socket-open socket)
  (let* ((bev (socket-c socket))
         (bev-data (deref-data-from-pointer bev))
         (bev-data-pointer (getf bev-data :data-pointer))
         (fd (le::bufferevent-getfd bev)))
    (le:bufferevent-disable bev (logior le:+ev-read+ le:+ev-write+))
    (if (eq (socket-direction socket) 'in)
        (decf *incoming-connection-count*)
        (decf *outgoing-connection-count*))
    (le:bufferevent-free bev)
    (le:evutil-closesocket fd)
    (setf (socket-closed socket) t)
    (free-pointer-data bev-data-pointer)
    (free-pointer-data bev :preserve-pointer t)))

(defun check-socket-open (socket)
  "Throw a socket-closed condition if given a socket that's closed."
  (when (subtypep (type-of socket) 'socket)
    (when (socket-closed socket)
      (error 'socket-closed :code -1 :msg "Trying to operate on a closed socket"))))

(defun free-listener (listener)
  "Free a socket listener and all associated data."
  (free-pointer-data listener :preserve-pointer t)
  (le:evconnlistener-free listener))

(defun set-socket-timeouts (socket read-sec write-sec &key socket-is-bufferevent)
  "Given a pointer to a libevent socket (bufferevent), set the read/write
   timeouts on the bufferevent."
  (check-socket-open socket)
  (let ((socket (if socket-is-bufferevent
                    socket
                    (socket-c socket))))
    (multiple-value-bind (read-sec read-usec) (split-usec-time read-sec)
      (multiple-value-bind (write-sec write-usec) (split-usec-time write-sec)
        (make-foreign-type (read-to (le::cffi-type le::timeval))
                           (('le::tv-sec read-sec)
                            ('le::tv-usec read-usec))
          (make-foreign-type (write-to (le::cffi-type le::timeval))
                             (('le::tv-sec write-sec)
                              ('le::tv-usec write-usec))
            (let ((read-to (if (numberp read-sec) read-to (cffi:null-pointer)))
                  (write-to (if (numberp write-sec) write-to (cffi:null-pointer))))
            (le:bufferevent-set-timeouts socket read-to write-to))))))))

(defun read-socket-data (socket data-cb &key socket-is-evbuffer)
  "Read the data out of a bufferevent (or directly, an evbuffer)."
  (check-socket-open socket)
  (let ((bufsize *buffer-size*)
        (buffer-c *socket-buffer-c*)
        (buffer-lisp *socket-buffer-lisp*)
        (input (if socket-is-evbuffer
                   socket
                   (le:bufferevent-get-input (socket-c socket)))))
    (loop for n = (le:evbuffer-remove input buffer-c bufsize)
          while (< 0 n) do
      (dotimes (i n)
        (setf (aref buffer-lisp i) (cffi:mem-aref buffer-c :char i)))
      (funcall data-cb (subseq buffer-lisp 0 n)))))

(defun write-socket-data (socket data &key write-cb socket-is-evbuffer)
  "Write data into libevent socket. Allows specifying if the given socket is
   actually just a libevent evbuffer, or a socket (aka bufferevent). Also allows
   assigning of a write-cb, which will be called when the data written to the
   evbuffer is flushed out to the socket. This can be useful for closing a
   connection after writing data to it."
  (check-socket-open socket)

  ;; if a write-cb was passed, set it into the socket's callbacks
  (when (and (not socket-is-evbuffer) write-cb (functionp write-cb))
    (let* ((bev (socket-c socket))
           (bev-data (deref-data-from-pointer bev))
           (socket-data-pointer (getf bev-data :data-pointer))
           (callbacks (get-callbacks socket-data-pointer))
           (read-cb (getf callbacks :read-cb (cffi:null-pointer)))
           (event-cb (getf callbacks :event-cb (cffi:null-pointer))))
    (le:bufferevent-enable bev (logior le:+ev-write+))
    (save-callbacks socket-data-pointer (list :read-cb read-cb :write-cb write-cb :event-cb event-cb))))
  
  ;; copy the data into an evbuffer and ship it off to the socket
  (let* ((data (if (stringp data)
                   (babel:string-to-octets data :encoding :utf-8)
                   data))
         (data-length (length data))
         (data-index 0)
         (evbuffer (if socket-is-evbuffer
                       socket
                       (le:bufferevent-get-output (socket-c socket))))
         (buffer-c *socket-buffer-c*))
    ;; copy into evbuffer using existing c buffer
    (loop while (< 0 data-length) do
      (let ((bufsize (min data-length *buffer-size*)))
      (dotimes (i bufsize)
        (setf (cffi:mem-aref buffer-c :unsigned-char i) (aref data data-index))
        (incf data-index))
      (le:evbuffer-add evbuffer buffer-c bufsize)
      (decf data-length bufsize)))))

(defun drain-evbuffer (evbuffer)
  "Grab all data in an evbuffer and put it into a byte array (returned)."
  (let ((body nil))
    (read-socket-data evbuffer
                      (lambda (data)
                        (if body
                            (setf body (append-array body data :element-type '(unsigned-byte 8)))
                            data))
                      :socket-is-evbuffer t)
    body))
<commit_msg>Fix drain-evbuffer following my APPEND-ARRAY optimization (sorry)<commit_after>(in-package :cl-async)

(define-condition socket-closed (connection-error) ()
  (:report (lambda (c s) (format s "Socket closed: ~a: ~a" (conn-errcode c) (conn-errmsg c))))
  (:documentation "Thrown when a closed socket is being operated on."))

(defclass socket ()
  ((c :accessor socket-c :initarg :c :initform (cffi:null-pointer))
   (closed :accessor socket-closed :initarg :closed :initform nil)
   (direction :accessor socket-direction :initarg :direction :initform nil))
  (:documentation "Wraps around a libevent bufferevent socket."))

(defun get-last-tcp-err ()
  "Since libevent provides a macro but not a function for getting the last error
   code, we have to essentially rebuild that macro here.

   NOTE: As libevent says, this function cannot be considered idempotent. In
   other words, calling it more than once may result in different values
   returned. So, call once and save the result until the next time an error
   happens."
  #+(or :windows :win32 :win64)
    (let ((code (cffi:foreign-funcall "WSAGetLastError" :int)))
      (values code
              (cffi:foreign-funcall "evutil_socket_error_to_string" :int code :string)))
  #-(or :windows :win32 :win64)
    (let ((code (cffi:mem-aref (cffi:foreign-symbol-pointer "errno") :int)))
      (values code
              (cffi:foreign-funcall "strerror" :int code :string))))

(defun close-socket (socket)
  "Free a socket (bufferevent) and clear out all associated data."
  ;; grab the data pointer associated with the bufferevent and free it. see
  ;; comment tcp-send about data-pointer for a better explanation.
  (check-socket-open socket)
  (let* ((bev (socket-c socket))
         (bev-data (deref-data-from-pointer bev))
         (bev-data-pointer (getf bev-data :data-pointer))
         (fd (le::bufferevent-getfd bev)))
    (le:bufferevent-disable bev (logior le:+ev-read+ le:+ev-write+))
    (if (eq (socket-direction socket) 'in)
        (decf *incoming-connection-count*)
        (decf *outgoing-connection-count*))
    (le:bufferevent-free bev)
    (le:evutil-closesocket fd)
    (setf (socket-closed socket) t)
    (free-pointer-data bev-data-pointer)
    (free-pointer-data bev :preserve-pointer t)))

(defun check-socket-open (socket)
  "Throw a socket-closed condition if given a socket that's closed."
  (when (subtypep (type-of socket) 'socket)
    (when (socket-closed socket)
      (error 'socket-closed :code -1 :msg "Trying to operate on a closed socket"))))

(defun free-listener (listener)
  "Free a socket listener and all associated data."
  (free-pointer-data listener :preserve-pointer t)
  (le:evconnlistener-free listener))

(defun set-socket-timeouts (socket read-sec write-sec &key socket-is-bufferevent)
  "Given a pointer to a libevent socket (bufferevent), set the read/write
   timeouts on the bufferevent."
  (check-socket-open socket)
  (let ((socket (if socket-is-bufferevent
                    socket
                    (socket-c socket))))
    (multiple-value-bind (read-sec read-usec) (split-usec-time read-sec)
      (multiple-value-bind (write-sec write-usec) (split-usec-time write-sec)
        (make-foreign-type (read-to (le::cffi-type le::timeval))
                           (('le::tv-sec read-sec)
                            ('le::tv-usec read-usec))
          (make-foreign-type (write-to (le::cffi-type le::timeval))
                             (('le::tv-sec write-sec)
                              ('le::tv-usec write-usec))
            (let ((read-to (if (numberp read-sec) read-to (cffi:null-pointer)))
                  (write-to (if (numberp write-sec) write-to (cffi:null-pointer))))
            (le:bufferevent-set-timeouts socket read-to write-to))))))))

(defun read-socket-data (socket data-cb &key socket-is-evbuffer)
  "Read the data out of a bufferevent (or directly, an evbuffer)."
  (check-socket-open socket)
  (let ((bufsize *buffer-size*)
        (buffer-c *socket-buffer-c*)
        (buffer-lisp *socket-buffer-lisp*)
        (input (if socket-is-evbuffer
                   socket
                   (le:bufferevent-get-input (socket-c socket)))))
    (loop for n = (le:evbuffer-remove input buffer-c bufsize)
          while (< 0 n) do
      (dotimes (i n)
        (setf (aref buffer-lisp i) (cffi:mem-aref buffer-c :char i)))
      (funcall data-cb (subseq buffer-lisp 0 n)))))

(defun write-socket-data (socket data &key write-cb socket-is-evbuffer)
  "Write data into libevent socket. Allows specifying if the given socket is
   actually just a libevent evbuffer, or a socket (aka bufferevent). Also allows
   assigning of a write-cb, which will be called when the data written to the
   evbuffer is flushed out to the socket. This can be useful for closing a
   connection after writing data to it."
  (check-socket-open socket)

  ;; if a write-cb was passed, set it into the socket's callbacks
  (when (and (not socket-is-evbuffer) write-cb (functionp write-cb))
    (let* ((bev (socket-c socket))
           (bev-data (deref-data-from-pointer bev))
           (socket-data-pointer (getf bev-data :data-pointer))
           (callbacks (get-callbacks socket-data-pointer))
           (read-cb (getf callbacks :read-cb (cffi:null-pointer)))
           (event-cb (getf callbacks :event-cb (cffi:null-pointer))))
    (le:bufferevent-enable bev (logior le:+ev-write+))
    (save-callbacks socket-data-pointer (list :read-cb read-cb :write-cb write-cb :event-cb event-cb))))
  
  ;; copy the data into an evbuffer and ship it off to the socket
  (let* ((data (if (stringp data)
                   (babel:string-to-octets data :encoding :utf-8)
                   data))
         (data-length (length data))
         (data-index 0)
         (evbuffer (if socket-is-evbuffer
                       socket
                       (le:bufferevent-get-output (socket-c socket))))
         (buffer-c *socket-buffer-c*))
    ;; copy into evbuffer using existing c buffer
    (loop while (< 0 data-length) do
      (let ((bufsize (min data-length *buffer-size*)))
      (dotimes (i bufsize)
        (setf (cffi:mem-aref buffer-c :unsigned-char i) (aref data data-index))
        (incf data-index))
      (le:evbuffer-add evbuffer buffer-c bufsize)
      (decf data-length bufsize)))))

(defun drain-evbuffer (evbuffer)
  "Grab all data in an evbuffer and put it into a byte array (returned)."
  (let ((body nil))
    (read-socket-data evbuffer
                      (lambda (data)
                        (if body
                            (setf body (append-array body data))
                            data))
                      :socket-is-evbuffer t)
    body))

{'fraction_non_alphanumeric': 0.1252150497419403, 'fraction_numerical': 0.0021691973969631237, 'mean_word_length': 2.651010376843255, 'pattern_counts': {'":': 0, '<': 7, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2571461', 'n_tokens_mistral': 4282, 'n_tokens_neox': 4176, 'n_words': 1307}