jdk-sandbox: comparison src/java.base/share/classes/sun/net/www/http/KeepAliveStream.java

equal deleted inserted replaced

-:e89c7aaf2906
+:8595871a5446
 /*
-* Copyright (c) 1996, 2012, Oracle and/or its affiliates. All rights reserved.
+* Copyright (c) 1996, 2019, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 */
 package sun.net.www.http;
 import java.io.*;
+import java.util.concurrent.locks.ReentrantLock;
 import sun.net.ProgressSource;
 import sun.net.www.MeteredStream;
 import jdk.internal.misc.InnocuousThread;
 /**
 HttpClient hc;
 boolean hurried;
 // has this KeepAliveStream been put on the queue for asynchronous cleanup.
-protected boolean queuedForCleanup = false;
+protected volatile boolean queuedForCleanup = false;
 private static final KeepAliveStreamCleaner queue = new KeepAliveStreamCleaner();
 private static Thread cleanerThread; // null
 /**
 public KeepAliveStream(InputStream is, ProgressSource pi, long expected, HttpClient hc)  {
 super(is, pi, expected);
 this.hc = hc;
 }
+final ReentrantLock readLock() {
+return readLock;
+}
 /**
 * Attempt to cache this connection
 */
 public void close() throws IOException  {
-// If the inputstream is closed already, just return.
+// If the inputstream is closed already, or if this stream
-if (closed) {
+// has already been queued for cleanup.just return.
-return;
+if (closed || queuedForCleanup) return;
-}
-// If this stream has already been queued for cleanup.
-if (queuedForCleanup) {
-return;
-}
 // Skip past the data that's left in the Inputstream because
 // some sort of error may have occurred.
 // Do this ONLY if the skip won't block. The stream may have
 // been closed at the beginning of a big file and we don't want
 // to hang around for nothing. So if we can't skip without blocking
 // we just close the socket and, therefore, terminate the keepAlive
 // NOTE: Don't close super class
+// For consistency, access to `expected` and `count` should be
+// protected by readLock
+readLock.lock();
 try {
-if (expected > count) {
+if (closed || queuedForCleanup) return;
-long nskip = expected - count;
+try {
-if (nskip <= available()) {
+if (expected > count) {
-do {} while ((nskip = (expected - count)) > 0L
+long nskip = expected - count;
-&& skip(Math.min(nskip, available())) > 0L);
+if (nskip <= available()) {
-} else if (expected <= KeepAliveStreamCleaner.MAX_DATA_REMAINING && !hurried) {
+do {
-//put this KeepAliveStream on the queue so that the data remaining
+} while ((nskip = (expected - count)) > 0L
-//on the socket can be cleanup asyncronously.
+&& skip(Math.min(nskip, available())) > 0L);
-queueForCleanup(new KeepAliveCleanerEntry(this, hc));
+} else if (expected <= KeepAliveStreamCleaner.MAX_DATA_REMAINING && !hurried) {
-} else {
+//put this KeepAliveStream on the queue so that the data remaining
-hc.closeServer();
+//on the socket can be cleanup asyncronously.
-}
+queueForCleanup(new KeepAliveCleanerEntry(this, hc));
-}
+} else {
-if (!closed && !hurried && !queuedForCleanup) {
+hc.closeServer();
-hc.finished();
+}
+}
+if (!closed && !hurried && !queuedForCleanup) {
+hc.finished();
+}
+} finally {
+if (pi != null)
+pi.finishTracking();
+if (!queuedForCleanup) {
+// nulling out the underlying inputstream as well as
+// httpClient to let gc collect the memories faster
+in = null;
+hc = null;
+closed = true;
+}
 }
 } finally {
-if (pi != null)
+readLock.unlock();
-pi.finishTracking();
-if (!queuedForCleanup) {
-// nulling out the underlying inputstream as well as
-// httpClient to let gc collect the memories faster
-in = null;
-hc = null;
-closed = true;
-}
 }
 }
 /* we explicitly do not support mark/reset */
 public void reset() throws IOException {
 throw new IOException("mark/reset not supported");
 }
-public synchronized boolean hurry() {
+public boolean hurry() {
+readLock.lock();
 try {
 /* CASE 0: we're actually already done */
 if (closed || count >= expected) {
 return false;
 } else if (in.available() < (expected - count)) {
 return true;
 }
 } catch (IOException e) {
 // e.printStackTrace();
 return false;
+} finally {
+readLock.unlock();
 }
 }
 private static void queueForCleanup(KeepAliveCleanerEntry kace) {
-synchronized(queue) {
+queue.queueLock.lock();
+try {
 if(!kace.getQueuedForCleanup()) {
 if (!queue.offer(kace)) {
 kace.getHttpClient().closeServer();
 return;
 }
 kace.setQueuedForCleanup();
-queue.notifyAll();
+queue.waiter.signalAll();
 }
 boolean startCleanupThread = (cleanerThread == null);
 if (!startCleanupThread) {
 if (!cleanerThread.isAlive()) {
 cleanerThread.start();
 return null;
 }
 });
 }
-} // queue
+} finally {
+queue.queueLock.unlock();
+}
 }
 protected long remainingToRead() {
+assert readLock.isHeldByCurrentThread();
 return expected - count;
 }
 protected void setClosed() {
+assert readLock.isHeldByCurrentThread();
 in = null;
 hc = null;
 closed = true;
 }
 }

branch	JDK-8229867-branch
changeset 57968	8595871a5446
parent 47216	71c04702a3d5