forked from pinpoint-apm/pinpoint
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[pinpoint-apm#11379] Add AsyncPollingPutWriter
- Loading branch information
Showing
12 changed files
with
738 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
88 changes: 88 additions & 0 deletions
88
commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/AsyncPollerOption.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,88 @@ | ||
package com.navercorp.pinpoint.common.hbase.async; | ||
|
||
import com.navercorp.pinpoint.common.util.CpuUtils; | ||
|
||
public class AsyncPollerOption { | ||
|
||
private int queueSize = 1000 * 100; | ||
|
||
private int writeBufferSize = 100; | ||
private int writeBufferPeriodicFlush = 100; | ||
private int parallelism = 0; | ||
private int cpuRatio = 1; | ||
private int minCpuCore = 2; | ||
|
||
|
||
public int getQueueSize() { | ||
return queueSize; | ||
} | ||
|
||
public void setQueueSize(int queueSize) { | ||
this.queueSize = queueSize; | ||
} | ||
|
||
public int getWriteBufferSize() { | ||
return writeBufferSize; | ||
} | ||
|
||
public void setWriteBufferSize(int writeBufferSize) { | ||
this.writeBufferSize = writeBufferSize; | ||
} | ||
|
||
public int getWriteBufferPeriodicFlush() { | ||
return writeBufferPeriodicFlush; | ||
} | ||
|
||
public void setWriteBufferPeriodicFlush(int writeBufferPeriodicFlush) { | ||
this.writeBufferPeriodicFlush = writeBufferPeriodicFlush; | ||
} | ||
|
||
public int getParallelism() { | ||
return cpu(parallelism, cpuRatio, minCpuCore); | ||
} | ||
|
||
int cpu(int parallelism, int cpuRatio, int minCpu) { | ||
if (parallelism <= 0) { | ||
final int cpuCount = getCpuCount(); | ||
int cpu = Math.floorDiv(cpuCount, cpuRatio); | ||
return Math.max(cpu, minCpu); | ||
} | ||
return parallelism; | ||
} | ||
|
||
int getCpuCount() { | ||
return CpuUtils.cpuCount(); | ||
} | ||
|
||
public void setParallelism(int parallelism) { | ||
this.parallelism = parallelism; | ||
} | ||
|
||
public int getCpuRatio() { | ||
return cpuRatio; | ||
} | ||
|
||
public void setCpuRatio(int cpuRatio) { | ||
this.cpuRatio = cpuRatio; | ||
} | ||
|
||
public int getMinCpuCore() { | ||
return minCpuCore; | ||
} | ||
|
||
public void setMinCpuCore(int minCpuCore) { | ||
this.minCpuCore = minCpuCore; | ||
} | ||
|
||
@Override | ||
public String toString() { | ||
return "AsyncPollerOption{" + | ||
"queueSize=" + queueSize + | ||
", writeBufferSize=" + writeBufferSize + | ||
", writeBufferPeriodicFlush=" + writeBufferPeriodicFlush + | ||
", parallelism=" + parallelism + | ||
", cpuRatio=" + cpuRatio + | ||
", minCpuCore=" + minCpuCore + | ||
'}'; | ||
} | ||
} |
203 changes: 203 additions & 0 deletions
203
commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/AsyncPollerThread.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,203 @@ | ||
package com.navercorp.pinpoint.common.hbase.async; | ||
|
||
import com.navercorp.pinpoint.common.hbase.RequestNotPermittedException; | ||
import com.navercorp.pinpoint.common.hbase.util.FutureUtils; | ||
import com.navercorp.pinpoint.common.profiler.logging.ThrottledLogger; | ||
import org.apache.hadoop.hbase.TableName; | ||
import org.apache.hadoop.hbase.client.Put; | ||
import org.apache.logging.log4j.LogManager; | ||
import org.apache.logging.log4j.Logger; | ||
|
||
import java.io.Closeable; | ||
import java.util.ArrayList; | ||
import java.util.HashMap; | ||
import java.util.List; | ||
import java.util.Map; | ||
import java.util.Objects; | ||
import java.util.concurrent.ArrayBlockingQueue; | ||
import java.util.concurrent.BlockingQueue; | ||
import java.util.concurrent.CompletableFuture; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.atomic.AtomicBoolean; | ||
|
||
public class AsyncPollerThread implements Closeable { | ||
|
||
private final Logger logger = LogManager.getLogger(this.getClass()); | ||
private final ThrottledLogger tLogger = ThrottledLogger.getLogger(logger, 100); | ||
|
||
private final TableWriterFactory writerFactory; | ||
|
||
private final BlockingQueue<WriteRequest<Void>> queue; | ||
private final int queueSize; | ||
private final int writeBufferSize; | ||
private final int writeBufferPeriodicFlush; | ||
private final int pollTimeout; | ||
|
||
private final Thread thread; | ||
private final AtomicBoolean runState = new AtomicBoolean(true); | ||
|
||
public static final RequestNotPermittedException OVERFLOW = new RequestNotPermittedException("write queue is full", false); | ||
|
||
public AsyncPollerThread(String id, TableWriterFactory writerFactory, | ||
AsyncPollerOption option) { | ||
this.writerFactory = Objects.requireNonNull(writerFactory, "writerFactory"); | ||
|
||
this.queueSize = option.getQueueSize(); | ||
this.queue = new ArrayBlockingQueue<>(queueSize); | ||
|
||
this.writeBufferSize = option.getWriteBufferSize(); | ||
this.writeBufferPeriodicFlush = option.getWriteBufferPeriodicFlush(); | ||
this.pollTimeout = Math.max(writeBufferPeriodicFlush / 4, 20); | ||
|
||
this.thread = new Thread(this::dispatch, id); | ||
this.thread.setDaemon(true); | ||
this.thread.start(); | ||
} | ||
|
||
public List<CompletableFuture<Void>> write(TableName tableName, List<Put> puts) { | ||
Objects.requireNonNull(tableName, "tableName"); | ||
Objects.requireNonNull(puts, "puts"); | ||
if (isShutdown()) { | ||
return FutureUtils.newFutureList(() -> CompletableFuture.failedFuture(new IllegalStateException("closed")), puts.size()); | ||
} | ||
|
||
WriteRequest<Void> writeRequest = new WriteRequest<>(tableName, puts); | ||
if (this.queue.offer(writeRequest)) { | ||
return writeRequest.getFutures(); | ||
} | ||
tLogger.info("write queue overflow"); | ||
return FutureUtils.newFutureList(() -> CompletableFuture.failedFuture(OVERFLOW), puts.size()); | ||
} | ||
|
||
|
||
public void dispatch() { | ||
while (isRun()) { | ||
try { | ||
List<WriteRequest<Void>> requests = poll(); | ||
if (requests.isEmpty()) { | ||
continue; | ||
} | ||
|
||
Map<TableName, List<WriteRequest<Void>>> map = tableGroup(requests); | ||
|
||
for (Map.Entry<TableName, List<WriteRequest<Void>>> entry : map.entrySet()) { | ||
TableName tableName = entry.getKey(); | ||
List<WriteRequest<Void>> writes = entry.getValue(); | ||
|
||
if (logger.isDebugEnabled()) { | ||
logger.debug("write {} {} requests:{}", this.thread.getName(), tableName, writes.size()); | ||
} | ||
List<Put> puts = getPuts(writes); | ||
|
||
AsyncTableWriterFactory.Writer writer = this.writerFactory.newWriter(tableName); | ||
List<CompletableFuture<Void>> hbaseResults = writer.batch(puts); | ||
addListeners(hbaseResults, writes); | ||
} | ||
} catch (InterruptedException e) { | ||
Thread.currentThread().interrupt(); | ||
logger.debug("Thread.interrupted {}", this.thread.getName()); | ||
if (isShutdown()) { | ||
break; | ||
} | ||
} catch (Throwable th) { | ||
logger.warn("Error {}", this.thread.getName(), th); | ||
if (isShutdown()) { | ||
break; | ||
} | ||
} | ||
} | ||
logger.info("dispatch terminated {}", this.thread.getName()); | ||
} | ||
|
||
private boolean isRun() { | ||
return runState.get(); | ||
} | ||
|
||
private boolean isShutdown() { | ||
return !runState.get(); | ||
} | ||
|
||
private Map<TableName, List<WriteRequest<Void>>> tableGroup(List<WriteRequest<Void>> requests) { | ||
Map<TableName, List<WriteRequest<Void>>> map = new HashMap<>(); | ||
for (WriteRequest<Void> req : requests) { | ||
TableName tableName = req.getTableName(); | ||
List<WriteRequest<Void>> puts = map.computeIfAbsent(tableName, (key) -> new ArrayList<>()); | ||
puts.add(req); | ||
} | ||
return map; | ||
} | ||
|
||
@Override | ||
public void close() { | ||
logger.debug("Close {}", this.thread.getName()); | ||
this.runState.set(false); | ||
this.thread.interrupt(); | ||
try { | ||
this.thread.join(3000); | ||
} catch (InterruptedException ignored) { | ||
Thread.currentThread().interrupt(); | ||
} | ||
} | ||
|
||
|
||
|
||
private List<Put> getPuts(List<WriteRequest<Void>> writes) { | ||
List<Put> puts = new ArrayList<>(writes.size()); | ||
for (WriteRequest<Void> write : writes) { | ||
puts.addAll(write.getPuts()); | ||
} | ||
return puts; | ||
} | ||
|
||
private static <V> void addListeners(List<CompletableFuture<V>> hbaseResults, List<WriteRequest<V>> requests) { | ||
int i = 0; | ||
for (WriteRequest<V> writeRequest : requests) { | ||
for (CompletableFuture<V> write : writeRequest.getFutures()) { | ||
CompletableFuture<V> hbaseFuture = hbaseResults.get(i++); | ||
FutureUtils.addListener(hbaseFuture, write); | ||
} | ||
} | ||
} | ||
|
||
private List<WriteRequest<Void>> poll() throws InterruptedException { | ||
final long startTime = System.currentTimeMillis(); | ||
|
||
List<WriteRequest<Void>> drain = new ArrayList<>(writeBufferSize); | ||
int drainSize = 0; | ||
while (isRun()) { | ||
WriteRequest<Void> request = queue.poll(pollTimeout, TimeUnit.MILLISECONDS); | ||
if (request != null) { | ||
drain.add(request); | ||
drainSize += request.getPuts().size(); | ||
if (bufferOverflow(drainSize)) { | ||
return drain; | ||
} | ||
} | ||
if (drainSize > 0) { | ||
if (timeout(startTime)) { | ||
return drain; | ||
} | ||
} | ||
} | ||
return drain; | ||
} | ||
|
||
private boolean timeout(long startTime) { | ||
return System.currentTimeMillis() - startTime > writeBufferPeriodicFlush; | ||
} | ||
|
||
private boolean bufferOverflow(int drainSize) { | ||
return drainSize >= writeBufferSize; | ||
} | ||
|
||
@Override | ||
public String toString() { | ||
return "AsyncPollerThread{" + | ||
", queueSize=" + queueSize + | ||
", writeBufferSize=" + writeBufferSize + | ||
", writeBufferPeriodicFlush=" + writeBufferPeriodicFlush + | ||
", pollTimeout=" + pollTimeout + | ||
", thread=" + thread + | ||
'}'; | ||
} | ||
} |
Oops, something went wrong.