/*
* Copyright 2010-2012 Ning, Inc.
*
* Ning licenses this file to you under the Apache License, version 2.0
* (the "License"); you may not use this file except in compliance with the
* License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package com.ning.billing.usage.timeline;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ning.billing.config.UsageConfig;
import com.ning.billing.usage.timeline.chunks.TimelineChunk;
import com.ning.billing.usage.timeline.persistent.TimelineDao;
import com.ning.billing.util.callcontext.CallOrigin;
import com.ning.billing.util.callcontext.InternalCallContext;
import com.ning.billing.util.callcontext.InternalCallContextFactory;
import com.ning.billing.util.callcontext.UserType;
import com.google.inject.Inject;
import com.google.inject.Singleton;
/**
* This class runs a thread that batch-writes TimelineChunks to the db.
* This class is thread-safe, and only holds up threads that want to queue
* TimelineChunks for the time it takes to copy the ArrayList of PendingChunkMaps.
* <p/>
* The background writing thread is scheduled every few seconds, as controlled by
* config.getBackgroundWriteCheckInterval(). It writes the current inventory of
* chunks if there are at least config.getBackgroundWriteBatchSize()
* TimelineChunks to be written, or if the time since the last write exceeds
* config.getBackgroundWriteMaxDelay().
*/
@Singleton
public class BackgroundDBChunkWriter {
private static final Logger log = LoggerFactory.getLogger(BackgroundDBChunkWriter.class);
private final TimelineDao timelineDAO;
private final UsageConfig config;
private final boolean performForegroundWrites;
private final InternalCallContextFactory internalCallContextFactory;
private final AtomicInteger pendingChunkCount = new AtomicInteger();
private final AtomicBoolean shuttingDown = new AtomicBoolean();
private List<PendingChunkMap> pendingChunks = new ArrayList<PendingChunkMap>();
private DateTime lastWriteTime = new DateTime();
private AtomicBoolean doingWritesNow = new AtomicBoolean();
private final ScheduledExecutorService backgroundWriteThread = Executors.newSingleThreadScheduledExecutor();
private final AtomicLong maybePerformBackgroundWritesCount = new AtomicLong();
private final AtomicLong backgroundWritesCount = new AtomicLong();
private final AtomicLong pendingChunkMapsAdded = new AtomicLong();
private final AtomicLong pendingChunksAdded = new AtomicLong();
private final AtomicLong pendingChunkMapsWritten = new AtomicLong();
private final AtomicLong pendingChunksWritten = new AtomicLong();
private final AtomicLong pendingChunkMapsMarkedConsumed = new AtomicLong();
private final AtomicLong foregroundChunkMapsWritten = new AtomicLong();
private final AtomicLong foregroundChunksWritten = new AtomicLong();
@Inject
public BackgroundDBChunkWriter(final TimelineDao timelineDAO, final UsageConfig config, final InternalCallContextFactory internalCallContextFactory) {
this(timelineDAO, config, config.getPerformForegroundWrites(), internalCallContextFactory);
}
public BackgroundDBChunkWriter(final TimelineDao timelineDAO, @Nullable final UsageConfig config,
final boolean performForegroundWrites, final InternalCallContextFactory internalCallContextFactory) {
this.timelineDAO = timelineDAO;
this.config = config;
this.performForegroundWrites = performForegroundWrites;
this.internalCallContextFactory = internalCallContextFactory;
}
public synchronized void addPendingChunkMap(final PendingChunkMap chunkMap) {
if (shuttingDown.get()) {
log.error("In addPendingChunkMap(), but finishBackgroundWritingAndExit is true!");
} else {
if (performForegroundWrites) {
foregroundChunkMapsWritten.incrementAndGet();
final List<TimelineChunk> chunksToWrite = new ArrayList<TimelineChunk>(chunkMap.getChunkMap().values());
foregroundChunksWritten.addAndGet(chunksToWrite.size());
timelineDAO.bulkInsertTimelineChunks(chunksToWrite, createCallContext());
chunkMap.getAccumulator().markPendingChunkMapConsumed(chunkMap.getPendingChunkMapId());
} else {
pendingChunkMapsAdded.incrementAndGet();
final int chunkCount = chunkMap.getChunkCount();
pendingChunksAdded.addAndGet(chunkCount);
pendingChunks.add(chunkMap);
pendingChunkCount.addAndGet(chunkCount);
}
}
}
private void performBackgroundWrites() {
backgroundWritesCount.incrementAndGet();
List<PendingChunkMap> chunkMapsToWrite = null;
synchronized (this) {
chunkMapsToWrite = pendingChunks;
pendingChunks = new ArrayList<PendingChunkMap>();
pendingChunkCount.set(0);
}
final List<TimelineChunk> chunks = new ArrayList<TimelineChunk>();
for (final PendingChunkMap map : chunkMapsToWrite) {
pendingChunkMapsWritten.incrementAndGet();
pendingChunksWritten.addAndGet(map.getChunkMap().size());
chunks.addAll(map.getChunkMap().values());
}
timelineDAO.bulkInsertTimelineChunks(chunks, createCallContext());
for (final PendingChunkMap map : chunkMapsToWrite) {
pendingChunkMapsMarkedConsumed.incrementAndGet();
map.getAccumulator().markPendingChunkMapConsumed(map.getPendingChunkMapId());
}
}
private void maybePerformBackgroundWrites() {
// If already running background writes, just return
maybePerformBackgroundWritesCount.incrementAndGet();
if (!doingWritesNow.compareAndSet(false, true)) {
return;
} else {
try {
if (shuttingDown.get()) {
performBackgroundWrites();
}
final int pendingCount = pendingChunkCount.get();
if (pendingCount > 0) {
if (pendingCount >= config.getBackgroundWriteBatchSize() ||
new DateTime().isBefore(lastWriteTime.plusMillis((int) config.getBackgroundWriteMaxDelay().getMillis()))) {
performBackgroundWrites();
lastWriteTime = new DateTime();
}
}
} finally {
doingWritesNow.set(false);
}
}
}
public synchronized boolean getShutdownFinished() {
return !doingWritesNow.get() && pendingChunks.size() == 0;
}
public void initiateShutdown() {
shuttingDown.set(true);
}
public void runBackgroundWriteThread() {
if (!performForegroundWrites) {
backgroundWriteThread.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
maybePerformBackgroundWrites();
}
},
config.getBackgroundWriteCheckInterval().getMillis(),
config.getBackgroundWriteCheckInterval().getMillis(),
TimeUnit.MILLISECONDS);
}
}
public void stopBackgroundWriteThread() {
if (!performForegroundWrites) {
backgroundWriteThread.shutdown();
}
}
public long getMaybePerformBackgroundWritesCount() {
return maybePerformBackgroundWritesCount.get();
}
public long getBackgroundWritesCount() {
return backgroundWritesCount.get();
}
public long getPendingChunkMapsAdded() {
return pendingChunkMapsAdded.get();
}
public long getPendingChunksAdded() {
return pendingChunksAdded.get();
}
public long getPendingChunkMapsWritten() {
return pendingChunkMapsWritten.get();
}
public long getPendingChunksWritten() {
return pendingChunksWritten.get();
}
public long getPendingChunkMapsMarkedConsumed() {
return pendingChunkMapsMarkedConsumed.get();
}
public long getForegroundChunkMapsWritten() {
return foregroundChunkMapsWritten.get();
}
public long getForegroundChunksWritten() {
return foregroundChunksWritten.get();
}
private InternalCallContext createCallContext() {
return internalCallContextFactory.createInternalCallContext("ChunkWriter", CallOrigin.INTERNAL, UserType.SYSTEM, null);
}
}