Hello Jeanfrancois,
I like this idea.
Just small comment. In code
1 public ThreadAttachment detach(boolean copyState) {
2 try{
3 threadAttachment.setByteBuffer(byteBuffer);
4 threadAttachment.setSSLEngine(sslEngine);
5 threadAttachment.setInputBB(inputBB);
6 threadAttachment.setOutputBB(outputBB);
7
8 return threadAttachment;
9 } finally {
10 // We cannot cache/re-use this object as it might be referenced
11 // by more than one thread.
12 if (copyState){
13 // Re-create a new ByteBuffer
14 byteBuffer = ByteBufferFactory.allocateView(8192,false);
15 threadAttachment = new ThreadAttachment();
16 }
17 threadAttachment.setThreadId(getName() + "-" + getId());
18 }
19 }
probably it makes sense to avoid such code like in lines 8 and 15...
from one side you're returning threadAttachment, from other reassigning
it in finally block. I looks confusing, at least to me :))
Good work!
WBR,
Alexey.
Jeanfrancois Arcand wrote:
> Hi,
>
> I would like to introduce a set of new APIs for handling the case when
> an ReadFilter is used and the temporary Selector trick for doing extra
> reads cannot be used. The use case is the following:
>
> + Inside a "ReadFilter", the socketChannel.read(...) reads some bytes
> and then starts returning 0 (no more bytes available). In that case,
> the SelectionKey must be registered back to its SelectorHandler, and
> the state of the transaction needs to be persisted. By state here I
> means the byteBuffer that contains the partial read (bytes) and maybe
> some application specific data.
>
> To support that use case, I would like to add two new method to the
> WorkerThread interface:
>
>> 49 /**
>> 50 * Detach the current set of attributes (state)
>> associated with this instance.
>> 51 * Invoking detach(true) will re-create all the
>> ByteBuffer associated with
>> 52 * this thread, hence this method must be called only
>> when required. If
>> 53 * you only need to invoke the object, call detach(false)
>> instead but make
>> 54 * sure you aren't caching or re-using the
>> ThreadAttachment with another
>> 55 * thread.
>> 56 * @param true to copy the attributes into the
>> ThreadAttachment and re-create
>> 57 * them
>> 58 * @return a new ThreadAttachment
>> 59 */
>> 60 public ThreadAttachment detach(boolean copyState);
>> 61
>> 62
>> 63 /**
>> 64 * Attach the ThreadAttachment to this instance. This
>> will configure this
>> 65 * Thread attributes like ByteBuffer, SSLEngine, etc.
>> 66 * @param ThreadAttachment the attachment.
>> 67 */
>> 68 public void attach(ThreadAttachment threadAttachment);
>
> Hence when an application needs more bytes and need to register back
> to the SelectorHandler:
>
>
> public class MyReadFilter{
>
> public boolean execute(Context ctx){
> ....
> WorkerThread wt = ((WorkerThread)Thread.currentThread());
> ByteBuffer bb = wt.getByteBuffer();
> int length = socketChannel.read(bb);
> ...
> if (length == 0){
>
> ThreadAttachment ta = wt.detach(true);
> ctx.getSelectionKey().attach(ta);
> ....
> ctx.setKeyRegistrationState(
> Context.KeyRegistrationState.REGISTER);
> ...
> }
>
> }
>
> Internally, the SelectorHandler will do, once bytes are available:
>
> WorkerThread wt = ((WorkerThread)Thread.currentThread());
> wt.attach((ThreadAttachment)selectionKey.attachment());
>
> and then invoke the Filter Again. Calling wt.attach() will properly
> configure the Thread attribute so when MyReadFilter.execute is called,
> the previously read bytes will be already available inside the
> ByteBuffer.
>
> I will apply the solution to the SSLReadFilter as well, which
> currentlt use the SSLSession to store the state of the transaction.
> I'm attaching the diff of the changes.
>
> Let me know what you think.
>
> Thanks
>
> -- Jeanfrancois
>
>
>
>
>
>
> ------------------------------------------------------------------------
>
> Index: WorkerThread.java
> ===================================================================
> --- WorkerThread.java (revision 308)
> +++ WorkerThread.java (working copy)
> @@ -25,7 +25,7 @@
> import java.nio.ByteBuffer;
>
> /**
> - * Simple interface to allow the addition of <code>Thread</code> attributes.
> + * Simple interface to allow the addition of <code>Thread</code> attributes.
> *
> * @author Jean-Francois Arcand
> */
> @@ -45,4 +45,25 @@
> */
> public ByteBuffer getByteBuffer();
>
> +
> + /**
> + * Detach the current set of attributes (state) associated with this instance.
> + * Invoking detach(true) will re-create all the ByteBuffer associated with
> + * this thread, hence this method must be called only when required. If
> + * you only need to invoke the object, call detach(false) instead but make
> + * sure you aren't caching or re-using the ThreadAttachment with another
> + * thread.
> + * @param true to copy the attributes into the ThreadAttachment and re-create
> + * them
> + * @return a new ThreadAttachment
> + */
> + public ThreadAttachment detach(boolean copyState);
> +
> +
> + /**
> + * Attach the ThreadAttachment to this instance. This will configure this
> + * Thread attributes like ByteBuffer, SSLEngine, etc.
> + * @param ThreadAttachment the attachment.
> + */
> + public void attach(ThreadAttachment threadAttachment);
> }
> Index: WorkerThreadImpl.java
> ===================================================================
> --- WorkerThreadImpl.java (revision 308)
> +++ WorkerThreadImpl.java (working copy)
> @@ -70,22 +70,28 @@
> /**
> * The encrypted ByteBuffer used for handshaking and reading request bytes.
> */
> - private ByteBuffer inputBB;
> + protected ByteBuffer inputBB;
>
>
> /**
> * The encrypted ByteBuffer used for handshaking and writing response bytes.
> */
> - private ByteBuffer outputBB;
> + protected ByteBuffer outputBB;
>
>
> /**
> * The <code>SSLEngine</code> used to manage the SSL over NIO request.
> */
> - private SSLEngine sslEngine;
> -
> + protected SSLEngine sslEngine;
>
> +
> /**
> + * The state/attributes on this WorkerThread.
> + */
> + private ThreadAttachment threadAttachment = new ThreadAttachment();
> +
> +
> + /**
> * Create a Thread that will synchronizes/block on
> * <code>Pipeline</code> instance.
> * @param threadGroup <code>ThreadGroup</code>
> @@ -233,5 +239,53 @@
> public void setSSLEngine(SSLEngine sslEngine) {
> this.sslEngine = sslEngine;
> }
> +
> +
> + /**
> + * Detach the current set of attributes (state) associated with this instance.
> + * Invoking detach(true) will re-create all the ByteBuffer associated with
> + * this thread, hence this method must be called only when required. If
> + * you only need to invoke the object, call detach(false) instead but make
> + * sure you aren't caching or re-using the ThreadAttachment with another
> + * thread.
> + * @param true to copy the attributes into the ThreadAttachment and re-create
> + * them.
> + * @return a new ThreadAttachment
> + */
> + public ThreadAttachment detach(boolean copyState) {
> + try{
> + threadAttachment.setByteBuffer(byteBuffer);
> + threadAttachment.setSSLEngine(sslEngine);
> + threadAttachment.setInputBB(inputBB);
> + threadAttachment.setOutputBB(outputBB);
> +
> + return threadAttachment;
> + } finally {
> + // We cannot cache/re-use this object as it might be referenced
> + // by more than one thread.
> + if (copyState){
> + // Re-create a new ByteBuffer
> + byteBuffer = ByteBufferFactory.allocateView(8192,false);
> + threadAttachment = new ThreadAttachment();
> + }
> + threadAttachment.setThreadId(getName() + "-" + getId());
> + }
> + }
> +
> +
> + /**
> + * Attach the ThreadAttachment to this instance. This will configure this
> + * Thread attributes like ByteBuffer, SSLEngine, etc.
> + * @param ThreadAttachment the attachment.
> + */
> + public void attach(ThreadAttachment threadAttachment) {
> + byteBuffer = threadAttachment.getByteBuffer();
> + sslEngine = threadAttachment.getSSLEngine();
> + inputBB = threadAttachment.getInputBB();
> + outputBB = threadAttachment.getOutputBB();
> +
> + this.threadAttachment = threadAttachment;
> + threadAttachment.setThreadId(getName() + "-" + getId());
> + }
> }
>
> /*
> * The contents of this file are subject to the terms
> * of the Common Development and Distribution License
> * (the License). You may not use this file except in
> * compliance with the License.
> *
> * You can obtain a copy of the license at
> * https://glassfish.dev.java.net/public/CDDLv1.0.html or
> * glassfish/bootstrap/legal/CDDLv1.0.txt.
> * See the License for the specific language governing
> * permissions and limitations under the License.
> *
> * When distributing Covered Code, include this CDDL
> * Header Notice in each file and include the License file
> * at glassfish/bootstrap/legal/CDDLv1.0.txt.
> * If applicable, add the following below the CDDL Header,
> * with the fields enclosed by brackets [] replaced by
> * you own identifying information:
> * "Portions Copyrighted [year] [name of copyright owner]"
> *
> * Copyright 2007 Sun Microsystems, Inc. All rights reserved.
> */
> package com.sun.grizzly.util;
>
> import java.nio.ByteBuffer;
> import java.util.WeakHashMap;
> import javax.net.ssl.SSLEngine;
>
> /**
> * This object represent the state of a <code>WorkerThread</code>. This include
> * the ByteBuffer binded to the WorkerThread, application data etc.
> * @author Jeanfrancois
> */
> public class ThreadAttachment {
>
>
> private long timeout;
>
>
> private String threadId;
>
>
> private WeakHashMap<String,Object> map;
>
>
> private ByteBuffer byteBuffer;
>
>
> /**
> * The encrypted ByteBuffer used for handshaking and reading request bytes.
> */
> private ByteBuffer inputBB;
>
>
> /**
> * The encrypted ByteBuffer used for handshaking and writing response bytes.
> */
> private ByteBuffer outputBB;
>
>
> /**
> * The <code>SSLEngine</code> used to manage the SSL over NIO request.
> */
> private SSLEngine sslEngine;
>
>
> public ThreadAttachment(){
> map = new WeakHashMap<String,Object>();
> }
>
>
> public void setAttribute(String key, Object value){
> map.put(key,value);
> }
>
>
> public Object getAttribute(String key){
> return map.get(key);
> }
>
>
> public Object removeAttribute(String key){
> return map.remove(key);
> }
>
> /**
> * Set the <code>ByteBuffer</code> shared this thread
> */
> public void setByteBuffer(ByteBuffer byteBuffer){
> this.byteBuffer = byteBuffer;
> }
>
>
> /**
> * Return the <code>ByteBuffer</code> shared this thread
> */
> public ByteBuffer getByteBuffer(){
> return byteBuffer;
> }
>
>
> /**
> * Return the encrypted <code>ByteBuffer</code> used to handle request.
> * @return <code>ByteBuffer</code>
> */
> public ByteBuffer getInputBB(){
> return inputBB;
> }
>
>
> /**
> * Set the encrypted <code>ByteBuffer</code> used to handle request.
> * @param inputBB <code>ByteBuffer</code>
> */
> public void setInputBB(ByteBuffer inputBB){
> this.inputBB = inputBB;
> }
>
>
> /**
> * Return the encrypted <code>ByteBuffer</code> used to handle response.
> * @return <code>ByteBuffer</code>
> */
> public ByteBuffer getOutputBB(){
> return outputBB;
> }
>
>
> /**
> * Set the encrypted <code>ByteBuffer</code> used to handle response.
> * @param outputBB <code>ByteBuffer</code>
> */
> public void setOutputBB(ByteBuffer outputBB){
> this.outputBB = outputBB;
> }
>
>
> /**
> * Set the <code>SSLEngine</code>.
> * @return <code>SSLEngine</code>
> */
> public SSLEngine getSSLEngine() {
> return sslEngine;
> }
>
>
> /**
> * Get the <code>SSLEngine</code>.
> * @param sslEngine <code>SSLEngine</code>
> */
> public void setSSLEngine(SSLEngine sslEngine) {
> this.sslEngine = sslEngine;
> }
>
>
> public String getThreadId() {
> return threadId;
> }
>
>
> public void setThreadId(String threadId) {
> this.threadId = threadId;
> }
>
>
> public void setTimeout(long timeout){
> this.timeout = timeout;
> }
>
> }
>
>
> ------------------------------------------------------------------------
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: dev-unsubscribe_at_grizzly.dev.java.net
> For additional commands, e-mail: dev-help_at_grizzly.dev.java.net
>