cxf-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dk...@apache.org
Subject svn commit: r1383103 - in /cxf/sandbox/dkulp_async_clients/http-hc/src: main/java/org/apache/cxf/transport/http/asyncclient/ test/java/org/apache/cxf/transport/http/asyncclient/ test/resources/
Date Mon, 10 Sep 2012 20:38:54 GMT
Author: dkulp
Date: Mon Sep 10 20:38:54 2012
New Revision: 1383103

URL: http://svn.apache.org/viewvc?rev=1383103&view=rev
Log:
Updates to reduce some byte[] copying into buffers if a large buffer is to be written.
Update some default socket buffer sizes.

Removed:
    cxf/sandbox/dkulp_async_clients/http-hc/src/test/resources/log4j.properties
Modified:
    cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java
    cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitFactory.java
    cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/CXFAsyncRequester.java
    cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/SharedOutputBuffer.java
    cxf/sandbox/dkulp_async_clients/http-hc/src/test/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitTest.java

Modified: cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java
URL: http://svn.apache.org/viewvc/cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java?rev=1383103&r1=1383102&r2=1383103&view=diff
==============================================================================
--- cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java
(original)
+++ cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java
Mon Sep 10 20:38:54 2012
@@ -128,10 +128,6 @@ public class AsyncHTTPConduit extends UR
             //o = "http".equals(s);
             o = Boolean.TRUE;
         }
-        if ("https".equals(s)) {
-            //https doesn't work yet
-            //o = Boolean.FALSE;
-        }
         if (!MessageUtils.isTrue(o)) {
             message.put(USE_ASYNC, Boolean.FALSE);
             super.setupConnection(message, uri, csPolicy);
@@ -223,8 +219,8 @@ public class AsyncHTTPConduit extends UR
             entity = message.get(CXFHttpRequest.class);
             basicEntity = (BasicHttpEntity)entity.getEntity();
             HeapByteBufferAllocator allocator = new HeapByteBufferAllocator();
-            inbuf = new SharedInputBuffer(4096 * 4, allocator);
-            outbuf = new SharedOutputBuffer(4096 * 4, allocator);
+            inbuf = new SharedInputBuffer(16320, allocator);
+            outbuf = new SharedOutputBuffer(16320, allocator);
         }
         
         protected void setProtocolHeaders() throws IOException {
@@ -587,8 +583,8 @@ public class AsyncHTTPConduit extends UR
             
             //reset the buffers
             HeapByteBufferAllocator allocator = new HeapByteBufferAllocator();
-            inbuf = new SharedInputBuffer(4096 * 4, allocator);
-            outbuf = new SharedOutputBuffer(4096 * 4, allocator);
+            inbuf = new SharedInputBuffer(16320, allocator);
+            outbuf = new SharedOutputBuffer(16320, allocator);
             try {
                 this.url = new URI(newURL);
                 setupConnection(outMessage, this.url, csPolicy);

Modified: cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitFactory.java
URL: http://svn.apache.org/viewvc/cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitFactory.java?rev=1383103&r1=1383102&r2=1383103&view=diff
==============================================================================
--- cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitFactory.java
(original)
+++ cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitFactory.java
Mon Sep 10 20:38:54 2012
@@ -55,7 +55,6 @@ import org.apache.http.nio.reactor.ssl.S
 import org.apache.http.nio.util.ByteBufferAllocator;
 import org.apache.http.nio.util.HeapByteBufferAllocator;
 import org.apache.http.params.BasicHttpParams;
-import org.apache.http.params.CoreConnectionPNames;
 import org.apache.http.params.HttpParams;
 
 /**
@@ -124,16 +123,12 @@ public class AsyncHTTPConduitFactory imp
         if (requester != null) {
             return;
         }
-     // HTTP parameters for the client
-        HttpParams params = new BasicHttpParams();
-        params.setIntParameter(CoreConnectionPNames.SOCKET_BUFFER_SIZE, 16 * 1024);
-       
         // Create client-side I/O reactor
         IOReactorConfig config = new IOReactorConfig();
         config.setTcpNoDelay(true);
         
         final IOEventDispatch ioEventDispatch = new DefaultHttpClientIODispatch(new HttpAsyncRequestExecutor(),
-                                                                                params);
+                                                                                new BasicHttpParams());
         ioReactor = new DefaultConnectingIOReactor(config);
 
         // Run the I/O reactor in a separate thread
@@ -191,7 +186,7 @@ public class AsyncHTTPConduitFactory imp
             }
             
         };
-        connectionManager.setDefaultMaxPerRoute(1000);
+        connectionManager.setDefaultMaxPerRoute(2500);
         connectionManager.setMaxTotal(5000);
         requester = new CXFAsyncRequester(connectionManager);
     }

Modified: cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/CXFAsyncRequester.java
URL: http://svn.apache.org/viewvc/cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/CXFAsyncRequester.java?rev=1383103&r1=1383102&r2=1383103&view=diff
==============================================================================
--- cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/CXFAsyncRequester.java
(original)
+++ cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/CXFAsyncRequester.java
Mon Sep 10 20:38:54 2012
@@ -28,6 +28,7 @@ import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLException;
 import javax.net.ssl.SSLSession;
 
+import org.apache.http.HttpVersion;
 import org.apache.http.client.protocol.ClientContext;
 import org.apache.http.concurrent.BasicFuture;
 import org.apache.http.concurrent.FutureCallback;
@@ -39,6 +40,10 @@ import org.apache.http.nio.conn.ssl.SSLL
 import org.apache.http.nio.protocol.HttpAsyncRequestProducer;
 import org.apache.http.nio.protocol.HttpAsyncResponseConsumer;
 import org.apache.http.nio.reactor.IOSession;
+import org.apache.http.params.HttpConnectionParams;
+import org.apache.http.params.HttpParams;
+import org.apache.http.params.HttpProtocolParams;
+import org.apache.http.params.SyncBasicHttpParams;
 import org.apache.http.protocol.HttpContext;
 
 public class CXFAsyncRequester {
@@ -97,7 +102,16 @@ public class CXFAsyncRequester {
                 e.printStackTrace();
             }
         }
-        DefaultHttpAsyncClient dhac = new DefaultHttpAsyncClient(caConMan);
+        DefaultHttpAsyncClient dhac = new DefaultHttpAsyncClient(caConMan) {
+            @Override
+            protected HttpParams createHttpParams() {
+                HttpParams params = new SyncBasicHttpParams();
+                HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1);
+                HttpConnectionParams.setTcpNoDelay(params, true);
+                HttpConnectionParams.setSocketBufferSize(params, 16332);
+                return params;
+            }
+        };
         context.setAttribute(ClientContext.SCHEME_REGISTRY, reg);
         dhac.execute(requestProducer, responseConsumer, context, callback);
 

Modified: cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/SharedOutputBuffer.java
URL: http://svn.apache.org/viewvc/cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/SharedOutputBuffer.java?rev=1383103&r1=1383102&r2=1383103&view=diff
==============================================================================
--- cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/SharedOutputBuffer.java
(original)
+++ cxf/sandbox/dkulp_async_clients/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/SharedOutputBuffer.java
Mon Sep 10 20:38:54 2012
@@ -21,6 +21,7 @@ package org.apache.cxf.transport.http.as
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.nio.ByteBuffer;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -52,6 +53,8 @@ public class SharedOutputBuffer extends 
     private volatile IOControl ioctrl;
     private volatile boolean shutdown;
     private volatile boolean endOfStream;
+    
+    private volatile ByteBuffer largeWrapper;
 
     public SharedOutputBuffer(int buffersize, final ByteBufferAllocator allocator) {
         super(buffersize, allocator);
@@ -121,13 +124,17 @@ public class SharedOutputBuffer extends 
             this.ioctrl = ioc;
             setOutputMode();
             int bytesWritten = 0;
-            if (super.hasData()) {
-                bytesWritten = encoder.write(this.buffer);
+            if (largeWrapper != null || super.hasData()) {
+                if (!this.buffer.hasRemaining() && largeWrapper != null) {
+                    bytesWritten = encoder.write(largeWrapper);
+                } else {
+                    bytesWritten = encoder.write(this.buffer);
+                }
                 if (encoder.isCompleted()) {
                     this.endOfStream = true;
                 }
             }
-            if (!super.hasData()) {
+            if ((largeWrapper == null || !largeWrapper.hasRemaining()) && !super.hasData())
{
                 // No more buffered content
                 // If at the end of the stream, terminate
                 if (this.endOfStream && !encoder.isCompleted()) {
@@ -138,7 +145,10 @@ public class SharedOutputBuffer extends 
                     this.ioctrl.suspendOutput();
                 }
             }
-            this.condition.signalAll();
+            // no need to signal if the large wrapper is present and has data remaining 
+            if (largeWrapper == null || !largeWrapper.hasRemaining()) {
+                this.condition.signalAll();
+            }
             return bytesWritten;
         } finally {
             this.lock.unlock();
@@ -178,10 +188,19 @@ public class SharedOutputBuffer extends 
                     flushContent();
                     setInputMode();
                 }
-                int chunk = Math.min(remaining, this.buffer.remaining());
-                this.buffer.put(b, off, chunk);
-                remaining -= chunk;
-                off += chunk;
+                if (buffer.position() == 0 && (this.buffer.remaining() * 2) <
remaining) {
+                    largeWrapper = ByteBuffer.wrap(b, off, remaining);
+                    while (largeWrapper.hasRemaining()) {
+                        flushContent();
+                    }
+                    largeWrapper = null;
+                    remaining = 0;
+                } else {
+                    int chunk = Math.min(remaining, this.buffer.remaining());
+                    this.buffer.put(b, off, chunk);
+                    remaining -= chunk;
+                    off += chunk;
+                }
             }
         } finally {
             this.lock.unlock();
@@ -219,7 +238,7 @@ public class SharedOutputBuffer extends 
         this.lock.lock();
         try {
             try {
-                while (super.hasData()) {
+                while ((largeWrapper != null && largeWrapper.hasRemaining()) || super.hasData())
{
                     if (this.shutdown) {
                         throw new InterruptedIOException("Output operation aborted");
                     }

Modified: cxf/sandbox/dkulp_async_clients/http-hc/src/test/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitTest.java
URL: http://svn.apache.org/viewvc/cxf/sandbox/dkulp_async_clients/http-hc/src/test/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitTest.java?rev=1383103&r1=1383102&r2=1383103&view=diff
==============================================================================
--- cxf/sandbox/dkulp_async_clients/http-hc/src/test/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitTest.java
(original)
+++ cxf/sandbox/dkulp_async_clients/http-hc/src/test/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitTest.java
Mon Sep 10 20:38:54 2012
@@ -86,7 +86,7 @@ public class AsyncHTTPConduitTest extend
             });
         
         StringBuilder builder = new StringBuilder("NaNaNa");
-        for (int x = 0; x < 10; x++) {
+        for (int x = 0; x < 500000; x++) {
             builder.append(" NaNaNa ");
         }
         request = builder.toString();



Mime
View raw message