Use the OpenDJ's worker-threads to process HTTP requests.

Yannick Lecaillez

02.01.2016 f0b2b6acbf6762c4a37ad7dad5c41e04b42f37db

Use the OpenDJ's worker-threads to process HTTP requests.

Inject a filter moving the processing of the HTTP requests from the
Grizzly NIO thread to the OpenDJ's worker thread. Previously this move
to worker thread was performed the ConnectionFactory.getConnection()
which is a bit too late since some blocking request processing
(i.e: authorization) were still performed in the NIO thread.

 opendj-server-legacy/src/main/java/org/forgerock/opendj/adapter/server3x/Adapters.java

@@ -17,8 +17,6 @@

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.List;

import org.forgerock.opendj.ldap.AbstractSynchronousConnection;
import org.forgerock.opendj.ldap.ByteString;
@@ -53,19 +51,16 @@
import org.forgerock.opendj.ldap.responses.Responses;
import org.forgerock.opendj.ldap.responses.Result;
import org.forgerock.util.promise.Promise;
import org.forgerock.util.promise.PromiseImpl;
import org.opends.server.core.AddOperation;
import org.opends.server.core.BindOperation;
import org.opends.server.core.CompareOperation;
import org.opends.server.core.DeleteOperation;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.ExtendedOperation;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.protocols.internal.InternalSearchListener;
import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.protocols.internal.Requests;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.OperationType;
import org.opends.server.types.SearchFilter;
import org.opends.server.types.SearchResultEntry;
import org.opends.server.types.SearchResultReference;
@@ -73,6 +68,7 @@
import static org.forgerock.opendj.adapter.server3x.Converters.*;
import static org.forgerock.opendj.ldap.ByteString.*;
import static org.forgerock.opendj.ldap.LdapException.*;
import static org.forgerock.util.promise.Promises.*;

/** This class provides a connection factory and an adapter for the OpenDJ 2.x server. */
public final class Adapters {
@@ -102,42 +98,26 @@
    public static ConnectionFactory newConnectionFactory(final InternalClientConnection icc) {
        return new ConnectionFactory() {
            @Override
            public void close() {
                // Nothing to do.
            }

            @Override
            public Promise<Connection, LdapException> getConnectionAsync() {
                final PromiseImpl<Connection, LdapException> promise = PromiseImpl.create();
                try
                {
                  DirectoryServer.getWorkQueue().submitOperation(new AsyncOperation<>(icc, new Runnable()
                  {
                    @Override
                    public void run()
                    {
                      try
                      {
                        promise.handleResult(getConnection());
                      }
                      catch (LdapException e)
                      {
                        promise.handleException(e);
                      }
                    }
                  }));
                }
                catch (DirectoryException e)
                {
                  promise.handleException(LdapException.newLdapException(e.getResultCode()));
                }
                return promise;
               try
              {
                return newResultPromise(getConnection());
              }
              catch (LdapException e)
              {
                return newExceptionPromise(e);
              }
            }

            @Override
            public Connection getConnection() throws LdapException {
                return newConnection(icc);
            }

            @Override
            public void close() {
                // Nothing to do.
            }
        };
    }

@@ -324,58 +304,4 @@
      };
  }

  /**
   * This operation is hack to be able to execute a {@link Runnable} in a
   * Directory Server's worker thread.
   */
  private static final class AsyncOperation<V> extends org.opends.server.types.AbstractOperation
  {
    private final Runnable runnable;

    AsyncOperation(InternalClientConnection icc, Runnable runnable)
    {
      super(icc, icc.nextOperationID(), icc.nextMessageID(), Collections.<org.opends.server.types.Control> emptyList());
      this.setInternalOperation(true);
      this.runnable = runnable;
    }

    @Override
    public void run()
    {
      runnable.run();
    }

    @Override
    public OperationType getOperationType()
    {
      return null;
    }
    @Override
    public List<org.opends.server.types.Control> getResponseControls()
    {
      return Collections.emptyList();
    }
    @Override
    public void addResponseControl(org.opends.server.types.Control control)
    {
    }
    @Override
    public void removeResponseControl(org.opends.server.types.Control control)
    {
    }
    @Override
    public DN getProxiedAuthorizationDN()
    {
      return null;
    }
    @Override
    public void setProxiedAuthorizationDN(DN proxiedAuthorizationDN)
    {
    }
    @Override
    public void toString(StringBuilder buffer)
    {
      buffer.append(AsyncOperation.class.getSimpleName());
    }
  }
}

 opendj-server-legacy/src/main/java/org/opends/server/protocols/http/HTTPConnectionHandler.java

@@ -25,6 +25,7 @@
import java.net.InetAddress;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
@@ -41,11 +42,15 @@
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;

import org.forgerock.http.Filter;
import org.forgerock.http.Handler;
import org.forgerock.http.HttpApplication;
import org.forgerock.http.HttpApplicationException;
import org.forgerock.http.handler.Handlers;
import org.forgerock.http.io.Buffer;
import org.forgerock.http.protocol.Request;
import org.forgerock.http.protocol.Response;
import org.forgerock.http.protocol.Status;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.config.server.ConfigChangeResult;
@@ -55,6 +60,10 @@
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.server.config.server.ConnectionHandlerCfg;
import org.forgerock.opendj.server.config.server.HTTPConnectionHandlerCfg;
import org.forgerock.services.context.Context;
import org.forgerock.util.promise.NeverThrowsException;
import org.forgerock.util.promise.Promise;
import org.forgerock.util.promise.PromiseImpl;
import org.forgerock.util.time.TimeService;
import org.glassfish.grizzly.http.HttpProbe;
import org.glassfish.grizzly.http.server.HttpServer;
@@ -77,9 +86,12 @@
import org.opends.server.extensions.NullTrustManagerProvider;
import org.opends.server.loggers.HTTPAccessLogger;
import org.opends.server.monitors.ClientConnectionMonitorProvider;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.types.AbstractOperation;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.HostPort;
import org.opends.server.types.InitializationException;
import org.opends.server.types.OperationType;
import org.opends.server.util.DynamicConstants;
import org.opends.server.util.SelectableCertificateKeyManager;
import org.opends.server.util.StaticUtils;
@@ -897,6 +909,7 @@
    {
      return Handlers.chainOf(
          serverContext.getHTTPRouter(),
          new ExecuteInWorkerThreadFilter(),
          new AllowDenyFilter(currentConfig.getDeniedClient(), currentConfig.getAllowedClient()),
          new CommonAuditTransactionIdFilter(serverContext),
          new CommonAuditHttpAccessCheckEnabledFilter(serverContext,
@@ -920,4 +933,91 @@
    }
  }

  /** Moves the processing of the request in this Directory Server's worker thread. */
  private static final class ExecuteInWorkerThreadFilter implements Filter
  {
    @Override
    public Promise<Response, NeverThrowsException> filter(final Context context, final Request request,
        final Handler next)
    {
      final PromiseImpl<Response, NeverThrowsException> promise = PromiseImpl.create();
      try
      {
        DirectoryServer.getWorkQueue().submitOperation(new AsyncOperation<>(
            InternalClientConnection.getRootConnection(),
            new Runnable()
            {
              @Override
              public void run()
              {
                next.handle(context, request).thenOnResultOrException(promise, promise);
              }
            }));
      }
      catch (Exception e)
      {
        promise.handleResult(new Response(Status.INTERNAL_SERVER_ERROR).setCause(e));
      }
      return promise;
    }

    /** This operation is hack to be able to execute a {@link Runnable} in a Directory Server's worker thread. */
    private static final class AsyncOperation<V> extends AbstractOperation
    {
      private final Runnable runnable;

      AsyncOperation(InternalClientConnection icc, Runnable runnable)
      {
        super(icc, icc.nextOperationID(), icc.nextMessageID(),
            Collections.<org.opends.server.types.Control> emptyList());
        this.setInternalOperation(true);
        this.runnable = runnable;
      }

      @Override
      public void run()
      {
        runnable.run();
      }

      @Override
      public OperationType getOperationType()
      {
        return null;
      }

      @Override
      public List<org.opends.server.types.Control> getResponseControls()
      {
        return Collections.emptyList();
      }

      @Override
      public void addResponseControl(org.opends.server.types.Control control)
      {
      }

      @Override
      public void removeResponseControl(org.opends.server.types.Control control)
      {
      }

      @Override
      public DN getProxiedAuthorizationDN()
      {
        return null;
      }

      @Override
      public void setProxiedAuthorizationDN(DN proxiedAuthorizationDN)
      {
      }

      @Override
      public void toString(StringBuilder buffer)
      {
        buffer.append(AsyncOperation.class.getSimpleName());
      }
    }
  }
}