Skip to content

Commit 5b405f8

Browse files
jean-philippe-martingarrettjonesgoogle
jean-philippe-martin
authored and
garrettjonesgoogle
committed
---
yaml --- r: 5477 b: refs/heads/master c: f72746c h: refs/heads/master i: 5475: 9bb87b0
1 parent 2f555b8 commit 5b405f8

7 files changed

Lines changed: 152 additions & 17 deletions

File tree

‎[refs]‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
---
2-
refs/heads/master: a467f2e9ab25ae53a2eee74e63cd5d2b24819555
2+
refs/heads/master: f72746c6183378664a9326ee406d679926a7def4
33
refs/heads/travis: dae77e558b884bc1b165155482d76c8e40b0fca4
44
refs/heads/gh-pages: 4936f6d1c43be1ab76229d2743bae07f4b4124b3
55
refs/tags/0.0.9: 22f1839238f66c39e67ed4dfdcd273b1ae2e8444

‎trunk/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageConfiguration.java‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,12 @@ public abstract class CloudStorageConfiguration {
5959
*/
6060
public abstract int blockSize();
6161

62+
/**
63+
* Returns the number of times we try re-opening a channel if it's closed unexpectedly
64+
* while reading.
65+
*/
66+
public abstract int maxChannelReopens();
67+
6268
/**
6369
* Creates a new builder, initialized with the following settings:
6470
*
@@ -82,6 +88,7 @@ public static final class Builder {
8288
private boolean stripPrefixSlash = true;
8389
private boolean usePseudoDirectories = true;
8490
private int blockSize = CloudStorageFileSystem.BLOCK_SIZE_DEFAULT;
91+
private int maxChannelReopens = 0;
8592

8693
/**
8794
* Changes current working directory for new filesystem. This defaults to the root directory.
@@ -134,6 +141,11 @@ public Builder blockSize(int value) {
134141
return this;
135142
}
136143

144+
public Builder maxChannelReopens(int value) {
145+
maxChannelReopens = value;
146+
return this;
147+
}
148+
137149
/**
138150
* Creates new instance without destroying builder.
139151
*/
@@ -143,7 +155,8 @@ public CloudStorageConfiguration build() {
143155
permitEmptyPathComponents,
144156
stripPrefixSlash,
145157
usePseudoDirectories,
146-
blockSize);
158+
blockSize,
159+
maxChannelReopens);
147160
}
148161

149162
Builder() {}
@@ -168,6 +181,9 @@ static CloudStorageConfiguration fromMap(Map<String, ?> env) {
168181
case "blockSize":
169182
builder.blockSize((Integer) entry.getValue());
170183
break;
184+
case "maxChannelReopens":
185+
builder.maxChannelReopens((Integer) entry.getValue());
186+
break;
171187
default:
172188
throw new IllegalArgumentException(entry.getKey());
173189
}

‎trunk/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageFileSystemProvider.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -141,7 +141,6 @@ public CloudStorageFileSystemProvider() {
141141

142142
CloudStorageFileSystemProvider(@Nullable StorageOptions gcsStorageOptions) {
143143
this.storageOptions = gcsStorageOptions;
144-
145144
}
146145

147146
// Initialize this.storage, once. This may throw an exception if default authentication
@@ -234,6 +233,7 @@ public SeekableByteChannel newByteChannel(
234233
private SeekableByteChannel newReadChannel(Path path, Set<? extends OpenOption> options)
235234
throws IOException {
236235
initStorage();
236+
int maxChannelReopens = ((CloudStorageFileSystem) path.getFileSystem()).config().maxChannelReopens();
237237
for (OpenOption option : options) {
238238
if (option instanceof StandardOpenOption) {
239239
switch ((StandardOpenOption) option) {
@@ -255,6 +255,8 @@ private SeekableByteChannel newReadChannel(Path path, Set<? extends OpenOption>
255255
default:
256256
throw new UnsupportedOperationException(option.toString());
257257
}
258+
} else if (option instanceof OptionMaxChannelReopens) {
259+
maxChannelReopens = ((OptionMaxChannelReopens) option).maxChannelReopens();
258260
} else {
259261
throw new UnsupportedOperationException(option.toString());
260262
}
@@ -263,7 +265,7 @@ private SeekableByteChannel newReadChannel(Path path, Set<? extends OpenOption>
263265
if (cloudPath.seemsLikeADirectoryAndUsePseudoDirectories()) {
264266
throw new CloudStoragePseudoDirectoryException(cloudPath);
265267
}
266-
return CloudStorageReadChannel.create(storage, cloudPath.getBlobId(), 0);
268+
return CloudStorageReadChannel.create(storage, cloudPath.getBlobId(), 0, maxChannelReopens);
267269
}
268270

269271
private SeekableByteChannel newWriteChannel(Path path, Set<? extends OpenOption> options)

‎trunk/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageOptions.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,5 +91,15 @@ public static CloudStorageOption.OpenCopy withBlockSize(int size) {
9191
return OptionBlockSize.create(size);
9292
}
9393

94+
/**
95+
* Sets the max number of times that the channel can be reopen if reading
96+
* fails because the channel unexpectedly closes.
97+
*
98+
* <p>The default is 0.
99+
*/
100+
public static CloudStorageOption.OpenCopy withChannelReopen(int count) {
101+
return OptionMaxChannelReopens.create(count);
102+
}
103+
94104
private CloudStorageOptions() {}
95105
}

‎trunk/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageReadChannel.java‎

Lines changed: 62 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import com.google.cloud.storage.BlobId;
2323
import com.google.cloud.storage.BlobInfo;
2424
import com.google.cloud.storage.Storage;
25+
import com.google.cloud.storage.StorageException;
2526

2627
import java.io.IOException;
2728
import java.nio.ByteBuffer;
@@ -41,27 +42,42 @@
4142
@ThreadSafe
4243
final class CloudStorageReadChannel implements SeekableByteChannel {
4344

44-
private final ReadChannel channel;
45+
private final Storage gcsStorage;
46+
private final BlobId file;
47+
// max # of times we may reopen the file
48+
private final int maxChannelReopens;
49+
// how many times we re-opened the file
50+
private int reopens;
51+
private ReadChannel channel;
4552
private long position;
4653
private long size;
4754

55+
/**
56+
* @param maxChannelReopens max number of times to try re-opening the channel if it closes on us unexpectedly.
57+
*/
4858
@CheckReturnValue
4959
@SuppressWarnings("resource")
50-
static CloudStorageReadChannel create(Storage gcsStorage, BlobId file, long position)
60+
static CloudStorageReadChannel create(Storage gcsStorage, BlobId file, long position, int maxChannelReopens)
5161
throws IOException {
62+
return new CloudStorageReadChannel(gcsStorage, file, position, maxChannelReopens);
63+
}
64+
65+
private CloudStorageReadChannel(Storage gcsStorage, BlobId file, long position, int maxChannelReopens) throws IOException {
66+
this.gcsStorage = gcsStorage;
67+
this.file = file;
68+
this.position = position;
69+
this.reopens = 0;
70+
this.maxChannelReopens = maxChannelReopens;
5271
// XXX: Reading size and opening file should be atomic.
53-
long size = fetchSize(gcsStorage, file);
54-
ReadChannel channel = gcsStorage.reader(file);
72+
this.size = fetchSize(gcsStorage, file);
73+
innerOpen();
74+
}
75+
76+
private void innerOpen() throws IOException {
77+
this.channel = gcsStorage.reader(file);
5578
if (position > 0) {
5679
channel.seek((int) position);
5780
}
58-
return new CloudStorageReadChannel(position, size, channel);
59-
}
60-
61-
private CloudStorageReadChannel(long position, long size, ReadChannel channel) {
62-
this.position = position;
63-
this.size = size;
64-
this.channel = channel;
6581
}
6682

6783
@Override
@@ -78,11 +94,36 @@ public void close() throws IOException {
7894
}
7995
}
8096

97+
8198
@Override
8299
public int read(ByteBuffer dst) throws IOException {
83100
synchronized (this) {
84101
checkOpen();
85-
int amt = channel.read(dst);
102+
int amt;
103+
int retries = 0;
104+
int maxRetries = 3;
105+
dst.mark();
106+
while (true) {
107+
try {
108+
dst.reset();
109+
amt = channel.read(dst);
110+
break;
111+
} catch (StorageException exs) {
112+
if (exs.getMessage().contains("Connection closed prematurely") && reopens < maxChannelReopens) {
113+
// this error isn't marked as retryable since the channel is closed;
114+
// but here at this higher level we can retry it.
115+
reopens++;
116+
sleepForAttempt(reopens);
117+
innerOpen();
118+
continue;
119+
} else if ((exs.getCode() == 500 || exs.getCode() == 503) && retries < maxRetries) {
120+
retries++;
121+
sleepForAttempt(retries);
122+
continue;
123+
}
124+
throw exs;
125+
}
126+
}
86127
if (amt > 0) {
87128
position += amt;
88129
// XXX: This would only ever happen if the fetchSize() race-condition occurred.
@@ -94,6 +135,15 @@ public int read(ByteBuffer dst) throws IOException {
94135
}
95136
}
96137

138+
private void sleepForAttempt(int attempt) {
139+
try {
140+
Thread.sleep((attempt - 1) * 500);
141+
} catch (InterruptedException iex) {
142+
// reset interrupt flag
143+
Thread.currentThread().interrupt();
144+
}
145+
}
146+
97147
@Override
98148
public long size() throws IOException {
99149
synchronized (this) {

‎trunk/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/OptionMaxChannelReopens.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/*
2+
* Copyright 2016 Google Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.storage.contrib.nio;
18+
19+
import com.google.auto.value.AutoValue;
20+
21+
@AutoValue
22+
abstract class OptionMaxChannelReopens implements CloudStorageOption.OpenCopy {
23+
24+
/**
25+
* Re-open the channel if it's closed unexpectedly while we're reading it.
26+
*/
27+
static OptionMaxChannelReopens create(int retryCount) {
28+
return new AutoValue_OptionMaxChannelReopens(retryCount);
29+
}
30+
31+
abstract int maxChannelReopens();
32+
}

‎trunk/google-cloud-contrib/google-cloud-nio/src/test/java/com/google/cloud/storage/contrib/nio/CloudStorageReadChannelTest.java‎

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import com.google.cloud.storage.BlobId;
3131
import com.google.cloud.storage.Storage;
3232

33+
import com.google.cloud.storage.StorageException;
3334
import org.junit.Before;
3435
import org.junit.Rule;
3536
import org.junit.Test;
@@ -63,7 +64,7 @@ public void before() throws IOException {
6364
when(gcsStorage.get(file)).thenReturn(metadata);
6465
when(gcsStorage.reader(eq(file))).thenReturn(gcsChannel);
6566
when(gcsChannel.isOpen()).thenReturn(true);
66-
chan = CloudStorageReadChannel.create(gcsStorage, file, 0);
67+
chan = CloudStorageReadChannel.create(gcsStorage, file, 0, 1);
6768
verify(gcsStorage).get(eq(file));
6869
verify(gcsStorage).reader(eq(file));
6970
}
@@ -79,6 +80,30 @@ public void testRead() throws IOException {
7980
verify(gcsChannel, times(3)).isOpen();
8081
}
8182

83+
@Test
84+
public void testReadRetry() throws IOException {
85+
ByteBuffer buffer = ByteBuffer.allocate(1);
86+
when(gcsChannel.read(eq(buffer)))
87+
.thenThrow(new StorageException(new IOException("Connection closed prematurely: bytesRead = 33554432, Content-Length = 41943040")))
88+
.thenReturn(1);
89+
assertThat(chan.position()).isEqualTo(0L);
90+
assertThat(chan.read(buffer)).isEqualTo(1);
91+
assertThat(chan.position()).isEqualTo(1L);
92+
verify(gcsChannel, times(2)).read(any(ByteBuffer.class));
93+
}
94+
95+
@Test
96+
public void testReadRetryEventuallyGivesUp() throws IOException {
97+
ByteBuffer buffer = ByteBuffer.allocate(1);
98+
when(gcsChannel.read(eq(buffer)))
99+
.thenThrow(new StorageException(new IOException("Connection closed prematurely: bytesRead = 33554432, Content-Length = 41943040")))
100+
.thenThrow(new StorageException(new IOException("Connection closed prematurely: bytesRead = 33554432, Content-Length = 41943040")))
101+
.thenReturn(1);
102+
assertThat(chan.position()).isEqualTo(0L);
103+
thrown.expect(StorageException.class);
104+
chan.read(buffer);
105+
}
106+
82107
@Test
83108
public void testRead_whenClosed_throwsCce() throws IOException {
84109
when(gcsChannel.isOpen()).thenReturn(false);

0 commit comments

Comments
 (0)