protected long _syncPull(final long startTime, RemoteClusterNode peer, SyncListResponse<?> listResponse)
throws InterruptedException, IOException
{
final long processUntil = _stuff.currentTimeMillis() + MAX_TIME_FOR_SYNCPULL_MSECS;
long total = 0L;
ActiveNodeState savedState = null;
int listCalls = 0;
ActiveNodeState pstate = peer.persisted();
// Let's try to limit damage from infinite loops by second check
// (ideally shouldn't need such ad hoc limit but...)
while (_running.get() && ++listCalls < 1000) {
final int count = listResponse.size();
total += count;
if (count == 0) {
break;
}
List<SyncListResponseEntry> newEntries = listResponse.entries;
/*int tombstoneCount =*/ _handleTombstones(newEntries);
// then filter out entries that we already have:
_filterSeen(newEntries);
if (!_running.get()) { // short-circuit during shutdown
break;
}
if (!newEntries.isEmpty()) {
int newCount = newEntries.size();
AtomicInteger rounds = new AtomicInteger(0);
/*long lastProcessed =*/ _fetchMissing(peer.getAddress(), newEntries, rounds);
int fetched = newCount - newEntries.size();
double secs = (_stuff.currentTimeMillis() - startTime) / 1000.0;
String timeDesc = String.format("%.2f", secs);
LOG.info("Fetched {}/{} missing entries ({} listed) from {} in {} seconds ({} rounds)",
new Object[] { fetched, newCount, count, peer.getAddress(), timeDesc, rounds.get()});
}
// One safety thing: let's persist synced-up-to after first round;
// this to reduce likelihood of 'poison pills' from blocking sync pipeline
long lastSeenTimestamp = listResponse.lastSeen();
if (lastSeenTimestamp > 0L) {
pstate = pstate.withSyncedUpTo(lastSeenTimestamp);
} else {
LOG.warn("Missing lastSeenTimestamp from sync-list to {}", peer.getAddress());
}
peer.setPersisted(pstate);
if (_stuff.currentTimeMillis() >= processUntil) {
// we use negative values to indicate time out... old-skool
total = -total;
break;
}
if (savedState == null) {
savedState = pstate;
_stores.getRemoteNodeStore().upsertEntry(peer.getAddress(), pstate);
}
// And then get more stuff...
/* Except for one more thing: if we seem to be running out of entries,
* let's not wait for trickles; inefficient to request stuff by ones and twos.
* Instead, let stuff aggregate and we ought to get more.
*
* ... would be good to be able to verify it has an effect too. But for now
* just need to assume it does.
*/
if (count < 50) {
if (listResponse.clientWait > 0L) {
break;
}
}
listResponse = _syncListAccessor
.fetchRemoteSyncList(_localState, peer.getAddress(),
pstate.getSyncedUpTo(), TIMEOUT_FOR_INITIAL_SYNCLIST_MSECS);
lastSeenTimestamp = listResponse.lastSeen();
}
// Also make sure to update this timestamp
if (savedState != pstate) {