zmq4: add context to semaphore lock

This change adds context to the semaphore.lock method so that read can be
unblocked when the context is canceled, instead of resulting in a deadlock.

msgio.go

@@ -87,7 +87,7 @@ func (q *qreader) rmConn(r *Conn) {
 }
 
 func (q *qreader) read(ctx context.Context, msg *Msg) error {
-	q.sem.lock()
+	q.sem.lock(ctx)
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
@@ -166,7 +166,7 @@ func (mw *mwriter) rmConn(w *Conn) {
 }
 
 func (w *mwriter) write(ctx context.Context, msg Msg) error {
-	w.sem.lock()
+	w.sem.lock(ctx)
 	grp, _ := errgroup.WithContext(ctx)
 	w.mu.Lock()
 	for i := range w.ws {
@@ -199,8 +199,11 @@ func (sem *semaphore) enable() {
 	}
 }
 
-func (sem *semaphore) lock() {
-	<-sem.ready
+func (sem *semaphore) lock(ctx context.Context) {
+	select {
+	case <-ctx.Done():
+	case <-sem.ready:
+	}
 }
 
 var (

pub.go

@@ -173,7 +173,7 @@ func (q *pubQReader) rmConn(r *Conn) {
 }
 
 func (q *pubQReader) read(ctx context.Context, msg *Msg) error {
-	q.sem.lock()
+	q.sem.lock(ctx)
 	select {
 	case <-ctx.Done():
 	case *msg = <-q.c:

router.go

@@ -143,7 +143,7 @@ func (q *routerQReader) rmConn(r *Conn) {
 }
 
 func (q *routerQReader) read(ctx context.Context, msg *Msg) error {
-	q.sem.lock()
+	q.sem.lock(ctx)
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
@@ -224,7 +224,7 @@ func (mw *routerMWriter) rmConn(w *Conn) {
 }
 
 func (w *routerMWriter) write(ctx context.Context, msg Msg) error {
-	w.sem.lock()
+	w.sem.lock(ctx)
 	grp, _ := errgroup.WithContext(ctx)
 	w.mu.Lock()
 	id := msg.Frames[0]