Skip to content

Commit a43992a

Browse files
aozarovaozarov
committed
Merge pull request #69 from aozarov/temp2
extract reader/writer impl, increase their default buffer and minimize footprint when possible.
2 parents c081812 + c394656 commit a43992a

File tree

4 files changed

+283
-224
lines changed

4 files changed

+283
-224
lines changed

‎src/main/java/com/google/gcloud/examples/StorageExample.java

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
package com.google.gcloud.examples;
1818

19+
import com.google.gcloud.RetryParams;
1920
import com.google.gcloud.spi.StorageRpc.Tuple;
2021
import com.google.gcloud.storage.BatchRequest;
2122
import com.google.gcloud.storage.BatchResponse;
@@ -498,7 +499,8 @@ public static void main(String... args) throws Exception {
498499
printUsage();
499500
return;
500501
}
501-
StorageServiceOptions.Builder optionsBuilder = StorageServiceOptions.builder();
502+
StorageServiceOptions.Builder optionsBuilder =
503+
StorageServiceOptions.builder().retryParams(RetryParams.getDefaultInstance());
502504
StorageAction action;
503505
if (args.length >= 2 && !ACTIONS.containsKey(args[0])) {
504506
optionsBuilder.projectId(args[0]);

‎src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
/*
2+
* Copyright 2015 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.gcloud.storage;
18+
19+
import static com.google.gcloud.RetryHelper.runWithRetries;
20+
21+
import com.google.api.services.storage.model.StorageObject;
22+
import com.google.gcloud.spi.StorageRpc;
23+
24+
import java.io.IOException;
25+
import java.io.ObjectInputStream;
26+
import java.io.ObjectOutputStream;
27+
import java.nio.ByteBuffer;
28+
import java.util.Map;
29+
import java.util.concurrent.Callable;
30+
31+
/**
32+
* Default implementation for BlobReadChannel.
33+
*/
34+
class BlobReadChannelImpl implements BlobReadChannel {
35+
36+
private static final int MIN_BUFFER_SIZE = 2 * 1024 * 1024;
37+
private static final long serialVersionUID = 4821762590742862669L;
38+
39+
private final StorageServiceOptions serviceOptions;
40+
private final Blob blob;
41+
private final Map<StorageRpc.Option, ?> requestOptions;
42+
private int position;
43+
private boolean isOpen;
44+
private boolean endOfStream;
45+
46+
private transient StorageRpc storageRpc;
47+
private transient StorageObject storageObject;
48+
private transient int bufferPos;
49+
private transient byte[] buffer;
50+
51+
BlobReadChannelImpl(StorageServiceOptions serviceOptions, Blob blob,
52+
Map<StorageRpc.Option, ?> requestOptions) {
53+
this.serviceOptions = serviceOptions;
54+
this.blob = blob;
55+
this.requestOptions = requestOptions;
56+
isOpen = true;
57+
initTransients();
58+
}
59+
60+
private void writeObject(ObjectOutputStream out) throws IOException {
61+
if (buffer != null) {
62+
position += bufferPos;
63+
buffer = null;
64+
bufferPos = 0;
65+
endOfStream = false;
66+
}
67+
out.defaultWriteObject();
68+
}
69+
70+
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
71+
in.defaultReadObject();
72+
initTransients();
73+
}
74+
75+
private void initTransients() {
76+
storageRpc = serviceOptions.storageRpc();
77+
storageObject = blob.toPb();
78+
}
79+
80+
@Override
81+
public boolean isOpen() {
82+
return isOpen;
83+
}
84+
85+
@Override
86+
public void close() {
87+
if (isOpen) {
88+
buffer = null;
89+
isOpen = false;
90+
}
91+
}
92+
93+
private void validateOpen() throws IOException {
94+
if (!isOpen) {
95+
throw new IOException("stream is closed");
96+
}
97+
}
98+
99+
@Override
100+
public void seek(int position) throws IOException {
101+
validateOpen();
102+
this.position = position;
103+
buffer = null;
104+
bufferPos = 0;
105+
endOfStream = false;
106+
}
107+
108+
@Override
109+
public int read(ByteBuffer byteBuffer) throws IOException {
110+
validateOpen();
111+
if (buffer == null) {
112+
if (endOfStream) {
113+
return -1;
114+
}
115+
final int toRead = Math.max(byteBuffer.remaining(), MIN_BUFFER_SIZE);
116+
buffer = runWithRetries(new Callable<byte[]>() {
117+
@Override
118+
public byte[] call() {
119+
return storageRpc.read(storageObject, requestOptions, position, toRead);
120+
}
121+
}, serviceOptions.retryParams(), StorageServiceImpl.EXCEPTION_HANDLER);
122+
if (toRead > buffer.length) {
123+
endOfStream = true;
124+
if (buffer.length == 0) {
125+
buffer = null;
126+
return -1;
127+
}
128+
}
129+
}
130+
int toWrite = Math.min(buffer.length - bufferPos, byteBuffer.remaining());
131+
byteBuffer.put(buffer, bufferPos, toWrite);
132+
bufferPos += toWrite;
133+
if (bufferPos >= buffer.length) {
134+
position += buffer.length;
135+
buffer = null;
136+
bufferPos = 0;
137+
}
138+
return toWrite;
139+
}
140+
}

‎src/main/java/com/google/gcloud/storage/BlobWriterChannelImpl.java

Lines changed: 138 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
/*
2+
* Copyright 2015 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.gcloud.storage;
18+
19+
import static com.google.gcloud.RetryHelper.runWithRetries;
20+
import static java.util.concurrent.Executors.callable;
21+
22+
import com.google.api.services.storage.model.StorageObject;
23+
import com.google.gcloud.spi.StorageRpc;
24+
25+
import java.io.IOException;
26+
import java.io.ObjectInputStream;
27+
import java.io.ObjectOutputStream;
28+
import java.nio.ByteBuffer;
29+
import java.util.Arrays;
30+
import java.util.Map;
31+
32+
/**
33+
* Default implementation for BlobWriteChannel.
34+
*/
35+
class BlobWriterChannelImpl implements BlobWriteChannel {
36+
37+
private static final long serialVersionUID = 8675286882724938737L;
38+
private static final int CHUNK_SIZE = 256 * 1024;
39+
private static final int MIN_BUFFER_SIZE = 8 * CHUNK_SIZE;
40+
41+
private final StorageServiceOptions options;
42+
private final Blob blob;
43+
private final String uploadId;
44+
private int position;
45+
private byte[] buffer = new byte[0];
46+
private int limit;
47+
private boolean isOpen = true;
48+
49+
private transient StorageRpc storageRpc;
50+
private transient StorageObject storageObject;
51+
52+
public BlobWriterChannelImpl(StorageServiceOptions options, Blob blob,
53+
Map<StorageRpc.Option, ?> optionsMap) {
54+
this.options = options;
55+
this.blob = blob;
56+
initTransients();
57+
uploadId = options.storageRpc().open(storageObject, optionsMap);
58+
}
59+
60+
private void writeObject(ObjectOutputStream out) throws IOException {
61+
if (isOpen) {
62+
flush(true);
63+
}
64+
out.defaultWriteObject();
65+
}
66+
67+
private void flush(boolean compact) {
68+
if (limit >= MIN_BUFFER_SIZE || compact && limit >= CHUNK_SIZE) {
69+
final int length = limit - limit % CHUNK_SIZE;
70+
runWithRetries(callable(new Runnable() {
71+
@Override
72+
public void run() {
73+
storageRpc.write(uploadId, buffer, 0, storageObject, position, length, false);
74+
}
75+
}), options.retryParams(), StorageServiceImpl.EXCEPTION_HANDLER);
76+
position += length;
77+
limit -= length;
78+
byte[] temp = new byte[compact ? limit : MIN_BUFFER_SIZE];
79+
System.arraycopy(buffer, length, temp, 0, limit);
80+
buffer = temp;
81+
}
82+
}
83+
84+
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
85+
in.defaultReadObject();
86+
if (isOpen) {
87+
initTransients();
88+
}
89+
}
90+
91+
private void initTransients() {
92+
storageRpc = options.storageRpc();
93+
storageObject = blob.toPb();
94+
}
95+
96+
private void validateOpen() throws IOException {
97+
if (!isOpen) {
98+
throw new IOException("stream is closed");
99+
}
100+
}
101+
102+
@Override
103+
public int write(ByteBuffer byteBuffer) throws IOException {
104+
validateOpen();
105+
int toWrite = byteBuffer.remaining();
106+
int spaceInBuffer = buffer.length - limit;
107+
if (spaceInBuffer >= toWrite) {
108+
byteBuffer.get(buffer, limit, toWrite);
109+
} else {
110+
buffer = Arrays.copyOf(buffer,
111+
Math.max(MIN_BUFFER_SIZE, buffer.length + toWrite - spaceInBuffer));
112+
byteBuffer.get(buffer, limit, toWrite);
113+
}
114+
limit += toWrite;
115+
flush(false);
116+
return toWrite;
117+
}
118+
119+
@Override
120+
public boolean isOpen() {
121+
return isOpen;
122+
}
123+
124+
@Override
125+
public void close() throws IOException {
126+
if (isOpen) {
127+
runWithRetries(callable(new Runnable() {
128+
@Override
129+
public void run() {
130+
storageRpc.write(uploadId, buffer, 0, storageObject, position, limit, true);
131+
}
132+
}), options.retryParams(), StorageServiceImpl.EXCEPTION_HANDLER);
133+
position += buffer.length;
134+
isOpen = false;
135+
buffer = null;
136+
}
137+
}
138+
}

0 commit comments

Comments
 (0)