Немного о dataflow

[love5an]

Вчера думал над тем, что бы написать о cells. Так и не придумал пока, но зато полез читать про разные dataflow-фреймворки, и решил накатать на лиспе какой-нибудь свой.

Вот что вышло: neural-flow
От cells отличается как минимум тем, что интерфейс невероятно простой, совершенно без макросов, используется метаобъектный протокол, а документации еще меньше.
Работает на SBCL, Clozure CL и clisp, как минимум.

Подробнее:

Почему `neural'?

Ну, я подумал, что потоки данных в контексте dataflow парадигмы очень похожи на импульсы в нейронах.

Вот, нашел такую картинку в гугле, и подрисовал кое-чего в пейнте:



Дендриты это такие штуковины, по которым нейрон из синапсов, от других нейронов, принимает какие-либо сигналы.
После того, как он их обрабатывает, через аксон он передает новые сигналы, другим нейронам и другим клеткам.

При чем здесь dataflow, опять же? При том, что обработчики данных связаны в сеть, каждый из них получает данные от других обработчиков, что-то с ними делает, и пересылает дальше, всем обработчикам, с которыми связан.

Вообще говоря, я даже не стал особо утруждать себя выбором названия для класса обработчиков.

(defclass neuron (closer-mop:funcallable-standard-object)
  ((name :initarg :name :type symbol)
   (owner :initarg :owner)
   (receivers :initarg :receivers :type list)
   (value :initarg :value)
   (%function :initarg :function :type function)
   (handler :initarg :handler :type function)
   (handler-filter :initarg :handler-filter))
  (:default-initargs
      :name nil
      :owner nil
      :function (lambda (value &optional sender)
                  (declare (ignore sender))
                  value)
      :receivers '()
      :handler (lambda (neuron condition)
                 (declare (ignore neuron condition))
                 (values))
      :handler-filter 'error)
  (:metaclass closer-mop:funcallable-standard-class))

Слот "receivers" это такой аксон - коллекция других нейронов или просто функций, в которые нейрон передает обработанные данные. Последние обработанные данные хранятся в слоте "value". Слот "%function" - функция, ответственная за процессинг данных. Слоты "handler" и "handler-filter" - для обработки ошибок -  первый из них хранит функцию-обработчик, второй - список классов и объектов ошибок, к которым обработчик применяется.

А где же дендриты?
В качестве них выступаем мы сами. Ну, или другие нейроны:

(defgeneric update-neuron (neuron)
  (:method ((neuron neuron))
    (loop :with neuron-value = (neuron-value neuron)
      :for receiver :in (slot-value neuron 'receivers)
      :do (funcall receiver neuron-value neuron))))

(defun (setf neuron-value) (new-value neuron &optional sender)
  (block function
    (let ((new-value (handler-bind
                       ((condition (lambda (c) (%neuron-handler neuron c))))
                       (prog ()
                           start
                           (restart-case
                             (return (funcall (neuron-function neuron) new-value sender))
                             (retry-processing (&optional (value new-value))
                                 (setf new-value value)
                                 (go start))
                             (abort-processing (&optional return-value)
                                 (return-from function return-value)))))))
      (prog1 (setf (slot-value neuron 'value) new-value)
       (update-neuron neuron)))))

В случае ошибки, или другого сигнала, управление передается функции-обработчику-сигнала(тот самый "handler"), которая может как минимум два действия совершить - прервать конкретную ветвь потока данных(перезапуск "abort-processing"), или же попробовать вызвать функцию, обрабатывающую данные, заново(возможно, с новым значением)(перезапуск "retry-processing").

Как видно, нейрон сделан funcallable-standard-object'ом не зря - с таким подходом мы можем выдавать обработанные данные не только другим нейронам, но и любым функциям, не утруждая себя typecase'ами и прочим.

Кстати, сама функция funcallable-объекта представляет собой по сути тот же самый (setf neuron-value):

(closer-mop:set-funcallable-instance-function
  object (lambda (value &optional sender)
           (setf (neuron-value object sender) value)))

Казалось бы - ну, обычный паттерн observer, что удивительного?
На самом деле, не все так просто.

Основная фича библиотеки - метакласс dataflow-class:

(defclass dataflow-class (closer-mop:standard-class)
  ()
  (:metaclass closer-mop:standard-class))

Конечно, он не просто так вот определяется. Там, в библиотеке, много магии из метаобъектного протокола - желающие могут посмотреть.

Вобщем, суть в том, что все слоты классов, которые являются экземплярами метакласса dataflow-class - представляют собой нейроны, описанные выше.

При этом, мы можем продолжать с ними работать как с обычными слотами обычных классов - через функцию slot-value и :reader/:writer/:accessor методы. В случае изменения значения слота (т.е. через (setf slot-value) или :writer-метод) нижележащий нейрон автоматически распространит поток данных по всем слотам других(и даже этого же самого) объектов dataflow-классов, по всем нейронам, и по всем функциям, с которыми связан.

Кстати, собственно связывание нейронов/функций делается с помощью функции add-connection:

(defun add-connection (source destination &key source-name destination-name (test #'eq))
  (declare (type (or neuron dataflow-object) source)
           (type (or neuron function symbol cons dataflow-object) destination)
           (type symbol source-name destination-name)
           (type (or symbol function) test))
  (let ((source (if (typep source 'neuron)
                  source
                  (slot-neuron source source-name)))
        (destination (typecase destination
                       ((or neuron function) destination)
                       ((or symbol cons) (fdefinition destination))
                       (T (slot-neuron destination destination-name)))))
    (pushnew destination (slot-value source 'receivers) :test test)
    (values source destination)))

А разрываются связи с помощью remove-connection.

Ну, теперь пара примеров.

Первый демонстрирует то, что event-driven парадигма является частным случаем подхода, реализованного в библиотеке:

(defclass my-object (dataflow-object)
  ;; :neuron-name позволяет задавать нейрону имя, отличное от имени слота
  ((%event :initform nil :neuron-name event))
  (:metaclass dataflow-class))

(defun fire-event (instance)
  ;;update-slot просто вызывает update-neuron,
  ;; не перевычисляя значение слота
  (update-slot instance 'event))

(defparameter *my-object* (make-instance 'my-object))

(add-connection *my-object* (lambda (data &optional sender)
                              (declare (ignore data sender))
                              (write-line "An event was fired"))
                :source-name 'event)

(defun event-example ()
  (fire-event *my-object*))

;; (event-example)
;; ==> `An event was fired' на *standard-output*

Второй пример посложнее, и показывает, как приятно было бы на основе моей библиотеки реализовать gui-фреймворк:

(defclass textbox (dataflow-object)
  ((text :initform "Hello, world!" :accessor textbox-text))
  (:metaclass dataflow-class))

(defparameter *textbox* (make-instance 'textbox))

;;устанавливаем слоту `text' функцию-обработчик-данных
(setf (slot-function *textbox* 'text)
      (lambda (data &optional sender)
        (declare (ignore sender))
        ;;если пользователь, или кто еще, вдруг решил положить
        ;;в слот что-то отличное от строки, сигнализируем ошибку
        (unless (typep data 'string)
          (error "Textbox `text' slot only accepts strings"))
        data))

;;функция-обработчик-ошибок. Просто пишет на *error-output* сведения об ошибке
;; и останавливает поток данных(не тот, который thread, и даже не тот, который
;; stream, а именно тот, который `flow of data')
;;Почему-то подумалось про монаду Maybe.
(defun abort-flow-handler (neuron cond)
  (declare (ignore neuron))
  (format *error-output* "~&Something gone wrong: ~a~%" cond)
  (invoke-restart 'abort-processing))

;; устанавливаем нейрону слота обработчик ошибок
(setf (slot-handler *textbox* 'text) #'abort-flow-handler)

;; Следующий объект в цепочке у нас просто нейрон, но точно так же можно
;; было бы сделать его и слотом объекта какого-либо класса, понятное дело.
(defparameter *text-processor*
    (make-instance 'neuron
      ;; функция, обрабатывающая данные
      :function (lambda (text &optional sender)
                  (declare (ignore sender))
                  ;;считываем какой-нибудь объект
                  ;; из строки с помощью лиспового reader'а
                  (let* ((*read-eval* nil)
                         (data (read-from-string text)))
                    ;; и если он вдруг не является числом,
                    ;; кидаем исключение
                    (unless (numberp data)
                      (error "Invalid input: ~s" data))
                    ;; обработанные данные - cons-ячейка, в которой
                    ;; car - считанное число, а cdr - оно же, помноженное на 2
                    (cons data (* data 2))))
      ;; обработчик ошибок у нейрона тот же, что и у слота объекта *textbox*
      ;; кстати, handler-filter по умолчанию - error, т.е. отлавливает
      ;; только сигналы оного класса
      :handler #'abort-flow-handler))

;; связываем слот *textbox* и нейрон *text-processor*
(add-connection *textbox* *text-processor* :source-name 'text)

;; Данные, полученные от *text-processor*, у нас поставляются
;; просто-напросто в функцию, которая печатает их на *standard-output*
(add-connection *text-processor* (lambda (data &optional sender)
                                   (declare (ignore sender))
                                   (format t "~&~a + ~:*~a = ~a~%"
                                           (car data) (cdr data))))

(defun textbox-example (&optional (text "123"))
  (setf (textbox-text *textbox*) text)
  (setf (textbox-text *textbox*) 123)
  (setf (textbox-text *textbox*) "abc"))

;; (textbox-example)
;; ==> 123 + 123 = 246
;;     Something gone wrong: Textbox `text' slot only accepts strings
;;     Something gone wrong: Invalid input: ABC

Вобщем, библиотека под MIT-лицензией, так что можно, ничего не боясь, тырить оттуда код в коммерческие closed-source проекты - главное не забывать говорить мне спасибо.