mirror of
https://github.com/berkeleydb/je.git
synced 2024-07-12 12:37:18 +00:00
830 lines
29 KiB
Java
830 lines
29 KiB
Java
|
/*-
|
|||
|
|
* Copyright (C) 2002, 2017, Oracle and/or its affiliates. All rights reserved.
|
|||
|
|
*
|
|||
|
|
* This file was distributed by Oracle as part of a version of Oracle Berkeley
|
|||
|
|
* DB Java Edition made available at:
|
|||
|
|
*
|
|||
|
|
* http://www.oracle.com/technetwork/database/database-technologies/berkeleydb/downloads/index.html
|
|||
|
|
*
|
|||
|
|
* Please see the LICENSE file included in the top-level directory of the
|
|||
|
|
* appropriate version of Oracle Berkeley DB Java Edition for a copy of the
|
|||
|
|
* license and additional information.
|
|||
|
|
*/
|
|||
|
|
|
|||
|
|
import static org.junit.Assert.assertEquals;
|
|||
|
|
import static org.junit.Assert.assertNotNull;
|
|||
|
|
import static org.junit.Assert.assertTrue;
|
|||
|
|
import static org.junit.Assert.fail;
|
|||
|
|
|
|||
|
|
import java.io.File;
|
|||
|
|
import java.text.DateFormat;
|
|||
|
|
import java.util.Arrays;
|
|||
|
|
import java.util.List;
|
|||
|
|
import java.util.Random;
|
|||
|
|
import java.util.concurrent.ArrayBlockingQueue;
|
|||
|
|
import java.util.concurrent.BlockingQueue;
|
|||
|
|
import java.util.concurrent.ThreadPoolExecutor;
|
|||
|
|
import java.util.concurrent.TimeUnit;
|
|||
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|||
|
|
import java.util.concurrent.atomic.AtomicInteger;
|
|||
|
|
import java.util.concurrent.atomic.AtomicLong;
|
|||
|
|
|
|||
|
|
import com.sleepycat.bind.tuple.LongBinding;
|
|||
|
|
import com.sleepycat.bind.tuple.TupleBase;
|
|||
|
|
import com.sleepycat.bind.tuple.TupleInput;
|
|||
|
|
import com.sleepycat.bind.tuple.TupleOutput;
|
|||
|
|
import com.sleepycat.je.CacheMode;
|
|||
|
|
import com.sleepycat.je.Cursor;
|
|||
|
|
import com.sleepycat.je.Database;
|
|||
|
|
import com.sleepycat.je.DatabaseConfig;
|
|||
|
|
import com.sleepycat.je.DatabaseEntry;
|
|||
|
|
import com.sleepycat.je.Durability;
|
|||
|
|
import com.sleepycat.je.Environment;
|
|||
|
|
import com.sleepycat.je.EnvironmentConfig;
|
|||
|
|
import com.sleepycat.je.Get;
|
|||
|
|
import com.sleepycat.je.OperationResult;
|
|||
|
|
import com.sleepycat.je.Put;
|
|||
|
|
import com.sleepycat.je.SecondaryConfig;
|
|||
|
|
import com.sleepycat.je.SecondaryCursor;
|
|||
|
|
import com.sleepycat.je.SecondaryDatabase;
|
|||
|
|
import com.sleepycat.je.SecondaryKeyCreator;
|
|||
|
|
import com.sleepycat.je.Transaction;
|
|||
|
|
import com.sleepycat.je.WriteOptions;
|
|||
|
|
import com.sleepycat.je.dbi.TTL;
|
|||
|
|
import com.sleepycat.je.test.SpeedyTTLTime;
|
|||
|
|
import com.sleepycat.je.utilint.TracerFormatter;
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* Tests concurrent access when using TTL.
|
|||
|
|
*
|
|||
|
|
* Goals:
|
|||
|
|
* - Read records that are expiring.
|
|||
|
|
* - Delete records that are expiring.
|
|||
|
|
* - Update TTL for records that are expiring, including changing TTL to zero.
|
|||
|
|
* - Lock a record multiple times per txn.
|
|||
|
|
* - Read key then data in separate ops.
|
|||
|
|
* - Use secondaries that expire.
|
|||
|
|
* - Read primary and secondary records individually in separate ops.
|
|||
|
|
* - Verify expected data.
|
|||
|
|
* - Tolerate only expected deviant behavior, only on expiration boundaries.
|
|||
|
|
*/
|
|||
|
|
public class TTLStress {
|
|||
|
|
|
|||
|
|
private static final WriteOptions ONE_HOUR_TTL =
|
|||
|
|
new WriteOptions().setTTL(1, TimeUnit.HOURS).setUpdateTTL(true);
|
|||
|
|
|
|||
|
|
/* Must be at least long enough to empty the full queue. */
|
|||
|
|
private static final int TERMINATION_SEC = 10 * 60;
|
|||
|
|
|
|||
|
|
/* Since a TTL of 1 hour is used, this is the TTL in millis. */
|
|||
|
|
private static final int FAKE_MILLIS_PER_HOUR = 100;
|
|||
|
|
|
|||
|
|
/*
|
|||
|
|
* Must be at least the time for a thread to wake up and do an insert. The
|
|||
|
|
* insert and a read operation for that record are queued one after the
|
|||
|
|
* other and may both be assigned to threads at about the same time.
|
|||
|
|
*/
|
|||
|
|
private static final int THREAD_SWITCH_TIME = 5000;
|
|||
|
|
|
|||
|
|
/*
|
|||
|
|
* The read operation expects to read the record, and then for it to
|
|||
|
|
* expire, all before EXPIRATION_MAX_MILLIS.
|
|||
|
|
*/
|
|||
|
|
private static final int EXPIRATION_MAX_MILLIS =
|
|||
|
|
(FAKE_MILLIS_PER_HOUR * 2) + THREAD_SWITCH_TIME;
|
|||
|
|
|
|||
|
|
private static final DateFormat DATE_FORMAT =
|
|||
|
|
TracerFormatter.makeDateFormat();
|
|||
|
|
|
|||
|
|
private static final int DEFAULT_TEST_THREADS = 10;
|
|||
|
|
private static final int DEFAULT_CLEANER_THREADS = 2;
|
|||
|
|
private static final int DEFAULT_DURATION_MINUTES = 30;
|
|||
|
|
private static final int DEFAULT_MAIN_CACHE_MB = 200;
|
|||
|
|
private static final int DEFAULT_OFFHEAP_CACHE_MB = 200;
|
|||
|
|
private static final int DEFAULT_QUEUE_SIZE = 1000;
|
|||
|
|
private static final String DEFAULT_HOME_DIR = "tmp";
|
|||
|
|
|
|||
|
|
public static void main(final String[] args) {
|
|||
|
|
try {
|
|||
|
|
printArgs(args);
|
|||
|
|
final TTLStress test = new TTLStress(args);
|
|||
|
|
test.runTest();
|
|||
|
|
test.close();
|
|||
|
|
System.exit(0);
|
|||
|
|
} catch (Throwable e) {
|
|||
|
|
e.printStackTrace(System.out);
|
|||
|
|
System.exit(-1);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private Environment env;
|
|||
|
|
private Database db;
|
|||
|
|
private SecondaryDatabase secDb;
|
|||
|
|
private int durationMinutes = DEFAULT_DURATION_MINUTES;
|
|||
|
|
private ThreadPoolExecutor executor;
|
|||
|
|
private final SpeedyTTLTime speedyTime =
|
|||
|
|
new SpeedyTTLTime(FAKE_MILLIS_PER_HOUR);
|
|||
|
|
private final AtomicInteger nInserts = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nUpdates = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nDeletions = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nDeleteExpired = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nPriReads = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nPriExpired = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nPriExpiredData = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nPriDeleted = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nPriNotFound = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nSecReads = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nSecExpired = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nSecExpiredData = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nSecDeleted = new AtomicInteger(0);
|
|||
|
|
private final AtomicInteger nSecNotFound = new AtomicInteger(0);
|
|||
|
|
|
|||
|
|
private TTLStress(String[] args) {
|
|||
|
|
|
|||
|
|
int nTestThreads = DEFAULT_TEST_THREADS;
|
|||
|
|
int queueSize = DEFAULT_QUEUE_SIZE;
|
|||
|
|
int nCleanerThreads = DEFAULT_CLEANER_THREADS;
|
|||
|
|
int mainCacheMb = DEFAULT_MAIN_CACHE_MB;
|
|||
|
|
int offheapCacheMb = DEFAULT_OFFHEAP_CACHE_MB;
|
|||
|
|
String homeDir = DEFAULT_HOME_DIR;
|
|||
|
|
|
|||
|
|
/* Parse arguments. */
|
|||
|
|
for (int i = 0; i < args.length; i += 1) {
|
|||
|
|
final String arg = args[i];
|
|||
|
|
final boolean moreArgs = i < args.length - 1;
|
|||
|
|
if (arg.equals("-h") && moreArgs) {
|
|||
|
|
homeDir = args[++i];
|
|||
|
|
} else if (arg.equals("-threads") && moreArgs) {
|
|||
|
|
nTestThreads = Integer.parseInt(args[++i]);
|
|||
|
|
} else if (arg.equals("-cleaners") && moreArgs) {
|
|||
|
|
nCleanerThreads = Integer.parseInt(args[++i]);
|
|||
|
|
} else if (arg.equals("-minutes") && moreArgs) {
|
|||
|
|
durationMinutes = Integer.parseInt(args[++i]);
|
|||
|
|
} else if (arg.equals("-cacheMB") && moreArgs) {
|
|||
|
|
mainCacheMb = Integer.parseInt(args[++i]);
|
|||
|
|
} else if (arg.equals("-offheapMB") && moreArgs) {
|
|||
|
|
offheapCacheMb = Integer.parseInt(args[++i]);
|
|||
|
|
} else if (arg.equals("-queueSize") && moreArgs) {
|
|||
|
|
queueSize = Integer.parseInt(args[++i]);
|
|||
|
|
} else {
|
|||
|
|
throw new IllegalArgumentException("Unknown arg: " + arg);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
executor = new ThreadPoolExecutor(
|
|||
|
|
nTestThreads, nTestThreads,
|
|||
|
|
0L, TimeUnit.SECONDS,
|
|||
|
|
new ArrayBlockingQueue<Runnable>(queueSize),
|
|||
|
|
new ThreadPoolExecutor.AbortPolicy());
|
|||
|
|
|
|||
|
|
executor.prestartAllCoreThreads();
|
|||
|
|
|
|||
|
|
open(homeDir, nCleanerThreads, mainCacheMb, offheapCacheMb);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private static void printArgs(String[] args) {
|
|||
|
|
System.out.print("\nCommand line arguments:");
|
|||
|
|
for (String arg : args) {
|
|||
|
|
System.out.print(' ');
|
|||
|
|
System.out.print(arg);
|
|||
|
|
}
|
|||
|
|
System.out.println();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void open(final String homeDir,
|
|||
|
|
final int nCleanerThreads,
|
|||
|
|
final int mainCacheMb,
|
|||
|
|
final int offheapCacheMb) {
|
|||
|
|
|
|||
|
|
final EnvironmentConfig envConfig = new EnvironmentConfig();
|
|||
|
|
envConfig.setAllowCreate(true);
|
|||
|
|
envConfig.setTransactional(true);
|
|||
|
|
envConfig.setDurability(Durability.COMMIT_NO_SYNC);
|
|||
|
|
envConfig.setConfigParam(
|
|||
|
|
EnvironmentConfig.LOG_FILE_MAX,
|
|||
|
|
String.valueOf(1024 * 1024));
|
|||
|
|
envConfig.setConfigParam(
|
|||
|
|
EnvironmentConfig.CLEANER_THREADS,
|
|||
|
|
String.valueOf(nCleanerThreads));
|
|||
|
|
envConfig.setCacheSize(mainCacheMb * (1024 * 1024));
|
|||
|
|
envConfig.setOffHeapCacheSize(offheapCacheMb * (1024 * 1024));
|
|||
|
|
/* Account for very slow test machines. */
|
|||
|
|
envConfig.setLockTimeout(30, TimeUnit.SECONDS);
|
|||
|
|
|
|||
|
|
env = new Environment(new File(homeDir), envConfig);
|
|||
|
|
|
|||
|
|
final DatabaseConfig dbConfig = new DatabaseConfig();
|
|||
|
|
dbConfig.setAllowCreate(true);
|
|||
|
|
dbConfig.setTransactional(true);
|
|||
|
|
dbConfig.setCacheMode(CacheMode.EVICT_LN);
|
|||
|
|
|
|||
|
|
db = env.openDatabase(null, "priDb", dbConfig);
|
|||
|
|
|
|||
|
|
final SecondaryConfig secConfig = new SecondaryConfig();
|
|||
|
|
secConfig.setAllowCreate(true);
|
|||
|
|
secConfig.setTransactional(true);
|
|||
|
|
secConfig.setCacheMode(CacheMode.EVICT_LN);
|
|||
|
|
secConfig.setSortedDuplicates(true);
|
|||
|
|
secConfig.setKeyCreator(new SecondaryKeyCreator() {
|
|||
|
|
@Override
|
|||
|
|
public boolean createSecondaryKey(final SecondaryDatabase secDb,
|
|||
|
|
final DatabaseEntry key,
|
|||
|
|
final DatabaseEntry data,
|
|||
|
|
final DatabaseEntry result) {
|
|||
|
|
result.setData(key.getData());
|
|||
|
|
return true;
|
|||
|
|
}
|
|||
|
|
});
|
|||
|
|
|
|||
|
|
secDb = env.openSecondaryDatabase(null, "secDb", db, secConfig);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void close() throws InterruptedException {
|
|||
|
|
|
|||
|
|
log("Starting shutdown");
|
|||
|
|
|
|||
|
|
executor.shutdown();
|
|||
|
|
|
|||
|
|
if (!executor.awaitTermination(TERMINATION_SEC, TimeUnit.SECONDS)) {
|
|||
|
|
|
|||
|
|
System.out.println(
|
|||
|
|
"Could not terminate gracefully after " +
|
|||
|
|
TERMINATION_SEC + " seconds");
|
|||
|
|
|
|||
|
|
final List<Runnable> stillRunning = executor.shutdownNow();
|
|||
|
|
|
|||
|
|
if (!stillRunning.isEmpty()) {
|
|||
|
|
|
|||
|
|
System.out.println(
|
|||
|
|
"Did not empty queue during close after " +
|
|||
|
|
TERMINATION_SEC + " seconds, " + stillRunning.size() +
|
|||
|
|
" tasks still running.");
|
|||
|
|
|
|||
|
|
System.exit(1);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
secDb.close();
|
|||
|
|
db.close();
|
|||
|
|
env.close();
|
|||
|
|
|
|||
|
|
log(String.format(
|
|||
|
|
"Test succeeded %n" +
|
|||
|
|
"nInserts: %,d %n" +
|
|||
|
|
"nUpdates: %,d %n" +
|
|||
|
|
"nDeletions: %,d %n" +
|
|||
|
|
"nPriReads: %,d %n" +
|
|||
|
|
"nPriExpired: %,d %n" +
|
|||
|
|
"nPriExpiredData: %,d %n" +
|
|||
|
|
"nPriDeleted: %,d %n" +
|
|||
|
|
"nPriNotFound: %,d %n" +
|
|||
|
|
"nSecReads: %,d %n" +
|
|||
|
|
"nSecExpired: %,d %n" +
|
|||
|
|
"nSecExpiredData: %,d %n" +
|
|||
|
|
"nSecDeleted: %,d %n" +
|
|||
|
|
"nSecNotFound: %,d %n" +
|
|||
|
|
"nDeleteExpired: %,d ",
|
|||
|
|
nInserts.get(),
|
|||
|
|
nUpdates.get(),
|
|||
|
|
nDeletions.get(),
|
|||
|
|
nPriReads.get(),
|
|||
|
|
nPriExpired.get(),
|
|||
|
|
nPriExpiredData.get(),
|
|||
|
|
nPriDeleted.get(),
|
|||
|
|
nPriNotFound.get(),
|
|||
|
|
nSecReads.get(),
|
|||
|
|
nSecExpired.get(),
|
|||
|
|
nSecExpiredData.get(),
|
|||
|
|
nSecDeleted.get(),
|
|||
|
|
nSecNotFound.get(),
|
|||
|
|
nDeleteExpired.get()));
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private static void log(final String msg) {
|
|||
|
|
synchronized (DATE_FORMAT) {
|
|||
|
|
System.out.println(
|
|||
|
|
DATE_FORMAT.format(System.currentTimeMillis()) + " " + msg);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void runTest() throws Throwable {
|
|||
|
|
|
|||
|
|
final long endTime =
|
|||
|
|
System.currentTimeMillis() + (durationMinutes * 60 * 1000);
|
|||
|
|
|
|||
|
|
final Random rnd = new Random(123);
|
|||
|
|
final BlockingQueue<Runnable> queue = executor.getQueue();
|
|||
|
|
|
|||
|
|
speedyTime.start();
|
|||
|
|
|
|||
|
|
while (System.currentTimeMillis() < endTime) {
|
|||
|
|
|
|||
|
|
final boolean doDelete;
|
|||
|
|
final boolean doUpdate;
|
|||
|
|
int nOps = 3; // insert, read primary and secondary
|
|||
|
|
|
|||
|
|
switch (rnd.nextInt(3)) {
|
|||
|
|
case 0:
|
|||
|
|
doDelete = false;
|
|||
|
|
doUpdate = false;
|
|||
|
|
break;
|
|||
|
|
case 1:
|
|||
|
|
doDelete = false;
|
|||
|
|
doUpdate = true;
|
|||
|
|
nOps += 1;
|
|||
|
|
break;
|
|||
|
|
case 2:
|
|||
|
|
doDelete = true;
|
|||
|
|
doUpdate = false;
|
|||
|
|
nOps += 1;
|
|||
|
|
break;
|
|||
|
|
default:
|
|||
|
|
throw new RuntimeException();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if (queue.remainingCapacity() < nOps) {
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
final long key = rnd.nextLong();
|
|||
|
|
final AtomicBoolean abandonOp = new AtomicBoolean(false);
|
|||
|
|
final AtomicLong expirationTime = new AtomicLong(0);
|
|||
|
|
|
|||
|
|
executor.execute(new Write(
|
|||
|
|
key, abandonOp, expirationTime, !doUpdate /*doInsert*/));
|
|||
|
|
|
|||
|
|
executor.execute(new PrimaryRead(
|
|||
|
|
key, abandonOp, expirationTime, doDelete));
|
|||
|
|
|
|||
|
|
executor.execute(new SecondaryRead(
|
|||
|
|
key, abandonOp, expirationTime, doDelete, doUpdate));
|
|||
|
|
|
|||
|
|
if (doUpdate) {
|
|||
|
|
executor.execute(new Write(
|
|||
|
|
key, abandonOp, expirationTime, false /*doInsert*/));
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if (doDelete) {
|
|||
|
|
executor.execute(new Delete(
|
|||
|
|
key, abandonOp, expirationTime));
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private class Write implements Runnable {
|
|||
|
|
|
|||
|
|
private static final boolean DEBUG_WRITE = false;
|
|||
|
|
|
|||
|
|
private final long key;
|
|||
|
|
private final AtomicBoolean abandonOp;
|
|||
|
|
private final AtomicLong expirationTime;
|
|||
|
|
private final boolean doInsert;
|
|||
|
|
|
|||
|
|
Write(final long key,
|
|||
|
|
final AtomicBoolean abandonOp,
|
|||
|
|
final AtomicLong expirationTime,
|
|||
|
|
final boolean doInsert) {
|
|||
|
|
this.key = key;
|
|||
|
|
this.abandonOp = abandonOp;
|
|||
|
|
this.doInsert = doInsert;
|
|||
|
|
this.expirationTime = expirationTime;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
@Override
|
|||
|
|
public void run() {
|
|||
|
|
|
|||
|
|
Thread.currentThread().setName("Write");
|
|||
|
|
|
|||
|
|
try {
|
|||
|
|
final DatabaseEntry keyEntry = new DatabaseEntry();
|
|||
|
|
LongBinding.longToEntry(key, keyEntry);
|
|||
|
|
|
|||
|
|
final boolean useSeparateLN = (key & 1) != 0;
|
|||
|
|
final byte[] dataBytes = new byte[useSeparateLN ? 20 : 8];
|
|||
|
|
final DatabaseEntry dataEntry = new DatabaseEntry();
|
|||
|
|
|
|||
|
|
TupleBase.outputToEntry(
|
|||
|
|
new TupleOutput(dataBytes).writeLong(key),
|
|||
|
|
dataEntry);
|
|||
|
|
|
|||
|
|
final Transaction txn = env.beginTransaction(null, null);
|
|||
|
|
|
|||
|
|
final long opBeforeTime =
|
|||
|
|
speedyTime.realTimeToFakeTime(System.currentTimeMillis());
|
|||
|
|
|
|||
|
|
OperationResult result = db.put(
|
|||
|
|
txn, keyEntry, dataEntry,
|
|||
|
|
doInsert ? Put.NO_OVERWRITE : Put.OVERWRITE,
|
|||
|
|
ONE_HOUR_TTL);
|
|||
|
|
|
|||
|
|
final long opAfterTime =
|
|||
|
|
speedyTime.realTimeToFakeTime(System.currentTimeMillis());
|
|||
|
|
|
|||
|
|
if (result == null && doInsert) {
|
|||
|
|
log("Apparent duplicate random number as key: " + key);
|
|||
|
|
txn.abort();
|
|||
|
|
abandonOp.set(true);
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
assertNotNull("Could not write: " + key, result);
|
|||
|
|
|
|||
|
|
final long opBeforeExpTime =
|
|||
|
|
writeTimeToExpirationTime(opBeforeTime);
|
|||
|
|
|
|||
|
|
final long opAfterExpTime =
|
|||
|
|
writeTimeToExpirationTime(opAfterTime);
|
|||
|
|
|
|||
|
|
final long resultExpTime = result.getExpirationTime();
|
|||
|
|
boolean resultExpTimeMatches = false;
|
|||
|
|
|
|||
|
|
for (long time = opBeforeExpTime;
|
|||
|
|
time <= opAfterExpTime;
|
|||
|
|
time += TTL.MILLIS_PER_HOUR) {
|
|||
|
|
|
|||
|
|
if (resultExpTime == time) {
|
|||
|
|
resultExpTimeMatches = true;
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if (!resultExpTimeMatches) {
|
|||
|
|
fail(
|
|||
|
|
"key: " + key +
|
|||
|
|
" opBeforeTime: " +
|
|||
|
|
formatTime(opBeforeTime) +
|
|||
|
|
" opAfterTime: " +
|
|||
|
|
formatTime(opAfterTime) +
|
|||
|
|
" opBeforeExpTime: " +
|
|||
|
|
formatTime(opBeforeExpTime) +
|
|||
|
|
" opAfterExpTime: " +
|
|||
|
|
formatTime(opAfterExpTime) +
|
|||
|
|
" resultExpTime: " +
|
|||
|
|
formatTime(resultExpTime));
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
expirationTime.set(resultExpTime);
|
|||
|
|
|
|||
|
|
if (DEBUG_WRITE) {
|
|||
|
|
log(
|
|||
|
|
(doInsert ? "Inserted " : "Updated ") +
|
|||
|
|
"key: " + key +
|
|||
|
|
" opBeforeTime: " +
|
|||
|
|
formatTime(opBeforeTime) +
|
|||
|
|
" opAfterTime: " +
|
|||
|
|
formatTime(opAfterTime) +
|
|||
|
|
" resultExpTime: " +
|
|||
|
|
formatTime(resultExpTime));
|
|||
|
|
|
|||
|
|
result = db.get(
|
|||
|
|
txn, keyEntry, dataEntry, Get.SEARCH, null);
|
|||
|
|
|
|||
|
|
assertNotNull("Could not read: " + key, result);
|
|||
|
|
|
|||
|
|
assertEquals(
|
|||
|
|
resultExpTime, result.getExpirationTime());
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
txn.commit();
|
|||
|
|
|
|||
|
|
if (doInsert) {
|
|||
|
|
nInserts.incrementAndGet();
|
|||
|
|
} else {
|
|||
|
|
nUpdates.incrementAndGet();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
} catch (Throwable e) {
|
|||
|
|
e.printStackTrace(System.out);
|
|||
|
|
System.exit(1);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private class Delete implements Runnable {
|
|||
|
|
|
|||
|
|
private final long key;
|
|||
|
|
private final AtomicBoolean abandonOp;
|
|||
|
|
private final AtomicLong expirationTime;
|
|||
|
|
|
|||
|
|
Delete(final long key,
|
|||
|
|
final AtomicBoolean abandonOp,
|
|||
|
|
final AtomicLong expirationTime) {
|
|||
|
|
this.key = key;
|
|||
|
|
this.abandonOp = abandonOp;
|
|||
|
|
this.expirationTime = expirationTime;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
@Override
|
|||
|
|
public void run() {
|
|||
|
|
|
|||
|
|
Thread.currentThread().setName("Delete");
|
|||
|
|
|
|||
|
|
final long startTime = System.currentTimeMillis();
|
|||
|
|
final DatabaseEntry keyEntry = new DatabaseEntry();
|
|||
|
|
LongBinding.longToEntry(key, keyEntry);
|
|||
|
|
|
|||
|
|
try {
|
|||
|
|
while (true) {
|
|||
|
|
|
|||
|
|
if (abandonOp.get()) {
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
final long sysTime = System.currentTimeMillis();
|
|||
|
|
final long opTime = speedyTime.realTimeToFakeTime(sysTime);
|
|||
|
|
|
|||
|
|
if (sysTime - startTime > EXPIRATION_MAX_MILLIS) {
|
|||
|
|
|
|||
|
|
if (TTL.isExpired(expirationTime.get())) {
|
|||
|
|
nDeleteExpired.incrementAndGet();
|
|||
|
|
|
|||
|
|
} else {
|
|||
|
|
fail("Did not expire: " + key +
|
|||
|
|
" opTime: " + formatTime(opTime) +
|
|||
|
|
" expirationTime: " +
|
|||
|
|
formatTime(expirationTime.get()));
|
|||
|
|
}
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
final OperationResult result = db.delete(
|
|||
|
|
null, keyEntry, null);
|
|||
|
|
|
|||
|
|
if (result == null) {
|
|||
|
|
|
|||
|
|
if (TTL.isExpired(expirationTime.get())) {
|
|||
|
|
nDeleteExpired.incrementAndGet();
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
assertEquals(
|
|||
|
|
expirationTime.get(), result.getExpirationTime());
|
|||
|
|
|
|||
|
|
nDeletions.incrementAndGet();
|
|||
|
|
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
} catch (Throwable e) {
|
|||
|
|
e.printStackTrace(System.out);
|
|||
|
|
System.exit(1);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private class PrimaryRead implements Runnable {
|
|||
|
|
|
|||
|
|
private final long key;
|
|||
|
|
private final AtomicBoolean abandonOp;
|
|||
|
|
private final AtomicLong expirationTime;
|
|||
|
|
private final boolean doDelete;
|
|||
|
|
|
|||
|
|
PrimaryRead(final long key,
|
|||
|
|
final AtomicBoolean abandonOp,
|
|||
|
|
final AtomicLong expirationTime,
|
|||
|
|
final boolean doDelete) {
|
|||
|
|
this.key = key;
|
|||
|
|
this.abandonOp = abandonOp;
|
|||
|
|
this.expirationTime = expirationTime;
|
|||
|
|
this.doDelete = doDelete;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
@Override
|
|||
|
|
public void run() {
|
|||
|
|
|
|||
|
|
Thread.currentThread().setName("PrimaryRead");
|
|||
|
|
|
|||
|
|
final DatabaseEntry keyEntry = new DatabaseEntry();
|
|||
|
|
final DatabaseEntry dataEntry = new DatabaseEntry();
|
|||
|
|
final long startTime = System.currentTimeMillis();
|
|||
|
|
OperationResult result;
|
|||
|
|
|
|||
|
|
while (true) {
|
|||
|
|
|
|||
|
|
if (abandonOp.get()) {
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
try (final Cursor cursor = db.openCursor(null, null)) {
|
|||
|
|
|
|||
|
|
final long sysTime = System.currentTimeMillis();
|
|||
|
|
final long opTime = speedyTime.realTimeToFakeTime(sysTime);
|
|||
|
|
|
|||
|
|
if (sysTime - startTime > EXPIRATION_MAX_MILLIS) {
|
|||
|
|
|
|||
|
|
if (TTL.isExpired(expirationTime.get())) {
|
|||
|
|
nPriExpired.incrementAndGet();
|
|||
|
|
|
|||
|
|
} else if (doDelete) {
|
|||
|
|
nPriDeleted.incrementAndGet();
|
|||
|
|
|
|||
|
|
} else {
|
|||
|
|
fail("Did not expire: " + key +
|
|||
|
|
" opTime: " + formatTime(opTime) +
|
|||
|
|
" expirationTime: " +
|
|||
|
|
formatTime(expirationTime.get()));
|
|||
|
|
}
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
nPriReads.incrementAndGet();
|
|||
|
|
|
|||
|
|
LongBinding.longToEntry(key, keyEntry);
|
|||
|
|
final boolean readDataSeparately = (key % 3) == 0;
|
|||
|
|
|
|||
|
|
result = cursor.get(
|
|||
|
|
keyEntry,
|
|||
|
|
readDataSeparately ? null : dataEntry,
|
|||
|
|
Get.SEARCH, null);
|
|||
|
|
|
|||
|
|
if (result == null) {
|
|||
|
|
if (TTL.isExpired(expirationTime.get())) {
|
|||
|
|
nPriExpired.incrementAndGet();
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
assertEquals(key, LongBinding.entryToLong(keyEntry));
|
|||
|
|
assertEquals(
|
|||
|
|
expirationTime.get(), result.getExpirationTime());
|
|||
|
|
|
|||
|
|
if (!readDataSeparately) {
|
|||
|
|
final TupleInput input =
|
|||
|
|
TupleBase.entryToInput(dataEntry);
|
|||
|
|
assertEquals(key, input.readLong());
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
result = cursor.get(
|
|||
|
|
keyEntry, dataEntry, Get.CURRENT, null);
|
|||
|
|
|
|||
|
|
if (result == null) {
|
|||
|
|
if (readDataSeparately &&
|
|||
|
|
TTL.isExpired(expirationTime.get())) {
|
|||
|
|
nPriExpiredData.incrementAndGet();
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
fail("Could not read locked record: " + key);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
assertEquals(key, LongBinding.entryToLong(keyEntry));
|
|||
|
|
final TupleInput input = TupleBase.entryToInput(dataEntry);
|
|||
|
|
assertEquals(key, input.readLong());
|
|||
|
|
assertEquals(
|
|||
|
|
expirationTime.get(), result.getExpirationTime());
|
|||
|
|
|
|||
|
|
} catch (Throwable e) {
|
|||
|
|
e.printStackTrace(System.out);
|
|||
|
|
System.exit(1);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private class SecondaryRead implements Runnable {
|
|||
|
|
|
|||
|
|
private final long key;
|
|||
|
|
private final AtomicBoolean abandonOp;
|
|||
|
|
private final AtomicLong expirationTime;
|
|||
|
|
private final boolean doDelete;
|
|||
|
|
private final boolean doUpdate;
|
|||
|
|
|
|||
|
|
SecondaryRead(final long key,
|
|||
|
|
final AtomicBoolean abandonOp,
|
|||
|
|
final AtomicLong expirationTime,
|
|||
|
|
final boolean doDelete,
|
|||
|
|
final boolean doUpdate) {
|
|||
|
|
this.key = key;
|
|||
|
|
this.abandonOp = abandonOp;
|
|||
|
|
this.expirationTime = expirationTime;
|
|||
|
|
this.doDelete = doDelete;
|
|||
|
|
this.doUpdate = doUpdate;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
@Override
|
|||
|
|
public void run() {
|
|||
|
|
|
|||
|
|
Thread.currentThread().setName("SecondaryRead");
|
|||
|
|
|
|||
|
|
final DatabaseEntry keyEntry = new DatabaseEntry();
|
|||
|
|
final DatabaseEntry pKeyEntry = new DatabaseEntry();
|
|||
|
|
final DatabaseEntry dataEntry = new DatabaseEntry();
|
|||
|
|
final long startTime = System.currentTimeMillis();
|
|||
|
|
OperationResult result;
|
|||
|
|
|
|||
|
|
while (true) {
|
|||
|
|
|
|||
|
|
if (abandonOp.get()) {
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
try (final SecondaryCursor cursor =
|
|||
|
|
secDb.openCursor(null, null)) {
|
|||
|
|
|
|||
|
|
final long sysTime = System.currentTimeMillis();
|
|||
|
|
final long opTime = speedyTime.realTimeToFakeTime(sysTime);
|
|||
|
|
|
|||
|
|
if (sysTime - startTime > EXPIRATION_MAX_MILLIS) {
|
|||
|
|
|
|||
|
|
if (TTL.isExpired(expirationTime.get())) {
|
|||
|
|
nSecExpired.incrementAndGet();
|
|||
|
|
|
|||
|
|
} else if (doDelete) {
|
|||
|
|
nSecDeleted.incrementAndGet();
|
|||
|
|
|
|||
|
|
} else {
|
|||
|
|
fail("Did not expire: " + key +
|
|||
|
|
" currentTime: " + formatTime(opTime) +
|
|||
|
|
" expirationTime: " +
|
|||
|
|
formatTime(expirationTime.get()));
|
|||
|
|
}
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
nSecReads.incrementAndGet();
|
|||
|
|
|
|||
|
|
LongBinding.longToEntry(key, keyEntry);
|
|||
|
|
|
|||
|
|
/*
|
|||
|
|
* If we read data separately when an update or deletion is
|
|||
|
|
* being done, this would cause deadlocks.
|
|||
|
|
*/
|
|||
|
|
final boolean readDataSeparately =
|
|||
|
|
!doUpdate && !doDelete && (key % 3) == 0;
|
|||
|
|
|
|||
|
|
result = cursor.get(
|
|||
|
|
keyEntry, pKeyEntry,
|
|||
|
|
readDataSeparately ? null : dataEntry,
|
|||
|
|
Get.SEARCH, null);
|
|||
|
|
|
|||
|
|
if (result == null) {
|
|||
|
|
if (TTL.isExpired(expirationTime.get())) {
|
|||
|
|
nSecExpired.incrementAndGet();
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
assertEquals(key, LongBinding.entryToLong(keyEntry));
|
|||
|
|
assertEquals(
|
|||
|
|
expirationTime.get(), result.getExpirationTime());
|
|||
|
|
|
|||
|
|
assertTrue(Arrays.equals(
|
|||
|
|
keyEntry.getData(), pKeyEntry.getData()));
|
|||
|
|
|
|||
|
|
if (!readDataSeparately) {
|
|||
|
|
final TupleInput input =
|
|||
|
|
TupleBase.entryToInput(dataEntry);
|
|||
|
|
assertEquals(key, input.readLong());
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
result = cursor.get(
|
|||
|
|
keyEntry, pKeyEntry, dataEntry, Get.CURRENT, null);
|
|||
|
|
|
|||
|
|
if (result == null) {
|
|||
|
|
if (readDataSeparately &&
|
|||
|
|
TTL.isExpired(expirationTime.get())) {
|
|||
|
|
nSecExpiredData.incrementAndGet();
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
fail("Could not read locked record: " + key +
|
|||
|
|
" readDataSeparately: " + readDataSeparately);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
assertEquals(key, LongBinding.entryToLong(keyEntry));
|
|||
|
|
|
|||
|
|
assertTrue(Arrays.equals(
|
|||
|
|
keyEntry.getData(), pKeyEntry.getData()));
|
|||
|
|
|
|||
|
|
final TupleInput input = TupleBase.entryToInput(dataEntry);
|
|||
|
|
assertEquals(key, input.readLong());
|
|||
|
|
assertEquals(
|
|||
|
|
expirationTime.get(), result.getExpirationTime());
|
|||
|
|
|
|||
|
|
} catch (Throwable e) {
|
|||
|
|
e.printStackTrace(System.out);
|
|||
|
|
System.exit(1);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private long writeTimeToExpirationTime(final long writeTime) {
|
|||
|
|
|
|||
|
|
final int expiration = TTL.systemTimeToExpiration(
|
|||
|
|
writeTime + TTL.MILLIS_PER_HOUR,
|
|||
|
|
true /*hours*/);
|
|||
|
|
|
|||
|
|
return TTL.expirationToSystemTime(expiration, true /*hours*/);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private String formatTime(final long time) {
|
|||
|
|
synchronized (DATE_FORMAT) {
|
|||
|
|
return DATE_FORMAT.format(time);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
}
|