1 // Package msgport adapts a message channel to the duplex that chaindata speaks.
2 //
3 // The browser has no shared memory across a worker boundary, only messages: a MessagePort, a
4 // worker's postMessage, or a nested channel. What crosses is an ArrayBuffer that can be
5 // transferred rather than copied, which is why the frames were made binary in the first place.
6 //
7 // This package is the whole adapter, and it is small on purpose. One direction pumps frames
8 // from the port into the duplex's send side, the other pumps answers from the duplex's
9 // receive side back out. Everything above it — the request ids, the cancel, the close — is
10 // chaindata's, unchanged, and everything below it is whatever the platform calls a message
11 // channel. Neither side knows the other exists.
12 package msgport
13 14 import (
15 "errors"
16 "sync"
17 18 "git.smesh.lol/tiny-spawn"
19 )
20 21 // ErrClosed is a port that is gone: closed locally, or its far end died.
22 var ErrClosed = errors.New("msgport: the port is closed")
23 24 // Buffer is how many frames may wait in one direction.
25 //
26 // The protocol is request/answer with one request in flight per chain in practice, so this is
27 // not a queue depth so much as a way to keep a burst of answers from blocking the JS event
28 // loop. A port whose buffer fills is a peer that has stopped reading, and the adapter treats
29 // that as death rather than as backpressure it cannot apply.
30 const Buffer = 64
31 32 // Port is one end of a message channel that carries whole frames.
33 //
34 // Receive returns a channel that is closed when the port is gone, which is how a worker that
35 // was terminated without a word reaches the protocol as a close instead of as silence.
36 type Port interface {
37 // Send delivers one frame, transferring its buffer if the platform allows it.
38 Send(frame []byte) error
39 // Receive is the incoming frames, closed when the port is gone.
40 Receive() <-chan []byte
41 // Close ends the port in both directions.
42 Close() error
43 }
44 45 // Duplex adapts a port to the duplex chaindata speaks.
46 //
47 // The returned value is the caller's end: writing to Tx sends a frame out, reading from Rx is
48 // the far side's answer. Closing Tx, a send failure, or the port closing all end the duplex
49 // the same way — the receive side is closed, so a reader sees a channel that is done rather
50 // than a channel that is quiet.
51 func Duplex(port Port) tinyspawn.Duplex[[]byte] {
52 out := make(chan []byte, Buffer)
53 in := make(chan []byte, Buffer)
54 55 var (
56 mu sync.Mutex
57 dead bool
58 )
59 60 // stop closes the receive side exactly once. It runs under the same lock a delivery
61 // takes, so a frame can never be sent to a closed channel: deliver either wins the
62 // lock and hands the frame over, or finds the port dead and drops it.
63 stop := func() {
64 mu.Lock()
65 defer mu.Unlock()
66 67 if dead {
68 return
69 }
70 71 dead = true
72 73 close(in)
74 }
75 76 deliver := func(frame []byte) bool {
77 mu.Lock()
78 defer mu.Unlock()
79 80 if dead {
81 return false
82 }
83 84 select {
85 case in <- frame:
86 return true
87 default:
88 // The reader is not keeping up. A protocol with one request in flight cannot
89 // catch up later, so the honest answer is that this channel is dead.
90 return false
91 }
92 }
93 94 // Outgoing: the caller's writes, sent to the platform.
95 go func() {
96 for frame := range out {
97 mu.Lock()
98 gone := dead
99 mu.Unlock()
100 101 if gone {
102 return
103 }
104 105 if err := port.Send(frame); err != nil {
106 stop()
107 108 return
109 }
110 }
111 112 // The caller closed Tx: it has stopped sending, which is how a terminated worker
113 // looks from the other end.
114 stop()
115 }()
116 117 // Incoming: the platform's frames, handed to the caller.
118 go func() {
119 for frame := range port.Receive() {
120 if !deliver(frame) {
121 stop()
122 123 return
124 }
125 }
126 127 stop()
128 }()
129 130 return tinyspawn.Duplex[[]byte]{Tx: out, Rx: in}
131 }
132