import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
private long getCurrentID(final StaticBuffer partitionKey) throws BackendException {
    List<Entry> blocks = BackendOperation.execute(new BackendOperation.Transactional<List<Entry>>() {
        @Override
        public List<Entry> call(StoreTransaction txh) throws BackendException {
            return idStore.getSlice(new KeySliceQuery(partitionKey, LOWER_SLICE, UPPER_SLICE).setLimit(5), txh);
        }
    },this,times);

    if (blocks == null) throw new TemporaryBackendException("Could not read from storage");
    long latest = BASE_ID;

    for (Entry e : blocks) {
        long counterVal = getBlockValue(e);
        if (latest < counterVal) {
            latest = counterVal;
        }
    }
    return latest;
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
/**
 * Simplest case test of the lock cleaner.
 */
@Test
public void testDeleteSingleLock() throws BackendException {
    Instant now = Instant.ofEpochMilli(1L);

    Entry expiredLockCol = StaticArrayEntry.of(codec.toLockCol(now,
            defaultLockRid, TimestampProviders.MILLI), BufferUtil.getIntBuffer(0));
    EntryList expiredSingleton = StaticArrayEntryList.of(expiredLockCol);

    now = now.plusMillis(1);
    del = new StandardLockCleanerRunnable(store, kc, tx, codec, now, TimestampProviders.MILLI);

    expect(store.getSlice(eq(ksq), eq(tx)))
            .andReturn(expiredSingleton);

    store.mutate(
            eq(key),
            eq(ImmutableList.<Entry> of()),
            eq(ImmutableList.<StaticBuffer> of(expiredLockCol.getColumn())),
            anyObject(StoreTransaction.class));

    ctrl.replay();
    del.run();
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
@Override
public void mutateMany(Map<String, KVMutation> mutations, StoreTransaction txh) throws BackendException {
    for (Map.Entry<String,KVMutation> muts : mutations.entrySet()) {
        BerkeleyJEKeyValueStore store = openDatabase(muts.getKey());
        KVMutation mut = muts.getValue();

        if (!mut.hasAdditions() && !mut.hasDeletions()) {
            log.debug("Empty mutation set for {}, doing nothing", muts.getKey());
        } else {
            log.debug("Mutating {}", muts.getKey());
        }

        if (mut.hasAdditions()) {
            for (KeyValueEntry entry : mut.getAdditions()) {
                store.insert(entry.getKey(),entry.getValue(),txh);
                log.trace("Insertion on {}: {}", muts.getKey(), entry);
            }
        }
        if (mut.hasDeletions()) {
            for (StaticBuffer del : mut.getDeletions()) {
                store.delete(del,txh);
                log.trace("Deletion on {}: {}", muts.getKey(), del);
            }
        }
    }
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
@Override
public StaticBuffer get(StaticBuffer key, StoreTransaction txh) throws BackendException {
    Transaction tx = getTransaction(txh);
    try {
        DatabaseEntry dbkey = key.as(ENTRY_FACTORY);
        DatabaseEntry data = new DatabaseEntry();

        log.trace("db={}, op=get, tx={}", name, txh);

        OperationStatus status = db.get(tx, dbkey, data, getLockMode(txh));

        if (status == OperationStatus.SUCCESS) {
            return getBuffer(data);
        } else {
            return null;
        }
    } catch (DatabaseException e) {
        throw new PermanentBackendException(e);
    }
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
public void insert(StaticBuffer key, StaticBuffer value, StoreTransaction txh, boolean allowOverwrite) throws BackendException {
    Transaction tx = getTransaction(txh);
    try {
        OperationStatus status;

        log.trace("db={}, op=insert, tx={}", name, txh);

        if (allowOverwrite)
            status = db.put(tx, key.as(ENTRY_FACTORY), value.as(ENTRY_FACTORY));
        else
            status = db.putNoOverwrite(tx, key.as(ENTRY_FACTORY), value.as(ENTRY_FACTORY));

        if (status != OperationStatus.SUCCESS) {
            if (status == OperationStatus.KEYEXIST) {
                throw new PermanentBackendException("Key already exists on no-overwrite.");
            } else {
                throw new PermanentBackendException("Could not write entity, return status: " + status);
            }
        }
    } catch (DatabaseException e) {
        throw new PermanentBackendException(e);
    }
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
@Override
public void clean(KeyColumn target, long cutoff, StoreTransaction tx) {
    Long b = blocked.putIfAbsent(target, cutoff);
    if (null == b) {
        log.info("Enqueuing expired lock cleaner task for target={}, tx={}, cutoff={}",
                new Object[] { target, tx, cutoff });
        try {
            exec.submit(new StandardLockCleanerRunnable(store, target, tx, serializer, cutoff));
        } catch (RejectedExecutionException e) {
            log.debug("Failed to enqueue expired lock cleaner for target={}, tx={}, cutoff={}",
                    new Object[] { target, tx, cutoff, e });
        }
    } else {
        log.debug("Blocked redundant attempt to enqueue lock cleaner task for target={}, tx={}, cutoff={}",
                new Object[] { target, tx, cutoff });
    }
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
@Override
public void clean(KeyColumn target, Instant cutoff, StoreTransaction tx) {
    Instant b = blocked.putIfAbsent(target, cutoff);
    if (null == b) {
        log.info("Enqueuing expired lock cleaner task for target={}, tx={}, cutoff={}",
                new Object[] { target, tx, cutoff });
        try {
            exec.submit(new StandardLockCleanerRunnable(store, target, tx, serializer, cutoff, times));
        } catch (RejectedExecutionException e) {
            log.debug("Failed to enqueue expired lock cleaner for target={}, tx={}, cutoff={}",
                    new Object[] { target, tx, cutoff, e });
        }
    } else {
        log.debug("Blocked redundant attempt to enqueue lock cleaner task for target={}, tx={}, cutoff={}",
                new Object[] { target, tx, cutoff });
    }
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
@Override
public void delete(StaticBuffer key, StoreTransaction txh) throws BackendException {
    log.trace("Deletion");
    Transaction tx = getTransaction(txh);
    try {
        log.trace("db={}, op=delete, tx={}", name, txh);
        OperationStatus status = db.delete(tx, key.as(ENTRY_FACTORY));
        if (status != OperationStatus.SUCCESS) {
            throw new PermanentBackendException("Could not remove: " + status);
        }
    } catch (DatabaseException e) {
        throw new PermanentBackendException(e);
    }
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
@Before
public void setupMocks() {
    relaxedCtrl = EasyMock.createControl();
    tx = relaxedCtrl.createMock(StoreTransaction.class);

    ctrl = EasyMock.createStrictControl();
    store = ctrl.createMock(KeyColumnValueStore.class);
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
private LockInfo recordSuccessfulLockWrite(StoreTransaction tx, long duration, TemporalUnit tu, StaticBuffer del) throws BackendException {
    currentTimeNS = currentTimeNS.plusNanos(1);
    expect(times.getTime()).andReturn(currentTimeNS);

    final Instant lockNS = currentTimeNS;

    StaticBuffer lockCol = codec.toLockCol(lockNS, defaultLockRid, times);
    Entry add = StaticArrayEntry.of(lockCol, defaultLockVal);

    StaticBuffer k = eq(defaultLockKey);
    final List<Entry> adds = eq(Arrays.<Entry>asList(add));
    final List<StaticBuffer> dels;
    if (null != del) {
        dels = eq(Arrays.<StaticBuffer>asList(del));
    } else {
        dels = eq(ImmutableList.<StaticBuffer>of());
    }

    store.mutate(k, adds, dels, eq(tx));
    expectLastCall().once();

    currentTimeNS = currentTimeNS.plus(duration, tu);


    expect(times.getTime()).andReturn(currentTimeNS);

    ConsistentKeyLockStatus status = new ConsistentKeyLockStatus(
            lockNS,
            lockNS.plus(defaultExpireNS));

    return new LockInfo(lockNS, status, lockCol);
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
public AbstractLocker(StaticBuffer rid, TimestampProvider times,
        ConsistentKeyLockerSerializer serializer,
        LocalLockMediator<StoreTransaction> llm, LockerState<S> lockState,
        Duration lockExpire, Logger log) {
    this.rid = rid;
    this.times = times;
    this.serializer = serializer;
    this.llm = llm;
    this.lockState = lockState;
    this.lockExpire = lockExpire;
    this.log = log;
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
@Override
public void mutateMany(Map<String, Map<StaticBuffer, KCVMutation>> mutations, StoreTransaction txh) throws BackendException {
    if (!manager.getFeatures().hasStoreTTL()) {
        assert manager.getFeatures().hasCellTTL();
        for (Map.Entry<String,Map<StaticBuffer, KCVMutation>> sentry : mutations.entrySet()) {
            Integer ttl = ttlEnabledStores.get(sentry.getKey());
            if (null != ttl && 0 < ttl) {
                for (KCVMutation mut : sentry.getValue().values()) {
                    if (mut.hasAdditions()) applyTTL(mut.getAdditions(), ttl);
                }
            }
        }
    }
    manager.mutateMany(mutations,txh);
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p/>
 * This implementation supports locking when {@code lockStore} is non-null.
 * <p/>
 * Consider the following scenario. This method is called twice with
 * identical key, column, and txh arguments, but with different
 * expectedValue arguments in each call. In testing, it seems titan's
 * graphdb requires that implementations discard the second expectedValue
 * and, when checking expectedValues vs actual values just prior to mutate,
 * only the initial expectedValue argument should be considered.
 */
@Override
public void acquireLock(StaticBuffer key, StaticBuffer column, StaticBuffer expectedValue, StoreTransaction txh) throws BackendException {
    if (locker != null) {
        ExpectedValueCheckingTransaction tx = (ExpectedValueCheckingTransaction) txh;
        if (tx.isMutationStarted())
            throw new PermanentLockingException("Attempted to obtain a lock after mutations had been persisted");
        KeyColumn lockID = new KeyColumn(key, column);
        log.debug("Attempting to acquireLock on {} ev={}", lockID, expectedValue);
        locker.writeLock(lockID, tx.getConsistentTx());
        tx.storeExpectedValue(this, lockID, expectedValue);
    } else {
        store.acquireLock(key, column, expectedValue, unwrapTx(txh));
    }
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
public StandardLockCleanerRunnable(KeyColumnValueStore store, KeyColumn target, StoreTransaction tx, ConsistentKeyLockerSerializer serializer, Instant cutoff, TimestampProvider times) {
    this.store = store;
    this.target = target;
    this.serializer = serializer;
    this.tx = tx;
    this.cutoff = cutoff;
    this.times = times;
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
/**
 * Retrieve the mediator associated with {@code name} via {@link LocalLockMediators#get(String)}.
 *
 * @param name the mediator name
 * @return this builder
 */
public B mediatorName(String name) {
    Preconditions.checkNotNull(name);
    Preconditions.checkNotNull(times, "Timestamp provider must be set before initializing local lock mediator");
    mediator(LocalLockMediators.INSTANCE.<StoreTransaction>get(name, times));
    return self();
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
public AbstractLocker(StaticBuffer rid, TimestampProvider times,
        ConsistentKeyLockerSerializer serializer,
        LocalLockMediator<StoreTransaction> llm, LockerState<S> lockState,
        Duration lockExpire, Logger log) {
    this.rid = rid;
    this.times = times;
    this.timeUnit = times.getUnit();
    this.serializer = serializer;
    this.llm = llm;
    this.lockState = lockState;
    this.lockExpire = lockExpire;
    this.log = log;
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
public CacheTransaction(StoreTransaction tx, KeyColumnValueStoreManager manager, int persistChunkSize,
                        Duration maxWriteTime, boolean batchLoading, int expectedNumStores) {
    Preconditions.checkArgument(tx != null && manager != null && persistChunkSize > 0);
    this.tx = tx;
    this.manager = manager;
    this.batchLoading = batchLoading;
    this.numMutations = 0;
    this.persistChunkSize = persistChunkSize;
    this.maxWriteTime = maxWriteTime;
    this.mutations = new HashMap<KCVSCache, Map<StaticBuffer, KCVEntryMutation>>(expectedNumStores);
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
@Override
public void deleteLocks(StoreTransaction tx) throws TemporaryLockingException, PermanentLockingException {
    if (null != tx.getConfiguration().getGroupName()) {
        MetricManager.INSTANCE.getCounter(tx.getConfiguration().getGroupName(), M_LOCKS, M_DELETE, M_CALLS).inc();
    }

    Map<KeyColumn, S> m = lockState.getLocksForTx(tx);

    Iterator<KeyColumn> iter = m.keySet().iterator();
    while (iter.hasNext()) {
        KeyColumn kc = iter.next();
        S ls = m.get(kc);
        try {
            deleteSingleLock(kc, ls, tx);
        } catch (AssertionError ae) {
            throw ae; // Concession to ease testing with mocks & behavior verification
        } catch (Throwable t) {
            log.error("Exception while deleting lock on " + kc, t);
            if (null != tx.getConfiguration().getGroupName()) {
                MetricManager.INSTANCE.getCounter(tx.getConfiguration().getGroupName(), M_LOCKS, M_DELETE, M_CALLS).inc();
            }
        }
        // Regardless of whether we successfully deleted the lock from storage, take it out of the local mediator
        llm.unlock(kc, tx);
        iter.remove();
    }
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
public Map<KeyColumn, S> getLocksForTx(StoreTransaction tx) {
    Map<KeyColumn, S> m = locks.get(tx);

    if (null == m) {
        m = new HashMap<KeyColumn, S>();
        final Map<KeyColumn, S> x = locks.putIfAbsent(tx, m);
        if (null != x) {
            m = x;
        }
    }

    return m;
}
 

import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction; //导入依赖的package包/类
public static<R> R execute(Transactional<R> exe, TransactionalProvider provider, TimestampProvider times) throws BackendException {
    StoreTransaction txh = null;
    try {
        txh = provider.openTx();
        if (!txh.getConfiguration().hasCommitTime()) txh.getConfiguration().setCommitTime(times.getTime());
        return exe.call(txh);
    } catch (BackendException e) {
        if (txh!=null) txh.rollback();
        txh=null;
        throw e;
    } finally {
        if (txh!=null) txh.commit();
    }
}