Coverage Report

Coverage Report - com.sun.grizzly.util.WorkerThreadImpl

Classes in this File

Line Coverage

Branch Coverage

Complexity

WorkerThreadImpl

81 %

69/85

89 %

34/38

 /*
  * 
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  * 
  * Copyright 2007-2008 Sun Microsystems, Inc. All rights reserved.
  * 
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
  * and Distribution License("CDDL") (collectively, 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/CDDL+GPL.html
  * or glassfish/bootstrap/legal/LICENSE.txt.  See the License for the specific
  * language governing permissions and limitations under the License.
  * 
  * When distributing the software, include this License Header Notice in each
  * file and include the License file at glassfish/bootstrap/legal/LICENSE.txt.
  * Sun designates this particular file as subject to the "Classpath" exception
  * as provided by Sun in the GPL Version 2 section of the License file that
  * accompanied this code.  If applicable, add the following below the License
  * Header, with the fields enclosed by brackets [] replaced by your own
  * identifying information: "Portions Copyrighted [year]
  * [name of copyright owner]"
  * 
  * Contributor(s):
  * 
  * If you wish your version of this file to be governed by only the CDDL or
  * only the GPL Version 2, indicate your decision by adding "[Contributor]
  * elects to include this software in this distribution under the [CDDL or GPL
  * Version 2] license."  If you don't indicate a single choice of license, a
  * recipient has the option to distribute your version of this file under
  * either the CDDL, the GPL Version 2 or to extend the choice of license to
  * its licensees as provided above.  However, if you add GPL Version 2 code
  * and therefore, elected the GPL Version 2 license, then the option applies
  * only if the new code is made subject to such option by the copyright
  * holder.
  *
  */
 package com.sun.grizzly.util;
 
 import com.sun.grizzly.Controller;
 import com.sun.grizzly.Pipeline;
 import java.util.concurrent.Callable;
 import java.util.logging.Level;
 import com.sun.grizzly.util.ByteBufferFactory.ByteBufferType;
 import com.sun.grizzly.util.ThreadAttachment.Mode;
 
 /**
  * Simple worker thread used for processing HTTP requests. All threads are
  * synchronized using a {@link Pipeline} object
  *
  * @author Jean-Francois Arcand
  */
 public class WorkerThreadImpl extends WorkerThread {
     
     private static final int DEFAULT_BYTE_BUFFER_SIZE = 8192;
     
     /**
      * What will be run.
      */
     protected Runnable target;
     
     
     /**
      * The {@link Pipeline} on which this thread synchronize.
      */
     protected Pipeline<Callable> pipeline;
     
     
     /**
      * Looing variable.
      */
     protected volatile boolean execute = true;
     
     
     /**
      * The <code>ThreadGroup</code> used.
      */
     protected final static ThreadGroup threadGroup = new ThreadGroup("Grizzly");
     
 
     /**
      * The state/attributes on this WorkerThread.
      */
     private ThreadAttachment threadAttachment;
     
     
     /**
      * The ByteBufferType used when creating the ByteBuffer attached to this object.
      */
     private ByteBufferType byteBufferType = ByteBufferType.HEAP_VIEW;
     
     
     /**
      * The size of the ByteBuffer attached to this object.
      */
     private int initialByteBufferSize;
     
     
     /**
      * Create a Thread that will synchronizes/block on
      * {@link Pipeline} instance.
      * @param threadGroup <code>ThreadGroup</code>
      * @param runnable <code>Runnable</code>
      */
     public WorkerThreadImpl(ThreadGroup threadGroup, Runnable runnable){
         this(threadGroup, runnable, DEFAULT_BYTE_BUFFER_SIZE);
     }
     
     /**
      * Create a Thread that will synchronizes/block on
      * {@link Pipeline} instance.
      * @param threadGroup <code>ThreadGroup</code>
      * @param runnable <code>Runnable</code>
      * @param initialByteBufferSize initial {@link ByteBuffer} size
      */
     public WorkerThreadImpl(ThreadGroup threadGroup, Runnable runnable, 
             int initialByteBufferSize){
         super(threadGroup, runnable);
         setDaemon(true);
         target = runnable;
         this.initialByteBufferSize = initialByteBufferSize;
     }
     
     /**
      * Create a Thread that will synchronizes/block on
      * {@link Pipeline} instance.
      * @param pipeline {@link Pipeline}
      * @param name <code>String</code>
      */
     public WorkerThreadImpl(Pipeline<Callable> pipeline, String name){
         this(pipeline, name, DEFAULT_BYTE_BUFFER_SIZE);
     }
     
     /**
      * Create a Thread that will synchronizes/block on
      * {@link Pipeline} instance.
      * @param pipeline {@link Pipeline}
      * @param name <code>String</code>
      * @param initialByteBufferSize initial {@link ByteBuffer} size
      */
     public WorkerThreadImpl(Pipeline<Callable> pipeline, String name, 
             int initialByteBufferSize){
         super(threadGroup, name);
         this.pipeline = pipeline;
         setDaemon(true);
         this.initialByteBufferSize = initialByteBufferSize;
     }
     
     /**
      * Execute a {@link Task}.
      */
     @Override
     public void run(){        
         if (byteBuffer == null){
             byteBuffer = ByteBufferFactory.allocate(byteBufferType,
                     initialByteBufferSize);
         }
         
         if (target != null){
             target.run();
             return;
         }
         
         while (execute) {
             try{
                 // Wait for a Task to be added to the pipeline.
                 Callable t = pipeline.waitForIoTask();
                 if (t != null){
                     t.call();
                     t = null;
                 }
             } catch (Throwable t) {
                 if (execute) {
                     Controller.logger().log(Level.SEVERE,
                             "WorkerThreadImpl unexpected exception: ",t);
                 } else {
                     Controller.logger().log(Level.FINE,
                             "WorkerThreadImpl unexpected exception, when WorderThread supposed to be closed: ",t);
                 }
             } finally {
                 reset();
             }
         }
     }
     
     
     /**
      * Stop this thread. If this Thread is performing atask, the task will be
      * completed.
      */
     public void terminate(){
         execute = false;
     }
             
     public ThreadAttachment updateAttachment(int mode) {
         ThreadAttachment currentAttachment = getAttachment();
         currentAttachment.reset();
 
         if ((mode & Mode.BYTE_BUFFER) != 0) {
             currentAttachment.setByteBuffer(byteBuffer);
         }
 
         if ((mode & Mode.SSL_ENGINE) != 0) {
             currentAttachment.setSSLEngine(sslEngine);
         }
 
         if ((mode & Mode.INPUT_BB) != 0) {
             currentAttachment.setInputBB(inputBB);
         }
 
         if ((mode & Mode.OUTPUT_BB) != 0) {
             currentAttachment.setOutputBB(outputBB);
         }
 
         currentAttachment.setMode(mode);
         
         return currentAttachment;
     }
     
     public ThreadAttachment getAttachment() {
         if (threadAttachment == null) {
             threadAttachment = new ThreadAttachment();
         }
 
         return threadAttachment;
     }
 
     public ThreadAttachment detach() {
         ThreadAttachment currentAttachment = getAttachment();
         int mode = currentAttachment.getMode();
         updateAttachment(mode);
         
         // Re-create a new ByteBuffer
         if ((mode & Mode.BYTE_BUFFER) != 0) {
             byteBuffer = ByteBufferFactory.allocate(byteBufferType,
                     initialByteBufferSize);
         }
         
         if ((mode & Mode.SSL_ENGINE) != 0) {
             sslEngine = null;
         }
         
         if ((mode & Mode.INPUT_BB) != 0) {
             inputBB = null;
         }
         
         if ((mode & Mode.OUTPUT_BB) != 0) {
             outputBB = null;
         }
         
         // Switch to the new ThreadAttachment.
         this.threadAttachment = null;
         
         currentAttachment.deassociate();
         return currentAttachment;
     }
 
 
     public void attach(ThreadAttachment threadAttachment) {
         threadAttachment.associate();
         int mode = threadAttachment.getMode();
         
         if ((mode & Mode.BYTE_BUFFER) != 0) {
             byteBuffer = threadAttachment.getByteBuffer();
         }
         
         if ((mode & Mode.SSL_ENGINE) != 0) {
             sslEngine = threadAttachment.getSSLEngine();
         }
         
         if ((mode & Mode.INPUT_BB) != 0) {
             inputBB = threadAttachment.getInputBB();
         }
         
         if ((mode & Mode.OUTPUT_BB) != 0) {
             outputBB = threadAttachment.getOutputBB();
         }
         
         this.threadAttachment = threadAttachment;   
     }
 
     
     /**
      * The <code>ByteBufferType</code> used to create the {@link ByteBuffer}
      * associated with this object.
      * @return The <code>ByteBufferType</code> used to create the {@link ByteBuffer}
      * associated with this object.
      */
     public ByteBufferType getByteBufferType() {
         return byteBufferType;
     }
 
     
     /**
      * Set the <code>ByteBufferType</code> to use when creating the
      * {@link ByteBuffer} associated with this object.
      * @param byteBufferType The ByteBuffer type.
      */
     public void setByteBufferType(ByteBufferType byteBufferType) {
         this.byteBufferType = byteBufferType;
     }
 
     @Override
     protected void reset() {
         if (threadAttachment != null) {
             /** 
              * ThreadAttachment was created during prev. processing and wasn't
              * detached. It could happen due to some error - we need to release
              * the ThreadAttachment association with the current thread
              */
             threadAttachment.deassociate();
         }
         
         threadAttachment = null;
         super.reset();
     }
 }
 

1		/*
2		*
3		* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
4		*
5		* Copyright 2007-2008 Sun Microsystems, Inc. All rights reserved.
6		*
7		* The contents of this file are subject to the terms of either the GNU
8		* General Public License Version 2 only ("GPL") or the Common Development
9		* and Distribution License("CDDL") (collectively, the "License"). You
10		* may not use this file except in compliance with the License. You can obtain
11		* a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
12		* or glassfish/bootstrap/legal/LICENSE.txt. See the License for the specific
13		* language governing permissions and limitations under the License.
14		*
15		* When distributing the software, include this License Header Notice in each
16		* file and include the License file at glassfish/bootstrap/legal/LICENSE.txt.
17		* Sun designates this particular file as subject to the "Classpath" exception
18		* as provided by Sun in the GPL Version 2 section of the License file that
19		* accompanied this code. If applicable, add the following below the License
20		* Header, with the fields enclosed by brackets [] replaced by your own
21		* identifying information: "Portions Copyrighted [year]
22		* [name of copyright owner]"
23		*
24		* Contributor(s):
25		*
26		* If you wish your version of this file to be governed by only the CDDL or
27		* only the GPL Version 2, indicate your decision by adding "[Contributor]
28		* elects to include this software in this distribution under the [CDDL or GPL
29		* Version 2] license." If you don't indicate a single choice of license, a
30		* recipient has the option to distribute your version of this file under
31		* either the CDDL, the GPL Version 2 or to extend the choice of license to
32		* its licensees as provided above. However, if you add GPL Version 2 code
33		* and therefore, elected the GPL Version 2 license, then the option applies
34		* only if the new code is made subject to such option by the copyright
35		* holder.
36		*
37		*/
38		package com.sun.grizzly.util;
39
40		import com.sun.grizzly.Controller;
41		import com.sun.grizzly.Pipeline;
42		import java.util.concurrent.Callable;
43		import java.util.logging.Level;
44		import com.sun.grizzly.util.ByteBufferFactory.ByteBufferType;
45		import com.sun.grizzly.util.ThreadAttachment.Mode;
46
47		/**
48		* Simple worker thread used for processing HTTP requests. All threads are
49		* synchronized using a {@link Pipeline} object
50		*
51		* @author Jean-Francois Arcand
52		*/
53		public class WorkerThreadImpl extends WorkerThread {
54
55		private static final int DEFAULT_BYTE_BUFFER_SIZE = 8192;
56
57		/**
58		* What will be run.
59		*/
60		protected Runnable target;
61
62
63		/**
64		* The {@link Pipeline} on which this thread synchronize.
65		*/
66		protected Pipeline<Callable> pipeline;
67
68
69		/**
70		* Looing variable.
71		*/
72	353	protected volatile boolean execute = true;
73
74
75		/**
76		* The <code>ThreadGroup</code> used.
77		*/
78	1	protected final static ThreadGroup threadGroup = new ThreadGroup("Grizzly");
79
80
81		/**
82		* The state/attributes on this WorkerThread.
83		*/
84		private ThreadAttachment threadAttachment;
85
86
87		/**
88		* The ByteBufferType used when creating the ByteBuffer attached to this object.
89		*/
90	353	private ByteBufferType byteBufferType = ByteBufferType.HEAP_VIEW;
91
92
93		/**
94		* The size of the ByteBuffer attached to this object.
95		*/
96		private int initialByteBufferSize;
97
98
99		/**
100		* Create a Thread that will synchronizes/block on
101		* {@link Pipeline} instance.
102		* @param threadGroup <code>ThreadGroup</code>
103		* @param runnable <code>Runnable</code>
104		*/
105		public WorkerThreadImpl(ThreadGroup threadGroup, Runnable runnable){
106	0	this(threadGroup, runnable, DEFAULT_BYTE_BUFFER_SIZE);
107	0	}
108
109		/**
110		* Create a Thread that will synchronizes/block on
111		* {@link Pipeline} instance.
112		* @param threadGroup <code>ThreadGroup</code>
113		* @param runnable <code>Runnable</code>
114		* @param initialByteBufferSize initial {@link ByteBuffer} size
115		*/
116		public WorkerThreadImpl(ThreadGroup threadGroup, Runnable runnable,
117		int initialByteBufferSize){
118	0	super(threadGroup, runnable);
119	0	setDaemon(true);
120	0	target = runnable;
121	0	this.initialByteBufferSize = initialByteBufferSize;
122	0	}
123
124		/**
125		* Create a Thread that will synchronizes/block on
126		* {@link Pipeline} instance.
127		* @param pipeline {@link Pipeline}
128		* @param name <code>String</code>
129		*/
130		public WorkerThreadImpl(Pipeline<Callable> pipeline, String name){
131	0	this(pipeline, name, DEFAULT_BYTE_BUFFER_SIZE);
132	0	}
133
134		/**
135		* Create a Thread that will synchronizes/block on
136		* {@link Pipeline} instance.
137		* @param pipeline {@link Pipeline}
138		* @param name <code>String</code>
139		* @param initialByteBufferSize initial {@link ByteBuffer} size
140		*/
141		public WorkerThreadImpl(Pipeline<Callable> pipeline, String name,
142		int initialByteBufferSize){
143	353	super(threadGroup, name);
144	353	this.pipeline = pipeline;
145	353	setDaemon(true);
146	353	this.initialByteBufferSize = initialByteBufferSize;
147	353	}
148
149		/**
150		* Execute a {@link Task}.
151		*/
152		@Override
153		public void run(){
154	353	if (byteBuffer == null){
155	353	byteBuffer = ByteBufferFactory.allocate(byteBufferType,
156		initialByteBufferSize);
157		}
158
159	353	if (target != null){
160	0	target.run();
161	0	return;
162		}
163
164	389238	while (execute) {
165		try{
166		// Wait for a Task to be added to the pipeline.
167	388885	Callable t = pipeline.waitForIoTask();
168	388885	if (t != null){
169	156189	t.call();
170	156189	t = null;
171		}
172	0	} catch (Throwable t) {
173	0	if (execute) {
174	0	Controller.logger().log(Level.SEVERE,
175		"WorkerThreadImpl unexpected exception: ",t);
176		} else {
177	0	Controller.logger().log(Level.FINE,
178		"WorkerThreadImpl unexpected exception, when WorderThread supposed to be closed: ",t);
179		}
180		} finally {
181	388885	reset();
182	388882	}
183		}
184	352	}
185
186
187		/**
188		* Stop this thread. If this Thread is performing atask, the task will be
189		* completed.
190		*/
191		public void terminate(){
192	353	execute = false;
193	353	}
194
195		public ThreadAttachment updateAttachment(int mode) {
196	161777	ThreadAttachment currentAttachment = getAttachment();
197	161777	currentAttachment.reset();
198
199	161777	if ((mode & Mode.BYTE_BUFFER) != 0) {
200	12	currentAttachment.setByteBuffer(byteBuffer);
201		}
202
203	161777	if ((mode & Mode.SSL_ENGINE) != 0) {
204	161774	currentAttachment.setSSLEngine(sslEngine);
205		}
206
207	161777	if ((mode & Mode.INPUT_BB) != 0) {
208	161658	currentAttachment.setInputBB(inputBB);
209		}
210
211	161777	if ((mode & Mode.OUTPUT_BB) != 0) {
212	582	currentAttachment.setOutputBB(outputBB);
213		}
214
215	161777	currentAttachment.setMode(mode);
216
217	161777	return currentAttachment;
218		}
219
220		public ThreadAttachment getAttachment() {
221	269563	if (threadAttachment == null) {
222	124	threadAttachment = new ThreadAttachment();
223		}
224
225	269563	return threadAttachment;
226		}
227
228		public ThreadAttachment detach() {
229	53894	ThreadAttachment currentAttachment = getAttachment();
230	53894	int mode = currentAttachment.getMode();
231	53894	updateAttachment(mode);
232
233		// Re-create a new ByteBuffer
234	53894	if ((mode & Mode.BYTE_BUFFER) != 0) {
235	11	byteBuffer = ByteBufferFactory.allocate(byteBufferType,
236		initialByteBufferSize);
237		}
238
239	53894	if ((mode & Mode.SSL_ENGINE) != 0) {
240	53892	sslEngine = null;
241		}
242
243	53894	if ((mode & Mode.INPUT_BB) != 0) {
244	53892	inputBB = null;
245		}
246
247	53894	if ((mode & Mode.OUTPUT_BB) != 0) {
248	200	outputBB = null;
249		}
250
251		// Switch to the new ThreadAttachment.
252	53894	this.threadAttachment = null;
253
254	53894	currentAttachment.deassociate();
255	53894	return currentAttachment;
256		}
257
258
259		public void attach(ThreadAttachment threadAttachment) {
260	53884	threadAttachment.associate();
261	53884	int mode = threadAttachment.getMode();
262
263	53884	if ((mode & Mode.BYTE_BUFFER) != 0) {
264	4	byteBuffer = threadAttachment.getByteBuffer();
265		}
266
267	53884	if ((mode & Mode.SSL_ENGINE) != 0) {
268	53881	sslEngine = threadAttachment.getSSLEngine();
269		}
270
271	53884	if ((mode & Mode.INPUT_BB) != 0) {
272	53881	inputBB = threadAttachment.getInputBB();
273		}
274
275	53884	if ((mode & Mode.OUTPUT_BB) != 0) {
276	194	outputBB = threadAttachment.getOutputBB();
277		}
278
279	53884	this.threadAttachment = threadAttachment;
280	53884	}
281
282
283		/**
284		* The <code>ByteBufferType</code> used to create the {@link ByteBuffer}
285		* associated with this object.
286		* @return The <code>ByteBufferType</code> used to create the {@link ByteBuffer}
287		* associated with this object.
288		*/
289		public ByteBufferType getByteBufferType() {
290	0	return byteBufferType;
291		}
292
293
294		/**
295		* Set the <code>ByteBufferType</code> to use when creating the
296		* {@link ByteBuffer} associated with this object.
297		* @param byteBufferType The ByteBuffer type.
298		*/
299		public void setByteBufferType(ByteBufferType byteBufferType) {
300	353	this.byteBufferType = byteBufferType;
301	353	}
302
303		@Override
304		protected void reset() {
305	388885	if (threadAttachment != null) {
306		/**
307		* ThreadAttachment was created during prev. processing and wasn't
308		* detached. It could happen due to some error - we need to release
309		* the ThreadAttachment association with the current thread
310		*/
311	114	threadAttachment.deassociate();
312		}
313
314	388885	threadAttachment = null;
315	388880	super.reset();
316	388885	}
317		}
318