frereth.cp.message — com.frereth/cp 0.0.3-g3b13aae

Translation of curvecpmessage.c

This is really a generic buffer program

The "parent" child/server reads/writes to pipes that this provides, in a specific (and apparently undocumented) communications protocol.

This, in turn, reads/writes data from/to a child that it spawns.

I keep wanting to think of this as a simple transducer and just skip the buffering pieces, but they (and the flow control) are really the main point.

action-trigger! clj

(action-trigger!
  {:keys [:frereth.cp.message/actual-next :frereth.cp.message/delta_f
          :frereth.cp.message/scheduling-time]
   :as timing-details}
  {:keys [:frereth.cp.message.specs/message-loop-name] :as io-handle}
  {:keys [:frereth.cp.message.specs/outgoing] :as state}
  log-state-atom
  next-action)

Some action triggered the main ioloop

build-un-ackd-blocks clj

(build-un-ackd-blocks {:keys [:frereth.weald.specs/logger]
                       log-state :frereth.weald.specs/state})

child->! clj

(child->! {:keys [:frereth.weald.specs/logger
                  :frereth.cp.message.specs/child-out
                  :frereth.cp.message.specs/from-child
                  :frereth.cp.message.specs/message-loop-name
                  :frereth.cp.message.specs/pipe-from-child-size]
           log-state-atom :frereth.weald.specs/state-atom
           :as io-handle}
          array-o-bytes)

Send bytes from a child buffer...if we have room

child-buffer-timeout clj

How long might we block child->

child-close! clj

(child-close! {:keys [:frereth.cp.message.specs/from-child] :as io-handle})

Notify parent that child is done sending

choose-next-scheduled-time clj

(choose-next-scheduled-time {{:keys [:frereth.cp.message.specs/n-sec-per-block
                                     :frereth.cp.message.specs/rtt-timeout]
                              :as flow-control}
                               :frereth.cp.message.specs/flow-control
                             {:keys [:frereth.cp.message.specs/->child-buffer
                                     :frereth.cp.message.specs/gap-buffer]}
                               :frereth.cp.message.specs/incoming
                             {:keys [:frereth.cp.message.specs/earliest-time
                                     :frereth.cp.message.specs/last-block-time
                                     :frereth.cp.message.specs/send-eof
                                     :frereth.cp.message.specs/un-sent-blocks
                                     :frereth.cp.message.specs/un-ackd-blocks
                                     :frereth.cp.message.specs/want-ping]
                              :as outgoing}
                               :frereth.cp.message.specs/outgoing
                             :keys [:frereth.cp.message.specs/message-loop-name
                                    :frereth.cp.message.specs/recent]
                             log-state :frereth.weald.specs/state
                             :as state}
                            to-child-done?)

clear-interrupt! clj

(clear-interrupt!)

condensed-choose-next-scheduled-time clj

(condensed-choose-next-scheduled-time
  {{:keys [:frereth.cp.message.specs/n-sec-per-block
           :frereth.cp.message.specs/rtt-timeout]}
     :frereth.cp.message.specs/flow-control
   {:keys [:frereth.cp.message.specs/->child-buffer
           :frereth.cp.message.specs/gap-buffer]}
     :frereth.cp.message.specs/incoming
   {:keys [:frereth.cp.message.specs/ackd-addr
           :frereth.cp.message.specs/earliest-time
           :frereth.cp.message.specs/last-block-time
           :frereth.cp.message.specs/send-eof
           :frereth.cp.message.specs/un-sent-blocks
           :frereth.cp.message.specs/un-ackd-blocks
           :frereth.cp.message.specs/want-ping]
    :as outgoing}
     :frereth.cp.message.specs/outgoing
   :keys [:frereth.cp.message.specs/message-loop-name
          :frereth.cp.message.specs/recent]
   log-state :frereth.weald.specs/state
   :as state}
  to-child-done?)

do-start clj

(do-start {:keys [:frereth.cp.message.specs/message-loop-name]
           {:keys [:frereth.cp.message.specs/pipe-from-child-size]
            :or {pipe-from-child-size K/k-64}
            :as outgoing}
             :frereth.cp.message.specs/outgoing
           {:keys [:frereth.cp.message.specs/pipe-to-child-size]
            :as incoming
            :or {pipe-to-child-size K/k-64}}
             :frereth.cp.message.specs/incoming
           :as state}
          logger
          parent-cb
          child-cb)

Trigger side-effects to start a ::state

fast-spins clj

get-state clj

(get-state stream-holder)

(get-state {:keys [:frereth.weald.specs/logger
                   :frereth.cp.message.specs/message-loop-name
                   :frereth.cp.message.specs/stream]
            log-state-atom :frereth.weald.specs/state-atom}
           timeout
           failure-signal)

Synchronous equivalent to a deref

halt! clj

(halt! {:keys [:frereth.weald.specs/logger
               :frereth.cp.message.specs/message-loop-name
               :frereth.cp.message.specs/stream
               :frereth.cp.message.specs/from-child
               :frereth.cp.message.specs/child-out]
        log-state-atom :frereth.weald.specs/state-atom
        :as io-handle})

This is supposed to be a hard-stop that means we're totally done.

initial-state clj

(initial-state human-name logger)

(initial-state human-name
               server?
               {{:keys [:frereth.cp.message.specs/pipe-to-child-size]
                 :or {pipe-to-child-size K/k-64}
                 :as incoming}
                  :frereth.cp.message.specs/incoming
                {:keys [:frereth.cp.message.specs/pipe-from-child-size]
                 :or {pipe-from-child-size K/k-64}
                 :as outgoing}
                  :frereth.cp.message.specs/outgoing
                log-state :frereth.weald.specs/state
                :as opts}
               logger)

Put together an initial state that's ready to start!

interrupt! clj

(interrupt!)

interrupted? clj

(interrupted?)

max-child-buffer-size clj

Maximum message blocks from parent to child that we'll buffer before dropping

must be power of 2 -- DJB

parent->! clj

(parent->! io-handle array-o-bytes)

(parent->! {:keys [:frereth.weald.specs/logger
                   :frereth.cp.message.specs/message-loop-name
                   :frereth.cp.message.specs/stream]
            log-state-atom :frereth.weald.specs/state-atom
            :as io-handle}
           array-o-bytes
           timeout)

Receive a byte array from parent

411-435: try receiving messages: (DJB)

The parent is going to call this. It should trigger the pieces that naturally fall downstream and lead to writing the bytes to the child.

It's replacing one of the polling triggers that set off the main() event loop. Need to account for that fundamental strategic change

pick-next-action clj

(pick-next-action this
                  {:keys [:frereth.cp.message/now
                          :frereth.cp.message.specs/recent
                          :frereth.cp.message.specs/stream]
                   actual-next :frereth.cp.message/next-action-time
                   log-state :frereth.weald.specs/state})

schedule-next-timeout! clj

(schedule-next-timeout!
  {:keys [:frereth.weald.specs/logger :frereth.cp.message.specs/->parent
          :frereth.cp.message.specs/child-output-loop
          :frereth.cp.message.specs/child-input-loop
          :frereth.cp.message.specs/to-child
          :frereth.cp.message.specs/to-child-done?
          :frereth.cp.message.specs/message-loop-name
          :frereth.cp.message.specs/stream]
   :as io-handle}
  {:keys [:frereth.cp.message.specs/recent]
   {:keys [:frereth.cp.message.specs/next-action] :as flow-control}
     :frereth.cp.message.specs/flow-control
   {:keys [:frereth.cp.message.specs/receive-eof
           :frereth.cp.message.specs/receive-total-bytes
           :frereth.cp.message.specs/receive-written]}
     :frereth.cp.message.specs/incoming
   {:keys [:frereth.cp.message.specs/send-eof-acked] :as outgoing}
     :frereth.cp.message.specs/outgoing
   log-state :frereth.weald.specs/state
   :as state})

start-event-loops! clj

(start-event-loops! {:keys [:frereth.weald.specs/logger
                            :frereth.cp.message.specs/->child
                            :frereth.cp.message.specs/message-loop-name]
                     :as io-handle}
                    {log-state :frereth.weald.specs/state :as state})

swap-parent-callback! clj

(swap-parent-callback! io-handle new-callback)

(swap-parent-callback! {:keys [:frereth.weald.specs/logger
                               :frereth.cp.message.specs/message-loop-name
                               :frereth.cp.message.specs/stream]
                        log-state-atom :frereth.weald.specs/state-atom
                        :as io-loop}
                       time-out
                       timed-out-value
                       new-callback)

Swap out the ->parent callback

For switching the Client from Initiate to Message packets

trigger-from-child clj

(trigger-from-child io-handle
                    callback
                    accepted?
                    {{:keys [:frereth.cp.message.specs/strm-hwm] :as outgoing}
                       :frereth.cp.message.specs/outgoing
                     :keys [:frereth.cp.message.specs/message-loop-name]
                     :as state})

trigger-from-parent clj

(trigger-from-parent {:keys [:frereth.weald.specs/logger
                             :frereth.cp.message.specs/message-loop-name]
                      :as io-handle}
                     message
                     {{:keys [:frereth.cp.message.specs/->child-buffer]}
                        :frereth.cp.message.specs/incoming
                      {:keys
                         [:frereth.cp.message.specs/client-waiting-on-response]}
                        :frereth.cp.message.specs/flow-control
                      log-state :frereth.weald.specs/state
                      :as state})

Message block arrived from parent. Work triggered by ioloop

trigger-from-timer clj

(trigger-from-timer io-handle
                    {:keys [:frereth.cp.message.specs/message-loop-name]
                     :as state})

trigger-output clj

(trigger-output {:keys [:frereth.cp.message.specs/to-child] :as io-handle}
                {{:keys [:frereth.cp.message.specs/next-action]}
                   :frereth.cp.message.specs/flow-control
                 :keys [:frereth.cp.message.specs/message-loop-name]
                 :as state})

write-from-child-timeout clj

milliseconds before we give up on writing a packet from child to parent

write-from-parent-timeout clj

milliseconds before we give up on writing a packet from parent to child