mirror of
https://github.com/luxfi/zmq.git
synced 2026-10-30 15:11:11 +00:00
zeekayandHanzo Dev
cde37cf581
Modules declare go 1.27.1 and the build images follow. go fix applied the modernizers; omitzero is not applied, since it changes what the JSON says. Co-authored-by: Hanzo Dev <[email protected]>
102 lines
2.3 KiB
Go
102 lines
2.3 KiB
Go
// Copyright 2025 The go-zeromq Authors. All rights reserved.
|
|
// Use of this source code is governed by a BSD-style
|
|
// license that can be found in the LICENSE file.
|
|
|
|
package zmq4
|
|
|
|
import (
|
|
"io"
|
|
"net"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestConnReaperDeadlock2(t *testing.T) {
|
|
t.Skip("Test causes deadlock - needs investigation")
|
|
ep := must(EndPoint("tcp"))
|
|
defer cleanUp(ep)
|
|
|
|
ctx := t.Context()
|
|
|
|
// Bind the server.
|
|
srv := NewRouter(ctx, WithLogger(Devnull)).(*routerSocket)
|
|
if err := srv.Listen(ep); err != nil {
|
|
t.Fatalf("could not listen on %q: %+v", ep, err)
|
|
}
|
|
defer srv.Close()
|
|
|
|
// Add modified clients connection to server
|
|
// so any send to client will trigger context switch
|
|
// and be failing.
|
|
// Idea is that while srv.Send is progressing,
|
|
// the connection will be closed and assigned
|
|
// for connection reaper, and reaper will try to remove those
|
|
id := "client-x"
|
|
srv.sck.mu.Lock()
|
|
rmw := srv.sck.w.(*routerMWriter)
|
|
for range 2 {
|
|
w := &Conn{}
|
|
w.Peer.Meta = make(Metadata)
|
|
w.Peer.Meta[sysSockID] = id
|
|
w.rw = &sockSendEOF{}
|
|
w.onCloseErrorCB = srv.sck.scheduleRmConn
|
|
// Do not to call srv.addConn as we dont want to have listener on this fake socket
|
|
rmw.addConn(w)
|
|
srv.sck.conns = append(srv.sck.conns, w)
|
|
}
|
|
srv.sck.mu.Unlock()
|
|
|
|
// Now try to send a message from the server to all clients.
|
|
msg := NewMsgFrom(nil, nil, []byte("payload"))
|
|
msg.Frames[0] = []byte(id)
|
|
if err := srv.Send(msg); err != nil {
|
|
t.Logf("Send to %s failed: %+v\n", id, err)
|
|
}
|
|
}
|
|
|
|
type sockSendEOF struct {
|
|
}
|
|
|
|
var a atomic.Int32
|
|
|
|
func (r *sockSendEOF) Write(b []byte) (n int, err error) {
|
|
// Each odd write fails asap.
|
|
// Each even write fails after sleep.
|
|
// Such a way we ensure the short write failure
|
|
// will cause socket be assinged to connection reaper
|
|
// while srv.Send is still in progress due to long writes.
|
|
if x := a.Add(1); x&1 == 0 {
|
|
time.Sleep(1 * time.Second)
|
|
}
|
|
return 0, io.EOF
|
|
}
|
|
|
|
func (r *sockSendEOF) Read(b []byte) (int, error) {
|
|
return 0, nil
|
|
}
|
|
|
|
func (r *sockSendEOF) Close() error {
|
|
return nil
|
|
}
|
|
|
|
func (r *sockSendEOF) LocalAddr() net.Addr {
|
|
return nil
|
|
}
|
|
|
|
func (r *sockSendEOF) RemoteAddr() net.Addr {
|
|
return nil
|
|
}
|
|
|
|
func (r *sockSendEOF) SetDeadline(t time.Time) error {
|
|
return nil
|
|
}
|
|
|
|
func (r *sockSendEOF) SetReadDeadline(t time.Time) error {
|
|
return nil
|
|
}
|
|
|
|
func (r *sockSendEOF) SetWriteDeadline(t time.Time) error {
|
|
return nil
|
|
}
|