Blocking in a connector algorithms

[emil14]

Current implementation of the connector algorithm has some downsides that leads to unnecessary blocking. Here's how it works

For every N connections, where connection is a one sender to many receivers relation, spawn N goroutines. Each goroutine reads an message from sender and distributes across M receivers. It does so in a cyclic manner, trying to avoid unnecessary blocking - if one receiver is busy it immediately goes to the next one and do so in a loop until every receiver got the message. However, if sender is fast enough to deliver a new message before all receivers got the previous one, then we have to wait. In other words - the speed of distribution of a message across M receivers is the speed of the slowest receiver.

It's possible in theory for "fast" receivers not to wait for the slower and continue to receive messages. Here's the example:

Let s1 sender sends message at the speed of 1 msg/second and let there be 2 receivers r1 and r2 - r1 receiver is as fast as s1 sender and is able to receive 1 msg/second but r2 receiver is slow and only can receive 0.5 msg/second (it needs 2 seconds to process a message).

• s1 sends a m1 first message to r1 and r2
• one seconds have passed
• s1 sends second m2 message
• at the same time r1 finishes processing of m1 and could immediately receive m1 but instead waits for r2 that needs one more second
• another second have passed
• s1 sends third message m3
• after two seconds r2 finishes processing of m1
• r1 and r2 immediately receives m2 message but there's already m3 waiting
• after another second situation happens again (and will happen over and over until program terminate)

However, keep in mind that order of messages must not be broken. Any optimization of the algorithm must keep the correctness of the current implementation.

---

Related to #250, #90, #243

[emil14]

problem with const

func (c Connector) broadcast(ctx context.Context, conn Connection) error {
	buf := make(chan Msg, 100)
	defer close(buf)
	go func() {
		for msg := range buf {
			c.distribute(ctx, msg, conn.Meta, conn.Receivers)
		}
	}()

	receiverSet := make(map[PortAddr]struct{}, len(conn.Meta.ReceiverPortAddrs))
	for _, receiverPortAddr := range conn.Meta.ReceiverPortAddrs {
		receiverSet[receiverPortAddr] = struct{}{}
	}

	for {
		select {
		case <-ctx.Done():
			return nil
		case msg := <-conn.Sender:
			msg = c.listener.Send(Event{
				Type: MessageSentEvent,
				MessageSent: &EventMessageSent{
					SenderPortAddr:    conn.Meta.SenderPortAddr,
					ReceiverPortAddrs: receiverSet,
				},
			}, msg)
			// instead of distributing msg directly, we use buffer, so sender can write faster than receivers read
			buf <- msg
		}
	}
}