snapshots[vertexId - fromVertex] = null;
}
@Override
public void restoreUngrabbed() {
final TimerContext _timer = restore.time();
// Restore such walks that were not grabbed (because the vertex
// was not initially scheduled)
int v = fromVertex;
int restoreCount = 0;
for(long[] snapshot : snapshots) {
if (snapshot != null && !processedBits[v - fromVertex]) {
for(int i=0; i<snapshot.length; i++) {
long w = snapshot[i];
moveWalk(w, v, trackBit(w));
restoreCount++;
}
}
v++;
}
logger.info("Restored " + restoreCount);
_timer.stop();
}
// Note: accurate number only before snapshot is being purged
public long numWalks() {
long sum = 0;
for(int b=fromBucket; b <= toBucket; b++) {
sum += walks[b].length;
}
return sum;
}
@Override
public WalkArray getWalksAtVertex(int vertexId, boolean processed) {
int bucketIdx = vertexId / bucketSize;
int localBucketIdx = bucketIdx - (fromVertex / bucketSize);
processedBits[vertexId - fromVertex] = true;
if (snapshotInitBits[localBucketIdx]) {
long[] array = snapshots[vertexId - fromVertex];
if (array == null) {
return null;
} else {
return new LongWalkArray(snapshots[vertexId - fromVertex]);
}
} else {
final TimerContext _timer = grabTimer.time();
long[] bucketToConsume = null;
int len = 0;
synchronized (bucketLocks[bucketIdx]) {
if (!snapshotInitBits[localBucketIdx]) {
int bucketFirstVertex = bucketSize * bucketIdx;
len = walkIndices[bucketIdx];
bucketToConsume = walks[bucketIdx];
if (bucketToConsume != null) {
walks[bucketIdx] = null;
walkIndices[bucketIdx] = 0;
final int[] snapshotSizes = new int[bucketSize];
final int[] snapshotIdxs = new int[bucketSize];
/* Calculate vertex-walks sizes */
for(int i=0; i < len; i++) {
long w = bucketToConsume[i];
snapshotSizes[off(w)]++;
}
int offt = bucketFirstVertex - fromVertex;
for(int i=0; i < snapshotSizes.length; i++) {
if (snapshotSizes[i] > 0 && i >= -offt && i + offt < snapshots.length)
snapshots[i + offt] = new long[snapshotSizes[i]];
}
for(int i=0; i < len; i++) {
long w = bucketToConsume[i];
int vertex = bucketFirstVertex + off(w);
if (vertex >= fromVertex && vertex <= toVertexInclusive) {
int snapshotOff = vertex - fromVertex;
int localOff = vertex - bucketFirstVertex;
snapshots[snapshotOff][snapshotIdxs[localOff]] = w;
snapshotIdxs[localOff]++;
} else {
// add back
moveWalk(w, vertex, trackBit(w));
}
}
}
snapshotInitBits[localBucketIdx] = true;
}
}
if (bucketConsumer != null && bucketToConsume != null && len > 0) {
bucketConsumer.consume(bucketIdx * bucketSize, new LongWalkArray(bucketToConsume), len);
if (len > 1000000) {
log((bucketIdx * bucketSize) + " - " + ((bucketIdx+1)) * bucketSize + ", " + len);
}
}
_timer.stop();
long[] array = snapshots[vertexId - fromVertex];
if (array == null) {
return null;
} else {
return new LongWalkArray(snapshots[vertexId - fromVertex]);