[jetty-user] 回复： [jetty-user] 回复： [je tty-user] why jetty7 is slower than jetty6?

thanks for the reply again.
I tried  -Dorg.eclipse.jetty.io.nio.JVMBUG_THRESHHOLD=51200 in the start.ini, and it didnt work.In fact,after having replicated more than 100M,it became greatly slow about 1M every a few seconds.

For solr's replication,it use a long http connection for one file and just use OutputStream to write and read data in batch. Jetty6 is really good for it.So,does Jetty7 change something  to make it bad?I want to know if you have test cases to transmit large file with OutputStream every 1M from one jetty server to another to reappear the problem? Of course, I can do this if I have some time this week.

BTW:the FastOutputStream is a OutputStream wraper like:
public class FastOutputStream extends OutputStream implements DataOutput {
  private final OutputStream out;
  private final byte[] buf;
  private long written;  // how many bytes written
  private int pos;

  public FastOutputStream(OutputStream w) {
  // use default BUFSIZE of BufferedOutputStream so if we wrap that
  // it won't cause double buffering.
    this(w, new byte[8192], 0);
  }

  public FastOutputStream(OutputStream sink, byte[] tempBuffer, int start) {
    this.out = sink;
    this.buf = tempBuffer;
    this.pos = start;
  }


  public static FastOutputStream wrap(OutputStream sink) {
   return (sink instanceof FastOutputStream) ? (FastOutputStream)sink : new FastOutputStream(sink);
  }

  @Override
  public void write(int b) throws IOException {
    write((byte)b);
  }

  public void write(byte b[]) throws IOException {
    write(b,0,b.length);
  }

  public void write(byte b) throws IOException {
    if (pos >= buf.length) {
      out.write(buf);
      written += pos;
      pos=0;
    }
    buf[pos++] = b;
  }

  @Override
  public void write(byte arr[], int off, int len) throws IOException {
    int space = buf.length - pos;
    if (len < space) {
      System.arraycopy(arr, off, buf, pos, len);
      pos += len;
    } else if (len<buf.length) {
      // if the data to write is small enough, buffer it.
      System.arraycopy(arr, off, buf, pos, space);
      out.write(buf);
      written += buf.length;
      pos = len-space;
      System.arraycopy(arr, off+space, buf, 0, pos);
    } else {
      if (pos>0) {
        out.write(buf,0,pos);  // flush
        written += pos;
        pos=0;
      }
      // don't buffer, just write to sink
      out.write(arr, off, len);
      written += len;            
    }
  }

  /** reserve at least len bytes at the end of the buffer.
   * Invalid if len > buffer.length
   * @param len
   */
  public void reserve(int len) throws IOException {
    if (len > (buf.length - pos))
      flushBuffer();
  }

  ////////////////// DataOutput methods ///////////////////
  public void writeBoolean(boolean v) throws IOException {
    write(v ? 1:0);
  }

  public void writeByte(int v) throws IOException {
    write((byte)v);
  }

  public void writeShort(int v) throws IOException {
    write((byte)(v >>> 8));
    write((byte)v);
  }

  public void writeChar(int v) throws IOException {
    writeShort(v);
  }

  public void writeInt(int v) throws IOException {
    reserve(4);
    buf[pos] = (byte)(v>>>24);
    buf[pos+1] = (byte)(v>>>16);
    buf[pos+2] = (byte)(v>>>8);
    buf[pos+3] = (byte)(v);
    pos+=4;
  }

  public void writeLong(long v) throws IOException {
    reserve(8);
    buf[pos] = (byte)(v>>>56);
    buf[pos+1] = (byte)(v>>>48);
    buf[pos+2] = (byte)(v>>>40);
    buf[pos+3] = (byte)(v>>>32);
    buf[pos+4] = (byte)(v>>>24);
    buf[pos+5] = (byte)(v>>>16);
    buf[pos+6] = (byte)(v>>>8);
    buf[pos+7] = (byte)(v);
    pos+=8;
  }

  public void writeFloat(float v) throws IOException {
    writeInt(Float.floatToRawIntBits(v));
  }

  public void writeDouble(double v) throws IOException {
    writeLong(Double.doubleToRawLongBits(v));
  }

  public void writeBytes(String s) throws IOException {
    // non-optimized version, but this shouldn't be used anyway
    for (int i=0; i<s.length(); i++)
      write((byte)s.charAt(i));
  }

  public void writeChars(String s) throws IOException {
    // non-optimized version
    for (int i=0; i<s.length(); i++)
      writeChar(s.charAt(i)); 
  }

  public void writeUTF(String s) throws IOException {
    // non-optimized version, but this shouldn't be used anyway
    DataOutputStream daos = new DataOutputStream(this);
    daos.writeUTF(s);
  }


  @Override
  public void flush() throws IOException {
    flushBuffer();
    out.flush();
  }

  @Override
  public void close() throws IOException {
    flushBuffer();
    out.close();
  }

  /** Only flushes the buffer of the FastOutputStream, not that of the
   * underlying stream.
   */
  public void flushBuffer() throws IOException {
    out.write(buf, 0, pos);
    written += pos;
    pos=0;
  }

  public long size() {
    return written + pos;
  }
}


Thanks.

----------------- 原始邮件 ------------------发件人: "Greg Wilkins"<[email protected]>;
发送时间: 2010年11月15日(星期一) 上午9:16
收件人: "user"<[email protected]>; 

主题: Re: [jetty-user] 回复： [jetty-user] why jetty7 is slower than jetty6?

 
AH!

I think your code is falling victim to a wrongly detected JVM bug (and
work around).

Can you try running with

 -Dorg.eclipse.jetty.io.nio.JVMBUG_THRESHHOLD=51200

and see if you get better results.




On 15 November 2010 11:47:21 UTC+11, gregw <[email protected]> wrote:
> Also can you tell my what the FastOutputStream is?
>
> thanks
>
>
> On 15 November 2010 09:33, Greg Wilkins <[email protected]> wrote:
>> Kafka,
>>
>> that 998ms is very suspiciously close to 1s. It looks like a write is
>> getting lost somehow and only unblocked as the selector wakes up.
>>
>> I'll try to replicate this, but I need some more information.  Can
>> you capture the headers at the start of a response.  How frequently
>> do you send a 1M lump? how long are the pauses between?  Can you
>> measure how much time each write takes (and the size of the byte array
>> written) plus the time the flush takes.
>>
>> I'd also consider not flushing in the loop - but that should not cause
>> the 1s delay.
>>
>> Note also, while we do need to find/fix this issue, there are much
>> much better ways of writing large data that will give you much better
>> performance and use a lot less memory.  It looks like your data is
>> coming from a file, so you really want to let jetty use NIO file
>> mapping to send the content and you typically will get a 10x
>> improvement in data rate plus no use space buffers are needed.   The
>> down side is that you can't use a long held request and stream out
>> data over time - you will have to send 1 file per response. But you
>> can long poll and hold onto a request while waiting for a file to be
>> ready to be sent, and then let the default servlet send it.    I
>> think the way you have written your writer, the data is probably being
>> copied File system to Buffer, Buffer to byte array, byte array to
>> jetty buffer, jetty buffer to OS Buffer, OS Buffer to network
>> interface.  If you can arrange to use NIO file mapped buffers, then
>> the data will be copied File system to network interface.
>>
>> But none the less.... I think there is a bug we need to fix.
>>
>> cheers
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> 2010/11/13 kafka0102 <[email protected]>:
>>> thanks for reply.
>>> I tried jetty-distribution-7.2.1.v20101111, but it still had poor
>>> performance for solr's replication like 7.1.6.v20100715. I embed jetty
>>> 6.1.25(maven) to get great performance.I can give some solr's replication
>>> codes. For master's write, it's like:
>>>
>>> public void write(final OutputStream out) throws IOException {
>>>  final String fileName = params.get(FILE);
>>>  final String cfileName = params.get(CONF_FILE_SHORT);
>>>  final String sOffset = params.get(OFFSET);
>>>  final String sLen = params.get(LEN);
>>>  final String compress = params.get(COMPRESSION);
>>>  final String sChecksum = params.get(CHECKSUM);
>>>  final String sindexVersion = params.get(CMD_INDEX_VERSION);
>>>  if (sindexVersion != null) {
>>>  indexVersion = Long.parseLong(sindexVersion);
>>>  }
>>>  if (Boolean.parseBoolean(compress)) {
>>>  fos = new FastOutputStream(new DeflaterOutputStream(out));
>>>  } else {
>>>  fos = new FastOutputStream(out);
>>>  }
>>>  FileInputStream inputStream = null;
>>>  int packetsWritten = 0;
>>>  try {
>>>  long offset = -1;
>>>  int len = -1;
>>>  //check if checksum is requested
>>>  final boolean useChecksum = Boolean.parseBoolean(sChecksum);
>>>  if (sOffset != null) {
>>>  offset = Long.parseLong(sOffset);
>>>  }
>>>  if (sLen != null) {
>>>  len = Integer.parseInt(sLen);
>>>  }
>>>  if (fileName == null && cfileName == null) {
>>>  //no filename do nothing
>>>  writeNothing();
>>>  }
>>>
>>>  File file = null;
>>>  if (cfileName != null) {
>>>  //if if is a conf file read from config diectory
>>>  file = new File(core.getResourceLoader().getConfigDir(),
>>> cfileName);
>>>  } else {
>>>  //else read from the indexdirectory
>>>  file = new File(core.getIndexDir(), fileName);
>>>  }
>>>  if (file.exists() && file.canRead()) {
>>>  inputStream = new FileInputStream(file);
>>>  final FileChannel channel = inputStream.getChannel();
>>>  //if offset is mentioned move the pointer to that point
>>>  if (offset != -1) {
>>>  channel.position(offset);
>>>  }
>>>  final byte[] buf = new byte[len == -1 || len > PACKET_SZ ?
>>> PACKET_SZ : len];
>>>  Checksum checksum = null;
>>>  if (useChecksum) {
>>>  checksum = new Adler32();
>>>  }
>>>  final ByteBuffer bb = ByteBuffer.wrap(buf);
>>>
>>>  while (true) {
>>>  bb.clear();
>>>  final long bytesRead = channel.read(bb);
>>>  if (bytesRead <= 0) {
>>>  writeNothing();
>>>  fos.close();
>>>  break;
>>>  }
>>>  fos.writeInt((int) bytesRead);
>>>  if (useChecksum) {
>>>  checksum.reset();
>>>  checksum.update(buf, 0, (int) bytesRead);
>>>  fos.writeLong(checksum.getValue());
>>>  }
>>>  // final long startTime = System.currentTimeMillis();
>>>  fos.write(buf, 0, (int) bytesRead);
>>>  fos.flush();
>>>  // LOG.info("write "+bytesRead+"
>>> cost:"+(System.currentTimeMillis() - startTime));
>>>  if (indexVersion != null && packetsWritten % 5 == 0) {
>>>  //after every 5 packets reserve the commitpoint for some time
>>>  delPolicy.setReserveDuration(indexVersion,
>>> reserveCommitDuration);
>>>  }
>>>  packetsWritten++;
>>>  }
>>>  } else {
>>>  writeNothing();
>>>  }
>>>  } catch (final IOException e) {
>>>  LOG.warn("Exception while writing response for params: " + params,
>>> e);
>>>  } finally {
>>>  IOUtils.closeQuietly(inputStream);
>>>  }
>>>  }
>>>
>>> for slave's read,it's like:
>>>
>>> private int fetchPackets(final FastInputStream fis) throws Exception {
>>>  final byte[] intbytes = new byte[4];
>>>  final byte[] longbytes = new byte[8];
>>>  try {
>>>  while (true) {
>>>  if (stop) {
>>>  stop = false;
>>>  aborted = true;
>>>  throw new ReplicationHandlerException("User aborted
>>> replication");
>>>  }
>>>  long checkSumServer = -1;
>>>  fis.readFully(intbytes);
>>>  //read the size of the packet
>>>  final int packetSize = readInt(intbytes);
>>>  if (packetSize <= 0) {
>>>  LOG.warn("No content recieved for file: " + currentFile);
>>>  return NO_CONTENT;
>>>  }
>>>  if (buf.length < packetSize) {
>>>  buf = new byte[packetSize];
>>>  }
>>>  if (checksum != null) {
>>>  //read the checksum
>>>  fis.readFully(longbytes);
>>>  checkSumServer = readLong(longbytes);
>>>  }
>>>  //then read the packet of bytes
>>>  final long startTime = System.currentTimeMillis();
>>>  fis.readFully(buf, 0, packetSize);
>>>  // LOG.info("readFully"+packetSize+" cost " +
>>> (System.currentTimeMillis() - startTime));
>>>  //compare the checksum as sent from the master
>>>  if (includeChecksum) {
>>>  checksum.reset();
>>>  checksum.update(buf, 0, packetSize);
>>>  final long checkSumClient = checksum.getValue();
>>>  if (checkSumClient != checkSumServer) {
>>>  LOG.error("Checksum not matched between client and server for:
>>> " + currentFile);
>>>  //if checksum is wrong it is a problem return for retry
>>>  return 1;
>>>  }
>>>  }
>>>  //if everything is fine, write down the packet to the file
>>>  fileChannel.write(ByteBuffer.wrap(buf, 0, packetSize));
>>>  bytesDownloaded += packetSize;
>>>  if (bytesDownloaded >= size) {
>>>  return 0;
>>>  }
>>>  //errorcount is always set to zero after a successful packet
>>>  errorCount = 0;
>>>  }
>>>  } catch (final ReplicationHandlerException e) {
>>>  throw e;
>>>  } catch (final Exception e) {
>>>  LOG.warn("Error in fetching packets ", e);
>>>  //for any failure , increment the error count
>>>  errorCount++;
>>>  //if it fails for the same pacaket for MAX_RETRIES fail and come
>>> out
>>>  if (errorCount > MAX_RETRIES) {
>>>  throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
>>>  "Fetch failed for file:" + fileName, e);
>>>  }
>>>  return ERR;
>>>  }
>>>  }
>>>
>>> solr writes PACKET_SZ(1M default) every time.I find one time is too
>>> slow(more than 900ms) every 3 times.When I change PACKET_SZ to 10M,it's
>>> about 3s every time.The effect is still like PACKET_SZ = 1M.
>>>
>>>
>>>
>>>
>>> ------------------原始邮件------------------
>>> 发件人:"Jesse McConnell"<[email protected]>;
>>> 发送时间:2010年11月12日(星期五) 晚上9:53
>>> 收件人:"user"<[email protected]>;
>>> 主题:Re: [jetty-user] why jetty7 is slower than jetty6?
>>>
>>> which version of jetty? we have resolved an issue that might apply to
>>> your case were performance was reduced, its resolved in the staged
>>> 7.2.1.v20101111 build that we'll likely promote soon..
>>>
>>> you can try it at:
>>>
>>> https://oss.sonatype.org/content/repositories/jetty-024/org/eclipse/jetty/jetty-distribution/7.2.1.v20101111/
>>>
>>> cheers,
>>> jesse
>>>
>>> --
>>> jesse mcconnell
>>> [email protected]
>>>
>>>
>>>
>>> On Fri, Nov 12, 2010 at 05:13, kafka0102 <[email protected]> wrote:
>>>> I use jetty as a web server for solr, and I find a problem. solr replicate
>>>> data from master to slave, and master uses a long http connection to write
>>>> data 1M every time.I print the time cost every 1M transit, and find
>>>> default's jetty7 cost is like 996ms,5ms,6ms,998ms,4ms,8ms,thus one time
>>>> cost
>>>> about 1s every 3 times. If I change to jetty6,it's so fast, and every time
>>>> just cost less then 1ms.For jetty6 and jetty7,I just use default's conf,
>>>> and
>>>> the problem consist in both embed and external jetty . Who can tell me
>>>> what
>>>> I'm wrong?
>>>>
>>>
>>> ---------------------------------------------------------------------
>>> To unsubscribe from this list, please visit:
>>>
>>>  http://xircles.codehaus.org/manage_email
>>>
>>>
>>>
>>
>
>