Dynamic Input Channels

Suppose a function produces some output on a channel, and that the output of this function is to be conditionally consumed dependant on the current system state. Also, lets suppose, the system contains a restriction in that the output channel of the producer function must remain fixed.

The question is then, how to route the output to the correct consuming logic, without reassigning the output channel.

One solution is to dynamically assign the consumer's input channel. Go provides a mechanism for this; the channel itself.

s := make(chan chan int)
c := make(chan int)
for {
    select {
    case c = <- s:
    case v := <- c:
        logic(v)                                                                                                  
    }
}  

        

With this logic in place the consumer's input channel can be dynamically reassigned whilst the producer remains agnostic. 

Here is a complete example of the mechanism at work.  

Bullhorn

Bullhorn is a golang package that provides lightweight type-agnostic publish / subscribe messaging to goroutines.

You can find it here.

The model has been extracted from various time sensitive messaging applications I have worked with over the past few years.

Here is an example limit order / price matching procedure written using bullhorn subscriptions and events.

Running the code, will give you something like...


Got order:{CBG 0 1.05 100}
Got order:{CBG 1 1.05 200}
Matching:{CBG 0 1.05 100} against:CBG:  1655 0.8704 1.4305  2797
Matching:{CBG 1 1.05 200} against:CBG:  1655 0.8704 1.4305  2797
Matching:{CBG 0 1.05 100} against:CBG:  9414 0.8847 1.9453  9074
Matching:{CBG 1 1.05 200} against:CBG:  9414 0.8847 1.9453  9074
Matching:{CBG 0 1.05 100} against:CBG:  8787 0.9490 1.0373 12530 - Execute Order!
Matching:{CBG 1 1.05 200} against:CBG:  8787 0.9490 1.0373 12530
Matching:{CBG 1 1.05 200} against:CBG:  5840 0.9519 1.5722 12054
Matching:{CBG 1 1.05 200} against:CBG:  2485 0.9425 1.7220 11042
Matching:{CBG 1 1.05 200} against:CBG:  5235 1.0583 1.5840  4333 - Execute Order!

Waiting for Golang channels to drain

Golang's channels easily map onto the producer consumer pattern. Lets assume that everything that is produced needs to be consumed, even if the process receives a SIGTERM.

The following example shows how we can register channels, monitor for a kill signal, and then wait for everything to be consumed.



package main

import (
"log"
"os"
"os/signal"
"reflect"
"syscall"
"time"
)

var (
BufferSize = 512
MaxIter    = 10
monitored  []interface{}
c          = make(chan int, BufferSize)
stopping   bool
)

func RegisterChannel(i interface{}) {
monitored = append(monitored, i)
}

func MonitorSigTerm() chan bool {
s := make(chan os.Signal, 1)
b := make(chan bool)
signal.Notify(s, syscall.SIGTERM)

go func(c chan os.Signal, b chan bool) {
_ = <-c
log.Println("Cleaning up")
// tell the caller
b <- true
for _, i := range monitored {
ch := reflect.ValueOf(i)
if ch.Kind() != reflect.Chan {
continue
}
prev := 0
iteration := 0
for {
if ch.Len() == 0 {
break
}

if prev == ch.Len() {
iteration++
// enough?
if iteration >= MaxIter {
log.Println("Dropping")
break
}
} else {
iteration = 0
}

prev = ch.Len()
log.Printf("Draining:%v\n", prev)
// other goroutines are working, let them
time.Sleep(1e9)
}
}
os.Exit(1)
}(s, b)
return b
}

func main() {
RegisterChannel(c)
stop := MonitorSigTerm()

go func() {
i := 0
for {
if stopping {
break
}
i++
c <- i
time.Sleep(1e9)
}
}()

go func() {
for {
i := <-c
log.Printf("rx:%v\n", i)
// slower read
time.Sleep(2e9)
}
}()

stopping = <-stop

// wait for cleanup to finish
select {}

}