// #define BLOCKING
// #define IMMEDIATE
#define THREADED
// #define SLEEP_EACH 20ms
#include <atomic>
#include <cassert>
#include <chrono>
#include <iostream>
#include <memory>
#include <thread>
#include <vector>
#include <zmq.h>
int main()
  using namespace std::chrono_literals;
  using std::chrono::steady_clock;
  std::atomic<bool> stopped{false};
  std::atomic<steady_clock::time_point> lastReceived{steady_clock::now()};
  std::atomic<unsigned long long>
    messagesReceived{0ull},
    messagesSent{0ull};
  constexpr unsigned long long TOTAL_MESSAGES = 1'000;
  unsigned long long lastMessage = std::numeric_limits<unsigned long long>::max();
  std::vector<unsigned long long> messageNumbers;
  auto ctx = zmq_ctx_new();
  // pull socket reading numbers
  std::thread receiver([&]() {
    auto socket = zmq_socket(ctx, ZMQ_PULL);
    int opt = 100'000;
    zmq_setsockopt(socket, ZMQ_RCVHWM, &opt, sizeof opt);
    zmq_bind(socket, "tcp://127.0.0.1:20000");
    char buf[256];
    while (!stopped) {
      int len = zmq_recv(socket, &buf, sizeof buf, 0);
      if (len > 0) {
        buf[len] = '\0';
        int number = std::atoi(buf);
        messageNumbers.push_back(number);
        lastMessage = number;
        if (++messagesReceived == TOTAL_MESSAGES)
          stopped.store(true);
        lastReceived.store(steady_clock::now());
    zmq_close(socket);
  // time for socket to get ready
  std::this_thread::sleep_for(2s);
  int rc = 0;
  for (unsigned long long i = 0ull; i < TOTAL_MESSAGES && rc != -1; ++i)
#ifdef THREADED
    std::thread t([&]()
#endif
#ifdef SLEEP_EACH
      std::this_thread::sleep_for(SLEEP_EACH);
#endif
      auto socket = zmq_socket(ctx, ZMQ_PUSH);
      int opt = 10'000;
      zmq_setsockopt(socket, ZMQ_SNDTIMEO, &opt, sizeof opt);
      opt = 30'000;
      zmq_setsockopt(socket, ZMQ_LINGER, &opt, sizeof opt);
#ifdef IMMEDIATE
      opt = 1;
      zmq_setsockopt(socket, ZMQ_IMMEDIATE, &opt, sizeof opt);
#endif
      zmq_connect(socket, "tcp://127.0.0.1:20000");
      std::string msg = std::to_string(i);
#ifdef BLOCKING
      int blocking = 0;
#else
      int blocking = ZMQ_NOBLOCK;
#endif
      rc = zmq_send(socket, msg.c_str(), msg.length(), blocking);
      if (rc == -1) {
        std::cerr << "Send failed: [" << zmq_errno() << "] " << zmq_strerror(zmq_errno()) << std::endl;
      assert(rc != -1);
      ++messagesSent;
      zmq_close(socket);
#ifdef THREADED
    t.join();
#endif
  while (!stopped && receiver.joinable()) {
    // if we haven't got a new message in 5 seconds, assume no more are coming
    if (steady_clock::now() - lastReceived.load() > 5s) {
      if (!stopped) {
        std::cerr << "No message in 5 seconds, stopping test early..." << std::endl;
      stopped.store(true);
    std::this_thread::sleep_for(100ms);
  std::cout << "Sent " << messagesSent.load() << ", read " << messagesReceived.load() << ", lost " << (messagesSent - messagesReceived);
  //assert(messagesReceived.load() == TOTAL_MESSAGES);
  zmq_term(ctx);
  return 0;
What's the actual result? (include assertion message & call stack if applicable)
Basically, non-immediate and non-sleeping tests drop a few messages often.  Blocking and immediate never drop in the test case, but in the production environment blocking and immediate still drops unless we add something like std::cout after the send then it never drops (maybe due to a syscall occurring?).  Using a thread makes the dropping worse but is not necessary to get some drops.  Adding a 20ms sleep seems to fix the issue regardless of the other settings.  Even a 1ms sleep seems to fix the issue (syscall?).
I ran three runs of each of these configurations, here are the results for Windows 10 (1000 indicates a pass, anything lower is a failure with dropped messages):
Blocking
Immediate
Threaded
Sleep
Received Messages (out of 1000)
          
Note this is TCP transport but on the loopback so bad connectivity should not be possible.  The behavior matches what I see if we have a 0 linger set.  Cannot reproduce on inproc transport.
A curious thing is the sleep fix is not between zmq_send and zmq_close, but before making each socket.
I also just noticed Issue #1922 of which this is a possible duplicate, except this test reproduces on Mac and does not use a low HWM.