Consuming HTTP endpoint using Common Lisp
Tags: commonlisp, parallel, productivity
Published: 2022-10-17 14:09:34

I'll discuss approach how to make things run in parallel to collect data from some http endpoint. Far from perfect but good as starter to exploring this are.

**Situation:** we have some old code (coded by me years ago) that was a quick fix to collect incidents data from Service Mananger 9 (previous HP product, now maybe Microfocus). The code was ugly hack done in approx 1 day to get this thing done somehow. I opted back then to use dumb shell solution. As we all know, living in linux shell means you can do a lot, albeit not so efficiently. So the code grabbed incident numbers from DB, that were not closed. For each incident it executed curl to get status (and the stupid design required you to ask for incident and state - either open or closed, so if you have 1000 incidents, you have to do 2000 queries). The code put cap on max 24 running queries at a time. This means we constantly bombed kernel with new processes just to get some data from https endpoint.

**Version 0.1**

I have decided it is time to have some fun and use **Common Lisp** to create algorithm representation that deals with parallel execution. For this I decided to use [Clozure common lisp](https://ccl.clozure.com/), put basic [Qucklisp](https://www.quicklisp.org/beta/) there and load **some** libraries to do this.

**Parallel**
First package to use is [lparallel](https://lparallel.org/) to enable parallel processing without much coding on my side. Thing are easy here, you define `lparallel:*kernel*` with number of workers available for parallel tasks, define channel to receive results and start coding. I have actually used approach that does not even require channel for results.

```
(defparameter worker-count 32)
(setf lparallel:*kernel* (lparallel:make-kernel worker-count :bindings
    `((*standard-output* . ,*standard-output*)
      (*error-output*    . ,*error-output*) 
      (*drakma-stream* . nil) ))))
(setf *master-channel* (lparallel:make-channel))

```
You might raise eyebrows what is binding for? Well any program has one input, one output and one error stream/file descriptor when running. We are basically telling parallel kernel to have the same thing in all parallel threads defined. The last piece, `*drakma-stream*` is initially set to nil so that each parallel task can have its own connection available.

**Drakma continous querying**

We will reuse connection between http requests to save time on SSL handshake and overall load of target server. We are talking to some **Tomcat** server that has setup of 120 seconds for keep-alive. This means that we should take care of that and properly close and re-open connection when keep-alive expires. After some experiments, we defined base minimum function that wraps [drakma](https://github.com/edicl/drakma) calls. 

First we define helper macro to add stream to `drakma` call if we are reusing it. We expect some `TARGET-URI` to hold value of endpoint url. The `content` is post form content tha is used in query.

```
(defmacro DRAKMA-REQUEST-CONT (content a-stream)
  "Macro constructing drakma:http-request according to a-stream value"
  (let* ((request-1 (list 'drakma:http-request (TARGET-URI)
                            :method :post
                            :verify nil
                            :content-type "application/xml"
                            :close nil
                            :content content))
         (request (if (null a-stream) 
                      request-1
                      (append request-1 (list :stream a-stream))))
         (response `(ignore-errors (multiple-value-list ,request))))
         (if response
             response
              `(multiple-value-list ,request-1))))

```

Next we define function that will do the handling

```
(defmacro PARSE-RESULT-OCTET(result-octet)
    `(flexi-streams:octets-to-string ,result-octet :external-format :utf-8))


(defun SM9-QUERY (content)
  "Query SM9 via rws reusing connection stored in drakma-sream variable"
  ; Check drakma-stream
  (cond ((null *drakma-stream*)    (log:info  "Stream not opened, creating new connection."))
        ((streamp *drakma-stream*) (if (not (open-stream-p *drakma-stream*))
                                     (progn 
                                         (log:debug "Stream is closed, setting stream to nil")
                                         (setf *drakma-stream* nil))
                                     ;(log:debug "Stream opened, keeping it.")
                                     ))
       (t                        (log:debug "Stream opened, keeping it.")))
  (let* ((result-values (DRAKMA-REQUEST-CONT content *drakma-stream*))
         (result-stream (if (nth 5 result-values) nil (nth 4 result-values))) ;must-close is true - return nil as stream
         (status-code (cadr result-values))) 
        (if (null result-stream)
            (progn (if *drakma-stream* (close *drakma-stream*)) (setf *drakma-stream* nil))
            (setf *drakma-stream* result-stream))
        ;(log:debug status-code)
        (format t ".")
        (values (PARSE-RESULT-OCTET (car result-values)) (cadr result-values))))

```

With this in hand we have the base to run parallel getting of information reusing connection. How do we get this rolling?

**Execution**

We will define some `result` for holding what we collected.
```
(setf result nil)

```
We somehow get the ticket-ids from db and hold it in `*open-tickets*` list. The easiest way is to do
```
(progn 
  (setf result (lparallel:pmapcar #'SM9-GET-TICKET-BY-ID *open-tickets*)) 
  (SM9-close-sockets) 
  "Execution finished"))

```
**It is very important to close streams we used during collection**. We defined `SM9-close-sockets` as function that will broadcast task to all parallel threads of execution.
```
(defun SM9-close-sockets ()
 "closes all sockets inside threads"
 (lparallel:broadcast-task #'(lambda() (if (and (not (null *drakma-stream*))  (open-stream-p *drakma-stream*)) (close *drakma-stream*))(setf *drakma-stream* nil))))

```
Last piece is `SM9-GET-TICKET-BY-ID` that takes ticket id and create some data (contetnt) to be send to target to obtain results.
It could look like this (expects defined `SM9-USER`)
```
(defun SM9-REQUEST-TICKET(ticketId status)
    (concatenate 'string
    "<?xml version=\"1.0\" encoding=\"UTF-8\"  standalone=\"yes\"?>
    <ticketRetrieveRequest>
        <criteria>
            <customerDetails>
                <companyId>****</companyId>
                <locationId></locationId>
                <managementRegion>*****</managementRegion>
                <requestedById>" SM9-USER "</requestedById>
                <requestedForId>" SM9-USER "</requestedForId>
            </customerDetails>
            <systemType>Incident</systemType>
            <status>" status "</status>
            <ticketId>" ticketId "</ticketId>
    </criteria>
    </ticketRetrieveRequest>"))

(defun SM9-GET-TICKET-BY-ID (ticketId)
  "Get given ticketId from SM9 via rws server, tries both open and closed statuses, returns parsed object or NIL"
  (let* ((result1 (SM9-QUERY (SM9-REQUEST-TICKET ticketId "open")))
         (result2 (SM9-QUERY (SM9-REQUEST-TICKET ticketId "closed"))))
        (or
             (if (ignore-errors (jsown:val (ignore-errors (jsown:parse result1)) "tickets")) result1)   ; try to parse result1
             (if (ignore-errors (jsown:val (ignore-errors (jsown:parse result2)) "tickets")) result2)))) ; try to parse result2, at least one should be ok

```

So now we have version 0.1 ready we can play and test. Obvious question might be: why do we have to wait for all processing to be done? Cannot we write as we go when we have something ready? 
Reasons are obvious - if the searched scope is too big (like millions of requests) we might run `out-of-memory` and we will have to wait for this to happen. The second might be we want to know data as we got them, even if we do `polling`.

**Let us make V 0.2**

We will explore further and create writing thread, that will receive result along the way (it can be obviously something else than write, but hey, this is basic you can easily check :))

Still not perfect, but slightly better.
Let's define `file-writer` function that works on queues.
```
(defun file-writer(output-queue signal-queue feedback-queue target-file-name)
  ;;truncate file first
  (let ((stream (open target-file-name :direction :output :if-exists :supersede :if-does-not-exist :create)))
       (close stream))
  ;;open file for results
  (with-open-file (stream target-file-name :direction :output :if-exists :append)
        ;;loop trhough output-queue , watch signal-queue fol final run
        (loop while (lparallel.queue:queue-empty-p signal-queue)
          do (multiple-value-bind (message exists) (lparallel.queue:try-pop-queue output-queue :timeout 1)
                  (if exists 
                    (format stream "~a~%" message))))
        ;read signal
        (if (lparallel.queue:queue-empty-p signal-queue)
          ;this should not happen, we are waiting for signal to finish
          t
          ;actually there should be a signal in queue, read it
          (format t "Got signal ~a~%" (lparallel.queue:pop-queue signal-queue))
          )
        ; we might get signal sooner than all results are ready
        (if (lparallel.queue:queue-empty-p output-queue)
          ; nothing to do 
          t
          ; else process other requests
          (loop while (not (lparallel.queue:queue-empty-p output-queue))
            do (let ((message (lparallel.queue:pop-queue output-queue)))
                    (format stream "~a~%" message)
                    )))
        )
  (lparallel.queue:push-queue "writer finished" feedback-queue))

```

Function uses `lparallel.queue` package that comes with `lparallel` when `quicklisp` loads it. It takes `target-file-name`, that it truncates at beginning. Next it looks at `signal-queue` (if it is empty) and tries to read data from `output-queue` with timeout. So function reads the output queue while watching if there is signal on signal-queue to finish processing. After receiving signal on signal-queue, it reads that signal, empties the output-queue (process it) and send message to feedback-queue for synchronizing. **All this is to ensure, we properly close output stream. if we don't, we might experience truncated output at filesystem** (like 16k or 72k or whatever it will be).

Lparallel uses `bordeaux-threads` so we do not have to ask for it separately (quicklisp will load it as dependency). In order to start our new function, we define file-name, and queues, and run the writer.
```
(setf *output-queue* (lparallel.queue:make-queue :fixed-capacity 500))
(setf *signal-queue* (lparallel.queue:make-queue))
(setf *feedback-queue* (lparallel.queue:make-queue))

(defun run-in-background (fn &optional (name "Generic Background task") &rest params)
    (let ((background-task (bt:make-thread (lambda()
                            (apply fn params)) :name name)))
           broadcast-task))
;run writer thread
(run-in-background #'file-writer "SM9-Writer-Thread" *output-queue* *signal-queue* *feedback-queue* target-file-name)
;-------------------------
;main processing goes here
;-------------------------
;you can instruct file-writer to finish and close file and let you know when finished, by executing
(lparallel.queue:push-queue "finish" *signal-queue* )

;and wait for it to finishes by waiting on pop
(lparallel.queue:pop-queue *feedback-queue* )
;-------------------------
;this piece can run only after we processed all records
;-------------------------


```
Now we have the basics in place, but do not know how to process the data. Well actually it is not that difficult. Instead of returning the result from querying, we push it to output queue [ (lparallel.queue:push-queue data *output-queue*) ].
```
(defun SM9-GET-TICKET-BY-ID (ticketId)
  "Get given ticketId from SM9 via rws server, tries both open and closed statuses, returns parsed object or NIL"
  (let* ((result1 (SM9-QUERY (SM9-REQUEST-TICKET ticketId "open")))
         (result2 (SM9-QUERY (SM9-REQUEST-TICKET ticketId "closed"))))
        (or
             (if (ignore-errors (jsown:val (ignore-errors (jsown:parse result1)) "tickets")) (lparallel.queue:push-queue result1 *output-queue*))   ; try to parse result1
             (if (ignore-errors (jsown:val (ignore-errors (jsown:parse result2)) "tickets")) (lparallel.queue:push-queue result2 *output-queue*))))) ; try to parse result2, at least one should be ok

```
Initial benchmark shows we went from overloaded system (forks in shell scripts doing curl, jq, awk and whatnot) to balance execution with much shorter execution time (keep-alive, reusing connections).

That's it for first write, expecting constructive criticism. If you like this, let me know, I'll appreciate this. 

Cheers!