1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.io;
20
21 import java.io.Closeable;
22 import java.io.IOException;
23 import java.nio.ByteBuffer;
24 import java.nio.channels.ByteChannel;
25 import java.nio.channels.CancelledKeyException;
26 import java.nio.channels.GatheringByteChannel;
27 import java.nio.channels.SelectionKey;
28
29 import org.eclipse.jetty.util.BufferUtil;
30 import org.eclipse.jetty.util.log.Log;
31 import org.eclipse.jetty.util.log.Logger;
32 import org.eclipse.jetty.util.thread.Invocable;
33 import org.eclipse.jetty.util.thread.Locker;
34 import org.eclipse.jetty.util.thread.Scheduler;
35
36
37
38
39
40 public abstract class ChannelEndPoint extends AbstractEndPoint implements ManagedSelector.Selectable
41 {
42 private static final Logger LOG = Log.getLogger(ChannelEndPoint.class);
43
44 private final Locker _locker = new Locker();
45 private final ByteChannel _channel;
46 private final GatheringByteChannel _gather;
47 protected final ManagedSelector _selector;
48 protected final SelectionKey _key;
49 private boolean _updatePending;
50
51
52
53
54 protected int _currentInterestOps;
55
56
57
58
59 protected int _desiredInterestOps;
60
61 private abstract class RunnableTask implements Runnable, Invocable
62 {
63 final String _operation;
64
65 protected RunnableTask(String op)
66 {
67 _operation=op;
68 }
69
70 @Override
71 public String toString()
72 {
73 return String.format("CEP:%s:%s:%s",ChannelEndPoint.this,_operation,getInvocationType());
74 }
75 }
76
77 private abstract class RunnableCloseable extends RunnableTask implements Closeable
78 {
79 protected RunnableCloseable(String op)
80 {
81 super(op);
82 }
83
84 @Override
85 public void close()
86 {
87 try
88 {
89 ChannelEndPoint.this.close();
90 }
91 catch (Throwable x)
92 {
93 LOG.warn(x);
94 }
95 }
96 }
97
98 private final Runnable _runUpdateKey = new RunnableTask("runUpdateKey")
99 {
100 @Override
101 public InvocationType getInvocationType()
102 {
103 return InvocationType.NON_BLOCKING;
104 }
105
106 @Override
107 public void run()
108 {
109 updateKey();
110 }
111 };
112
113 private final Runnable _runFillable = new RunnableCloseable("runFillable")
114 {
115 @Override
116 public InvocationType getInvocationType()
117 {
118 return getFillInterest().getCallbackInvocationType();
119 }
120
121 @Override
122 public void run()
123 {
124 getFillInterest().fillable();
125 }
126 };
127
128 private final Runnable _runCompleteWrite = new RunnableCloseable("runCompleteWrite")
129 {
130 @Override
131 public InvocationType getInvocationType()
132 {
133 return getWriteFlusher().getCallbackInvocationType();
134 }
135
136 @Override
137 public void run()
138 {
139 getWriteFlusher().completeWrite();
140 }
141
142 @Override
143 public String toString()
144 {
145 return String.format("CEP:%s:%s:%s->%s",ChannelEndPoint.this,_operation,getInvocationType(),getWriteFlusher());
146 }
147
148 };
149
150 private final Runnable _runCompleteWriteFillable = new RunnableCloseable("runCompleteWriteFillable")
151 {
152 @Override
153 public InvocationType getInvocationType()
154 {
155 InvocationType fillT = getFillInterest().getCallbackInvocationType();
156 InvocationType flushT = getWriteFlusher().getCallbackInvocationType();
157 if (fillT==flushT)
158 return fillT;
159
160 if (fillT==InvocationType.EITHER && flushT==InvocationType.NON_BLOCKING)
161 return InvocationType.EITHER;
162
163 if (fillT==InvocationType.NON_BLOCKING && flushT==InvocationType.EITHER)
164 return InvocationType.EITHER;
165
166 return InvocationType.BLOCKING;
167 }
168
169 @Override
170 public void run()
171 {
172 getWriteFlusher().completeWrite();
173 getFillInterest().fillable();
174 }
175 };
176
177 public ChannelEndPoint(ByteChannel channel, ManagedSelector selector, SelectionKey key, Scheduler scheduler)
178 {
179 super(scheduler);
180 _channel=channel;
181 _selector=selector;
182 _key=key;
183 _gather=(channel instanceof GatheringByteChannel)?(GatheringByteChannel)channel:null;
184 }
185
186 @Override
187 public boolean isOptimizedForDirectBuffers()
188 {
189 return true;
190 }
191
192 @Override
193 public boolean isOpen()
194 {
195 return _channel.isOpen();
196 }
197
198 @Override
199 public void doClose()
200 {
201 if (LOG.isDebugEnabled())
202 LOG.debug("doClose {}", this);
203 try
204 {
205 _channel.close();
206 }
207 catch (IOException e)
208 {
209 LOG.debug(e);
210 }
211 finally
212 {
213 super.doClose();
214 }
215 }
216
217 @Override
218 public void onClose()
219 {
220 try
221 {
222 super.onClose();
223 }
224 finally
225 {
226 if (_selector!=null)
227 _selector.destroyEndPoint(this);
228 }
229 }
230
231 @Override
232 public int fill(ByteBuffer buffer) throws IOException
233 {
234 if (isInputShutdown())
235 return -1;
236
237 int pos=BufferUtil.flipToFill(buffer);
238 try
239 {
240 int filled = _channel.read(buffer);
241 if (LOG.isDebugEnabled())
242 LOG.debug("filled {} {}", filled, this);
243
244 if (filled>0)
245 notIdle();
246 else if (filled==-1)
247 shutdownInput();
248
249 return filled;
250 }
251 catch(IOException e)
252 {
253 LOG.debug(e);
254 shutdownInput();
255 return -1;
256 }
257 finally
258 {
259 BufferUtil.flipToFlush(buffer,pos);
260 }
261 }
262
263 @Override
264 public boolean flush(ByteBuffer... buffers) throws IOException
265 {
266 long flushed=0;
267 try
268 {
269 if (buffers.length==1)
270 flushed=_channel.write(buffers[0]);
271 else if (_gather!=null && buffers.length>1)
272 flushed=_gather.write(buffers,0,buffers.length);
273 else
274 {
275 for (ByteBuffer b : buffers)
276 {
277 if (b.hasRemaining())
278 {
279 int l=_channel.write(b);
280 if (l>0)
281 flushed+=l;
282 if (b.hasRemaining())
283 break;
284 }
285 }
286 }
287 if (LOG.isDebugEnabled())
288 LOG.debug("flushed {} {}", flushed, this);
289 }
290 catch (IOException e)
291 {
292 throw new EofException(e);
293 }
294
295 if (flushed>0)
296 notIdle();
297
298 for (ByteBuffer b : buffers)
299 if (!BufferUtil.isEmpty(b))
300 return false;
301
302 return true;
303 }
304
305 public ByteChannel getChannel()
306 {
307 return _channel;
308 }
309
310 @Override
311 public Object getTransport()
312 {
313 return _channel;
314 }
315
316
317 @Override
318 protected void needsFillInterest()
319 {
320 changeInterests(SelectionKey.OP_READ);
321 }
322
323 @Override
324 protected void onIncompleteFlush()
325 {
326 changeInterests(SelectionKey.OP_WRITE);
327 }
328
329 @Override
330 public Runnable onSelected()
331 {
332
333
334
335
336 int readyOps = _key.readyOps();
337 int oldInterestOps;
338 int newInterestOps;
339 try (Locker.Lock lock = _locker.lock())
340 {
341 _updatePending = true;
342
343 oldInterestOps = _desiredInterestOps;
344 newInterestOps = oldInterestOps & ~readyOps;
345 _desiredInterestOps = newInterestOps;
346 }
347
348 boolean fillable = (readyOps & SelectionKey.OP_READ) != 0;
349 boolean flushable = (readyOps & SelectionKey.OP_WRITE) != 0;
350
351 if (LOG.isDebugEnabled())
352 LOG.debug("onSelected {}->{} r={} w={} for {}", oldInterestOps, newInterestOps, fillable, flushable, this);
353
354
355 Runnable task= fillable
356 ? (flushable
357 ? _runCompleteWriteFillable
358 : _runFillable)
359 : (flushable
360 ? _runCompleteWrite
361 : null);
362
363 if (LOG.isDebugEnabled())
364 LOG.debug("task {}",task);
365 return task;
366 }
367
368 @Override
369 public void updateKey()
370 {
371
372
373
374
375 try
376 {
377 int oldInterestOps;
378 int newInterestOps;
379 try (Locker.Lock lock = _locker.lock())
380 {
381 _updatePending = false;
382 oldInterestOps = _currentInterestOps;
383 newInterestOps = _desiredInterestOps;
384 if (oldInterestOps != newInterestOps)
385 {
386 _currentInterestOps = newInterestOps;
387 _key.interestOps(newInterestOps);
388 }
389 }
390
391 if (LOG.isDebugEnabled())
392 LOG.debug("Key interests updated {} -> {} on {}", oldInterestOps, newInterestOps, this);
393 }
394 catch (CancelledKeyException x)
395 {
396 LOG.debug("Ignoring key update for concurrently closed channel {}", this);
397 close();
398 }
399 catch (Throwable x)
400 {
401 LOG.warn("Ignoring key update for " + this, x);
402 close();
403 }
404 }
405
406 private void changeInterests(int operation)
407 {
408
409
410
411
412
413 int oldInterestOps;
414 int newInterestOps;
415 boolean pending;
416 try (Locker.Lock lock = _locker.lock())
417 {
418 pending = _updatePending;
419 oldInterestOps = _desiredInterestOps;
420 newInterestOps = oldInterestOps | operation;
421 if (newInterestOps != oldInterestOps)
422 _desiredInterestOps = newInterestOps;
423 }
424
425 if (LOG.isDebugEnabled())
426 LOG.debug("changeInterests p={} {}->{} for {}", pending, oldInterestOps, newInterestOps, this);
427
428 if (!pending && _selector!=null)
429 _selector.submit(_runUpdateKey);
430 }
431
432
433 @Override
434 public String toEndPointString()
435 {
436
437 try
438 {
439 boolean valid = _key != null && _key.isValid();
440 int keyInterests = valid ? _key.interestOps() : -1;
441 int keyReadiness = valid ? _key.readyOps() : -1;
442 return String.format("%s{io=%d/%d,kio=%d,kro=%d}",
443 super.toEndPointString(),
444 _currentInterestOps,
445 _desiredInterestOps,
446 keyInterests,
447 keyReadiness);
448 }
449 catch (Throwable x)
450 {
451 return String.format("%s{io=%s,kio=-2,kro=-2}", super.toString(), _desiredInterestOps);
452 }
453 }
454 }