OPENDJ-832 (CR-1545) Leverage the work queue for processing requests receiv...

Jean-Noel Rouvignac

15.08.2013 c96af91cdd11f79c11e56d6c7fe33d0edc653ea3

 opends/src/server/org/opends/server/api/ClientConnection.java

@@ -23,7 +23,7 @@
 *
 *
 *      Copyright 2006-2009 Sun Microsystems, Inc.
 *      Portions copyright 2011-2012 ForgeRock AS
 *      Portions copyright 2011-2013 ForgeRock AS
 */
package org.opends.server.api;

@@ -109,41 +109,47 @@
   */
  protected AtomicBoolean bindOrStartTLSInProgress;

  // Indicates whether any necessary finalization work has been done
  // for this client connection.
  /**
   *  Indicates whether any necessary finalization work has been done for this
   *  client connection.
   */
  private boolean finalized;

  // The set of privileges assigned to this client connection.
  /** The set of privileges assigned to this client connection. */
  private HashSet<Privilege> privileges;

  // The size limit for use with this client connection.
  /** The size limit for use with this client connection. */
  private int sizeLimit;

  // The time limit for use with this client connection.
  /** The time limit for use with this client connection. */
  private int timeLimit;

  // The lookthrough limit for use with this client connection.
  /** The lookthrough limit for use with this client connection. */
  private int lookthroughLimit;

  // The time that this client connection was established.
  /** The time that this client connection was established. */
  private final long connectTime;

  // The idle time limit for this client connection.
  /** The idle time limit for this client connection. */
  private long idleTimeLimit;

  // The opaque information used for storing intermediate state
  // information needed across multi-stage SASL binds.
  /**
   * The opaque information used for storing intermediate state information
   * needed across multi-stage SASL binds.
   */
  private Object saslAuthState;

  // A string representation of the time that this client connection
  // was established.
  /**
   * A string representation of the time that this client connection was
   * established.
   */
  private final String connectTimeString;

  // A set of persistent searches registered for this client.
  /** A set of persistent searches registered for this client. */
  private final CopyOnWriteArrayList<PersistentSearch>
      persistentSearches;

  // The network group to which the connection belongs to.
  /** The network group to which the connection belongs to. */
  private NetworkGroup networkGroup;

  /** Need to evaluate the network group for the first operation. */
@@ -1767,5 +1773,17 @@
  {
    saslBindInProgress.set(false);
  }
}

  /**
   * Returns whether this connection is used for inner work not directly
   * requested by an external client.
   *
   * @return {@code true} if this is an inner connection, {@code false}
   *         otherwise
   */
  public boolean isInnerConnection()
  {
    return getConnectionID() < 0;
  }

}

 opends/src/server/org/opends/server/core/OperationWrapper.java

@@ -33,8 +33,8 @@
import org.opends.messages.Message;
import org.opends.messages.MessageBuilder;
import org.opends.server.api.ClientConnection;
import org.opends.server.types.*;
import org.opends.server.controls.ControlDecoder;
import org.opends.server.types.*;


/**
@@ -355,6 +355,15 @@
   * {@inheritDoc}
   */
  @Override
  public boolean isInnerOperation()
  {
    return operation.isInnerOperation();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isInternalOperation()
  {
    return operation.isInternalOperation();
@@ -454,6 +463,15 @@
   * {@inheritDoc}
   */
  @Override
  public void setInnerOperation(boolean isInnerOperation)
  {
    operation.setInnerOperation(isInnerOperation);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void setInternalOperation(boolean isInternalOperation)
  {
    operation.setInternalOperation(isInternalOperation);

 opends/src/server/org/opends/server/loggers/AbstractTextAccessLogPublisher.java

@@ -962,8 +962,7 @@
    @Override
    public boolean isConnectLoggable(final ClientConnection connection)
    {
      final long connectionID = connection.getConnectionID();
      if (connectionID >= 0 || !suppressInternalOperations)
      if (!connection.isInnerConnection() || !suppressInternalOperations)
      {
        switch (policy)
        {
@@ -989,8 +988,7 @@
    @Override
    public boolean isDisconnectLoggable(final ClientConnection connection)
    {
      final long connectionID = connection.getConnectionID();
      if (connectionID >= 0 || !suppressInternalOperations)
      if (!connection.isInnerConnection() || !suppressInternalOperations)
      {
        switch (policy)
        {
@@ -1071,21 +1069,9 @@
     */
    boolean isLoggable(final Operation operation)
    {
      final long connectionID = operation.getConnectionID();
      if (connectionID < 0)
      {
        // This is an internal operation.
        if (operation.isSynchronizationOperation())
        {
          return !suppressSynchronizationOperations;
        }
        else
        {
          return !suppressInternalOperations;
        }
      }

      return true;
      return !((suppressInternalOperations && operation.isInnerOperation())
          || (suppressSynchronizationOperations
              && operation.isSynchronizationOperation()));
    }
  }


 opends/src/server/org/opends/server/protocols/http/CollectClientConnectionsFilter.java

@@ -28,6 +28,7 @@

import static org.forgerock.opendj.adapter.server2x.Converters.*;
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.loggers.AccessLogger.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.util.StaticUtils.*;
@@ -37,7 +38,6 @@
import java.net.UnknownHostException;
import java.text.ParseException;
import java.util.Collection;
import java.util.Map;

import javax.servlet.FilterChain;
import javax.servlet.FilterConfig;
@@ -61,7 +61,6 @@
import org.forgerock.opendj.rest2ldap.servlet.Rest2LDAPContextFactory;
import org.opends.messages.Message;
import org.opends.server.admin.std.server.ConnectionHandlerCfg;
import org.opends.server.api.ClientConnection;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.schema.SchemaConstants;
import org.opends.server.types.AddressMask;
@@ -120,17 +119,18 @@
  public void doFilter(ServletRequest request, ServletResponse response,
      FilterChain chain)
  {
    final Map<ClientConnection, ClientConnection> clientConnections =
        this.connectionHandler.getClientConnectionsMap();
    final HTTPClientConnection clientConnection =
        new HTTPClientConnection(this.connectionHandler, request);
    clientConnections.put(clientConnection, clientConnection);
    this.connectionHandler.addClientConnection(clientConnection);
    try
    {
      if (!canProcessRequest(request, clientConnection))
      {
        return;
      }
      // logs the connect after all the possible disconnect reasons have been
      // checked.
      logConnect(clientConnection);

      Connection connection = new SdkConnectionAdapter(clientConnection);

@@ -157,6 +157,8 @@
      // The user could not be authenticated. Send an HTTP Basic authentication
      // challenge if HTTP Basic authentication is enabled.
      sendUnauthorizedResponseWithHTTPBasicAuthChallenge(response);
      clientConnection.disconnect(DisconnectReason.INVALID_CREDENTIALS, false,
          null);
    }
    catch (Exception e)
    {
@@ -174,10 +176,6 @@
      clientConnection
          .disconnect(DisconnectReason.SERVER_ERROR, false, message);
    }
    finally
    {
      clientConnections.remove(clientConnection);
    }
  }

  private boolean canProcessRequest(ServletRequest request,
@@ -190,7 +188,8 @@
    // established).
    if (clientConnection.getConnectionID() < 0)
    {
      // The connection will have already been closed.
      clientConnection.disconnect(DisconnectReason.ADMIN_LIMIT_EXCEEDED, true,
          ERR_CONNHANDLER_REJECTED_BY_SERVER.get());
      return false;
    }


 opends/src/server/org/opends/server/protocols/http/HTTPClientConnection.java

@@ -28,6 +28,7 @@

import static org.forgerock.opendj.adapter.server2x.Converters.*;
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.loggers.AccessLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;

import java.net.InetAddress;
@@ -47,7 +48,6 @@
import org.opends.messages.Message;
import org.opends.messages.MessageBuilder;
import org.opends.server.api.ClientConnection;
import org.opends.server.api.ConnectionHandler;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.SearchOperation;
import org.opends.server.loggers.debug.DebugTracer;
@@ -171,11 +171,6 @@
    this.request = request;

    this.connectionID = DirectoryServer.newConnectionAccepted(this);
    if (this.connectionID < 0)
    {
      disconnect(DisconnectReason.ADMIN_LIMIT_EXCEEDED, true,
          ERR_CONNHANDLER_REJECTED_BY_SERVER.get());
    }
  }

  /** {@inheritDoc} */
@@ -187,7 +182,7 @@

  /** {@inheritDoc} */
  @Override
  public ConnectionHandler<?> getConnectionHandler()
  public HTTPConnectionHandler getConnectionHandler()
  {
    return connectionHandler;
  }
@@ -315,8 +310,7 @@
  protected boolean sendIntermediateResponseMessage(
      IntermediateResponse intermediateResponse)
  {
    // TODO Auto-generated method stub
    return false;
    throw new RuntimeException("Not implemented");
  }

  /**
@@ -372,6 +366,10 @@
          .getClosureMessage()));
    }
    finalizeConnectionInternal();


    this.connectionHandler.removeClientConnection(this);
    logDisconnect(this, disconnectReason, message);
  }

  /** {@inheritDoc} */
@@ -475,7 +473,7 @@
            op.operation.abort(cancelRequest);
          }
          catch (Exception e)
          { // make sure all operations are cancelled, no mattter what
          { // make sure all operations are cancelled, no matter what
            if (debugEnabled())
            {
              TRACER.debugCaught(DebugLogLevel.ERROR, e);
@@ -593,4 +591,11 @@
  {
    return connectionValid;
  }

  /** {@inheritDoc} */
  @Override
  public boolean isInnerConnection()
  {
    return true;
  }
}

 opends/src/server/org/opends/server/protocols/http/HTTPConnectionHandler.java

@@ -200,6 +200,17 @@
        && !this.currentConfig.isAuthenticationRequired();
  }

  /**
   * Registers a client connection to track it.
   *
   * @param clientConnection
   *          the client connection to register
   */
  void addClientConnection(ClientConnection clientConnection)
  {
    clientConnections.put(clientConnection, clientConnection);
  }

  /** {@inheritDoc} */
  @Override
  public ConfigChangeResult applyConfigurationChange(
@@ -347,16 +358,6 @@
    return clientConnections.keySet();
  }

  /**
   * Gives access to the clientConnections to classes in this package.
   *
   * @return the Map containing the current client connections
   */
  Map<ClientConnection, ClientConnection> getClientConnectionsMap()
  {
    return clientConnections;
  }

  /** {@inheritDoc} */
  @Override
  public DN getComponentEntryDN()
@@ -644,6 +645,17 @@
    }
  }

  /**
   * Unregisters a client connection to stop tracking it.
   *
   * @param clientConnection
   *          the client connection to unregister
   */
  void removeClientConnection(ClientConnection clientConnection)
  {
    clientConnections.remove(clientConnection);
  }

  /** {@inheritDoc} */
  @Override
  public void run()

 opends/src/server/org/opends/server/protocols/http/SdkConnectionAdapter.java

@@ -69,8 +69,10 @@
import org.opends.server.core.UnbindOperationBasis;
import org.opends.server.core.WorkQueueStrategy;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.types.AuthenticationInfo;
import org.opends.server.types.ByteString;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.types.DisconnectReason;
import org.opends.server.types.Operation;

import com.forgerock.opendj.util.AsynchronousFutureResult;
@@ -125,6 +127,8 @@

    try
    {
      operation.setInnerOperation(this.clientConnection.isInnerConnection());

      // need this raw cast here to fool the compiler's generic type safety
      // Problem here is due to the generic type R on enqueueOperation()
      clientConnection.addOperationInProgress(operation,
@@ -201,14 +205,21 @@
  @Override
  public void close(UnbindRequest request, String reason)
  {
    final int messageID = nextMessageID.get();
    UnbindOperationBasis operation =
        new UnbindOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()));
    AuthenticationInfo authInfo = this.clientConnection.getAuthenticationInfo();
    if (authInfo != null && authInfo.isAuthenticated())
    {
      final int messageID = nextMessageID.get();
      UnbindOperationBasis operation =
          new UnbindOperationBasis(clientConnection, messageID, messageID,
              to(request.getControls()));

    // run synchronous
    operation.run();

      // run synchronous
      operation.run();
    }
    else
    {
      this.clientConnection.disconnect(DisconnectReason.UNBIND, false, null);
    }
    isClosed = true;
  }


 opends/src/server/org/opends/server/protocols/ldap/LDAPClientConnection.java

@@ -519,11 +519,6 @@
    }

    connectionID = DirectoryServer.newConnectionAccepted(this);
    if (connectionID < 0)
    {
      disconnect(DisconnectReason.ADMIN_LIMIT_EXCEEDED, true,
          ERR_CONNHANDLER_REJECTED_BY_SERVER.get());
    }
  }

  /**

 opends/src/server/org/opends/server/protocols/ldap/LDAPConnectionHandler.java

@@ -1239,7 +1239,8 @@
        clientChannel, getProtocol());
    if (clientConnection.getConnectionID() < 0)
    {
      // The connection will have already been closed.
      clientConnection.disconnect(DisconnectReason.ADMIN_LIMIT_EXCEEDED, true,
          ERR_CONNHANDLER_REJECTED_BY_SERVER.get());
      return;
    }


 opends/src/server/org/opends/server/types/AbstractOperation.java

@@ -116,6 +116,7 @@
   * itself rather than requested by an external client.
   */
  private boolean isInternalOperation;
  private Boolean isInnerOperation;

  /**
   * Indicates whether this operation is involved in data synchronization
@@ -621,65 +622,47 @@



  /**
   * Indicates whether this is an internal operation rather than one
   * that was requested by an external client.
   *
   * @return  {@code true} if this is an internal operation, or
   *          {@code false} if it is not.
   */
  /** {@inheritDoc} */
  @Override
  public final boolean isInternalOperation()
  {
    return isInternalOperation;
  }



  /**
   * Specifies whether this is an internal operation rather than one
   * that was requested by an external client.  This may not be called
   * from within a plugin.
   *
   * @param  isInternalOperation  Specifies whether this is an
   *                              internal operation rather than one
   *                              that was requested by an external
   *                              client.
   */
  /** {@inheritDoc} */
  @Override
  public final void setInternalOperation(boolean isInternalOperation)
  {
    this.isInternalOperation = isInternalOperation;
  }

  /** {@inheritDoc} */
  @Override
  public boolean isInnerOperation()
  {
    if (this.isInnerOperation != null)
    {
      return this.isInnerOperation;
    }
    return isInternalOperation();
  }

  /** {@inheritDoc} */
  @Override
  public void setInnerOperation(boolean isInnerOperation)
  {
    this.isInnerOperation = isInnerOperation;
  }


  /**
   * Indicates whether this is a synchronization operation rather than
   * one that was requested by an external client.
   *
   * @return  {@code true} if this is a data synchronization
   *          operation, or {@code false} if it is not.
   */
  /** {@inheritDoc} */
  @Override
  public final boolean isSynchronizationOperation()
  {
    return isSynchronizationOperation;
  }



  /**
   * Specifies whether this is a synchronization operation rather than
   * one that was requested by an external client.  This method may
   * not be called from within a plugin.
   *
   * @param  isSynchronizationOperation  Specifies whether this is a
   *                                     synchronization operation
   *                                     rather than one that was
   *                                     requested by an external
   *                                     client.
   */
  /** {@inheritDoc} */
  @Override
  public final void setSynchronizationOperation(
                         boolean isSynchronizationOperation)

 opends/src/server/org/opends/server/types/Operation.java

@@ -367,6 +367,31 @@
      isInternalOperation);

  /**
   * Indicates whether this is an inner operation rather than one that was
   * directly requested by an external client. Said otherwise, inner operations
   * include internal operations, but also operations in the server indirectly
   * mandated by external requests like Rest2LDAP for example. This may not be
   * called from within a plugin.
   *
   * @return {@code true} if this is an inner operation, or {@code false} if it
   *         is not.
   */
  boolean isInnerOperation();

  /**
   * Specifies whether this is an inner operation rather than one that was
   * directly requested by an external client. Said otherwise, inner operations
   * include internal operations, but also operations in the server indirectly
   * mandated by external requests like Rest2LDAP for example. This may not be
   * called from within a plugin.
   *
   * @param isInnerOperation
   *          Specifies whether this is an inner operation rather than one that
   *          was requested by an external client.
   */
  void setInnerOperation(boolean isInnerOperation);

  /**
   * Indicates whether this is a synchronization operation rather than
   * one that was requested by an external client.
   *

 opends/tests/unit-tests-testng/src/server/org/opends/server/loggers/AbstractTextAccessLogPublisherTest.java

@@ -42,40 +42,42 @@
  @DataProvider(name = "isLoggableData")
  public Object[][] getIsLoggableData()
  {
    // when suppress is set to true and the corresponding operation is set to
    // true too, then the operation is not loggable.
    // You can read the array like this: read two by two from line start, if
    // both are true in a pair, then the expected result is false (not loggable)
    return new Object[][] {
      { 1L, false, false, false, false, true },
      { -1L, true, true, true, true, false },
      { -1L, true, true, true, false, false },
      { -1L, true, true, false, true, false },
      { -1L, true, true, false, false, false },
      { -1L, true, false, true, true, false },
      { -1L, true, false, true, false, false },// this will change
      { -1L, true, false, false, true, true },
      { -1L, true, false, false, false, true },
      { -1L, false, true, true, true, true },// this will change
      { -1L, false, true, true, false, true },
      { -1L, false, true, false, true, true },
      { -1L, false, true, false, false, true },
      { -1L, false, false, true, true, false },
      { -1L, false, false, true, false, false },// this will change
      { -1L, false, false, false, true, true },
      { -1L, false, false, false, false, true }, };
      { true, true, true, true, false },
      { true, true, true, false, false },
      { true, true, false, true, false },
      { true, true, false, false, false },
      { true, false, true, true, false },
      { true, false, true, false, true },
      { true, false, false, true, true },
      { true, false, false, false, true },
      { false, true, true, true, false },
      { false, true, true, false, true },
      { false, true, false, true, true },
      { false, true, false, false, true },
      { false, false, true, true, false },
      { false, false, true, false, true },
      { false, false, false, true, true },
      { false, false, false, false, true }, };
  }

  @Test(dataProvider = "isLoggableData")
  public void rootFilterIsLoggable(long connectionID,
      boolean suppressSynchronization, boolean isSynchronizationOp,
      boolean suppressInternal, boolean isInternalOp, boolean testResult)
  public void rootFilterIsLoggable(boolean suppressSynchronization,
      boolean isSynchronizationOp, boolean suppressInternal,
      boolean isInternalOp, boolean expectedTestResult)
  {
    final Operation operation = mock(Operation.class);
    when(operation.getConnectionID()).thenReturn(connectionID);
    when(operation.isSynchronizationOperation())
        .thenReturn(isSynchronizationOp);
    when(operation.isInternalOperation()).thenReturn(isInternalOp);
    when(operation.isInnerOperation()).thenReturn(isInternalOp);

    final RootFilter filter =
        new RootFilter(suppressInternal, suppressSynchronization, null, null);
    assertEquals(filter.isLoggable(operation), testResult);
    assertEquals(filter.isLoggable(operation), expectedTestResult);
  }

}