{"action":"created","issue":{"url":"https://api.github.com/repos/nanomsg/nanomsg/issues/390","labels_url":"https://api.github.com/repos/nanomsg/nanomsg/issues/390/labels{/name}","comments_url":"https://api.github.com/repos/nanomsg/nanomsg/issues/390/comments","events_url":"https://api.github.com/repos/nanomsg/nanomsg/issues/390/events","html_url":"https://github.com/nanomsg/nanomsg/issues/390","id":62760275,"number":390,"title":"Pub/sub appears to be silently dropping messages when sending in bursts","user":{"login":"dasques","id":11542756,"avatar_url":"https://avatars.githubusercontent.com/u/11542756?v=3","gravatar_id":"","url":"https://api.github.com/users/dasques","html_url":"https://github.com/dasques","followers_url":"https://api.github.com/users/dasques/followers","following_url":"https://api.github.com/users/dasques/following{/other_user}","gists_url":"https://api.github.com/users/dasques/gists{/gist_id}","starred_url":"https://api.github.com/users/dasques/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dasques/subscriptions","organizations_url":"https://api.github.com/users/dasques/orgs","repos_url":"https://api.github.com/users/dasques/repos","events_url":"https://api.github.com/users/dasques/events{/privacy}","received_events_url":"https://api.github.com/users/dasques/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2015-03-18T18:00:26Z","updated_at":"2015-04-27T18:40:01Z","closed_at":null,"body":"My publisher sends a large number of messages (e.g. 5000) in a single burst, and occasionally some of the messages are being missed by the subscribers. \r\n\r\nI have boiled down the issue to two simple programs, a publisher and a subscriber, described in the below pseudo-code. (I will post the actual code later below.)\r\n\r\nPublisher:\r\n```c\r\nseqNumber = 0;\r\nwhile (true)\r\n{\r\n    do 5000 times\r\n    {\r\n        publish a message containing seqNumber.\r\n        ++seqNumber;\r\n    }\r\n    Sleep 1 second.\r\n}\r\n```\r\n\r\nSubscriber:\r\n```c\r\nwhile (true)\r\n{\r\n    Receive message.\r\n    Assert message seqNumber is 1 higher than last.\r\n}\r\n```\r\n\r\nWhen I run the publisher, and 3 subscribers, very quickly (within seconds) I see one of the subscribers complain of missed sequence numbers.\r\n\r\nSome caveats:\r\n1. My test machine was a linux VM, with 2 cpus, 4GB ram.\r\n2. I can't see the issue on my physical linux box.\r\n(However, if you run remove the \"Sleep 1 second\" above and run it for a long time, it still exhibits the same issue, but I'm thinking this is not necessarily as serious, given we are pushing the system to the limit.)\r\n3. I tried upping NN_SNDBUF and NN_RCVBUF to 32MB (from 128k), but this did not help.\r\n4. While I'm sending the burst, if every 10 messages, I put in sleep of 1 *micro second*, the problem goes away.\r\n5. When I run 3 subscribers, it's usually only one that fails after a while.\r\n6. When I use nanocat to subscribe and pipe the messages to a file, I see the same issue, messages being missed. It's to eliminate nanocat that I made my own simple subscriber.\r\n7. Version nanomsg-0.5-beta.\r\n8. The message size is 250 bytes, including the sequence number.\r\n\r\nSo my questions are:\r\n1. Given the code shown below, is there anything I'm doing anything wrong?\r\n2. Shouldn't something e.g. nn_send, return an error if for whatever reason it can't send all the messages to all the subscribers?\r\n\r\nPublisher:\r\n```c\r\n#include <assert.h>\r\n#include <string.h>\r\n#include <stdio.h>\r\n#include <stdlib.h>\r\n#include \"nn.h\"\r\n#include \"pubsub.h\"\r\n\r\nchar* tmplt = \r\n  \"abcdefghijabcdefghijabcdefghijabcdefghijabcdefghij\"\r\n  \"abcdefghijabcdefghijabcdefghijabcdefghijabcdefghij\"\r\n  \"abcdefghijabcdefghijabcdefghijabcdefghijabcdefghij\"\r\n  \"abcdefghijabcdefghijabcdefghijabcdefghijabcdefghij\"\r\n  \"abcdefghijabcdefghijabcdefghijabcdefgh %010d\"; // 249 bytes.\r\n\r\nint counter = 0;\r\n\r\nint server(const char *url, int msgsPerBurst)\r\n{\r\n  int sock = nn_socket(AF_SP, NN_PUB);\r\n  assert(sock >= 0);\r\n\r\n  // Set NN_SNDBUF size. Note: this is not required to reproduce the issue.\r\n  int setBufSize = 32 * 1024 * 1024;\r\n  assert(nn_setsockopt(sock, NN_SOL_SOCKET, NN_SNDBUF, &setBufSize, sizeof(setBufSize)) == 0);\r\n\r\n  assert(nn_bind(sock, url) >= 0);\r\n  while (1)\r\n  {\r\n    int ix = 0;\r\n    for (ix = 0; ix < msgsPerBurst; ++ix)\r\n    {\r\n      char* buf = nn_allocmsg(250, 0);\r\n      snprintf(buf, 250, tmplt, counter);\r\n      int szBuf = 250;\r\n      int bytes = nn_send(sock, &buf, NN_MSG, 0);\r\n\r\n      assert(bytes == szBuf);\r\n\r\n      // Uncommenting this makes it a lot better!\r\n      //if (counter % 10 == 0)\r\n      //  usleep(1); // Note: 1 microsecond.\r\n\r\n      ++counter;\r\n    }\r\n\r\n    printf(\"Published %d\\n\", counter);\r\n    sleep(1); // Note: 1 full second.\r\n  }\r\n  return nn_shutdown(sock, 0);\r\n}\r\n\r\n\r\nint main(const int argc, const char **argv)\r\n{\r\n  if (argc >= 3)\r\n    return server(argv[1], atoi(argv[2]));\r\n  else\r\n  {\r\n    fprintf (stderr, \"Usage: burst_publisher <URL> <msgsPerBurst>\\n\");\r\n    return 1;\r\n  }\r\n}\r\n```\r\n\r\nSubscriber:\r\n```c\r\n#include <assert.h>\r\n#include <string.h>\r\n#include <stdio.h>\r\n#include <stdlib.h>\r\n#include \"nn.h\"\r\n#include \"pubsub.h\"\r\n\r\nint lastSeqNum = -1;\r\nint count = 0;\r\n\r\nvoid process(char* rec, int bytes)\r\n{\r\n\tchar* space = strstr(rec, \" \");\r\n\tint seqNum = atoi(space + 1);\r\n\r\n\tif (lastSeqNum == -1)\r\n\t\tprintf(\"Saw first msg, seqNum: %d\\n\", seqNum);\r\n\telse if (seqNum != lastSeqNum + 1)\r\n\t{\r\n\t\tprintf(\"Oops! seqNum %d following %d, exiting after receiving %d messages.\\n\", seqNum, lastSeqNum, count);\r\n\t\texit(1);\r\n\t}\r\n\r\n\tlastSeqNum = seqNum;\r\n\t++count;\r\n}\r\n\r\nint subscribe(const char *url)\r\n{\r\n  int sock = nn_socket(AF_SP, NN_SUB);\r\n  assert(sock >= 0);\r\n\r\n  // Set NN_RCVBUF size. Note: this is not required to reproduce the issue.\r\n  int setBufSize = 32 * 1024 * 1024;\r\n  assert(nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVBUF, &setBufSize, sizeof(setBufSize)) == 0);\r\n\r\n  assert(nn_setsockopt(sock, NN_SUB, NN_SUB_SUBSCRIBE, \"\", 0) >= 0);\r\n  assert(nn_connect(sock, url) >= 0);\r\n  while (1)\r\n  {\r\n    char *buf = NULL;\r\n    int bytes = nn_recv(sock, &buf, NN_MSG, 0);\r\n    assert(bytes >= 0);\r\n\r\n    if (bytes > 0)\r\n \t\tprocess(buf, bytes);\r\n      \r\n    nn_freemsg(buf);\r\n  }\r\n  return nn_shutdown(sock, 0);\r\n}\r\n\r\nint main(const int argc, const char **argv)\r\n{\r\n  if (argc >= 2)\r\n    return subscribe(argv[1]);\r\n  else\r\n  {\r\n    fprintf(stderr, \"Usage: burst_subscriber <URL>\\n\");\r\n    return 1;\r\n  }\r\n}\r\n````\r\n\r\nHere's a typical output from a subscriber:\r\n```c\r\nSaw first msg, seqNum: 20000\r\nOops! seqNum 47858 following 47830, exiting after receiving 27831 messages.\r\n```"},"comment":{"url":"https://api.github.com/repos/nanomsg/nanomsg/issues/comments/96775320","html_url":"https://github.com/nanomsg/nanomsg/issues/390#issuecomment-96775320","issue_url":"https://api.github.com/repos/nanomsg/nanomsg/issues/390","id":96775320,"user":{"login":"vseryakov","id":1746546,"avatar_url":"https://avatars.githubusercontent.com/u/1746546?v=3","gravatar_id":"","url":"https://api.github.com/users/vseryakov","html_url":"https://github.com/vseryakov","followers_url":"https://api.github.com/users/vseryakov/followers","following_url":"https://api.github.com/users/vseryakov/following{/other_user}","gists_url":"https://api.github.com/users/vseryakov/gists{/gist_id}","starred_url":"https://api.github.com/users/vseryakov/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vseryakov/subscriptions","organizations_url":"https://api.github.com/users/vseryakov/orgs","repos_url":"https://api.github.com/users/vseryakov/repos","events_url":"https://api.github.com/users/vseryakov/events{/privacy}","received_events_url":"https://api.github.com/users/vseryakov/received_events","type":"User","site_admin":false},"created_at":"2015-04-27T18:40:01Z","updated_at":"2015-04-27T18:40:01Z","body":"can it be that messages arrive in different order?"}}