natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Decide on a gcBefore before sending out TreeRequests 

patch by marcuse, reviewed by yukim, for CASSANDRA-4932
This commit is contained in:
Marcus Eriksson 2013-04-09 08:08:32 +02:00
parent 3e0b21c45a
commit efbdee2370
8 changed files with 30 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -28,7 +28,8 @@
* remove row-level bloom filters (CASSANDRA-4885)
* Change Kernel Page Cache skipping into row preheating (disabled by default)
(CASSANDRA-4937)
* Improve repair by deciding on a gcBefore before sending
out TreeRequests (CASSANDRA-4932)
1.2.4
* Ensure that PerRowSecondaryIndex updates see the most recent values

5
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -618,7 +618,10 @@ public class CompactionManager implements CompactionManagerMBean
// we don't mark validating sstables as compacting in DataTracker, so we have to mark them referenced
// instead so they won't be cleaned up if they do get compacted during the validation
sstables = cfs.markCurrentSSTablesReferenced();
gcBefore = getDefaultGcBefore(cfs);
if (validator.request.gcBefore > 0)
gcBefore = validator.request.gcBefore;
else
gcBefore = getDefaultGcBefore(cfs);
}
CompactionIterable ci = new ValidationCompactionIterable(cfs, sstables, validator.request.range, gcBefore);

27
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -442,7 +442,7 @@ public class ActiveRepairService
public void doVerb(MessageIn<TreeRequest> message, int id)
{
TreeRequest remotereq = message.payload;
TreeRequest request = new TreeRequest(remotereq.sessionid, message.from, remotereq.range, remotereq.cf);
TreeRequest request = new TreeRequest(remotereq.sessionid, message.from, remotereq.range, remotereq.gcBefore, remotereq.cf);
// trigger read-only compaction
ColumnFamilyStore store = Table.open(request.cf.left).getColumnFamilyStore(request.cf.right);
@ -462,7 +462,7 @@ public class ActiveRepairService
{
// deserialize the remote tree, and register it
Validator response = message.payload;
TreeRequest request = new TreeRequest(response.request.sessionid, message.from, response.request.range, response.request.cf);
TreeRequest request = new TreeRequest(response.request.sessionid, message.from, response.request.range, response.request.gcBefore, response.request.cf);
ActiveRepairService.instance.rendezvous(request, response.tree);
}
}
@ -489,20 +489,22 @@ public class ActiveRepairService
public final String sessionid;
public final InetAddress endpoint;
public final Range<Token> range;
public final int gcBefore;
public final CFPair cf;
public TreeRequest(String sessionid, InetAddress endpoint, Range<Token> range, CFPair cf)
public TreeRequest(String sessionid, InetAddress endpoint, Range<Token> range, int gcBefore, CFPair cf)
{
this.sessionid = sessionid;
this.endpoint = endpoint;
this.cf = cf;
this.gcBefore = gcBefore;
this.range = range;
}
@Override
public final int hashCode()
{
return Objects.hashCode(sessionid, endpoint, cf, range);
return Objects.hashCode(sessionid, endpoint, gcBefore, cf, range);
}
@Override
@ -512,13 +514,13 @@ public class ActiveRepairService
return false;
TreeRequest that = (TreeRequest)o;
// handles nulls properly
return Objects.equal(sessionid, that.sessionid) && Objects.equal(endpoint, that.endpoint) && Objects.equal(cf, that.cf) && Objects.equal(range, that.range);
return Objects.equal(sessionid, that.sessionid) && Objects.equal(endpoint, that.endpoint) && gcBefore == that.gcBefore && Objects.equal(cf, that.cf) && Objects.equal(range, that.range);
}
@Override
public String toString()
{
return "#<TreeRequest " + sessionid + ", " + endpoint + ", " + cf + ", " + range + ">";
return "#<TreeRequest " + sessionid + ", " + endpoint + ", " + gcBefore + ", " + cf + ", " + range + ">";
}
public MessageOut<TreeRequest> createMessage()
@ -532,6 +534,9 @@ public class ActiveRepairService
{
out.writeUTF(request.sessionid);
CompactEndpointSerializationHelper.serialize(request.endpoint, out);
if (version >= MessagingService.VERSION_20)
out.writeInt(request.gcBefore);
out.writeUTF(request.cf.left);
out.writeUTF(request.cf.right);
AbstractBounds.serializer.serialize(request.range, out, version);
@ -541,17 +546,21 @@ public class ActiveRepairService
{
String sessId = in.readUTF();
InetAddress endpoint = CompactEndpointSerializationHelper.deserialize(in);
int gcBefore = -1;
if (version >= MessagingService.VERSION_20)
gcBefore = in.readInt();
CFPair cfpair = new CFPair(in.readUTF(), in.readUTF());
Range<Token> range;
range = (Range<Token>) AbstractBounds.serializer.deserialize(in, version);
return new TreeRequest(sessId, endpoint, range, cfpair);
return new TreeRequest(sessId, endpoint, range, gcBefore, cfpair);
}
public long serializedSize(TreeRequest request, int version)
{
return TypeSizes.NATIVE.sizeof(request.sessionid)
+ CompactEndpointSerializationHelper.serializedSize(request.endpoint)
+ TypeSizes.NATIVE.sizeof(request.gcBefore)
+ TypeSizes.NATIVE.sizeof(request.cf.left)
+ TypeSizes.NATIVE.sizeof(request.cf.right)
+ AbstractBounds.serializer.serializedSize(request.range, version);
@ -833,8 +842,10 @@ public class ActiveRepairService
if (isSequential)
makeSnapshots(endpoints);
int gcBefore = (int)(System.currentTimeMillis()/1000) - Table.open(tablename).getColumnFamilyStore(cfname).metadata.getGcGraceSeconds();
for (InetAddress endpoint : allEndpoints)
treeRequests.add(new TreeRequest(getName(), endpoint, range, new CFPair(tablename, cfname)));
treeRequests.add(new TreeRequest(getName(), endpoint, range, gcBefore, new CFPair(tablename, cfname)));
logger.info(String.format("[repair #%s] requesting merkle trees for %s (to %s)", getName(), cfname, allEndpoints));
treeRequests.start();

BIN
test/data/serialization/2.0/service.TreeRequest.bin
View File

Binary file not shown.

BIN
test/data/serialization/2.0/service.TreeResponse.bin
View File

Binary file not shown.

3
test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
View File

@ -91,7 +91,8 @@ public class LeveledCompactionStrategyTest extends SchemaLoader
ActiveRepairService.CFPair p = new ActiveRepairService.CFPair(ksname, cfname);
Range<Token> range = new Range<Token>(Util.token(""), Util.token(""));
ActiveRepairService.TreeRequest req = new ActiveRepairService.TreeRequest("1", FBUtilities.getLocalAddress(), range, p);
int gcBefore = (int)(System.currentTimeMillis()/1000) - table.getColumnFamilyStore(cfname).metadata.getGcGraceSeconds();
ActiveRepairService.TreeRequest req = new ActiveRepairService.TreeRequest("1", FBUtilities.getLocalAddress(), range, gcBefore, p);
ActiveRepairService.Validator validator = new ActiveRepairService.Validator(req);
CompactionManager.instance.submitValidation(store, validator).get();
}

3
test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java
View File

@ -105,7 +105,8 @@ public abstract class AntiEntropyServiceTestAbstract extends SchemaLoader
local_range = StorageService.instance.getLocalPrimaryRange();
// (we use REMOTE instead of LOCAL so that the reponses for the validator.complete() get lost)
request = new TreeRequest(UUID.randomUUID().toString(), REMOTE, local_range, new CFPair(tablename, cfname));
int gcBefore = (int)(System.currentTimeMillis()/1000) - store.metadata.getGcGraceSeconds();
request = new TreeRequest(UUID.randomUUID().toString(), REMOTE, local_range, gcBefore, new CFPair(tablename, cfname));
// Set a fake session corresponding to this fake request
ActiveRepairService.instance.submitArtificialRepairSession(request, tablename, cfname);
}

2
test/unit/org/apache/cassandra/service/SerializationsTest.java
View File

@ -106,6 +106,6 @@ public class SerializationsTest extends AbstractSerializationsTester
private static class Statics
{
private static final ActiveRepairService.CFPair pair = new ActiveRepairService.CFPair("Keyspace1", "Standard1");
private static final ActiveRepairService.TreeRequest req = new ActiveRepairService.TreeRequest("sessionId", FBUtilities.getBroadcastAddress(), FULL_RANGE, pair);
private static final ActiveRepairService.TreeRequest req = new ActiveRepairService.TreeRequest("sessionId", FBUtilities.getBroadcastAddress(), FULL_RANGE, 1234, pair);
}
}