Java.lang.ArrayIndexOutOfBoundsException: 256 с версией jeromq 0.3.6

Я использую Jeromq в многопоточной среде, как показано ниже. Ниже мой код, в котором конструктор SocketManager сначала подключается ко всем доступным сокетам, и я помещаю их в карту liveSocketsByDatacenter в методе connectToZMQSockets. После этого я запускаю фоновый поток в том же конструкторе, который запускается каждые 30 секунд, и он вызывает метод updateLiveSockets для ping всех тех сокетов, которые уже были на карте liveSocketsByDatacenter, и обновляет карту liveSocketsByDatacenter с тем, были ли эти сокеты живы или нет.

И метод getNextSocket() вызывается несколькими потоками чтения одновременно, чтобы получить следующий доступный сотовый доступ, а затем мы используем этот сокет для отправки данных на нем. Итак, мой вопрос в том, правильно ли мы используем Jeromq в многопоточной среде? Поскольку мы просто увидели исключение в нашей производственной среде с этой stacktrace, пока мы пытались отправить данные в этот живой сокет, поэтому я не уверен, что это ошибка или что-то еще?

java.lang.ArrayIndexOutOfBoundsException: 256
at zmq.YQueue.push(YQueue.java:97)
at zmq.YPipe.write(YPipe.java:47)
at zmq.Pipe.write(Pipe.java:232)
at zmq.LB.send(LB.java:83)
at zmq.Push.xsend(Push.java:48)
at zmq.SocketBase.send(SocketBase.java:590)
at org.zeromq.ZMQ$Socket.send(ZMQ.java:1271)
at org.zeromq.ZFrame.send(ZFrame.java:131)
at org.zeromq.ZFrame.sendAndKeep(ZFrame.java:146)
at org.zeromq.ZMsg.send(ZMsg.java:191)
at org.zeromq.ZMsg.send(ZMsg.java:163)

Ниже мой код:

public class SocketManager {
    private static final Random random = new Random();
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private final Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter = new ConcurrentHashMap<>();
    private final ZContext ctx = new ZContext();

    private static class Holder {
        private static final SocketManager instance = new SocketManager();
    }

    public static SocketManager getInstance() {
        return Holder.instance;
    }

    private SocketManager() {
      connectToZMQSockets();
      scheduler.scheduleAtFixedRate(this::updateLiveSockets, 30, 30, TimeUnit.SECONDS);
    }

    // during startup, making a connection and populate once
    private void connectToZMQSockets() {
      Map<Datacenters, List<String>> socketsByDatacenter = Utils.SERVERS;
      for (Map.Entry<Datacenters, List<String>> entry : socketsByDatacenter.entrySet()) {
        List<SocketHolder> addedColoSockets = connect(entry.getValue(), ZMQ.PUSH);
        liveSocketsByDatacenter.put(entry.getKey(), addedColoSockets);
      }
    }

    private List<SocketHolder> connect(List<String> addresses, int socketType) {
        List<SocketHolder> socketList = new ArrayList<>();
        for (String address : addresses) {
          try {
            Socket client = ctx.createSocket(socketType);
            // Set random identity to make tracing easier
            String identity = String.format("%04X-%04X", random.nextInt(), random.nextInt());
            client.setIdentity(identity.getBytes(ZMQ.CHARSET));
            client.setTCPKeepAlive(1);
            client.setSendTimeOut(7);
            client.setLinger(0);
            client.connect(address);

            SocketHolder zmq = new SocketHolder(client, ctx, address, true);
            socketList.add(zmq);
          } catch (Exception ex) {
            // log error
          }
        }
        return socketList;
    }

    // this method will be called by multiple threads concurrently to get the next live socket
    // is there any concurrency or thread safety issue or race condition here?
    public Optional<SocketHolder> getNextSocket() {
      for (Datacenters dc : Datacenters.getOrderedDatacenters()) {
        Optional<SocketHolder> liveSocket = getLiveSocket(liveSocketsByDatacenter.get(dc));
        if (liveSocket.isPresent()) {
          return liveSocket;
        }
      }
      return Optional.absent();
    }

    private Optional<SocketHolder> getLiveSocket(final List<SocketHolder> listOfEndPoints) {
      if (!CollectionUtils.isEmpty(listOfEndPoints)) {
        // The list of live sockets
        List<SocketHolder> liveOnly = new ArrayList<>(listOfEndPoints.size());
        for (SocketHolder obj : listOfEndPoints) {
          if (obj.isLive()) {
            liveOnly.add(obj);
          }
        }
        if (!liveOnly.isEmpty()) {
          // The list is not empty so we shuffle it an return the first element
          return Optional.of(liveOnly.get(random.nextInt(liveOnly.size()))); // just pick one
        }
      }
      return Optional.absent();
    }

    // runs every 30 seconds to ping all the socket to make sure whether they are alive or not
    private void updateLiveSockets() {
      Map<Datacenters, List<String>> socketsByDatacenter = Utils.SERVERS;

      for (Map.Entry<Datacenters, List<String>> entry : socketsByDatacenter.entrySet()) {
        List<SocketHolder> liveSockets = liveSocketsByDatacenter.get(entry.getKey());
        List<SocketHolder> liveUpdatedSockets = new ArrayList<>();
        for (SocketHolder liveSocket : liveSockets) { // LINE A
          Socket socket = liveSocket.getSocket();
          String endpoint = liveSocket.getEndpoint();
          Map<byte[], byte[]> holder = populateMap();
          Message message = new Message(holder, Partition.COMMAND);

          // pinging to see whether a socket is live or not
          boolean status = SendToSocket.getInstance().execute(message.getAdd(), holder, socket);
          boolean isLive = (status) ? true : false;

          SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, isLive);
          liveUpdatedSockets.add(zmq);
        }
        liveSocketsByDatacenter.put(entry.getKey(), Collections.unmodifiableList(liveUpdatedSockets));
      }
    }
}

И вот как я использую метод getNextSocket() класса SocketManager одновременно из нескольких потоков чтения:

// this method will be called from multiple threads
public boolean sendAsync(final long addr, final byte[] reco) {
  Optional<SocketHolder> liveSockets = SocketManager.getInstance().getNextSocket();
  return sendAsync(addr, reco, liveSockets.get().getSocket(), false);
}

public boolean sendAsync(final long addr, final byte[] reco, final Socket socket,
    final boolean messageA) {
  ZMsg msg = new ZMsg();
  msg.add(reco);
  boolean sent = msg.send(socket);
  msg.destroy();
  retryHolder.put(addr, reco);
  return sent;
}

  public boolean send(final long address, final byte[] encodedRecords, final Socket socket) {
    boolean sent = sendAsync(address, encodedRecords, socket, true);
    // if the record was sent successfully, then only sleep for timeout period
    if (sent) {
      try {
        TimeUnit.MILLISECONDS.sleep(500);
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
      }
    }
    // ...
    return sent;
  }

Я не думаю, что это правильно, я верю. Кажется, getNextSocket() может вернуть 0MQ socket в thread A. Одновременно поток таймера может получить доступ к тому же 0MQ socket, чтобы выполнить ping. В этом случае thread A и поток таймера мутирует тот же 0MQ socket, что приведет к проблемам. Итак, какой лучший и эффективный способ решить эту проблему?

Примечание. SocketHolder - это неизменный класс

Update:

Я заметил, что одна и та же проблема произошла в моем другом ящике с тем же ArrayIndexOutOfBoundsException, но на этот раз его 71 номер строки в файле "YQueue". Единственная последовательная вещь - 256 всегда. Таким образом, должно быть что-то, связанное с 256 точно, и я не могу понять, что это такое здесь 256?

java.lang.ArrayIndexOutOfBoundsException: 256
    at zmq.YQueue.backPos(YQueue.java:71)
    at zmq.YPipe.write(YPipe.java:51)
    at zmq.Pipe.write(Pipe.java:232)
    at zmq.LB.send(LB.java:83)
    at zmq.Push.xsend(Push.java:48)
    at zmq.SocketBase.send(SocketBase.java:590)
    at org.zeromq.ZMQ$Socket.send(ZMQ.java:1271)
    at org.zeromq.ZFrame.send(ZFrame.java:131)
    at org.zeromq.ZFrame.sendAndKeep(ZFrame.java:146)
    at org.zeromq.ZMsg.send(ZMsg.java:191)
    at org.zeromq.ZMsg.send(ZMsg.java:163)

Ответы

Ответ 1

Факт № 0: ZeroMQ не является потокобезопасным - по определению

В то время как документация ZeroMQ и превосходная книга Pieter HINTJENS "Код, подключенный. Том 1", не забудьте напомнить об этом, когда это возможно, идея возвращения или даже совместного использования экземпляра сокета ZeroMQ среди потоков появляется время от времени. Разумеется, методы класса-экземпляра могут доставить это почти "скрытое" внутри своих внутренних методов и атрибутов, но надлежащие проектные усилия должны предотвращать любые такие побочные эффекты без каких-либо исключений, никаких оправданий.

Совместное использование, если это обоснованно поддерживается количественными фактами, может быть способом для общего экземпляра zmq.Context(), но кристально чистая распределенная система может жить по действительно мультиагентной схеме, где каждый агент управляет собственным Context() -engine, точно настроенный для соответствующего сочетания настроек конфигурации и производительности.

Итак, каков наилучший и эффективный способ решить эту проблему?

Никогда не используйте сокет ZeroMQ. Никогда. Даже если новейшая разработка начала обещать некоторые близкие будущие изменения в этом направлении. Это плохая привычка загрязнять любой высокопроизводительный дизайн с низкой задержкой распределенной системы с совместным использованием. Доля ничего не является лучшим дизайном для этого домена.

Да, я вижу, что мы не должны делиться сокетами между потоками, но в моем коде
как вы думаете, лучший способ разрешить это?

Да, лучший и эффективный способ решить эту проблему - никогда не делиться сокетов ZeroMQ.

Это означает, что никогда не возвращать какой-либо объект, атрибуты которого являются сокетами ZeroMQ (которые вы активно создаете и возвращаете массивным образом из класса .connect(){...}). В вашем случае весь класс -методы, как представляется, сохраняются private, что может сгладить проблему разрешения "других потоков" касаться экземпляров класса-приватного сокета, но тот же принцип должен быть одобрен также для всех атрибутов- уровень, чтобы быть эффективным. Наконец, это "слияние" получает ярлык и нарушается public static SocketManager getInstance(),
который promiscuitively предлагает любому внешнему аферу получить прямой доступ к совместному использованию частных экземпляров сокетов ZeroMQ.

Если в какой-либо документации явным образом предупреждает почти каждую главу о том, чтобы не делиться вещами, скорее не следует делиться вещами.

Итак, переконструируйте методы таким образом, чтобы SocketManager получал больше функциональных возможностей, чем методы класса, которые будут выполнять встроенные функции must-have, чтобы явно запретить любой поток внешнего мира коснитесь экземпляра, не имеющего общего доступа, как описано в публикациях ZeroMQ.

Далее идет инвентаризация ресурсов: ваш код, кажется, пересматривает каждые 30 секунд состояние мира во всех DataCenters-of-Interest. Это фактически создает новые объекты List два раза в минуту. Хотя вы можете спекулировать java Сборщик мусора убирать все трэш, о чем больше не упоминается нигде, это не очень хорошая идея для объектов, связанных с ZeroMQ, встроенных внутри List-s из ваших предыдущих проверок повторной проверки. Объекты ZeroMQ по-прежнему ссылаются изнутри Zcontext() - созданных в ZeroMQ Context() -core- factory потоков ввода-вывода, которые также можно рассматривать как диспетчер ресурсов сокета-инвентаря ZeroMQ. Таким образом, все new -созданные сокеты-экземпляры получают не только внешний дескриптор из java -side, но и внутренний-дескриптор изнутри (Z)Context(). Все идет нормально. Но то, что не видно нигде в коде, - это какой-либо метод, который будет деактивировать любые и все сокеты ZeroMQ в объектных экземплярах, которые были отключены от java -side, но все же остаются ссылкой на (Z)Context() -side. Явное выделение ресурсов из выделенных ресурсов является справедливой практикой на стороне разработки, тем больше ресурсов, которые ограничены или иным образом ограничены. Способ для этого может отличаться для { "cheap" | "дорогостоящие" затраты на такую ​​обработку ресурсов (обработка ZeroMQ-сокетов чрезвычайно выгодна для обработки в виде небольшого "расходного/одноразового"... но это уже другая история).

Итак, добавьте также набор правильных методов повторного использования/ресурсов-демонтажа ресурсов, которые вернут общую сумму new -созданных сокетов обратно под вашу ответственность за контроль (ваш код отвечает за количество сокетов -handlers внутри (Z)Context()-область-ресурсы-контроль может создаваться и должен оставаться управляемым - будь то сознательно или нет).

Можно возразить, что может быть некоторая "promises" из автоматического обнаружения и (возможно, отложенная) сборка мусора, но тем не менее ваш код отвечает за правильное управление ресурсами, и даже ребята из LMAX никогда не получат такую ​​храбрую производительность, если они полагаются на "promises" из стандартного gc. Ваша проблема хуже, чем LMAX с максимальной производительностью. Ваш код (до сих пор опубликованный) ничего не делает для .close() и .term() связанных с ZeroMQ ресурсов. Это - прямая невозможная практика внутри экосистемы с неконтролируемым (распределенным спросом) - потреблением. Вы должны защитить свою лодку от перегрузки за пределы, которые, как вы знаете, могут безопасно обрабатывать и динамически разгружать каждую коробку, у которой нет получателя на "противоположном берегу".

Это капитан (ваш разработчик кода) .

Не указывая явным образом матрос-менеджмент управления запасами на самом низком уровне (ZeroMQ Context() -floor), чтобы некоторые ящики не загружались, проблема по-прежнему остается вашей. Стандартная gc -связка команды не будет делать это "автоматически", независимо от того, что "promises" может выглядеть так, как было бы, это не так. Поэтому будьте откровенны в отношении управления ресурсами ZeroMQ, оцените коды возврата, чтобы упорядочить эти шаги, которые необходимо предпринять, и надлежащим образом обрабатывать любые и все исключения, возникающие при выполнении этих операций управления ресурсами под явным контролем кода.

Нижняя (если вообще не самая низкая достижимая) использование ресурсов - более высокие (если не самые высокие достижимые) производительность - это бонус от выполнения этой работы правильно. Ребята из LMAX - хороший пример того, что это замечательно выходит за рамки стандартного java "promises", поэтому можно учиться на ладбах лучших.

Объявленные сигнатуры подписей, а также используемые, похоже, не соответствуют:
в то время как я могу ошибаться в этом вопросе, так как большинство моих проектных усилий не находятся в java полиморфных интерфейсов вызова, похоже, в подписи есть неправильное совпадение, опубликованное как:

private List<SocketHolder> connect( Datacenters  dc,                     // 1-st
                                    List<String> addresses,              // 2-nd
                                    int          socketType              // 3-rd
                                    ) {
        ... /* implementation */
}

и фактический вызов метода, вызванный внутри метода connectToZMQSockets(), просто:

        List<SocketHolder> addedColoSockets = connect( entry.getValue(), // 1-st
                                                       ZMQ.PUSH          // 2-nd
                                                       );