Repository: nanomsg/nng (c)

Issue:
Non-blocking send on rep sockets always fails
We have been working on a Qt project that utilises a host of req/rep sockets. Up until now we had been using nng v1.2.6 with no issues but upon updating to the latest release, v1.3.2, our pre-existing unit tests have shown a reproducible error; non-blocking calls to `nng_sendmsg` _always_ fail  (code `NNG_EAGAIN`) when using rep sockets that have received a message.  This occurs even in isolation, with no other ipc traffic, and while also allowing for multiple send attempts. Blocking calls still work as usual so this has been our workaround for the time being.

## NNG & Platform details.

- nng v1.3.2
- Ubuntu 18.04.5 LTS
- C/C++

## Expected Behaviour

0 return value indicating successful send of a received message when using a non-blocking call to `nng_sendmsg` on a rep socket.

## Actual Behaviour

Return value of 8 (`NNG_EAGAIN`) every time.

## Steps to Reproduce

1. Open req0 and rep0 sockets
2. Start listening on the rep socket
3. Dial using the req socket
4. Send an `nng_msg` from the req socket (blocking or non-blocking, matters not)
5. Receive the message using the rep socket (blocking or non-blocking, matters not)
6. Attempt to send a reply message using a non-blocking call to `nng_sendmsg`

Typically, under normal use, you can repeat step 6 if the message is not accepted at first (possibly due to backpressure). However, this is not the case here. Replacing step 6 with a blocking send or reverting back to nng v1.2.6 allows the test to pass again.

A verbose demonstration can be observed below.

 ```
int32_t testRepNonBlockSend(void)
{
    const char * const testUrl   = "ipc:///tmp/ut_nng_rep0.ipc";
    const int32_t      testValue = 265338;
    nng_socket         reqSock   = NNG_SOCKET_INITIALIZER;
    nng_socket         repSock   = NNG_SOCKET_INITIALIZER;
    nng_msg *          pTxMsg    = NULL;
    nng_msg *          pRxMsg    = NULL;

    // Start listening on the rep socket and dial
    int32_t rv = nng_rep0_open(&repSock);
    if (0 == rv)
    {
        rv = nng_listen(repSock, testUrl, NULL, 0);
    }
    if (0 == rv)
    {
        rv = nng_req0_open(&reqSock);
    }
    if (0 == rv)
    {
        rv = nng_dial(reqSock, testUrl, NULL, 0);
    }

    // Prepare the tx and rx messages
    if (0 == rv)
    {
        rv = nng_msg_alloc(&pTxMsg, sizeof(testValue));
    }
    if (0 == rv)
    {
        rv = nng_msg_alloc(&pRxMsg, sizeof(testValue));
        if (0 != rv)
        {
            nng_msg_free(pTxMsg);
        }
    }

    // Append an arbitrary value to the request message then send it
    if (0 == rv)
    {
        rv = nng_msg_append(pTxMsg, &testValue, sizeof(testValue));
    }
    if (0 == rv)
    {
        rv = nng_sendmsg(reqSock, pTxMsg, 0);
        if (0 != rv)
        {
            nng_msg_free(pTxMsg);
        }
    }

    // Receive the message
    if (0 == rv)
    {
        rv = nng_recvmsg(repSock, &pRxMsg, 0);
    }

    // Attempt to send the reply using a non-blocking call. This will always fail (even with
    // multiple attempts) when using nng v1.3.2. When using nng v1.2.6, the send works.
    if (0 == rv)
    {
        rv = nng_sendmsg(repSock, pRxMsg, NNG_FLAG_NONBLOCK);
        if (0 != rv)
        {
            nng_msg_free(pRxMsg);
        }
    }

    nng_close(repSock);
    nng_close(reqSock);

    return rv;
}
```

This was tested from C++ when wrapped using the gtest framework:

```
TEST(TestFixture, testRepNonBlockSend)
{
    EXPECT_EQ(0, testRepNonBlockSend());
}
```

Console Output:
```
Expected equality of these values:
  0
  testRepNonBlockSend()
    Which is: 8
```




Pull request:
fixes #1304 Non-blocking send on rep sockets always fails

Patch:
diff --git a/src/protocol/reqrep0/rep.c b/src/protocol/reqrep0/rep.c
index 6f859ee6..86593c26 100644
--- a/src/protocol/reqrep0/rep.c
+++ b/src/protocol/reqrep0/rep.c
@@ -161,12 +161,6 @@ rep0_ctx_send(void *arg, nni_aio *aio)
 		// reply for the single request we got.
 		nni_pollable_clear(&s->writable);
 	}
-	if ((rv = nni_aio_schedule(aio, rep0_ctx_cancel_send, ctx)) != 0) {
-		nni_mtx_unlock(&s->lk);
-		nni_aio_finish_error(aio, rv);
-		return;
-	}
-
 	if (len == 0) {
 		nni_mtx_unlock(&s->lk);
 		nni_aio_finish_error(aio, NNG_ESTATE);
@@ -199,6 +193,12 @@ rep0_ctx_send(void *arg, nni_aio *aio)
 		return;
 	}
 
+	if ((rv = nni_aio_schedule(aio, rep0_ctx_cancel_send, ctx)) != 0) {
+		nni_mtx_unlock(&s->lk);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
 	ctx->saio  = aio;
 	ctx->spipe = p;
 	nni_list_append(&p->sendq, ctx);


Test patch:
diff --git a/src/protocol/reqrep0/rep_test.c b/src/protocol/reqrep0/rep_test.c
index 367ec66e..dcdee423 100644
--- a/src/protocol/reqrep0/rep_test.c
+++ b/src/protocol/reqrep0/rep_test.c
@@ -434,12 +434,12 @@ test_rep_ctx_send_nonblock(void)
 	nng_socket req;
 	nng_ctx    ctx;
 	nng_aio *  aio;
-	nng_msg *msg;
+	nng_msg *  msg;
 
 	TEST_NNG_PASS(nng_req0_open(&req));
 	TEST_NNG_PASS(nng_rep0_open(&rep));
 	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
-	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 2000));
 	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000));
 	TEST_NNG_PASS(nng_ctx_open(&ctx, rep));
 	TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL));
@@ -456,12 +456,103 @@ test_rep_ctx_send_nonblock(void)
 	nng_ctx_send(ctx, aio);
 
 	nng_aio_wait(aio);
-	TEST_NNG_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_aio_result(aio));
 	TEST_NNG_PASS(nng_close(rep));
 	TEST_NNG_PASS(nng_close(req));
 	nng_aio_free(aio);
-	nng_msg_free(msg);
+}
+
+static void
+test_rep_ctx_send_nonblock2(void)
+{
+	nng_socket rep;
+	nng_socket req;
+	nng_ctx    rep_ctx[10];
+	nng_aio *  rep_aio[10];
+	int        num_good = 0;
+	int        num_fail = 0;
+
+	// We are going to send a bunch of requests, receive them,
+	// but then see that non-block pressure exerts for some, but
+	// that at least one non-blocking send works.
+	TEST_NNG_PASS(nng_req0_open_raw(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000));
+	for (int i = 0; i < 10; i++) {
+		TEST_NNG_PASS(nng_ctx_open(&rep_ctx[i], rep));
+		TEST_NNG_PASS(nng_aio_alloc(&rep_aio[i], NULL, NULL));
+	}
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	for (int i = 0; i < 10; i++) {
+		nng_msg *msg;
+		TEST_NNG_PASS(nng_msg_alloc(&msg, 4));
+                TEST_NNG_PASS(
+                    nng_msg_append_u32(msg, (unsigned) i | 0x80000000u));
+                nng_ctx_recv(rep_ctx[i], rep_aio[i]);
+		TEST_NNG_PASS(nng_sendmsg(req, msg, 0));
+	}
+	for (int i = 0; i < 10; i++) {
+		nng_msg *msg;
+		nng_aio_wait(rep_aio[i]);
+		TEST_NNG_PASS(nng_aio_result(rep_aio[i]));
+		msg = nng_aio_get_msg(rep_aio[i]);
+		nng_aio_set_timeout(rep_aio[i], 0);
+		nng_aio_set_msg(rep_aio[i], msg);
+		nng_ctx_send(rep_ctx[i], rep_aio[i]);
+	}
+
+	for (int i = 0; i < 10; i++) {
+		int rv;
+		nng_aio_wait(rep_aio[i]);
+		rv = nng_aio_result(rep_aio[i]);
+		if (rv == 0) {
+			num_good++;
+		} else {
+			TEST_NNG_FAIL(rv, NNG_ETIMEDOUT);
+			nng_msg_free(nng_aio_get_msg(rep_aio[i]));
+			num_fail++;
+		}
+	}
 
+	TEST_ASSERT(num_good > 0);
+	TEST_ASSERT(num_fail > 0);
+
+	for (int i = 0; i < 10; i++) {
+		nng_aio_free(rep_aio[i]);
+		nng_ctx_close(rep_ctx[i]);
+	}
+	TEST_NNG_PASS(nng_close(rep));
+	TEST_NNG_PASS(nng_close(req));
+}
+
+static void
+test_rep_send_nonblock(void)
+{
+	nng_socket rep;
+	nng_socket req;
+	int        rv;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	TEST_NNG_SEND_STR(req, "SEND");
+	TEST_NNG_RECV_STR(rep, "SEND");
+
+	// Use the nonblock flag
+	rv = nng_send(rep, "RECV", 5, NNG_FLAG_NONBLOCK);
+
+	TEST_NNG_PASS(rv);
+	TEST_NNG_RECV_STR(req, "RECV");
+	TEST_NNG_PASS(nng_close(rep));
+	TEST_NNG_PASS(nng_close(req));
 }
 
 void
@@ -496,12 +587,14 @@ TEST_LIST = {
 	{ "rep context does not poll", test_rep_context_no_poll },
 	{ "rep validate peer", test_rep_validate_peer },
 	{ "rep double recv", test_rep_double_recv },
+	{ "rep send nonblock", test_rep_send_nonblock },
 	{ "rep close pipe before send", test_rep_close_pipe_before_send },
 	{ "rep close pipe during send", test_rep_close_pipe_during_send },
 	{ "rep recv aio ctx stopped", test_rep_ctx_recv_aio_stopped },
 	{ "rep close pipe context send", test_rep_close_pipe_context_send },
 	{ "rep close context send", test_rep_close_context_send },
 	{ "rep context send nonblock", test_rep_ctx_send_nonblock },
+	{ "rep context send nonblock 2", test_rep_ctx_send_nonblock2 },
 	{ "rep context recv nonblock", test_rep_ctx_recv_nonblock },
 	{ "rep recv garbage", test_rep_recv_garbage },
 	{ NULL, NULL },