distributed-v2/src/main/java/com/orientechnologies/orient/distributed/impl/ODistributedDatabaseImpl.java
public void processRequest(final ODistributedRequest request, final boolean waitForAcceptingRequests) {
    if (!running) {
      throw new ODistributedException("Server is going down or is removing the database:'" + getDatabaseName() + "' discarding");
    }

    final ORemoteTask task = request.getTask();

    if (waitForAcceptingRequests) {
      waitIsReady(task);

      if (!running) {
        throw new ODistributedException("Server is going down or is removing the database:'" + getDatabaseName() + "' discarding");
      }
    }

    totalReceivedRequests.incrementAndGet();

    // final ODistributedMomentum lastMomentum = filterByMomentum.get();
    // if (lastMomentum != null && task instanceof OAbstractReplicatedTask) {
    // final OLogSequenceNumber taskLastLSN = ((OAbstractReplicatedTask) task).getLastLSN();
    //
    // final String sourceServer = manager.getNodeNameById(request.getId().getNodeId());
    // final OLogSequenceNumber lastLSNFromMomentum = lastMomentum.getLSN(sourceServer);
    //
    // if (taskLastLSN != null && lastLSNFromMomentum != null && taskLastLSN.compareTo(lastLSNFromMomentum) < 0) {
    // // SKIP REQUEST BECAUSE CONTAINS AN OLD LSN
    // final String msg = String.format("Skipped request %s on database '%s' because %s < current %s", request, databaseName,
    // taskLastLSN, lastLSNFromMomentum);
    // ODistributedServerLog.info(this, localNodeName, null, DIRECTION.NONE, msg);
    // ODistributedWorker.sendResponseBack(this, manager, request, new ODistributedException(msg));
    // return;
    // }
    // }

    final int[] partitionKeys = task.getPartitionKey();

    if (ODistributedServerLog.isDebugEnabled())
      ODistributedServerLog
          .debug(this, localNodeName, task.getNodeSource(), DIRECTION.IN, "Request %s on database '%s' partitionKeys=%s task=%s",
              request, databaseName, Arrays.toString(partitionKeys), task);

    if (partitionKeys.length > 1 || partitionKeys[0] == -1) {

      final Set<Integer> involvedWorkerQueues;
      if (partitionKeys.length > 1)
        involvedWorkerQueues = getInvolvedQueuesByPartitionKeys(partitionKeys);
      else
        // LOCK ALL THE QUEUES
        involvedWorkerQueues = ALL_QUEUES;

      // if (ODistributedServerLog.isDebugEnabled())
      ODistributedServerLog
          .debug(this, localNodeName, null, DIRECTION.NONE, "Request %s on database '%s' involvedQueues=%s", request, databaseName,
              involvedWorkerQueues);

      if (involvedWorkerQueues.size() == 1)
        // JUST ONE QUEUE INVOLVED: PROCESS IT IMMEDIATELY
        processRequest(involvedWorkerQueues.iterator().next(), request);
      else {
        // INVOLVING MULTIPLE QUEUES

        // if (ODistributedServerLog.isDebugEnabled())
        ODistributedServerLog.debug(this, localNodeName, null, DIRECTION.NONE,
            "Request %s on database '%s' waiting for all the previous requests to be completed", request, databaseName);

        // WAIT ALL THE INVOLVED QUEUES ARE FREE AND SYNCHRONIZED
        final CountDownLatch syncLatch = new CountDownLatch(involvedWorkerQueues.size());
        final ODistributedRequest syncRequest = new ODistributedRequest(null, request.getId().getNodeId(), -1, databaseName,
            new OSynchronizedTaskWrapper(syncLatch));
        for (int queue : involvedWorkerQueues) {
          ODistributedWorker worker = workerThreads.get(queue);
          worker.processRequest(syncRequest);
        }

        // Make infinite timeout everytime
        long taskTimeout = 0;
        try {
          if (taskTimeout <= 0)
            syncLatch.await();
          else {
            // WAIT FOR COMPLETION. THE TIMEOUT IS MANAGED IN SMALLER CYCLES TO PROPERLY RECOGNIZE WHEN THE DB IS REMOVED
            final long start = System.currentTimeMillis();
            final long cycleTimeout = Math.min(taskTimeout, 2000);

            boolean locked = false;
            do {
              if (syncLatch.await(cycleTimeout, TimeUnit.MILLISECONDS)) {
                // DONE
                locked = true;
                break;
              }

              if (this.workerThreads.size() == 0)
                // DATABASE WAS SHUTDOWN
                break;

            } while (System.currentTimeMillis() - start < taskTimeout);

            if (!locked) {
              final String msg = String.format(
                  "Cannot execute distributed request (%s) because all worker threads (%d) are busy (pending=%d timeout=%d)",
                  request, workerThreads.size(), syncLatch.getCount(), taskTimeout);
              ODistributedWorker.sendResponseBack(this, manager, request, new ODistributedOperationException(msg));
              return;
            }
          }
        } catch (InterruptedException e) {
          // IGNORE
          Thread.currentThread().interrupt();
          final String msg = String
              .format("Cannot execute distributed request (%s) because all worker threads (%d) are busy", request,
                  workerThreads.size());
          ODistributedWorker.sendResponseBack(this, manager, request, new ODistributedOperationException(msg));
          return;
        }

        // PUT THE TASK TO EXECUTE ONLY IN THE FIRST QUEUE AND PUT WAIT-FOR TASKS IN THE OTHERS. WHEN THE REAL TASK IS EXECUTED,
        // ALL THE OTHER TASKS WILL RETURN, SO THE QUEUES WILL BE BUSY DURING THE EXECUTION OF THE TASK. THIS AVOID CONCURRENT
        // EXECUTION FOR THE SAME PARTITION
        final CountDownLatch queueLatch = new CountDownLatch(1);

        int i = 0;
        for (int queue : involvedWorkerQueues) {
          final ODistributedRequest req;
          if (i++ == 0) {
            // USE THE FIRST QUEUE TO PROCESS THE REQUEST
            final String senderNodeName = manager.getNodeNameById(request.getId().getNodeId());
            request.setTask(new OSynchronizedTaskWrapper(queueLatch, senderNodeName, task));
            req = request;
          } else
            req = new ODistributedRequest(manager, request.getId().getNodeId(), -1, databaseName, new OWaitForTask(queueLatch));

          workerThreads.get(queue).processRequest(req);
        }
      }
    } else if (partitionKeys.length == 1 && partitionKeys[0] == -2) {
      // ANY PARTITION: USE THE FIRST EMPTY IF ANY, OTHERWISE THE FIRST IN THE LIST
      boolean found = false;

      for (ODistributedWorker q : workerThreads) {
        if (q.isWaitingForNextRequest() && q.localQueue.isEmpty()) {
          q.processRequest(request);
          found = true;
          break;
        }
      }

      if (!found)
        // ALL THE THREADS ARE BUSY, SELECT THE FIRST EMPTY ONE
        for (ODistributedWorker q : workerThreads) {
          if (q.localQueue.isEmpty()) {
            q.processRequest(request);
            found = true;
            break;
          }
        }

      if (!found)
        // EXEC ON THE FIRST QUEUE
        workerThreads.get(0).processRequest(request);

    } else if (partitionKeys.length == 1 && partitionKeys[0] == -3) {
      // SERVICE - LOCK
      ODistributedServerLog.debug(this, localNodeName, request.getTask().getNodeSource(), DIRECTION.IN,
          "Request %s on database '%s' dispatched to the lock worker", request, databaseName);

      lockThread.processRequest(request);

    } else if (partitionKeys.length == 1 && partitionKeys[0] == -4) {
      // SERVICE - FAST_NOLOCK
      ODistributedServerLog.debug(this, localNodeName, request.getTask().getNodeSource(), DIRECTION.IN,
          "Request %s on database '%s' dispatched to the nowait worker", request, databaseName);

      nowaitThread.processRequest(request);

    } else {
      processRequest(partitionKeys[0], request);
    }
  }
public void processRequest(final ODistributedRequest request, final boolean waitForAcceptingRequests) {
    if (!running) {
      throw new ODistributedException("Server is going down or is removing the database:'" + getDatabaseName() + "' discarding");
    }

    final ORemoteTask task = request.getTask();

    if (waitForAcceptingRequests) {
      waitIsReady(task);

      if (!running) {
        throw new ODistributedException("Server is going down or is removing the database:'" + getDatabaseName() + "' discarding");
      }
    }

    totalReceivedRequests.incrementAndGet();

    // final ODistributedMomentum lastMomentum = filterByMomentum.get();
    // if (lastMomentum != null && task instanceof OAbstractReplicatedTask) {
    // final OLogSequenceNumber taskLastLSN = ((OAbstractReplicatedTask) task).getLastLSN();
    //
    // final String sourceServer = manager.getNodeNameById(request.getId().getNodeId());
    // final OLogSequenceNumber lastLSNFromMomentum = lastMomentum.getLSN(sourceServer);
    //
    // if (taskLastLSN != null && lastLSNFromMomentum != null && taskLastLSN.compareTo(lastLSNFromMomentum) < 0) {
    // // SKIP REQUEST BECAUSE CONTAINS AN OLD LSN
    // final String msg = String.format("Skipped request %s on database '%s' because %s < current %s", request, databaseName,
    // taskLastLSN, lastLSNFromMomentum);
    // ODistributedServerLog.info(this, localNodeName, null, DIRECTION.NONE, msg);
    // ODistributedWorker.sendResponseBack(this, manager, request, new ODistributedException(msg));
    // return;
    // }
    // }

    final int[] partitionKeys = task.getPartitionKey();

    if (ODistributedServerLog.isDebugEnabled())
      ODistributedServerLog
          .debug(this, localNodeName, task.getNodeSource(), DIRECTION.IN, "Request %s on database '%s' partitionKeys=%s task=%s",
              request, databaseName, Arrays.toString(partitionKeys), task);

    if (partitionKeys.length > 1 || partitionKeys[0] == -1) {

      final Set<Integer> involvedWorkerQueues;
      if (partitionKeys.length > 1)
        involvedWorkerQueues = getInvolvedQueuesByPartitionKeys(partitionKeys);
      else
        // LOCK ALL THE QUEUES
        involvedWorkerQueues = ALL_QUEUES;

      // if (ODistributedServerLog.isDebugEnabled())
      ODistributedServerLog
          .debug(this, localNodeName, null, DIRECTION.NONE, "Request %s on database '%s' involvedQueues=%s", request, databaseName,
              involvedWorkerQueues);

      if (involvedWorkerQueues.size() == 1)
        // JUST ONE QUEUE INVOLVED: PROCESS IT IMMEDIATELY
        processRequest(involvedWorkerQueues.iterator().next(), request);
      else {
        // INVOLVING MULTIPLE QUEUES

        // if (ODistributedServerLog.isDebugEnabled())
        ODistributedServerLog.debug(this, localNodeName, null, DIRECTION.NONE,
            "Request %s on database '%s' waiting for all the previous requests to be completed", request, databaseName);

        // WAIT ALL THE INVOLVED QUEUES ARE FREE AND SYNCHRONIZED
        final CountDownLatch syncLatch = new CountDownLatch(involvedWorkerQueues.size());
        final ODistributedRequest syncRequest = new ODistributedRequest(null, request.getId().getNodeId(), -1, databaseName,
            new OSynchronizedTaskWrapper(syncLatch));
        for (int queue : involvedWorkerQueues) {
          ODistributedWorker worker = workerThreads.get(queue);
          worker.processRequest(syncRequest);
        }

        // Make infinite timeout everytime
        long taskTimeout = 0;
        try {
          if (taskTimeout <= 0)
            syncLatch.await();
          else {
            // WAIT FOR COMPLETION. THE TIMEOUT IS MANAGED IN SMALLER CYCLES TO PROPERLY RECOGNIZE WHEN THE DB IS REMOVED
            final long start = System.currentTimeMillis();
            final long cycleTimeout = Math.min(taskTimeout, 2000);

            boolean locked = false;
            do {
              if (syncLatch.await(cycleTimeout, TimeUnit.MILLISECONDS)) {
                // DONE
                locked = true;
                break;
              }

              if (this.workerThreads.size() == 0)
                // DATABASE WAS SHUTDOWN
                break;

            } while (System.currentTimeMillis() - start < taskTimeout);

            if (!locked) {
              final String msg = String.format(
                  "Cannot execute distributed request (%s) because all worker threads (%d) are busy (pending=%d timeout=%d)",
                  request, workerThreads.size(), syncLatch.getCount(), taskTimeout);
              ODistributedWorker.sendResponseBack(this, manager, request, new ODistributedOperationException(msg));
              return;
            }
          }
        } catch (InterruptedException e) {
          // IGNORE
          Thread.currentThread().interrupt();
          final String msg = String
              .format("Cannot execute distributed request (%s) because all worker threads (%d) are busy", request,
                  workerThreads.size());
          ODistributedWorker.sendResponseBack(this, manager, request, new ODistributedOperationException(msg));
          return;
        }

        // PUT THE TASK TO EXECUTE ONLY IN THE FIRST QUEUE AND PUT WAIT-FOR TASKS IN THE OTHERS. WHEN THE REAL TASK IS EXECUTED,
        // ALL THE OTHER TASKS WILL RETURN, SO THE QUEUES WILL BE BUSY DURING THE EXECUTION OF THE TASK. THIS AVOID CONCURRENT
        // EXECUTION FOR THE SAME PARTITION
        final CountDownLatch queueLatch = new CountDownLatch(1);

        int i = 0;
        for (int queue : involvedWorkerQueues) {
          final ODistributedRequest req;
          if (i++ == 0) {
            // USE THE FIRST QUEUE TO PROCESS THE REQUEST
            final String senderNodeName = manager.getNodeNameById(request.getId().getNodeId());
            request.setTask(new OSynchronizedTaskWrapper(queueLatch, senderNodeName, task));
            req = request;
          } else
            req = new ODistributedRequest(manager, request.getId().getNodeId(), -1, databaseName, new OWaitForTask(queueLatch));

          workerThreads.get(queue).processRequest(req);
        }
      }
    } else if (partitionKeys.length == 1 && partitionKeys[0] == -2) {
      // ANY PARTITION: USE THE FIRST EMPTY IF ANY, OTHERWISE THE FIRST IN THE LIST
      boolean found = false;

      for (ODistributedWorker q : workerThreads) {
        if (q.isWaitingForNextRequest() && q.localQueue.isEmpty()) {
          q.processRequest(request);
          found = true;
          break;
        }
      }

      if (!found)
        // ALL THE THREADS ARE BUSY, SELECT THE FIRST EMPTY ONE
        for (ODistributedWorker q : workerThreads) {
          if (q.localQueue.isEmpty()) {
            q.processRequest(request);
            found = true;
            break;
          }
        }

      if (!found)
        // EXEC ON THE FIRST QUEUE
        workerThreads.get(0).processRequest(request);

    } else if (partitionKeys.length == 1 && partitionKeys[0] == -3) {
      // SERVICE - LOCK
      ODistributedServerLog.debug(this, localNodeName, request.getTask().getNodeSource(), DIRECTION.IN,
          "Request %s on database '%s' dispatched to the lock worker", request, databaseName);

      lockThread.processRequest(request);

    } else if (partitionKeys.length == 1 && partitionKeys[0] == -4) {
      // SERVICE - FAST_NOLOCK
      ODistributedServerLog.debug(this, localNodeName, request.getTask().getNodeSource(), DIRECTION.IN,
          "Request %s on database '%s' dispatched to the nowait worker", request, databaseName);

      nowaitThread.processRequest(request);

    } else {
      processRequest(partitionKeys[0], request);
    }
  }
Distributed requests against the available workers by using one queue per worker. This guarantee the sequence of the operations
against the same record cluster.