Jetty线程“互锁”导致数据传输性能降低问题分析

Jetty线程“互锁”导致数据传输性能降低问题分析

/* ------------------------------------------------------------ */

    * Allows thread to block waiting for further events.

   @Override

   public boolean blockWritable(long timeoutMs) throws IOException

       synchronized (this)

           long start=_selectSet.getNow();

               _writeBlocked=true;

               while (isOpen() && _writeBlocked)

                       updateKey();

                       this.wait(timeoutMs);

                       timeoutMs -= _selectSet.getNow()-start;

                       if (_writeBlocked && timeoutMs<=0)

                           return false;

                   catch (InterruptedException e)

                       Log.warn(e);

           finally

               _writeBlocked=false;

               if (_idleTimestamp!=-1)

                   scheduleIdle();

       return true;

/* ------------------------------------------------------------ */

    * Select and dispatch tasks found from changes and the selector.

    * @throws IOException

   public void doSelect() throws IOException

           _selecting=Thread.currentThread();

           final Selector selector=_selector;

           // Make any key changes required

           Object change;

           int changes=_changes.size();

           while (changes-->0 && (change=_changes.poll())!=null)

                   if (change instanceof EndPoint)

                       // Update the operations for a key.

                       SelectChannelEndPoint endpoint = (SelectChannelEndPoint)change;

                       endpoint.doUpdateKey();

                   else if (change instanceof ChannelAndAttachment)

                       // finish accepting/connecting this connection

                       final ChannelAndAttachment asc = (ChannelAndAttachment)change;

                       final SelectableChannel channel=asc._channel;

                       final Object att = asc._attachment;

                       SelectionKey key = channel.register(selector,SelectionKey.OP_READ,att);

                       SelectChannelEndPoint endpoint = createEndPoint((SocketChannel)channel,key);

                       key.attach(endpoint);

                       endpoint.schedule();

                   else if (change instanceof SocketChannel)

                       // Newly registered channel

                       final SocketChannel channel=(SocketChannel)change;

                       SelectionKey key = channel.register(selector,SelectionKey.OP_READ,null);

                       SelectChannelEndPoint endpoint = createEndPoint(channel,key);

                       key.attach(endpoint);

                       endpoint.schedule();

                   else if (change instanceof Runnable)

                       dispatch((Runnable)change);

                       throw new IllegalArgumentException(change.toString());

               catch (Exception e)

                   if (isRunning())

                       Log.warn(e);

                       Log.debug(e);

               catch (Error e)

                   if (isRunning())

                       Log.warn(e);

                       Log.debug(e);

           // Do and instant select to see if any connections can be handled.

           int selected=selector.selectNow();

           _selects++;

           long now=System.currentTimeMillis();

           // if no immediate things to do

           if (selected==0)

               // If we are in pausing mode

               if (_pausing)

                       Thread.sleep(__BUSY_PAUSE); // pause to reduce impact of  busy loop

                   catch(InterruptedException e)

                       Log.ignore(e);

                   now=System.currentTimeMillis();

               // workout how long to wait in select

               _timeout.setNow(now);

               long to_next_timeout=_timeout.getTimeToNext();

               long wait = _changes.size()==0?__IDLE_TICK:0L;  

               if (wait > 0 && to_next_timeout >= 0 && wait > to_next_timeout)

                   wait = to_next_timeout;

               // If we should wait with a select

               if (wait>0)

                   long before=now;

                   selected=selector.select(wait);

                   _selects++;

                   now = System.currentTimeMillis();

                   _timeout.setNow(now);

                   checkJvmBugs(before, now, wait, selected);

           // have we been destroyed while sleeping

           if (_selector==null || !selector.isOpen())

               return;

           // Look for things to do

           for (SelectionKey key: selector.selectedKeys())

                   if (!key.isValid())

                       key.cancel();

                       SelectChannelEndPoint endpoint = (SelectChannelEndPoint)key.attachment();

                       if (endpoint != null)

                           endpoint.doUpdateKey();

                       continue;

                   Object att = key.attachment();

                   if (att instanceof SelectChannelEndPoint)

                       ((SelectChannelEndPoint)att).schedule();

                       // Wrap readable registered channel in an endpoint

                       SocketChannel channel = (SocketChannel)key.channel();

                       SelectChannelEndPoint endpoint = createEndPoint(channel,key);

                       key.attach(endpoint);

                       if (key.isReadable())

                           endpoint.schedule();                          

                   key = null;

               catch (CancelledKeyException e)

                   Log.ignore(e);

               catch (Exception e)

                   if (isRunning())

                       Log.warn(e);

                       Log.ignore(e);

                   if (key != null && !(key.channel() instanceof ServerSocketChannel) && key.isValid())

                       key.cancel();

           // Everything always handled

           selector.selectedKeys().clear();

           now=System.currentTimeMillis();

           _timeout.setNow(now);

           Task task = _timeout.expired();

           while (task!=null)

               if (task instanceof Runnable)

                   dispatch((Runnable)task);

               task = _timeout.expired();

           // Idle tick

           if (now-_idleTick>__IDLE_TICK)

               _idleTick=now;

               final long idle_now=((_lowResourcesConnections>0 && selector.keys().size()>_lowResourcesConnections))

                   ?(now+_maxIdleTime-_lowResourcesMaxIdleTime)

                   :now;

               dispatch(new Runnable()

                   public void run()

                       for (SelectChannelEndPoint endp:_endPoints.keySet())

                           endp.checkIdleTimestamp(idle_now);

       catch (CancelledKeyException e)

           Log.ignore(e);

       finally

           _selecting=null;

   /* ------------------------------------------------------------ */

   private void checkJvmBugs(long before, long now, long wait, int selected)

       throws IOException

       Selector selector = _selector;

       if (selector==null)

           return;

       // Look for JVM bugs over a monitor period.

       // http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6403933

       // http://bugs.sun.com/view_bug.do?bug_id=6693490

       if (now>_monitorNext)

           _selects=(int)(_selects*__MONITOR_PERIOD/(now-_monitorStart));

           _pausing=_selects>__MAX_SELECTS;

           if (_pausing)

               _paused++;

           _selects=0;

           _jvmBug=0;

           _monitorStart=now;

           _monitorNext=now+__MONITOR_PERIOD;

       if (now>_log)

           if (_paused>0)  

               Log.debug(this+" Busy selector - injecting delay "+_paused+" times");

           if (_jvmFix2>0)

               Log.debug(this+" JVM BUG(s) - injecting delay"+_jvmFix2+" times");

           if (_jvmFix1>0)

               Log.debug(this+" JVM BUG(s) - recreating selector "+_jvmFix1+" times, cancelled keys "+_jvmFix0+" times");

           else if(Log.isDebugEnabled() && _jvmFix0>0)

               Log.debug(this+" JVM BUG(s) - cancelled keys "+_jvmFix0+" times");

           _paused=0;

           _jvmFix2=0;

           _jvmFix1=0;

           _jvmFix0=0;

           _log=now+60000;

       // If we see signature of possible JVM bug, increment count.

       if (selected==0 && wait>10 && (now-before)<(wait/2))

           // Increment bug count and try a work around

           _jvmBug++;

           if (_jvmBug>(__JVMBUG_THRESHHOLD))

                   if (_jvmBug==__JVMBUG_THRESHHOLD+1)

                       _jvmFix2++;

                   Thread.sleep(__BUSY_PAUSE); // pause to avoid busy loop

               catch(InterruptedException e)

                   Log.ignore(e);

           else if (_jvmBug==__JVMBUG_THRESHHOLD)

               synchronized (this)

                   // BLOODY SUN BUG !!!  Try refreshing the entire selector.

                   final Selector new_selector = Selector.open();

                   for (SelectionKey k: selector.keys())

                       if (!k.isValid() || k.interestOps()==0)

                           continue;

                       final SelectableChannel channel = k.channel();

                       final Object attachment = k.attachment();

                       if (attachment==null)

                           addChange(channel);

                           addChange(channel,attachment);

                   _selector.close();

                   _selector=new_selector;

                   return;

           else if (_jvmBug%32==31) // heuristic attempt to cancel key 31,63,95,... loops

               // Cancel keys with 0 interested ops

               int cancelled=0;

               for (SelectionKey k: selector.keys())

                   if (k.isValid()&&k.interestOps()==0)

                       k.cancel();

                       cancelled++;

               if (cancelled>0)

                   _jvmFix0++;

               return;

       else if (__BUSY_KEY>0 && selected==1 && _selects>__MAX_SELECTS)

           // Look for busy key

           SelectionKey busy = selector.selectedKeys().iterator().next();

           if (busy==_busyKey)

               if (++_busyKeyCount>__BUSY_KEY && !(busy.channel() instanceof ServerSocketChannel))

                   final SelectChannelEndPoint endpoint = (SelectChannelEndPoint)busy.attachment();

                   Log.warn("Busy Key "+busy.channel()+" "+endpoint);

                   busy.cancel();

                   if (endpoint!=null)

                       dispatch(new Runnable()

                           public void run()

                                   endpoint.close();

                               catch (IOException e)

                                   Log.ignore(e);

               _busyKeyCount=0;

           _busyKey=busy;