Compare commits
No commits in common. "d60beccbcd53d86ee9373d35472be8a3e2155e1a" and "7a0bf64c17a4dd86e8ad518addda487805b83fe7" have entirely different histories.
d60beccbcd
...
7a0bf64c17
89
metadapta.go
89
metadapta.go
@ -4,13 +4,9 @@ import "io"
|
||||
import "fmt"
|
||||
import "net"
|
||||
import "sync"
|
||||
import "sync/atomic"
|
||||
import "git.tebibyte.media/sashakoshka/hopp/tape"
|
||||
import "git.tebibyte.media/sashakoshka/go-util/sync"
|
||||
|
||||
// TODO investigate why 30 never reaches the server, causing it to wait for ever
|
||||
// and never close the connection, causing the client to also wait forever
|
||||
|
||||
const closeMethod = 0xFFFF
|
||||
const int64Max = int64((^uint64(0)) >> 1)
|
||||
const defaultChunkSize = 0x1000
|
||||
@ -21,14 +17,6 @@ type Party bool; const (
|
||||
ClientSide Party = true
|
||||
)
|
||||
|
||||
func (party Party) String() string {
|
||||
if party == ServerSide {
|
||||
return "server"
|
||||
} else {
|
||||
return "client"
|
||||
}
|
||||
}
|
||||
|
||||
type a struct {
|
||||
sizeLimit int64
|
||||
underlying net.Conn
|
||||
@ -64,7 +52,7 @@ func AdaptA(underlying net.Conn, party Party) Conn {
|
||||
|
||||
func (this *a) Close() error {
|
||||
close(this.done)
|
||||
return nil
|
||||
return this.underlying.Close()
|
||||
}
|
||||
|
||||
func (this *a) LocalAddr() net.Addr {
|
||||
@ -78,34 +66,27 @@ func (this *a) RemoteAddr() net.Addr {
|
||||
func (this *a) OpenTrans() (Trans, error) {
|
||||
this.transLock.Lock()
|
||||
defer this.transLock.Unlock()
|
||||
if this.transID == int64Max {
|
||||
return nil, fmt.Errorf("could not open transaction: %w", ErrIntegerOverflow)
|
||||
}
|
||||
id := this.transID
|
||||
this.transID ++
|
||||
trans := &transA {
|
||||
parent: this,
|
||||
id: id,
|
||||
incoming: usync.NewGate[incomingMessage](),
|
||||
}
|
||||
this.transMap[id] = trans
|
||||
if this.party == ClientSide {
|
||||
this.transID ++
|
||||
} else {
|
||||
this.transID --
|
||||
if this.transID == int64Max {
|
||||
return nil, fmt.Errorf("could not open transaction: %w", ErrIntegerOverflow)
|
||||
}
|
||||
this.transID ++
|
||||
return trans, nil
|
||||
}
|
||||
|
||||
func (this *a) AcceptTrans() (Trans, error) {
|
||||
eof := fmt.Errorf("could not accept transaction: %w", io.EOF)
|
||||
select {
|
||||
case trans := <- this.transChan:
|
||||
if trans == nil {
|
||||
return nil, eof
|
||||
}
|
||||
return trans, nil
|
||||
case <- this.done:
|
||||
return nil, eof
|
||||
return nil, fmt.Errorf("could not accept transaction: %w", io.EOF)
|
||||
}
|
||||
}
|
||||
|
||||
@ -135,11 +116,7 @@ func (this *a) receive() {
|
||||
trans.closeDontUnlist()
|
||||
}
|
||||
clear(this.transMap)
|
||||
this.underlying.Close()
|
||||
}()
|
||||
|
||||
// receive MMBs in a loop and forward them to transactions until shit
|
||||
// starts closing
|
||||
for {
|
||||
transID, method, chunked, payload, err := decodeMessageA(this.underlying, this.sizeLimit)
|
||||
if err != nil {
|
||||
@ -147,7 +124,7 @@ func (this *a) receive() {
|
||||
return
|
||||
}
|
||||
|
||||
err = this.multiplexMMB(transID, method, chunked, payload)
|
||||
err = this.receiveMultiplex(transID, method, chunked, payload)
|
||||
if err != nil {
|
||||
this.err = fmt.Errorf("could not receive message: %w", err)
|
||||
return
|
||||
@ -155,7 +132,7 @@ func (this *a) receive() {
|
||||
}
|
||||
}
|
||||
|
||||
func (this *a) multiplexMMB(transID int64, method uint16, chunked bool, payload []byte) error {
|
||||
func (this *a) receiveMultiplex(transID int64, method uint16, chunked bool, payload []byte) error {
|
||||
if transID == 0 { return ErrMessageMalformed }
|
||||
|
||||
trans, err := func() (*transA, error) {
|
||||
@ -164,12 +141,6 @@ func (this *a) multiplexMMB(transID int64, method uint16, chunked bool, payload
|
||||
|
||||
trans, ok := this.transMap[transID]
|
||||
if !ok {
|
||||
// check if this is a superfluous close message and just
|
||||
// do nothing if so
|
||||
if method == closeMethod {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// it is forbidden for the other party to initiate a transaction
|
||||
// with an ID from this party
|
||||
if this.party == partyFromTransID(transID) {
|
||||
@ -187,24 +158,14 @@ func (this *a) multiplexMMB(transID int64, method uint16, chunked bool, payload
|
||||
}()
|
||||
if err != nil { return err }
|
||||
|
||||
if trans == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if method == closeMethod {
|
||||
return trans.Close()
|
||||
} else {
|
||||
trans.incoming.Send(incomingMessage {
|
||||
method: method,
|
||||
chunked: chunked,
|
||||
payload: payload,
|
||||
})
|
||||
}
|
||||
trans.incoming.Send(incomingMessage {
|
||||
method: method,
|
||||
chunked: chunked,
|
||||
payload: payload,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// most methods in transA don't need to be goroutine safe except those marked
|
||||
// as such
|
||||
type transA struct {
|
||||
parent *a
|
||||
id int64
|
||||
@ -212,24 +173,17 @@ type transA struct {
|
||||
currentReader io.Reader
|
||||
currentWriter io.Closer
|
||||
writeBuffer []byte
|
||||
closed atomic.Bool
|
||||
}
|
||||
|
||||
func (this *transA) Close() error {
|
||||
// MUST be goroutine safe
|
||||
err := this.closeDontUnlist()
|
||||
this.parent.unlistTransactionSafe(this.ID())
|
||||
return err
|
||||
}
|
||||
|
||||
func (this *transA) closeDontUnlist() (err error) {
|
||||
// MUST be goroutine safe
|
||||
this.incoming.Close()
|
||||
if !this.closed.Load() {
|
||||
err = this.Send(closeMethod, nil)
|
||||
}
|
||||
this.closed.Store(true)
|
||||
return err
|
||||
func (this *transA) closeDontUnlist() error {
|
||||
this.Send(closeMethod, nil)
|
||||
return this.incoming.Close()
|
||||
}
|
||||
|
||||
func (this *transA) ID() int64 {
|
||||
@ -270,11 +224,6 @@ func (this *transA) Receive() (method uint16, data []byte, err error) {
|
||||
}
|
||||
|
||||
func (this *transA) ReceiveReader() (uint16, io.Reader, error) {
|
||||
// if the transaction has been closed, return an io.EOF
|
||||
if this.closed.Load() {
|
||||
return 0, nil, io.EOF
|
||||
}
|
||||
|
||||
// drain previous reader if necessary
|
||||
if this.currentReader != nil {
|
||||
io.Copy(io.Discard, this.currentReader)
|
||||
@ -296,14 +245,13 @@ type readerA struct {
|
||||
eof bool
|
||||
}
|
||||
|
||||
// pull pulls the next MMB in this message from the transaction.
|
||||
func (this *readerA) pull() (uint16, error) {
|
||||
// if the previous message ended the chain, return an io.EOF
|
||||
if this.eof {
|
||||
return 0, io.EOF
|
||||
}
|
||||
|
||||
// get an MMB from the transaction we are a part of
|
||||
// get a message from the transaction we are a part of
|
||||
receive := this.parent.incoming.Receive()
|
||||
if receive != nil {
|
||||
if message, ok := <- receive; ok {
|
||||
@ -313,9 +261,6 @@ func (this *readerA) pull() (uint16, error) {
|
||||
this.eof = true
|
||||
}
|
||||
return message.method, nil
|
||||
} else {
|
||||
// signal parent transaction of closure
|
||||
this.parent.closed.Store(true)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -25,36 +25,20 @@ func TestConnA(test *testing.T) {
|
||||
"When the impostor is sus!",
|
||||
}
|
||||
|
||||
clientFunc := func(a Conn) {
|
||||
test.Log("CLIENT accepting transaction")
|
||||
trans, err := a.AcceptTrans()
|
||||
if err != nil { test.Fatal("CLIENT", err) }
|
||||
test.Log("CLIENT accepted transaction")
|
||||
test.Cleanup(func() { trans.Close() })
|
||||
for method, payload := range payloads {
|
||||
test.Log("CLIENT waiting...")
|
||||
gotMethod, gotPayloadBytes, err := trans.Receive()
|
||||
if err != nil { test.Fatal("CLIENT", err) }
|
||||
gotPayload := string(gotPayloadBytes)
|
||||
test.Log("CLIENT m:", gotMethod, "p:", gotPayload)
|
||||
if int(gotMethod) != method {
|
||||
test.Errorf("CLIENT method not equal")
|
||||
}
|
||||
if gotPayload != payload {
|
||||
test.Errorf("CLIENT payload not equal")
|
||||
}
|
||||
}
|
||||
test.Log("CLIENT waiting for transaction close...")
|
||||
gotMethod, gotPayload, err := trans.Receive()
|
||||
if !errors.Is(err, io.EOF) {
|
||||
test.Error("CLIENT wrong error:", err)
|
||||
test.Error("CLIENT method:", gotMethod)
|
||||
test.Error("CLIENT payload:", gotPayload)
|
||||
test.Fatal("CLIENT ok byeeeeeeeeeeeee")
|
||||
}
|
||||
}
|
||||
network := "tcp"
|
||||
addr := "localhost:7959"
|
||||
|
||||
serverFunc := func(a Conn) {
|
||||
// server
|
||||
listener, err := net.Listen(network, addr)
|
||||
if err != nil { test.Fatal(err) }
|
||||
test.Cleanup(func() { listener.Close() })
|
||||
go func() {
|
||||
test.Log("SERVER listening")
|
||||
conn, err := listener.Accept()
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
defer conn.Close()
|
||||
test.Cleanup(func() { conn.Close() })
|
||||
a := AdaptA(conn, ServerSide)
|
||||
trans, err := a.OpenTrans()
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
test.Cleanup(func() { trans.Close() })
|
||||
@ -64,82 +48,40 @@ func TestConnA(test *testing.T) {
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
}
|
||||
test.Log("SERVER closing connection")
|
||||
}()
|
||||
|
||||
// client
|
||||
test.Log("CLIENT dialing")
|
||||
conn, err := net.Dial(network, addr)
|
||||
if err != nil { test.Fatal("CLIENT", err) }
|
||||
test.Log("CLIENT dialed")
|
||||
a := AdaptA(conn, ClientSide)
|
||||
test.Cleanup(func() { a.Close() })
|
||||
test.Log("CLIENT accepting transaction")
|
||||
trans, err := a.AcceptTrans()
|
||||
if err != nil { test.Fatal("CLIENT", err) }
|
||||
test.Log("CLIENT accepted transaction")
|
||||
test.Cleanup(func() { trans.Close() })
|
||||
for method, payload := range payloads {
|
||||
test.Log("CLIENT waiting...")
|
||||
gotMethod, gotPayloadBytes, err := trans.Receive()
|
||||
if err != nil { test.Fatal("CLIENT", err) }
|
||||
gotPayload := string(gotPayloadBytes)
|
||||
test.Log("CLIENT m:", gotMethod, "p:", gotPayload)
|
||||
if int(gotMethod) != method {
|
||||
test.Errorf("CLIENT method not equal")
|
||||
}
|
||||
if gotPayload != payload {
|
||||
test.Errorf("CLIENT payload not equal")
|
||||
}
|
||||
}
|
||||
|
||||
clientServerEnvironment(test, clientFunc, serverFunc)
|
||||
}
|
||||
|
||||
func TestTransOpenCloseA(test *testing.T) {
|
||||
// currently:
|
||||
//
|
||||
// | data sent | data recvd | close sent | close recvd
|
||||
// 10 | X | X | X | server hangs
|
||||
// 20 | X | X | X | client hangs
|
||||
// 30 | X | | X |
|
||||
//
|
||||
// when a close message is recvd, it tries to push to the trans and
|
||||
// hangs on trans.incoming.Send, which hangs on sending the value to the
|
||||
// underlying channel. why is this?
|
||||
//
|
||||
// check if we are really getting values from the channel when pulling
|
||||
// from the trans channel when we are expecting a close.
|
||||
|
||||
clientFunc := func(conn Conn) {
|
||||
// 10
|
||||
trans, err := conn.OpenTrans()
|
||||
if err != nil { test.Error("CLIENT", err); return }
|
||||
test.Log("CLIENT sending 10")
|
||||
trans.Send(10, []byte("hi"))
|
||||
trans.Close()
|
||||
|
||||
// 20
|
||||
test.Log("CLIENT awaiting 20")
|
||||
trans, err = conn.AcceptTrans()
|
||||
if err != nil { test.Error("CLIENT", err); return }
|
||||
test.Cleanup(func() { trans.Close() })
|
||||
gotMethod, gotPayload, err := trans.Receive()
|
||||
if err != nil { test.Error("CLIENT", err); return }
|
||||
test.Logf("CLIENT m: %d p: %s", gotMethod, gotPayload)
|
||||
if gotMethod != 20 { test.Error("CLIENT wrong method")}
|
||||
|
||||
// 30
|
||||
trans, err = conn.OpenTrans()
|
||||
if err != nil { test.Error("CLIENT", err); return }
|
||||
test.Log("CLIENT sending 30")
|
||||
trans.Send(30, []byte("good"))
|
||||
trans.Close()
|
||||
test.Log("CLIENT waiting for connection close...")
|
||||
_, _, err = trans.Receive()
|
||||
if !errors.Is(err, io.EOF) {
|
||||
test.Fatal("CLIENT wrong error:", err)
|
||||
}
|
||||
|
||||
serverFunc := func(conn Conn) {
|
||||
// 10
|
||||
test.Log("SERVER awaiting 10")
|
||||
trans, err := conn.AcceptTrans()
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
test.Cleanup(func() { trans.Close() })
|
||||
gotMethod, gotPayload, err := trans.Receive()
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
test.Logf("SERVER m: %d p: %s", gotMethod, gotPayload)
|
||||
if gotMethod != 10 { test.Error("SERVER wrong method")}
|
||||
|
||||
// 20
|
||||
trans, err = conn.OpenTrans()
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
test.Log("SERVER sending 20")
|
||||
trans.Send(20, []byte("hi how r u"))
|
||||
trans.Close()
|
||||
|
||||
// 30
|
||||
test.Log("SERVER awaiting 30")
|
||||
trans, err = conn.AcceptTrans()
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
test.Cleanup(func() { trans.Close() })
|
||||
gotMethod, gotPayload, err = trans.Receive()
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
test.Logf("SERVER m: %d p: %s", gotMethod, gotPayload)
|
||||
if gotMethod != 30 { test.Error("SERVER wrong method")}
|
||||
}
|
||||
|
||||
clientServerEnvironment(test, clientFunc, serverFunc)
|
||||
test.Log("CLIENT done")
|
||||
conn.Close()
|
||||
}
|
||||
|
||||
func TestEncodeMessageA(test *testing.T) {
|
||||
@ -202,66 +144,3 @@ func TestDecodeMessageAErr(test *testing.T) {
|
||||
test.Fatalf("wrong error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEncodeDecodeMessageA(test *testing.T) {
|
||||
correctTransID := int64(2)
|
||||
correctMethod := uint16(30)
|
||||
correctPayload := []byte("good")
|
||||
buffer := bytes.Buffer { }
|
||||
err := encodeMessageA(&buffer, defaultSizeLimit, correctTransID, correctMethod, correctPayload)
|
||||
if err != nil { test.Fatal(err) }
|
||||
transID, method, chunked, payload, err := decodeMessageA(&buffer, defaultSizeLimit)
|
||||
if got, correct := transID, int64(2); got != correct {
|
||||
test.Fatalf("not equal: %v %v", got, correct)
|
||||
}
|
||||
if got, correct := method, uint16(30); got != correct {
|
||||
test.Fatalf("not equal: %v %v", got, correct)
|
||||
}
|
||||
if chunked {
|
||||
test.Fatalf("message should not be chunked")
|
||||
}
|
||||
if got, correct := payload, correctPayload; !slices.Equal(got, correct) {
|
||||
test.Fatalf("not equal: %v %v", got, correct)
|
||||
}
|
||||
}
|
||||
|
||||
func clientServerEnvironment(test *testing.T, clientFunc func(conn Conn), serverFunc func(conn Conn)) {
|
||||
network := "tcp"
|
||||
addr := "localhost:7959"
|
||||
|
||||
// server
|
||||
listener, err := net.Listen(network, addr)
|
||||
if err != nil { test.Fatal(err) }
|
||||
test.Cleanup(func() { listener.Close() })
|
||||
go func() {
|
||||
test.Log("SERVER listening")
|
||||
conn, err := listener.Accept()
|
||||
if err != nil { test.Error("SERVER", err); return }
|
||||
defer conn.Close()
|
||||
test.Cleanup(func() { conn.Close() })
|
||||
a := AdaptA(conn, ServerSide)
|
||||
test.Cleanup(func() { a.Close() })
|
||||
|
||||
serverFunc(a)
|
||||
test.Log("SERVER closing")
|
||||
}()
|
||||
|
||||
// client
|
||||
test.Log("CLIENT dialing")
|
||||
conn, err := net.Dial(network, addr)
|
||||
if err != nil { test.Fatal("CLIENT", err) }
|
||||
test.Log("CLIENT dialed")
|
||||
a := AdaptA(conn, ClientSide)
|
||||
test.Cleanup(func() { a.Close() })
|
||||
|
||||
clientFunc(a)
|
||||
|
||||
test.Log("CLIENT waiting for connection close...")
|
||||
trans, err := a.AcceptTrans()
|
||||
if !errors.Is(err, io.EOF) {
|
||||
test.Error("CLIENT wrong error:", err)
|
||||
test.Fatal("CLIENT trans:", trans)
|
||||
}
|
||||
test.Log("CLIENT DONE")
|
||||
conn.Close()
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user