distributed/src/main/java/com/orientechnologies/orient/server/distributed/impl/ODistributedDatabaseImpl.java
public synchronized 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);
        CyclicBarrier started = new CyclicBarrier(involvedWorkerQueues.size());
        CyclicBarrier finished = new CyclicBarrier(involvedWorkerQueues.size());
        // WAIT ALL THE INVOLVED QUEUES ARE FREE AND SYNCHRONIZED
        for (int queue : involvedWorkerQueues) {
          ODistributedWorker worker = workerThreads.get(queue);
          OWaitPartitionsReadyTask waitRequest = new OWaitPartitionsReadyTask(started, task, finished);

          final ODistributedRequest syncRequest = new ODistributedRequest(null, request.getId().getNodeId(),
              request.getId().getMessageId(), databaseName, waitRequest);
          worker.processRequest(syncRequest);
        }

      }
    } 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 synchronized 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);
        CyclicBarrier started = new CyclicBarrier(involvedWorkerQueues.size());
        CyclicBarrier finished = new CyclicBarrier(involvedWorkerQueues.size());
        // WAIT ALL THE INVOLVED QUEUES ARE FREE AND SYNCHRONIZED
        for (int queue : involvedWorkerQueues) {
          ODistributedWorker worker = workerThreads.get(queue);
          OWaitPartitionsReadyTask waitRequest = new OWaitPartitionsReadyTask(started, task, finished);

          final ODistributedRequest syncRequest = new ODistributedRequest(null, request.getId().getNodeId(),
              request.getId().getMessageId(), databaseName, waitRequest);
          worker.processRequest(syncRequest);
        }

      }
    } 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 synchronized 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 ) ; CyclicBarrier started = new CyclicBarrier ( involvedWorkerQueues . size ( ) ) ; CyclicBarrier finished = new CyclicBarrier ( involvedWorkerQueues . size ( ) ) ; // WAIT ALL THE INVOLVED QUEUES ARE FREE AND SYNCHRONIZED for ( int queue : involvedWorkerQueues ) { ODistributedWorker worker = workerThreads . get ( queue ) ; OWaitPartitionsReadyTask waitRequest = new OWaitPartitionsReadyTask ( started , task , finished ) ; final ODistributedRequest syncRequest = new ODistributedRequest ( null , request . getId ( ) . getNodeId ( ) , request . getId ( ) . getMessageId ( ) , databaseName , waitRequest ) ; worker . processRequest ( syncRequest ) ; } } } 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 .