Data still drained (?) with `init-tcp-socket`

[jd]

I'm trying to incorporate a cl-postgres into an event loop, in order to receive notification.

Here's my code:

(require :cl-async)
(require :postmodern)
(require :usocket)

(postmodern:connect-toplevel "db" "user" "pass" "localhost")

(defun get-socket-fd (connection)
  "Gets the fd from a socket connection."
  (let* ((type (type-of connection))
         (stream (cond ((subtypep type 'usocket:stream-usocket)
                        (usocket:socket-stream connection))
                       ((subtypep type 'stream)
                        connection))))
    (when stream
      #+sbcl
        (sb-sys:fd-stream-fd stream)
      #+cmu
        (system:fd-stream-fd stream)
      #+ccl
        (ccl::ioblock-device (ccl::stream-ioblock stream t))
      #+clisp
        (ext:stream-handles stream))))

(defun notification-handle (socket data)
  (declare (ignore socket data))
  ;; (format t "read char from socket ~a~%" (cl-postgres::read-str (cl-postgres::connection-socket postmodern:*database*))))
  (format t "Reading notification~%")
  (multiple-value-bind (channel payload pid) (cl-postgres:wait-for-notification
                                              postmodern:*database*)
    (format t "Notification received: ~a ~a ~a~%" channel payload pid)))

(defun start ()
  (postmodern:execute "LISTEN foobar")
  (as:init-tcp-socket #'notification-handle
                      (lambda (ev)
                        (format t "Event ~a ~%" ev))
                      :fd (get-socket-fd
                           (cl-postgres::connection-socket postmodern:*database*))
                      :dont-drain-read-buffer t)
  (format t "Started~%"))

(cl-async:start-event-loop #'start)

It's actually pretty simple: listen on foobar, and as soon as the socket receive a message, we read notification from the wire and print it out.

But this doesn't work as expected. If I run the program it prints

Started

Then I send a notification from psql with:

NOTIFY foobar, 'notif1';

The program prints Reading notification… but blocks, whereas it should read notif1 right away!

So I send a second notification with:

NOTIFY foobar, 'notif2';

And now it prints:

Notification received: foobar notif2 15438

So something ate up the first notification. If I do this with a 3rd and 4th notification, the 3th notification never appears too, and I get the 4th, etc.

Any hint?

[orthecreedence]

Just read this in the libevent docs: Make sure that the socket you provide to bufferevent_socket_new is in non-blocking mode. Libevent provides the convenience method evutil_make_socket_nonblocking for this. Oops. I pushed an update to master to make sure this happens. Please do a pull and see if this fixes your issue.

[orthecreedence]

Oh! I just re-read what you're trying to do, and I think I know what's happening. Although you still need to pull for that non-blocking flag to get set in init-tcp-socket, also what's happening is that when libevent gets the data on the socket, it still reads it off the socket into its buffer. :dont-drain-read-buffer simple tells cl-async's TCP implementation not to drain that read buffer when notifying that there's data on the socket (meaning the data param of the read-cb will be empty), and is mainly used for dealing with streams (ie if you want to wrap the socket in a async-io-stream stream manually and read data that way). However, the data is still being taken off the raw socket, which is what cl-postgres is trying to read (it doesn't have any concept of the libevent socket's read buffer). I might update the docs to make this more clear since I admit this is a bit confusing.

I don't suppose there's any way to get cl-postgres to work over a stream? You might get more traction that way since it's a more standardized interface for sending/receiving data.

[jd]

cl-postgres work over a stream, but I can't specify it which one to use here. At least not from the standard API. Isn't there anyway to tell libevent to not touch the socket at all? I just want being notified. :)

[orthecreedence]

I'll do a bit of research and get back to you, but my guess is probably not. Stay tuned!

[orthecreedence]

Wrong again. I realized that you can do this using plain old events. Note that at the moment, these aren't abstracted by cl-async and it would take a bit of work to do this. In the meantime, I'll provide some example code that should hopefully get you up and running:

(cffi:defcallback event-cb :void ((fd :int) (what :short) (arg :pointer))
  (cond
    ((< 0 (logand what le:+ev-read+))
     ;; fd ready to read from
     )
    ((< 0 (logand what le:+ev-write+))
     ;; fd ready to write to
     )
    ((< 0 (logand what le:+ev-timeout+))
     ;; fd/socket timeout
     )
    ((< 0 (logand what le:+ev-signal+))
     ;; signal on fd
     )))

(defun listen-to-fd (fd)
  (let ((event (le:event-new cl-async-util::*event-base*
                             fd
                             ;; listen to read/timeout events, and keep listening
                             (logior le:+ev-timeout+ le:+ev-read+ le:+ev-persist+)
                             (cffi:callback event-cb)
                             (cffi:null-pointer))))  ;; passed in as the "arg" param to the callback

    ;; store `event` somewhere here so that you can free it later

    ;; add the event to libevent to monitor the fd
    (le:event-add event (cffi:null-pointer))))

In this example, you pass the fd to listen-to-fd and it should trigger the event-cb when it times out or is ready to read from, at which point you can call the cl-postgres function you need to. Also remember that once the socket is closed, you'll have to free the event yourself via le:event-free to avoid memory leaks, so you may have to store it globally or reference it to the fd somehow.

Give this a shot and let me know how this turns out.

[jd]

Yeah, that definitely make things work, awesome!

I'll use that for the time being, but I think that gave you the kind of low-level feature you could add to cl-async. With this, I should be able to use cl-postgres in a async manner for notification.

Thanks again @orthecreedence!

[orthecreedence]

Glad to help! I'll work on an API for cl-async for wrapping raw events over the next few days as I get time. Let me know if you run into any more issues.