enm goes into infinite loop with pipeline tcp connection - when tcp service is not running.

[harishdeshmukh]

enm goes into infinite loop with pipeline tcp connection - when tcp service is not running.

start() ->
    enm:start_link(),
    Url = "tcp://localhost:8789",
    {ok,Push} = enm:push([{connect,Url},list]),
    Send1 = "Hello, World!",
    io:format("pushing message \"~s\"~n", [Send1]),
    ok = enm:send(Push, Send1),
    Send2 = "Goodbye.",
    io:format("pushing message \"~s\"~n", [Send2]),
    ok = enm:send(Push, Send2),
    enm:close(Push),
    enm:stop().

Scenario:
a. Connect Url is tcp
b. The tcp service is not running
c. Run the following code. I just did

$ erl -pa ebin deps/*/ebin
1> c("src/pipeline.erl").
{ok,pipeline}
3> pipeline:start().
pushing message "Hello, World!"
pushing message "Goodbye."

d. Look at top - the erlang process is taking 100% CPU.
e. CentOS - Linux version 3.10.0-123.20.1.el7.x86_64 (builder@kbuilder.dev.centos.org)
f. gcc version 4.8.2 20140120

Is it a known issue? Please let me know if its a user error.

Thanks.

[vinoski]

Thanks, will take a look.

[harishdeshmukh]

Here the rc from enm_do_send comes back as -1. This is the function sequence that repeats.

(gdb) where
#0  enm_ready_output (drv_data=0x7fae7bd0b338, event=0xa) at c_src/enm_drv.c:831
#1  0x00000000004c27eb in erts_port_task_execute ()
#2  0x00000000004ba2b4 in schedule ()
#3  0x0000000000572373 in process_main ()
#4  0x00000000004aa3ea in sched_thread_func ()
#5  0x0000000000606275 in thr_wrapper ()
#6  0x00007fae7cc15df5 in start_thread (arg=0x7fae73dbf700) at pthread_create.c:308
#7  0x00007fae7c73b1ad in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:113
(gdb) list
826     memset(&msghdr, 0, sizeof msghdr);
827     msghdr.msg_iov = (struct nn_iovec*)ev.iov;
828     msghdr.msg_iovlen = ev.vsize;
829     msghdr.msg_control = 0;
830     rc = enm_do_send(d, &msghdr);
831     if (rc > 0)
832         driver_deq(d->port, rc);
833     if (rc == total && d->b.writable) {
834         if (d->b.busy) {
835             d->b.busy = 0;
(gdb) n
833     if (rc == total && d->b.writable) {
(gdb) p rc
$14 = -1
(gdb) n
840 }

I've not done any erl port and new to erlang. But i think somehow erlport needs to be notified that operation failed.

static void
enm_ready_output(ErlDrvData drv_data, ErlDrvEvent event)
{
    EnmData* d = enm_sockets[(long)event];
    struct nn_msghdr msghdr;
    ErlIOVec ev;
    ErlDrvSizeT total;
    int rc;

    d->b.writable = 1;
    total = driver_peekqv(d->port, &ev);
    if (total == 0)
        return;
    memset(&msghdr, 0, sizeof msghdr);
    msghdr.msg_iov = (struct nn_iovec*)ev.iov;
    msghdr.msg_iovlen = ev.vsize;
    msghdr.msg_control = 0;
    rc = enm_do_send(d, &msghdr);
    if (rc > 0)
        driver_deq(d->port, rc);
    if (rc == total && d->b.writable) {
        if (d->b.busy) {
            d->b.busy = 0;
            set_busy_port(d->port, d->b.busy);
        }
        enm_write_select(d, 0);
    }
}

Anyway - you'll know it best. Please let me know if I can help.

[harishdeshmukh]

adding driver_failure(d->port, errno); is making it crash as oppose to hang.

static void
enm_ready_output(ErlDrvData drv_data, ErlDrvEvent event)
{
    EnmData* d = enm_sockets[(long)event];
    struct nn_msghdr msghdr;
    ErlIOVec ev;
    ErlDrvSizeT total;
    int rc;

    d->b.writable = 1;
    total = driver_peekqv(d->port, &ev);
    if (total == 0)
        return;
    memset(&msghdr, 0, sizeof msghdr);
    msghdr.msg_iov = (struct nn_iovec*)ev.iov;
    msghdr.msg_iovlen = ev.vsize;
    msghdr.msg_control = 0;
    rc = enm_do_send(d, &msghdr);
    if (rc > 0)
        driver_deq(d->port, rc);
    if (rc == total && d->b.writable) {
        if (d->b.busy) {
            d->b.busy = 0;
            set_busy_port(d->port, d->b.busy);
        }
        enm_write_select(d, 0);
    }
    if(rc <0){
        driver_failure(d->port, errno);
    }
}

[vinoski]

The problem with driver_failure is that it's too harsh for this situation. Please look at branch vinoski/fix7, which I think is a better fix, and try it out. @rzezeski can you have a look at that branch too and let me know your thoughts on this?

[harishdeshmukh]

Thanks Steve. I also thought driver_failure was not harsh but was not sure about eof. Will let test it next week and let you know.

[harishdeshmukh]

Hi Steve,

This fix works for us. We effectively get the closed socket effect and are going to retry with backoff.

Thanks a lot for help. When would you be able to push it to master?

Regards,
Harish.

[vinoski]

Thanks for testing it. I'll need to get the change reviewed before it can go to master. I'll try to get that done as soon as possible.

[supershal]

I wrote a small program to test this fix. It gives me {error, close} on second message even if the tcp server is listening properly. I tested the same code against older build(before the patch), it worked fine.
Please let me know if the test is doing anything wrong.

-module(enm_stress).

-export([stress/0]).

stress() ->
  enm:start_link(),
  Url = "tcp://127.0.0.1:9090",
  %Url = "inproc://pipeline",
  spawn(fun() -> acceptor(Url) end),
  timer:sleep(1000),
  spawn(fun() -> sender(Url) end).

acceptor(Url) ->
{ok, Socket} = enm:pull([{bind, Url}]),
io:format("accepting on socket ~p~n", [Socket]),
listen(Socket).

listen(Socket) ->
  receive
      {nnpull, Socket, <<"quit">>} ->
        io:format("quitting server~n"),
        enm:close(Socket);
      {nnpull, Socket, Message} ->
        io:format("~p : ~p~n", [erlang:now(), Message]),
        listen(Socket)
  after 10000 ->
        io:format("timeout. quitting server~n"),
        enm:close(Socket)
  end.

sender(Url) ->
 {ok, Client} = enm:push([{connect, Url}]),
 send(Client, 1000).

send(Socket, 0) ->
  io:format("Quitting client~n"),
  ok=enm:send(Socket, <<"quit">>),
  timer:sleep(2000),
  enm:close(Socket);
send(Socket, Tick) ->
  io:format("sending ~p ~n", [Tick]),
  ok=enm:send(Socket, integer_to_binary(Tick)),
  timer:sleep(5), % milliseconds 
  send(Socket, Tick-1).

[vinoski]

@supershal thanks for reporting this. Sometimes when I run your test code, it runs correctly to completion, but other times it hits the same error you describe. I'll need to look into it some more.

[vinoski]

I have a fix that preserves the behavior of the original change in this thread but that also avoids the problem @supershal has seen. The fix involves counting the number of EAGAIN spins we take through the enm_ready_output function, and if we reach 64 without having sent any data, a number I admittedly chose unscientifically, we close the socket. With this change the new test for @harishdeshmukh's issue still works and I've been able to run @supershal's test program for hundreds of iterations without failure. I'll commit the change soon. Given the nanomsg API, I don't know of another alternative that works here.

[vinoski]

@supershal can you try this fix and let me know if it works for you?