Hugegraph DistributedStoreManager Class Architecture

/**
* Generic interface to a backend storage engine.
*
*/

public interface StoreManager {

/**
* Returns a transaction handle for a new transaction according to the given configuration.
*
* @return New Transaction Handle
*/
public StoreTransaction beginTransaction(BaseTransactionConfig config) throws BackendException;

/**
* Closes the Storage Manager and all databases that have been opened.
*/
public void close() throws BackendException;

/**
* Deletes and clears all database in this storage manager.
* <p/>
* ATTENTION: Invoking this method will delete ALL your data!!
*/
public void clearStorage() throws BackendException;

/**
* Returns the features supported by this storage manager
*
* @return The supported features of this storage manager
* @see StoreFeatures
*/
public StoreFeatures getFeatures();

/**
* Return an identifier for the StoreManager. Two managers with the same
* name would open databases that read and write the same underlying data;
* two store managers with different names should be, for data read/write
* purposes, completely isolated from each other.
* <p/>
* Examples:
* <ul>
* <li>Cassandra keyspace</li>
* <li>HBase tablename</li>
* <li>InMemoryStore heap address (i.e. default toString()).</li>
* </ul>
*
* @return Name for this StoreManager
*/
public String getName();

/**
* Returns {@code KeyRange}s locally hosted on this machine. The start of
* each {@code KeyRange} is inclusive. The end is exclusive. The start and
* end must each be at least 4 bytes in length.
*
* @return A list of local key ranges
* @throws UnsupportedOperationException
*             if the underlying store does not support this operation.
*             Check {@link StoreFeatures#hasLocalKeyPartition()} first.
*/
public List<KeyRange> getLocalKeyPartition() throws BackendException;
}

AbstractStoreManager

*

public abstract class AbstractStoreManager implements StoreManager {

protected final boolean transactional;
protected final boolean batchLoading;
protected final Configuration storageConfig;

public AbstractStoreManager(Configuration storageConfig) {
batchLoading = storageConfig.get(STORAGE_BATCH);
boolean transactional = storageConfig.get(STORAGE_TRANSACTIONAL);
if (batchLoading) {
transactional = false;
}
this.transactional = transactional;
this.storageConfig = storageConfig;
}

public Configuration getStorageConfig() {
return storageConfig;
}

public EntryMetaData[] getMetaDataSchema(String storeName) {
List<EntryMetaData> schemaBuilder = Lists.newArrayList();
StoreFeatures features = getFeatures();
if (features.hasTimestamps() && storageConfig.get(STORE_META_TIMESTAMPS,storeName))
schemaBuilder.add(EntryMetaData.TIMESTAMP);
if (features.hasCellTTL() && storageConfig.get(STORE_META_TTL,storeName))
schemaBuilder.add(EntryMetaData.TTL);
if (features.hasVisibility() && storageConfig.get(STORE_META_VISIBILITY,storeName))
schemaBuilder.add(EntryMetaData.VISIBILITY);

if (schemaBuilder.isEmpty()) return StaticArrayEntry.EMPTY_SCHEMA;
return schemaBuilder.toArray(new EntryMetaData[schemaBuilder.size()]);
}

}

DistributedStoreManager

Deployment: Enum in DistributedStoreManager

public enum Deployment {

/**
* Connects to storage backend over the network or some other connection with significant latency
*/
REMOTE,

/**
* Connects to storage backend over localhost or some other connection with very low latency
*/
LOCAL,

/**
* Embedded with storage backend and communicates inside the JVM
*/
EMBEDDED

}

Fields in DistributedStoreManager

protected final TimestampProvider times;
private static final Random random = new Random();

protected final String[] hostnames;
protected final int port;
protected final Duration connectionTimeoutMS;
protected final int pageSize;

protected final String username;
protected final String password;

Methods in DistributedStoreManager.

/**
* Returns a randomly chosen host name. This is used to pick one host when multiple are configured
*
* @return
*/
protected String getSingleHostname() {
return hostnames[random.nextInt(hostnames.length)];
}

/**
* Whether authentication is enabled for this storage backend
*
* @return
*/
public boolean hasAuthentication() {
return username!=null;
}

/**
* Returns the default configured page size for this storage backend. The page size is used to determine
* the number of records to request at a time when streaming result data.
* @return
*/
public int getPageSize() {
return pageSize;
}

/*
* TODO this should go away once we have a HugeGraphConfig that encapsulates TimestampProvider
*/
public TimestampProvider getTimestampProvider() {
return times;
}

/**
* Returns the {@link Deployment} mode of this connection to the storage backend
*
* @return
*/
public abstract Deployment getDeployment();

@Override
public String toString() {
String hn = getSingleHostname();
return hn.substring(0, Math.min(hn.length(), 256)) + ":" + port;
}

protected void sleepAfterWrite(StoreTransaction txh, MaskedTimestamp mustPass) throws BackendException {
assert mustPass.getDeletionTime(times) < mustPass.getAdditionTime(times);
try {
times.sleepPast(mustPass.getAdditionTimeInstant(times));
} catch (InterruptedException e) {
throw new PermanentBackendException("Unexpected interrupt", e);
}

/**
* Helper class to create the deletion and addition timestamps for a particular transaction.
* It needs to be ensured that the deletion time is prior to the addition time since
* some storage backends use the time to resolve conflicts.
*/
public static class MaskedTimestamp {

private final Instant t;

public MaskedTimestamp(Instant commitTime) {
Preconditions.checkNotNull(commitTime);
this.t=commitTime;
}

public MaskedTimestamp(StoreTransaction txh) {
this(txh.getConfiguration().getCommitTime());
}

public long getDeletionTime(TimestampProvider times) {
return times.getTime(t)   & 0xFFFFFFFFFFFFFFFEL; // zero the LSB
}

public long getAdditionTime(TimestampProvider times) {
return (times.getTime(t)   & 0xFFFFFFFFFFFFFFFEL) | 1L; // force the LSB to 1
}

public Instant getAdditionTimeInstant(TimestampProvider times) {
return times.getTime(getAdditionTime(times));
}
}