/
AbstractEndpoint.java
989 lines (849 loc) · 33.1 KB
/
AbstractEndpoint.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.tomcat.util.net;
import java.io.OutputStreamWriter;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.apache.juli.logging.Log;
import org.apache.tomcat.util.IntrospectionUtils;
import org.apache.tomcat.util.net.AbstractEndpoint.Acceptor.AcceptorState;
import org.apache.tomcat.util.res.StringManager;
import org.apache.tomcat.util.threads.LimitLatch;
import org.apache.tomcat.util.threads.ResizableExecutor;
import org.apache.tomcat.util.threads.TaskQueue;
import org.apache.tomcat.util.threads.TaskThreadFactory;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;
/**
* @param <S> The type for the sockets managed by this endpoint.
*
* @author Mladen Turk
* @author Remy Maucherat
*/
public abstract class AbstractEndpoint<S> {
// -------------------------------------------------------------- Constants
protected static final StringManager sm = StringManager.getManager(
AbstractEndpoint.class.getPackage().getName());
public static interface Handler<S> {
/**
* Different types of socket states to react upon.
*/
public enum SocketState {
// TODO Add a new state to the AsyncStateMachine and remove
// ASYNC_END (if possible)
OPEN, CLOSED, LONG, ASYNC_END, SENDFILE, UPGRADING, UPGRADED
}
/**
* Process the provided socket with the given current status.
*
* @param socket The socket to process
* @param status The current socket status
*
* @return The state of the socket after processing
*/
public SocketState process(SocketWrapperBase<S> socket,
SocketStatus status);
/**
* Obtain the GlobalRequestProcessor associated with the handler.
*
* @return the GlobalRequestProcessor
*/
public Object getGlobal();
/**
* Recycle resources associated with the handler.
*/
public void recycle();
}
protected enum BindState {
UNBOUND, BOUND_ON_INIT, BOUND_ON_START
}
public abstract static class Acceptor implements Runnable {
public enum AcceptorState {
NEW, RUNNING, PAUSED, ENDED
}
protected volatile AcceptorState state = AcceptorState.NEW;
public final AcceptorState getState() {
return state;
}
private String threadName;
protected final void setThreadName(final String threadName) {
this.threadName = threadName;
}
protected final String getThreadName() {
return threadName;
}
}
private static final int INITIAL_ERROR_DELAY = 50;
private static final int MAX_ERROR_DELAY = 1600;
/**
* Async timeout thread
*/
protected class AsyncTimeout implements Runnable {
private volatile boolean asyncTimeoutRunning = true;
/**
* The background thread that checks async requests and fires the
* timeout if there has been no activity.
*/
@Override
public void run() {
// Loop until we receive a shutdown command
while (asyncTimeoutRunning) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// Ignore
}
long now = System.currentTimeMillis();
for (SocketWrapperBase<S> socket : waitingRequests) {
long asyncTimeout = socket.getAsyncTimeout();
if (asyncTimeout > 0) {
long asyncStart = socket.getLastAsyncStart();
if ((now - asyncStart) > asyncTimeout) {
// Avoid multiple timeouts
socket.setAsyncTimeout(-1);
processSocket(socket, SocketStatus.TIMEOUT, true);
}
}
}
// Loop if endpoint is paused
while (paused && asyncTimeoutRunning) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// Ignore
}
}
}
}
protected void stop() {
asyncTimeoutRunning = false;
}
}
// ----------------------------------------------------------------- Fields
/**
* Running state of the endpoint.
*/
protected volatile boolean running = false;
/**
* Will be set to true whenever the endpoint is paused.
*/
protected volatile boolean paused = false;
/**
* Are we using an internal executor
*/
protected volatile boolean internalExecutor = false;
/**
* counter for nr of connections handled by an endpoint
*/
private volatile LimitLatch connectionLimitLatch = null;
/**
* Socket properties
*/
protected SocketProperties socketProperties = new SocketProperties();
public SocketProperties getSocketProperties() {
return socketProperties;
}
/**
* Threads used to accept new connections and pass them to worker threads.
*/
protected Acceptor[] acceptors;
// ----------------------------------------------------------------- Properties
protected Map<String,SSLHostConfig> sslHostConfigs = new ConcurrentHashMap<>();
public void addSslHostConfig(SSLHostConfig sslHostConfig) {
String key = sslHostConfig.getHostName();
if (key == null || key.length() == 0) {
throw new IllegalArgumentException(sm.getString("endpoint.noSslHostName"));
}
SSLHostConfig duplicate = sslHostConfigs.put(key, sslHostConfig);
if (duplicate != null) {
throw new IllegalArgumentException(sm.getString("endpoint.duplicateSslHostName", key));
}
sslHostConfig.setConfigType(getSslConfigType());
}
protected abstract SSLHostConfig.Type getSslConfigType();
protected SSLHostConfig getSSLHostConfig(String sniHostName) {
SSLHostConfig result = null;
if (sniHostName != null) {
// First choice - direct match
result = sslHostConfigs.get(sniHostName);
if (result != null) {
return result;
}
// Second choice, wildcard match
int indexOfDot = sniHostName.indexOf('.');
if (indexOfDot > -1) {
result = sslHostConfigs.get("*" + sniHostName.substring(indexOfDot));
}
}
// Fall-back. Use the default
if (result == null) {
result = sslHostConfigs.get(SSLHostConfig.DEFAULT_SSL_HOST_NAME);
}
if (result == null) {
// Should never happen.
throw new IllegalStateException();
}
return result;
}
/**
* Has the user requested that send file be used where possible?
*/
private boolean useSendfile = true;
public boolean getUseSendfile() {
return useSendfile;
}
public void setUseSendfile(boolean useSendfile) {
this.useSendfile = useSendfile;
}
/**
* Time to wait for the internal executor (if used) to terminate when the
* endpoint is stopped in milliseconds. Defaults to 5000 (5 seconds).
*/
private long executorTerminationTimeoutMillis = 5000;
public long getExecutorTerminationTimeoutMillis() {
return executorTerminationTimeoutMillis;
}
public void setExecutorTerminationTimeoutMillis(
long executorTerminationTimeoutMillis) {
this.executorTerminationTimeoutMillis = executorTerminationTimeoutMillis;
}
/**
* Acceptor thread count.
*/
protected int acceptorThreadCount = 0;
public void setAcceptorThreadCount(int acceptorThreadCount) {
this.acceptorThreadCount = acceptorThreadCount;
}
public int getAcceptorThreadCount() { return acceptorThreadCount; }
/**
* Priority of the acceptor threads.
*/
protected int acceptorThreadPriority = Thread.NORM_PRIORITY;
public void setAcceptorThreadPriority(int acceptorThreadPriority) {
this.acceptorThreadPriority = acceptorThreadPriority;
}
public int getAcceptorThreadPriority() { return acceptorThreadPriority; }
private int maxConnections = 10000;
public void setMaxConnections(int maxCon) {
this.maxConnections = maxCon;
LimitLatch latch = this.connectionLimitLatch;
if (latch != null) {
// Update the latch that enforces this
if (maxCon == -1) {
releaseConnectionLatch();
} else {
latch.setLimit(maxCon);
}
} else if (maxCon > 0) {
initializeConnectionLatch();
}
}
public int getMaxConnections() { return this.maxConnections; }
/**
* Return the current count of connections handled by this endpoint, if the
* connections are counted (which happens when the maximum count of
* connections is limited), or <code>-1</code> if they are not. This
* property is added here so that this value can be inspected through JMX.
* It is visible on "ThreadPool" MBean.
*
* <p>The count is incremented by the Acceptor before it tries to accept a
* new connection. Until the limit is reached and thus the count cannot be
* incremented, this value is more by 1 (the count of acceptors) than the
* actual count of connections that are being served.
*
* @return The count
*/
public long getConnectionCount() {
LimitLatch latch = connectionLimitLatch;
if (latch != null) {
return latch.getCount();
}
return -1;
}
/**
* External Executor based thread pool.
*/
private Executor executor = null;
public void setExecutor(Executor executor) {
this.executor = executor;
this.internalExecutor = (executor==null);
}
public Executor getExecutor() { return executor; }
/**
* Server socket port.
*/
private int port;
public int getPort() { return port; }
public void setPort(int port ) { this.port=port; }
public abstract int getLocalPort();
/**
* Address for the server socket.
*/
private InetAddress address;
public InetAddress getAddress() { return address; }
public void setAddress(InetAddress address) { this.address = address; }
/**
* Allows the server developer to specify the backlog that
* should be used for server sockets. By default, this value
* is 100.
*/
private int backlog = 100;
public void setBacklog(int backlog) { if (backlog > 0) this.backlog = backlog; }
public int getBacklog() { return backlog; }
/**
* Controls when the Endpoint binds the port. <code>true</code>, the default
* binds the port on {@link #init()} and unbinds it on {@link #destroy()}.
* If set to <code>false</code> the port is bound on {@link #start()} and
* unbound on {@link #stop()}.
*/
private boolean bindOnInit = true;
public boolean getBindOnInit() { return bindOnInit; }
public void setBindOnInit(boolean b) { this.bindOnInit = b; }
private BindState bindState = BindState.UNBOUND;
/**
* Keepalive timeout, if not set the soTimeout is used.
*/
private Integer keepAliveTimeout = null;
public int getKeepAliveTimeout() {
if (keepAliveTimeout == null) {
return getSoTimeout();
} else {
return keepAliveTimeout.intValue();
}
}
public void setKeepAliveTimeout(int keepAliveTimeout) {
this.keepAliveTimeout = Integer.valueOf(keepAliveTimeout);
}
/**
* Socket TCP no delay.
*
* @return The current TCP no delay setting for sockets created by this
* endpoint
*/
public boolean getTcpNoDelay() { return socketProperties.getTcpNoDelay();}
public void setTcpNoDelay(boolean tcpNoDelay) { socketProperties.setTcpNoDelay(tcpNoDelay); }
/**
* Socket linger.
*
* @return The current socket linger time for sockets created by this
* endpoint
*/
public int getSoLinger() { return socketProperties.getSoLingerTime(); }
public void setSoLinger(int soLinger) {
socketProperties.setSoLingerTime(soLinger);
socketProperties.setSoLingerOn(soLinger>=0);
}
/**
* Socket timeout.
*
* @return The current socket timeout for sockets created by this endpoint
*/
public int getSoTimeout() { return socketProperties.getSoTimeout(); }
public void setSoTimeout(int soTimeout) { socketProperties.setSoTimeout(soTimeout); }
/**
* SSL engine.
*/
private boolean SSLEnabled = false;
public boolean isSSLEnabled() { return SSLEnabled; }
public void setSSLEnabled(boolean SSLEnabled) { this.SSLEnabled = SSLEnabled; }
private int minSpareThreads = 10;
public int getMinSpareThreads() {
return Math.min(minSpareThreads,getMaxThreads());
}
public void setMinSpareThreads(int minSpareThreads) {
this.minSpareThreads = minSpareThreads;
if (running && executor!=null) {
if (executor instanceof java.util.concurrent.ThreadPoolExecutor) {
((java.util.concurrent.ThreadPoolExecutor)executor).setCorePoolSize(minSpareThreads);
} else if (executor instanceof ResizableExecutor) {
((ResizableExecutor)executor).resizePool(minSpareThreads, maxThreads);
}
}
}
/**
* Maximum amount of worker threads.
*/
private int maxThreads = 200;
public void setMaxThreads(int maxThreads) {
this.maxThreads = maxThreads;
if (running && executor!=null) {
if (executor instanceof java.util.concurrent.ThreadPoolExecutor) {
((java.util.concurrent.ThreadPoolExecutor)executor).setMaximumPoolSize(maxThreads);
} else if (executor instanceof ResizableExecutor) {
((ResizableExecutor)executor).resizePool(minSpareThreads, maxThreads);
}
}
}
public int getMaxThreads() {
return getMaxThreadsExecutor(running);
}
protected int getMaxThreadsExecutor(boolean useExecutor) {
if (useExecutor && executor != null) {
if (executor instanceof java.util.concurrent.ThreadPoolExecutor) {
return ((java.util.concurrent.ThreadPoolExecutor)executor).getMaximumPoolSize();
} else if (executor instanceof ResizableExecutor) {
return ((ResizableExecutor)executor).getMaxThreads();
} else {
return -1;
}
} else {
return maxThreads;
}
}
/**
* Max keep alive requests
*/
private int maxKeepAliveRequests=100; // as in Apache HTTPD server
public int getMaxKeepAliveRequests() {
return maxKeepAliveRequests;
}
public void setMaxKeepAliveRequests(int maxKeepAliveRequests) {
this.maxKeepAliveRequests = maxKeepAliveRequests;
}
/**
* The maximum number of headers in a request that are allowed.
* 100 by default. A value of less than 0 means no limit.
*/
private int maxHeaderCount = 100; // as in Apache HTTPD server
public int getMaxHeaderCount() {
return maxHeaderCount;
}
public void setMaxHeaderCount(int maxHeaderCount) {
this.maxHeaderCount = maxHeaderCount;
}
/**
* Name of the thread pool, which will be used for naming child threads.
*/
private String name = "TP";
public void setName(String name) { this.name = name; }
public String getName() { return name; }
/**
* The default is true - the created threads will be
* in daemon mode. If set to false, the control thread
* will not be daemon - and will keep the process alive.
*/
private boolean daemon = true;
public void setDaemon(boolean b) { daemon = b; }
public boolean getDaemon() { return daemon; }
/**
* Priority of the worker threads.
*/
protected int threadPriority = Thread.NORM_PRIORITY;
public void setThreadPriority(int threadPriority) { this.threadPriority = threadPriority; }
public int getThreadPriority() { return threadPriority; }
protected abstract boolean getDeferAccept();
protected final List<String> negotiableProtocols = new ArrayList<>();
public void addNegotiatedProtocol(String negotiableProtocol) {
negotiableProtocols.add(negotiableProtocol);
}
/**
* Attributes provide a way for configuration to be passed to sub-components
* without the {@link org.apache.coyote.ProtocolHandler} being aware of the
* properties available on those sub-components.
*/
protected HashMap<String, Object> attributes = new HashMap<>();
/**
* Generic property setter called when a property for which a specific
* setter already exists within the
* {@link org.apache.coyote.ProtocolHandler} needs to be made available to
* sub-components. The specific setter will call this method to populate the
* attributes.
*
* @param name Name of property to set
* @param value The value to set the property to
*/
public void setAttribute(String name, Object value) {
if (getLog().isTraceEnabled()) {
getLog().trace(sm.getString("endpoint.setAttribute", name, value));
}
attributes.put(name, value);
}
/**
* Used by sub-components to retrieve configuration information.
*
* @param key The name of the property for which the value should be
* retrieved
*
* @return The value of the specified property
*/
public Object getAttribute(String key) {
Object value = attributes.get(key);
if (getLog().isTraceEnabled()) {
getLog().trace(sm.getString("endpoint.getAttribute", key, value));
}
return value;
}
public boolean setProperty(String name, String value) {
setAttribute(name, value);
final String socketName = "socket.";
try {
if (name.startsWith(socketName)) {
return IntrospectionUtils.setProperty(socketProperties, name.substring(socketName.length()), value);
} else {
return IntrospectionUtils.setProperty(this,name,value,false);
}
}catch ( Exception x ) {
getLog().error("Unable to set attribute \""+name+"\" to \""+value+"\"",x);
return false;
}
}
public String getProperty(String name) {
return (String) getAttribute(name);
}
/**
* Return the amount of threads that are managed by the pool.
*
* @return the amount of threads that are managed by the pool
*/
public int getCurrentThreadCount() {
if (executor!=null) {
if (executor instanceof ThreadPoolExecutor) {
return ((ThreadPoolExecutor)executor).getPoolSize();
} else if (executor instanceof ResizableExecutor) {
return ((ResizableExecutor)executor).getPoolSize();
} else {
return -1;
}
} else {
return -2;
}
}
/**
* Return the amount of threads that are in use
*
* @return the amount of threads that are in use
*/
public int getCurrentThreadsBusy() {
if (executor!=null) {
if (executor instanceof ThreadPoolExecutor) {
return ((ThreadPoolExecutor)executor).getActiveCount();
} else if (executor instanceof ResizableExecutor) {
return ((ResizableExecutor)executor).getActiveCount();
} else {
return -1;
}
} else {
return -2;
}
}
public boolean isRunning() {
return running;
}
public boolean isPaused() {
return paused;
}
public void createExecutor() {
internalExecutor = true;
TaskQueue taskqueue = new TaskQueue();
TaskThreadFactory tf = new TaskThreadFactory(getName() + "-exec-", daemon, getThreadPriority());
executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), 60, TimeUnit.SECONDS,taskqueue, tf);
taskqueue.setParent( (ThreadPoolExecutor) executor);
}
public void shutdownExecutor() {
if ( executor!=null && internalExecutor ) {
if ( executor instanceof ThreadPoolExecutor ) {
//this is our internal one, so we need to shut it down
ThreadPoolExecutor tpe = (ThreadPoolExecutor) executor;
tpe.shutdownNow();
long timeout = getExecutorTerminationTimeoutMillis();
if (timeout > 0) {
try {
tpe.awaitTermination(timeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
// Ignore
}
if (tpe.isTerminating()) {
getLog().warn(sm.getString("endpoint.warn.executorShutdown", getName()));
}
}
TaskQueue queue = (TaskQueue) tpe.getQueue();
queue.setParent(null);
}
executor = null;
}
}
/**
* Unlock the server socket accept using a bogus connection.
*/
protected void unlockAccept() {
// Only try to unlock the acceptor if it is necessary
boolean unlockRequired = false;
for (Acceptor acceptor : acceptors) {
if (acceptor.getState() == AcceptorState.RUNNING) {
unlockRequired = true;
break;
}
}
if (!unlockRequired) {
return;
}
InetSocketAddress saddr = null;
try {
// Need to create a connection to unlock the accept();
if (address == null) {
saddr = new InetSocketAddress("localhost", getLocalPort());
} else {
saddr = new InetSocketAddress(address, getLocalPort());
}
try (java.net.Socket s = new java.net.Socket()) {
int stmo = 2 * 1000;
int utmo = 2 * 1000;
if (getSocketProperties().getSoTimeout() > stmo)
stmo = getSocketProperties().getSoTimeout();
if (getSocketProperties().getUnlockTimeout() > utmo)
utmo = getSocketProperties().getUnlockTimeout();
s.setSoTimeout(stmo);
s.setSoLinger(getSocketProperties().getSoLingerOn(),getSocketProperties().getSoLingerTime());
if (getLog().isDebugEnabled()) {
getLog().debug("About to unlock socket for:"+saddr);
}
s.connect(saddr,utmo);
if (getDeferAccept()) {
/*
* In the case of a deferred accept / accept filters we need to
* send data to wake up the accept. Send OPTIONS * to bypass
* even BSD accept filters. The Acceptor will discard it.
*/
OutputStreamWriter sw;
sw = new OutputStreamWriter(s.getOutputStream(), "ISO-8859-1");
sw.write("OPTIONS * HTTP/1.0\r\n" +
"User-Agent: Tomcat wakeup connection\r\n\r\n");
sw.flush();
}
if (getLog().isDebugEnabled()) {
getLog().debug("Socket unlock completed for:"+saddr);
}
// Wait for upto 1000ms acceptor threads to unlock
long waitLeft = 1000;
for (Acceptor acceptor : acceptors) {
while (waitLeft > 0 &&
acceptor.getState() == AcceptorState.RUNNING) {
Thread.sleep(50);
waitLeft -= 50;
}
}
}
} catch(Exception e) {
if (getLog().isDebugEnabled()) {
getLog().debug(sm.getString("endpoint.debug.unlock", "" + getPort()), e);
}
}
}
// ---------------------------------------------- Request processing methods
/**
* Process the given SocketWrapper with the given status. Used to trigger
* processing as if the Poller (for those endpoints that have one)
* selected the socket.
*
* @param socketWrapper The socket wrapper to process
* @param socketStatus The input status to the processing
* @param dispatch Should the processing be performed on a new
* container thread
*/
public abstract void processSocket(SocketWrapperBase<S> socketWrapper,
SocketStatus socketStatus, boolean dispatch);
public void executeNonBlockingDispatches(SocketWrapperBase<S> socketWrapper) {
/*
* This method is called when non-blocking IO is initiated by defining
* a read and/or write listener in a non-container thread. It is called
* once the non-container thread completes so that the first calls to
* onWritePossible() and/or onDataAvailable() as appropriate are made by
* the container.
*
* Processing the dispatches requires (for APR/native at least)
* that the socket has been added to the waitingRequests queue. This may
* not have occurred by the time that the non-container thread completes
* triggering the call to this method. Therefore, the coded syncs on the
* SocketWrapper as the container thread that initiated this
* non-container thread holds a lock on the SocketWrapper. The container
* thread will add the socket to the waitingRequests queue before
* releasing the lock on the socketWrapper. Therefore, by obtaining the
* lock on socketWrapper before processing the dispatches, we can be
* sure that the socket has been added to the waitingRequests queue.
*/
synchronized (socketWrapper) {
Iterator<DispatchType> dispatches = socketWrapper.getIteratorAndClearDispatches();
while (dispatches != null && dispatches.hasNext()) {
DispatchType dispatchType = dispatches.next();
processSocket(socketWrapper, dispatchType.getSocketStatus(), false);
}
}
}
// ------------------------------------------------------- Lifecycle methods
/*
* NOTE: There is no maintenance of state or checking for valid transitions
* within this class other than ensuring that bind/unbind are called in the
* right place. It is expected that the calling code will maintain state and
* prevent invalid state transitions.
*/
public abstract void bind() throws Exception;
public abstract void unbind() throws Exception;
public abstract void startInternal() throws Exception;
public abstract void stopInternal() throws Exception;
public final void init() throws Exception {
if (bindOnInit) {
bind();
bindState = BindState.BOUND_ON_INIT;
}
}
public final void start() throws Exception {
if (bindState == BindState.UNBOUND) {
bind();
bindState = BindState.BOUND_ON_START;
}
startInternal();
}
protected final void startAcceptorThreads() {
int count = getAcceptorThreadCount();
acceptors = new Acceptor[count];
for (int i = 0; i < count; i++) {
acceptors[i] = createAcceptor();
String threadName = getName() + "-Acceptor-" + i;
acceptors[i].setThreadName(threadName);
Thread t = new Thread(acceptors[i], threadName);
t.setPriority(getAcceptorThreadPriority());
t.setDaemon(getDaemon());
t.start();
}
}
/**
* Hook to allow Endpoints to provide a specific Acceptor implementation.
*/
protected abstract Acceptor createAcceptor();
/**
* Pause the endpoint, which will stop it accepting new connections.
*/
public void pause() {
if (running && !paused) {
paused = true;
unlockAccept();
}
}
/**
* Resume the endpoint, which will make it start accepting new connections
* again.
*/
public void resume() {
if (running) {
paused = false;
}
}
public final void stop() throws Exception {
stopInternal();
if (bindState == BindState.BOUND_ON_START) {
unbind();
bindState = BindState.UNBOUND;
}
}
public final void destroy() throws Exception {
if (bindState == BindState.BOUND_ON_INIT) {
unbind();
bindState = BindState.UNBOUND;
}
}
protected abstract Log getLog();
protected LimitLatch initializeConnectionLatch() {
if (maxConnections==-1) return null;
if (connectionLimitLatch==null) {
connectionLimitLatch = new LimitLatch(getMaxConnections());
}
return connectionLimitLatch;
}
protected void releaseConnectionLatch() {
LimitLatch latch = connectionLimitLatch;
if (latch!=null) latch.releaseAll();
connectionLimitLatch = null;
}
protected void countUpOrAwaitConnection() throws InterruptedException {
if (maxConnections==-1) return;
LimitLatch latch = connectionLimitLatch;
if (latch!=null) latch.countUpOrAwait();
}
protected long countDownConnection() {
if (maxConnections==-1) return -1;
LimitLatch latch = connectionLimitLatch;
if (latch!=null) {
long result = latch.countDown();
if (result<0) {
getLog().warn("Incorrect connection count, multiple socket.close called on the same socket." );
}
return result;
} else return -1;
}
/**
* Provides a common approach for sub-classes to handle exceptions where a
* delay is required to prevent a Thread from entering a tight loop which
* will consume CPU and may also trigger large amounts of logging. For
* example, this can happen with the Acceptor thread if the ulimit for open
* files is reached.
*
* @param currentErrorDelay The current delay being applied on failure
* @return The delay to apply on the next failure
*/
protected int handleExceptionWithDelay(int currentErrorDelay) {
// Don't delay on first exception
if (currentErrorDelay > 0) {
try {
Thread.sleep(currentErrorDelay);
} catch (InterruptedException e) {
// Ignore
}
}
// On subsequent exceptions, start the delay at 50ms, doubling the delay
// on every subsequent exception until the delay reaches 1.6 seconds.
if (currentErrorDelay == 0) {
return INITIAL_ERROR_DELAY;
} else if (currentErrorDelay < MAX_ERROR_DELAY) {
return currentErrorDelay * 2;
} else {
return MAX_ERROR_DELAY;
}
}
// -------------------- SSL related properties --------------------
private String sslImplementationName = null;
public String getSslImplementationName() { return sslImplementationName; }
public void setSslImplementationName(String s) {
this.sslImplementationName = s;
}
private String sessionCacheSize = null;
public String getSessionCacheSize() { return sessionCacheSize;}
public void setSessionCacheSize(String s) { sessionCacheSize = s;}
private String sessionTimeout = "86400";
public String getSessionTimeout() { return sessionTimeout;}
public void setSessionTimeout(String s) { sessionTimeout = s;}
protected final Set<SocketWrapperBase<S>> waitingRequests = Collections
.newSetFromMap(new ConcurrentHashMap<SocketWrapperBase<S>, Boolean>());
/**
* The async timeout thread.
*/
private AsyncTimeout asyncTimeout = null;
public AsyncTimeout getAsyncTimeout() {
return asyncTimeout;
}
public void setAsyncTimeout(AsyncTimeout asyncTimeout) {
this.asyncTimeout = asyncTimeout;
}
}