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);
    }
  }
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.
Distributed requests against the available workers by using one queue per worker . This guarantee the sequence of the operations against the same record cluster .