Skip to content

Commit 5e3f405

Browse files
pongadpongad
committed
RFC: logging: experimental batching writer
Manually written batching implementation. Unlike GAX batching, it implements flushing and does not deal with partition keys. If we're OK with where this is heading, we should make this work with batching settings (shouldn't be hard) and load-test before migrating things to it. If not, this can serve as a starting point for a better GAX batching implementation. In an experiment, I publish 1M messages of 300 bytes each. Using LoggingHandler gives me ~14,000 msg/s. BatchingWriter gives ~67,000, using similar BatchingSettings. Letting users configure these settings might be important though; I have observed ~300K msg/s given enough CPU and memory.
1 parent 191d6f4 commit 5e3f405

3 files changed

Lines changed: 240 additions & 0 deletions

File tree

‎google-cloud-logging/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,11 @@
4646
<groupId>io.grpc</groupId>
4747
<artifactId>grpc-auth</artifactId>
4848
</dependency>
49+
<dependency>
50+
<groupId>com.google.truth</groupId>
51+
<artifactId>truth</artifactId>
52+
<scope>test</scope>
53+
</dependency>
4954
<dependency>
5055
<groupId>${project.groupId}</groupId>
5156
<artifactId>google-cloud-core</artifactId>

‎google-cloud-logging/src/main/java/com/google/cloud/logging/BatchingWriter.java‎

Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
1+
/*
2+
* Copyright 2018 Google LLC
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+
package com.google.cloud.logging;
17+
18+
import com.google.api.core.ApiFuture;
19+
import com.google.api.core.ApiFutureCallback;
20+
import com.google.api.core.ApiFutures;
21+
import com.google.common.base.Preconditions;
22+
import com.google.logging.v2.WriteLogEntriesRequest;
23+
import java.util.ArrayList;
24+
import java.util.List;
25+
import java.util.concurrent.ConcurrentHashMap;
26+
import java.util.concurrent.Future;
27+
import java.util.concurrent.ScheduledThreadPoolExecutor;
28+
import java.util.concurrent.Semaphore;
29+
import java.util.concurrent.TimeUnit;
30+
31+
class BatchingWriter {
32+
33+
interface Rpc {
34+
ApiFuture<Void> call(WriteLogEntriesRequest request);
35+
}
36+
37+
private final Rpc rpc;
38+
private final int batchSize;
39+
private final Semaphore pending;
40+
private final WriteLogEntriesRequest requestPrototype;
41+
private final ScheduledThreadPoolExecutor executor;
42+
43+
// Use Boolean, since concurrent maps don't support nulls.
44+
private final ConcurrentHashMap<ApiFuture<Void>, Boolean> pendingWrites =
45+
new ConcurrentHashMap<>();
46+
47+
private final ArrayList<com.google.logging.v2.LogEntry> currentBatch;
48+
49+
private Future<?> flushJob = null;
50+
51+
BatchingWriter(
52+
Rpc rpc,
53+
int batchSize,
54+
int maxPending,
55+
WriteLogEntriesRequest requestPrototype,
56+
ScheduledThreadPoolExecutor executor) {
57+
Preconditions.checkArgument(batchSize > 0, "batchSize must be positive");
58+
this.batchSize = batchSize;
59+
Preconditions.checkArgument(maxPending > 0, "maxPending must be positive");
60+
this.pending = new Semaphore(maxPending);
61+
62+
this.rpc = Preconditions.checkNotNull(rpc);
63+
this.requestPrototype = Preconditions.checkNotNull(requestPrototype);
64+
this.executor = Preconditions.checkNotNull(executor);
65+
this.currentBatch = new ArrayList<>(batchSize);
66+
}
67+
68+
synchronized void startJob() {
69+
Preconditions.checkArgument(flushJob == null, "job already started");
70+
flushJob =
71+
executor.scheduleWithFixedDelay(
72+
new Runnable() {
73+
@Override
74+
public void run() {
75+
initFlush();
76+
}
77+
},
78+
100,
79+
100,
80+
TimeUnit.MILLISECONDS);
81+
}
82+
83+
synchronized void stopJob() {
84+
Preconditions.checkArgument(flushJob != null, "job not started");
85+
flushJob.cancel(false);
86+
}
87+
88+
void add(com.google.logging.v2.LogEntry entry) throws InterruptedException {
89+
pending.acquire(1);
90+
synchronized (currentBatch) {
91+
currentBatch.add(entry);
92+
if (currentBatch.size() == batchSize) {
93+
final WriteLogEntriesRequest request =
94+
requestPrototype.toBuilder().addAllEntries(currentBatch).build();
95+
currentBatch.clear();
96+
97+
// Whoever calls send serializes the proto; so we do it off-thread.
98+
// This gives better CPU utilization if there are few producer threads
99+
// on a many-core machine.
100+
executor.execute(
101+
new Runnable() {
102+
@Override
103+
public void run() {
104+
send(request);
105+
}
106+
});
107+
}
108+
}
109+
}
110+
111+
void initFlush() {
112+
WriteLogEntriesRequest request;
113+
synchronized (currentBatch) {
114+
request = requestPrototype.toBuilder().addAllEntries(currentBatch).build();
115+
currentBatch.clear();
116+
}
117+
send(request);
118+
}
119+
120+
private void send(WriteLogEntriesRequest request) {
121+
final int count = request.getEntriesCount();
122+
if (count == 0) {
123+
return;
124+
}
125+
126+
final ApiFuture<Void> writeFuture = rpc.call(request);
127+
pendingWrites.put(writeFuture, Boolean.TRUE);
128+
ApiFutures.addCallback(
129+
writeFuture,
130+
new ApiFutureCallback<Void>() {
131+
private void onBoth() {
132+
pendingWrites.remove(writeFuture);
133+
pending.release(count);
134+
}
135+
136+
@Override
137+
public void onSuccess(Void v) {
138+
onBoth();
139+
}
140+
141+
@Override
142+
public void onFailure(Throwable t) {
143+
// Report failure.
144+
onBoth();
145+
}
146+
});
147+
}
148+
149+
List<ApiFuture<Void>> pendingRpcs() {
150+
return new ArrayList<>(pendingWrites.keySet());
151+
}
152+
}

‎google-cloud-logging/src/test/java/com/google/cloud/logging/BatchingWriterTest.java‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
/*
2+
* Copyright 2018 Google LLC
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+
package com.google.cloud.logging;
17+
18+
import static com.google.common.truth.Truth.assertThat;
19+
20+
import com.google.api.core.ApiFuture;
21+
import com.google.api.core.SettableApiFuture;
22+
import com.google.logging.v2.WriteLogEntriesRequest;
23+
import java.util.List;
24+
import java.util.concurrent.ScheduledThreadPoolExecutor;
25+
import org.junit.BeforeClass;
26+
import org.junit.Test;
27+
28+
public class BatchingWriterTest {
29+
30+
private static final WriteLogEntriesRequest EMPTY_REQUEST =
31+
WriteLogEntriesRequest.newBuilder().build();
32+
private static final com.google.logging.v2.LogEntry EMPTY_ENTRY =
33+
com.google.logging.v2.LogEntry.newBuilder().build();
34+
private static ScheduledThreadPoolExecutor EXECUTOR;
35+
36+
@BeforeClass
37+
public static void beforeClass() {
38+
// Executor is never used.
39+
EXECUTOR = new ScheduledThreadPoolExecutor(1);
40+
EXECUTOR.shutdownNow();
41+
}
42+
43+
@Test
44+
public void testEmptyWrite() {
45+
BatchingWriter.Rpc rpc =
46+
new BatchingWriter.Rpc() {
47+
@Override
48+
public ApiFuture<Void> call(WriteLogEntriesRequest request) {
49+
throw new UnsupportedOperationException("should never be called");
50+
}
51+
};
52+
53+
BatchingWriter writer = new BatchingWriter(rpc, 10, 10, EMPTY_REQUEST, EXECUTOR);
54+
writer.initFlush();
55+
56+
// If there's no message, there's no RPC.
57+
assertThat(writer.pendingRpcs()).isEmpty();
58+
}
59+
60+
@Test
61+
public void testFlush() throws Exception {
62+
final SettableApiFuture<Void> fakeFuture = SettableApiFuture.create();
63+
BatchingWriter.Rpc rpc =
64+
new BatchingWriter.Rpc() {
65+
@Override
66+
public ApiFuture<Void> call(WriteLogEntriesRequest request) {
67+
return fakeFuture;
68+
}
69+
};
70+
71+
BatchingWriter writer = new BatchingWriter(rpc, 10, 10, EMPTY_REQUEST, EXECUTOR);
72+
writer.add(EMPTY_ENTRY);
73+
writer.initFlush();
74+
75+
List<ApiFuture<Void>> futures = writer.pendingRpcs();
76+
assertThat(futures).hasSize(1);
77+
assertThat(futures.get(0).isDone()).isFalse();
78+
79+
fakeFuture.set(null);
80+
assertThat(futures.get(0).isDone()).isTrue();
81+
assertThat(writer.pendingRpcs()).isEmpty();
82+
}
83+
}

0 commit comments

Comments
 (0)