From 269a1d06ff820c287bb21a03fa76e3314110516a Mon Sep 17 00:00:00 2001
From: Jean-Noel Rouvignac <jean-noel.rouvignac@forgerock.com>
Date: Mon, 22 Apr 2013 10:28:50 +0000
Subject: [PATCH] OPENDJ-832 Leverage the work queue for processing requests received on the HTTP connection handler
---
opends/src/server/org/opends/server/extensions/ParallelWorkQueue.java | 51 +++++++++++++++++++++++++--------------------------
1 files changed, 25 insertions(+), 26 deletions(-)
diff --git a/opends/src/server/org/opends/server/extensions/ParallelWorkQueue.java b/opends/src/server/org/opends/server/extensions/ParallelWorkQueue.java
index 5214b64..eaefb78 100644
--- a/opends/src/server/org/opends/server/extensions/ParallelWorkQueue.java
+++ b/opends/src/server/org/opends/server/extensions/ParallelWorkQueue.java
@@ -29,6 +29,11 @@
+import static org.opends.messages.ConfigMessages.*;
+import static org.opends.messages.CoreMessages.*;
+import static org.opends.server.loggers.ErrorLogger.*;
+import static org.opends.server.loggers.debug.DebugLogger.*;
+
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -52,11 +57,6 @@
import org.opends.server.types.Operation;
import org.opends.server.types.ResultCode;
-import static org.opends.messages.ConfigMessages.*;
-import static org.opends.messages.CoreMessages.*;
-import static org.opends.server.loggers.ErrorLogger.*;
-import static org.opends.server.loggers.debug.DebugLogger.*;
-
/**
@@ -146,7 +146,8 @@
configuration.addParallelChangeListener(this);
// Get the necessary configuration from the provided entry.
- numWorkerThreads = getNumWorkerThreads(configuration);
+ numWorkerThreads =
+ computeNumWorkerThreads(configuration.getNumWorkerThreads());
// Create the actual work queue.
opQueue = new ConcurrentLinkedQueue<Operation>();
@@ -286,6 +287,14 @@
opsSubmitted.incrementAndGet();
}
+ /** {@inheritDoc} */
+ @Override
+ public boolean trySubmitOperation(Operation operation)
+ throws DirectoryException
+ {
+ submitOperation(operation);
+ return true;
+ }
/**
@@ -515,7 +524,8 @@
ParallelWorkQueueCfg configuration)
{
ArrayList<Message> resultMessages = new ArrayList<Message>();
- int newNumThreads = getNumWorkerThreads(configuration);
+ int newNumThreads =
+ computeNumWorkerThreads(configuration.getNumWorkerThreads());
// Apply a change to the number of worker threads if appropriate.
int currentThreads = workerThreads.size();
@@ -583,25 +593,14 @@
}
}
-
-
- // Determine the number of worker threads.
- private int getNumWorkerThreads(ParallelWorkQueueCfg configuration)
+ /**
+ * Return the number of worker threads used by this WorkQueue.
+ *
+ * @return the number of worker threads used by this WorkQueue
+ */
+ @Override
+ public int getNumWorkerThreads()
{
- if (configuration.getNumWorkerThreads() == null)
- {
- // Automatically choose based on the number of processors.
- int cpus = Runtime.getRuntime().availableProcessors();
- int value = Math.max(24, cpus * 2);
-
- Message message = INFO_ERGONOMIC_SIZING_OF_WORKER_THREAD_POOL.get(value);
- logError(message);
-
- return value;
- }
- else
- {
- return configuration.getNumWorkerThreads();
- }
+ return this.numWorkerThreads;
}
}
--
Gitblit v1.10.0