assert(request.hasTransactionId()) : "Got " + request.getClass().getSimpleName() + " without a txn id!";
Long txn_id = request.getTransactionId();
if (debug.val)
LOG.debug(String.format("Got %s for txn #%d", request.getClass().getSimpleName(), txn_id));
AbstractTransaction ts = this.hstore_site.getTransaction(txn_id);
assert(ts == null || ts instanceof LocalTransaction) :
String.format("Got init request for remote txn #%d but we already have one [%s]",
txn_id, ts);
// This allocation is unnecessary if we're on the same site
PartitionSet partitions = null;
if (ts instanceof LocalTransaction) {
partitions = ((LocalTransaction)ts).getPredictTouchedPartitions();
} else {
// We first need all of the partitions so that we know
// what it's actually going to touch
// The init callback obviously only needs to have the
// partitions that are local at this site.
partitions = new PartitionSet(request.getPartitionsList());
ParameterSet procParams = null;
if (request.hasProcParams()) {
FastDeserializer fds = new FastDeserializer(request.getProcParams().asReadOnlyByteBuffer());
try {
procParams = fds.readObject(ParameterSet.class);
} catch (Exception ex) {
String msg = String.format("Failed to deserialize procedure ParameterSet for txn #%d from %s",
txn_id, request.getClass().getSimpleName());
throw new ServerFaultException(msg, ex, txn_id);
}
}
// If we don't have a handle, we need to make one so that we can stick in the
// things that we need to keep track of at this site. At this point we know that we're on
// a remote site from the txn's base partition
ts = this.hstore_site.getTransactionInitializer()
.createRemoteTransaction(txn_id,
partitions,
procParams,
request.getBasePartition(),
request.getProcedureId());
// Make sure that we initialize the RemoteTransactionInitCallback too!
RemoteInitQueueCallback initCallback = ts.getInitCallback();
initCallback.init((RemoteTransaction)ts, partitions, callback);
}
// If (request.getPrefetchFragmentsCount() > 0), then we need to
// make a RemoteTransaction handle for ourselves so that we can keep track of
// our state when prefetching queries.
if (request.getPrefetchFragmentsCount() > 0) {
// Stick the prefetch information into the transaction
if (debug.val) {
PartitionSet prefetchPartitions = new PartitionSet();
for (WorkFragment fragment : request.getPrefetchFragmentsList())
prefetchPartitions.add(fragment.getPartitionId());
LOG.debug(String.format("%s - Attaching %d prefetch %s at partitions %s",
ts, request.getPrefetchFragmentsCount(),
WorkFragment.class.getSimpleName(), prefetchPartitions));
}
// for (int i = 0; i < request.getPrefetchParamsCount(); i++) {
// LOG.info(String.format("%s - XXX INBOUND PREFETCH RAW [%02d]: %s",
// ts, i,
// StringUtil.md5sum(request.getPrefetchParams(i).asReadOnlyByteBuffer())));
// }
ts.initializePrefetch();
ts.attachPrefetchQueries(request.getPrefetchFragmentsList(),
request.getPrefetchParamsList());
}
// We don't need to send back a response right here.
// The init callback will wait until it has results from all of the partitions