You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by el...@apache.org on 2013/10/09 00:57:51 UTC
git commit: ACCUMULO-1566 Pass down the readaheadThreshold parameter
from the client to the server so that the same limit is adhered to by the
server in regards to pipelining.
Updated Branches:
refs/heads/ACCUMULO-1566 [created] ff95c7147
ACCUMULO-1566 Pass down the readaheadThreshold parameter from the client to the
server so that the same limit is adhered to by the server in regards to
pipelining.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/ff95c714
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/ff95c714
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/ff95c714
Branch: refs/heads/ACCUMULO-1566
Commit: ff95c7147d210171fef3824eae399b7384cdaff9
Parents: e70a40d
Author: Josh Elser <el...@apache.org>
Authored: Tue Oct 8 18:54:30 2013 -0400
Committer: Josh Elser <el...@apache.org>
Committed: Tue Oct 8 18:54:30 2013 -0400
----------------------------------------------------------------------
.../core/client/impl/ThriftScanner.java | 15 ++-
.../thrift/TabletClientService.java | 124 +++++++++++++++++--
core/src/main/thrift/tabletserver.thrift | 3 +-
.../server/tabletserver/TabletServer.java | 10 +-
.../test/performance/thrift/NullTserver.java | 2 +-
5 files changed, 132 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff95c714/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java
index efdd142..efb31e8 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java
@@ -27,6 +27,7 @@ import java.util.Set;
import java.util.SortedMap;
import java.util.SortedSet;
+import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Instance;
@@ -97,7 +98,7 @@ public class ThriftScanner {
boolean waitForWrites = !serversWaitedForWrites.get(ttype).contains(server);
InitialScan isr = client.startScan(tinfo, scanState.credentials.toThrift(instance), extent.toThrift(), scanState.range.toThrift(),
Translator.translate(scanState.columns, Translator.CT), scanState.size, scanState.serverSideIteratorList, scanState.serverSideIteratorOptions,
- scanState.authorizations.getAuthorizationsBB(), waitForWrites, scanState.isolated);
+ scanState.authorizations.getAuthorizationsBB(), waitForWrites, scanState.isolated, scanState.readaheadThreshold);
if (waitForWrites)
serversWaitedForWrites.get(ttype).add(server);
@@ -132,6 +133,7 @@ public class ThriftScanner {
Text tableId;
Text startRow;
boolean skipStartRow;
+ long readaheadThreshold;
Range range;
@@ -150,9 +152,15 @@ public class ThriftScanner {
List<IterInfo> serverSideIteratorList;
Map<String,Map<String,String>> serverSideIteratorOptions;
-
+
public ScanState(Instance instance, Credentials credentials, Text tableId, Authorizations authorizations, Range range, SortedSet<Column> fetchedColumns,
int size, List<IterInfo> serverSideIteratorList, Map<String,Map<String,String>> serverSideIteratorOptions, boolean isolated) {
+ this(instance, credentials, tableId, authorizations, range, fetchedColumns, size, serverSideIteratorList, serverSideIteratorOptions, isolated,
+ Constants.SCANNER_DEFAULT_READAHEAD_THRESHOLD);
+ }
+
+ public ScanState(Instance instance, Credentials credentials, Text tableId, Authorizations authorizations, Range range, SortedSet<Column> fetchedColumns,
+ int size, List<IterInfo> serverSideIteratorList, Map<String,Map<String,String>> serverSideIteratorOptions, boolean isolated, long readaheadThreshold) {
this.instance = instance;
this.credentials = credentials;
this.authorizations = authorizations;
@@ -179,6 +187,7 @@ public class ThriftScanner {
this.serverSideIteratorOptions = serverSideIteratorOptions;
this.isolated = isolated;
+ this.readaheadThreshold = readaheadThreshold;
}
}
@@ -389,7 +398,7 @@ public class ThriftScanner {
boolean waitForWrites = !serversWaitedForWrites.get(ttype).contains(loc.tablet_location);
InitialScan is = client.startScan(tinfo, scanState.credentials.toThrift(scanState.instance), loc.tablet_extent.toThrift(), scanState.range.toThrift(),
Translator.translate(scanState.columns, Translator.CT), scanState.size, scanState.serverSideIteratorList, scanState.serverSideIteratorOptions,
- scanState.authorizations.getAuthorizationsBB(), waitForWrites, scanState.isolated);
+ scanState.authorizations.getAuthorizationsBB(), waitForWrites, scanState.isolated, scanState.readaheadThreshold);
if (waitForWrites)
serversWaitedForWrites.get(ttype).add(loc.tablet_location);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff95c714/core/src/main/java/org/apache/accumulo/core/tabletserver/thrift/TabletClientService.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/tabletserver/thrift/TabletClientService.java b/core/src/main/java/org/apache/accumulo/core/tabletserver/thrift/TabletClientService.java
index bd6578d..d02b5da 100644
--- a/core/src/main/java/org/apache/accumulo/core/tabletserver/thrift/TabletClientService.java
+++ b/core/src/main/java/org/apache/accumulo/core/tabletserver/thrift/TabletClientService.java
@@ -50,7 +50,7 @@ import org.slf4j.LoggerFactory;
public interface Iface extends org.apache.accumulo.core.client.impl.thrift.ClientService.Iface {
- public org.apache.accumulo.core.data.thrift.InitialScan startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated) throws org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException, NotServingTabletException, TooManyFilesException, org.apache.thrift.TException;
+ public org.apache.accumulo.core.data.thrift.InitialScan startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, long readaheadThreshold) throws org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException, NotServingTabletException, TooManyFilesException, org.apache.thrift.TException;
public org.apache.accumulo.core.data.thrift.ScanResult continueScan(org.apache.accumulo.trace.thrift.TInfo tinfo, long scanID) throws NoSuchScanIDException, NotServingTabletException, TooManyFilesException, org.apache.thrift.TException;
@@ -114,7 +114,7 @@ import org.slf4j.LoggerFactory;
public interface AsyncIface extends org.apache.accumulo.core.client.impl.thrift.ClientService .AsyncIface {
- public void startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.startScan_call> resultHandler) throws org.apache.thrift.TException;
+ public void startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, long readaheadThreshold, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.startScan_call> resultHandler) throws org.apache.thrift.TException;
public void continueScan(org.apache.accumulo.trace.thrift.TInfo tinfo, long scanID, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.continueScan_call> resultHandler) throws org.apache.thrift.TException;
@@ -196,13 +196,13 @@ import org.slf4j.LoggerFactory;
super(iprot, oprot);
}
- public org.apache.accumulo.core.data.thrift.InitialScan startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated) throws org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException, NotServingTabletException, TooManyFilesException, org.apache.thrift.TException
+ public org.apache.accumulo.core.data.thrift.InitialScan startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, long readaheadThreshold) throws org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException, NotServingTabletException, TooManyFilesException, org.apache.thrift.TException
{
- send_startScan(tinfo, credentials, extent, range, columns, batchSize, ssiList, ssio, authorizations, waitForWrites, isolated);
+ send_startScan(tinfo, credentials, extent, range, columns, batchSize, ssiList, ssio, authorizations, waitForWrites, isolated, readaheadThreshold);
return recv_startScan();
}
- public void send_startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated) throws org.apache.thrift.TException
+ public void send_startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, long readaheadThreshold) throws org.apache.thrift.TException
{
startScan_args args = new startScan_args();
args.setTinfo(tinfo);
@@ -216,6 +216,7 @@ import org.slf4j.LoggerFactory;
args.setAuthorizations(authorizations);
args.setWaitForWrites(waitForWrites);
args.setIsolated(isolated);
+ args.setReadaheadThreshold(readaheadThreshold);
sendBase("startScan", args);
}
@@ -922,9 +923,9 @@ import org.slf4j.LoggerFactory;
super(protocolFactory, clientManager, transport);
}
- public void startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, org.apache.thrift.async.AsyncMethodCallback<startScan_call> resultHandler) throws org.apache.thrift.TException {
+ public void startScan(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, long readaheadThreshold, org.apache.thrift.async.AsyncMethodCallback<startScan_call> resultHandler) throws org.apache.thrift.TException {
checkReady();
- startScan_call method_call = new startScan_call(tinfo, credentials, extent, range, columns, batchSize, ssiList, ssio, authorizations, waitForWrites, isolated, resultHandler, this, ___protocolFactory, ___transport);
+ startScan_call method_call = new startScan_call(tinfo, credentials, extent, range, columns, batchSize, ssiList, ssio, authorizations, waitForWrites, isolated, readaheadThreshold, resultHandler, this, ___protocolFactory, ___transport);
this.___currentMethod = method_call;
___manager.call(method_call);
}
@@ -941,7 +942,8 @@ import org.slf4j.LoggerFactory;
private List<ByteBuffer> authorizations;
private boolean waitForWrites;
private boolean isolated;
- public startScan_call(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, org.apache.thrift.async.AsyncMethodCallback<startScan_call> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
+ private long readaheadThreshold;
+ public startScan_call(org.apache.accumulo.trace.thrift.TInfo tinfo, org.apache.accumulo.core.security.thrift.TCredentials credentials, org.apache.accumulo.core.data.thrift.TKeyExtent extent, org.apache.accumulo.core.data.thrift.TRange range, List<org.apache.accumulo.core.data.thrift.TColumn> columns, int batchSize, List<org.apache.accumulo.core.data.thrift.IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, long readaheadThreshold, org.apache.thrift.async.AsyncMethodCallback<startScan_call> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
super(client, protocolFactory, transport, resultHandler, false);
this.tinfo = tinfo;
this.credentials = credentials;
@@ -954,6 +956,7 @@ import org.slf4j.LoggerFactory;
this.authorizations = authorizations;
this.waitForWrites = waitForWrites;
this.isolated = isolated;
+ this.readaheadThreshold = readaheadThreshold;
}
public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
@@ -970,6 +973,7 @@ import org.slf4j.LoggerFactory;
args.setAuthorizations(authorizations);
args.setWaitForWrites(waitForWrites);
args.setIsolated(isolated);
+ args.setReadaheadThreshold(readaheadThreshold);
args.write(prot);
prot.writeMessageEnd();
}
@@ -2170,7 +2174,7 @@ import org.slf4j.LoggerFactory;
public startScan_result getResult(I iface, startScan_args args) throws org.apache.thrift.TException {
startScan_result result = new startScan_result();
try {
- result.success = iface.startScan(args.tinfo, args.credentials, args.extent, args.range, args.columns, args.batchSize, args.ssiList, args.ssio, args.authorizations, args.waitForWrites, args.isolated);
+ result.success = iface.startScan(args.tinfo, args.credentials, args.extent, args.range, args.columns, args.batchSize, args.ssiList, args.ssio, args.authorizations, args.waitForWrites, args.isolated, args.readaheadThreshold);
} catch (org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException sec) {
result.sec = sec;
} catch (NotServingTabletException nste) {
@@ -2846,6 +2850,7 @@ import org.slf4j.LoggerFactory;
private static final org.apache.thrift.protocol.TField AUTHORIZATIONS_FIELD_DESC = new org.apache.thrift.protocol.TField("authorizations", org.apache.thrift.protocol.TType.LIST, (short)8);
private static final org.apache.thrift.protocol.TField WAIT_FOR_WRITES_FIELD_DESC = new org.apache.thrift.protocol.TField("waitForWrites", org.apache.thrift.protocol.TType.BOOL, (short)9);
private static final org.apache.thrift.protocol.TField ISOLATED_FIELD_DESC = new org.apache.thrift.protocol.TField("isolated", org.apache.thrift.protocol.TType.BOOL, (short)10);
+ private static final org.apache.thrift.protocol.TField READAHEAD_THRESHOLD_FIELD_DESC = new org.apache.thrift.protocol.TField("readaheadThreshold", org.apache.thrift.protocol.TType.I64, (short)12);
private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
static {
@@ -2864,6 +2869,7 @@ import org.slf4j.LoggerFactory;
public List<ByteBuffer> authorizations; // required
public boolean waitForWrites; // required
public boolean isolated; // required
+ public long readaheadThreshold; // required
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
@SuppressWarnings("all") public enum _Fields implements org.apache.thrift.TFieldIdEnum {
@@ -2877,7 +2883,8 @@ import org.slf4j.LoggerFactory;
SSIO((short)7, "ssio"),
AUTHORIZATIONS((short)8, "authorizations"),
WAIT_FOR_WRITES((short)9, "waitForWrites"),
- ISOLATED((short)10, "isolated");
+ ISOLATED((short)10, "isolated"),
+ READAHEAD_THRESHOLD((short)12, "readaheadThreshold");
private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
@@ -2914,6 +2921,8 @@ import org.slf4j.LoggerFactory;
return WAIT_FOR_WRITES;
case 10: // ISOLATED
return ISOLATED;
+ case 12: // READAHEAD_THRESHOLD
+ return READAHEAD_THRESHOLD;
default:
return null;
}
@@ -2957,6 +2966,7 @@ import org.slf4j.LoggerFactory;
private static final int __BATCHSIZE_ISSET_ID = 0;
private static final int __WAITFORWRITES_ISSET_ID = 1;
private static final int __ISOLATED_ISSET_ID = 2;
+ private static final int __READAHEADTHRESHOLD_ISSET_ID = 3;
private byte __isset_bitfield = 0;
public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
static {
@@ -2990,6 +3000,8 @@ import org.slf4j.LoggerFactory;
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.BOOL)));
tmpMap.put(_Fields.ISOLATED, new org.apache.thrift.meta_data.FieldMetaData("isolated", org.apache.thrift.TFieldRequirementType.DEFAULT,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.BOOL)));
+ tmpMap.put(_Fields.READAHEAD_THRESHOLD, new org.apache.thrift.meta_data.FieldMetaData("readaheadThreshold", org.apache.thrift.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I64)));
metaDataMap = Collections.unmodifiableMap(tmpMap);
org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(startScan_args.class, metaDataMap);
}
@@ -3008,7 +3020,8 @@ import org.slf4j.LoggerFactory;
Map<String,Map<String,String>> ssio,
List<ByteBuffer> authorizations,
boolean waitForWrites,
- boolean isolated)
+ boolean isolated,
+ long readaheadThreshold)
{
this();
this.tinfo = tinfo;
@@ -3025,6 +3038,8 @@ import org.slf4j.LoggerFactory;
setWaitForWritesIsSet(true);
this.isolated = isolated;
setIsolatedIsSet(true);
+ this.readaheadThreshold = readaheadThreshold;
+ setReadaheadThresholdIsSet(true);
}
/**
@@ -3096,6 +3111,7 @@ import org.slf4j.LoggerFactory;
}
this.waitForWrites = other.waitForWrites;
this.isolated = other.isolated;
+ this.readaheadThreshold = other.readaheadThreshold;
}
public startScan_args deepCopy() {
@@ -3118,6 +3134,8 @@ import org.slf4j.LoggerFactory;
this.waitForWrites = false;
setIsolatedIsSet(false);
this.isolated = false;
+ setReadaheadThresholdIsSet(false);
+ this.readaheadThreshold = 0;
}
public org.apache.accumulo.trace.thrift.TInfo getTinfo() {
@@ -3437,6 +3455,29 @@ import org.slf4j.LoggerFactory;
__isset_bitfield = EncodingUtils.setBit(__isset_bitfield, __ISOLATED_ISSET_ID, value);
}
+ public long getReadaheadThreshold() {
+ return this.readaheadThreshold;
+ }
+
+ public startScan_args setReadaheadThreshold(long readaheadThreshold) {
+ this.readaheadThreshold = readaheadThreshold;
+ setReadaheadThresholdIsSet(true);
+ return this;
+ }
+
+ public void unsetReadaheadThreshold() {
+ __isset_bitfield = EncodingUtils.clearBit(__isset_bitfield, __READAHEADTHRESHOLD_ISSET_ID);
+ }
+
+ /** Returns true if field readaheadThreshold is set (has been assigned a value) and false otherwise */
+ public boolean isSetReadaheadThreshold() {
+ return EncodingUtils.testBit(__isset_bitfield, __READAHEADTHRESHOLD_ISSET_ID);
+ }
+
+ public void setReadaheadThresholdIsSet(boolean value) {
+ __isset_bitfield = EncodingUtils.setBit(__isset_bitfield, __READAHEADTHRESHOLD_ISSET_ID, value);
+ }
+
public void setFieldValue(_Fields field, Object value) {
switch (field) {
case TINFO:
@@ -3527,6 +3568,14 @@ import org.slf4j.LoggerFactory;
}
break;
+ case READAHEAD_THRESHOLD:
+ if (value == null) {
+ unsetReadaheadThreshold();
+ } else {
+ setReadaheadThreshold((Long)value);
+ }
+ break;
+
}
}
@@ -3565,6 +3614,9 @@ import org.slf4j.LoggerFactory;
case ISOLATED:
return Boolean.valueOf(isIsolated());
+ case READAHEAD_THRESHOLD:
+ return Long.valueOf(getReadaheadThreshold());
+
}
throw new IllegalStateException();
}
@@ -3598,6 +3650,8 @@ import org.slf4j.LoggerFactory;
return isSetWaitForWrites();
case ISOLATED:
return isSetIsolated();
+ case READAHEAD_THRESHOLD:
+ return isSetReadaheadThreshold();
}
throw new IllegalStateException();
}
@@ -3714,6 +3768,15 @@ import org.slf4j.LoggerFactory;
return false;
}
+ boolean this_present_readaheadThreshold = true;
+ boolean that_present_readaheadThreshold = true;
+ if (this_present_readaheadThreshold || that_present_readaheadThreshold) {
+ if (!(this_present_readaheadThreshold && that_present_readaheadThreshold))
+ return false;
+ if (this.readaheadThreshold != that.readaheadThreshold)
+ return false;
+ }
+
return true;
}
@@ -3840,6 +3903,16 @@ import org.slf4j.LoggerFactory;
return lastComparison;
}
}
+ lastComparison = Boolean.valueOf(isSetReadaheadThreshold()).compareTo(typedOther.isSetReadaheadThreshold());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetReadaheadThreshold()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.readaheadThreshold, typedOther.readaheadThreshold);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
return 0;
}
@@ -3935,6 +4008,10 @@ import org.slf4j.LoggerFactory;
sb.append("isolated:");
sb.append(this.isolated);
first = false;
+ if (!first) sb.append(", ");
+ sb.append("readaheadThreshold:");
+ sb.append(this.readaheadThreshold);
+ first = false;
sb.append(")");
return sb.toString();
}
@@ -4140,6 +4217,14 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
+ case 12: // READAHEAD_THRESHOLD
+ if (schemeField.type == org.apache.thrift.protocol.TType.I64) {
+ struct.readaheadThreshold = iprot.readI64();
+ struct.setReadaheadThresholdIsSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
default:
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
@@ -4241,6 +4326,9 @@ import org.slf4j.LoggerFactory;
struct.tinfo.write(oprot);
oprot.writeFieldEnd();
}
+ oprot.writeFieldBegin(READAHEAD_THRESHOLD_FIELD_DESC);
+ oprot.writeI64(struct.readaheadThreshold);
+ oprot.writeFieldEnd();
oprot.writeFieldStop();
oprot.writeStructEnd();
}
@@ -4292,7 +4380,10 @@ import org.slf4j.LoggerFactory;
if (struct.isSetIsolated()) {
optionals.set(10);
}
- oprot.writeBitSet(optionals, 11);
+ if (struct.isSetReadaheadThreshold()) {
+ optionals.set(11);
+ }
+ oprot.writeBitSet(optionals, 12);
if (struct.isSetTinfo()) {
struct.tinfo.write(oprot);
}
@@ -4358,12 +4449,15 @@ import org.slf4j.LoggerFactory;
if (struct.isSetIsolated()) {
oprot.writeBool(struct.isolated);
}
+ if (struct.isSetReadaheadThreshold()) {
+ oprot.writeI64(struct.readaheadThreshold);
+ }
}
@Override
public void read(org.apache.thrift.protocol.TProtocol prot, startScan_args struct) throws org.apache.thrift.TException {
TTupleProtocol iprot = (TTupleProtocol) prot;
- BitSet incoming = iprot.readBitSet(11);
+ BitSet incoming = iprot.readBitSet(12);
if (incoming.get(0)) {
struct.tinfo = new org.apache.accumulo.trace.thrift.TInfo();
struct.tinfo.read(iprot);
@@ -4463,6 +4557,10 @@ import org.slf4j.LoggerFactory;
struct.isolated = iprot.readBool();
struct.setIsolatedIsSet(true);
}
+ if (incoming.get(11)) {
+ struct.readaheadThreshold = iprot.readI64();
+ struct.setReadaheadThresholdIsSet(true);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff95c714/core/src/main/thrift/tabletserver.thrift
----------------------------------------------------------------------
diff --git a/core/src/main/thrift/tabletserver.thrift b/core/src/main/thrift/tabletserver.thrift
index 4f9f13a..25e0b10 100644
--- a/core/src/main/thrift/tabletserver.thrift
+++ b/core/src/main/thrift/tabletserver.thrift
@@ -139,7 +139,8 @@ service TabletClientService extends client.ClientService {
7:map<string, map<string, string>> ssio,
8:list<binary> authorizations
9:bool waitForWrites,
- 10:bool isolated) throws (1:client.ThriftSecurityException sec, 2:NotServingTabletException nste, 3:TooManyFilesException tmfe),
+ 10:bool isolated,
+ 12:i64 readaheadThreshold) throws (1:client.ThriftSecurityException sec, 2:NotServingTabletException nste, 3:TooManyFilesException tmfe),
data.ScanResult continueScan(2:trace.TInfo tinfo, 1:data.ScanID scanID) throws (1:NoSuchScanIDException nssi, 2:NotServingTabletException nste, 3:TooManyFilesException tmfe),
oneway void closeScan(2:trace.TInfo tinfo, 1:data.ScanID scanID),
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff95c714/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java b/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
index abb8750..56f03af 100644
--- a/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
+++ b/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
@@ -795,6 +795,7 @@ public class TabletServer extends AbstractMetricsImpl implements org.apache.accu
public volatile ScanTask<ScanBatch> nextBatchTask;
public AtomicBoolean interruptFlag;
public Scanner scanner;
+ public long readaheadThreshold = Constants.SCANNER_DEFAULT_READAHEAD_THRESHOLD;
@Override
public void cleanup() {
@@ -1156,9 +1157,9 @@ public class TabletServer extends AbstractMetricsImpl implements org.apache.accu
@Override
public InitialScan startScan(TInfo tinfo, TCredentials credentials, TKeyExtent textent, TRange range, List<TColumn> columns, int batchSize,
- List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated)
- throws NotServingTabletException, ThriftSecurityException, org.apache.accumulo.core.tabletserver.thrift.TooManyFilesException {
-
+ List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated,
+ long readaheadThreshold) throws NotServingTabletException, ThriftSecurityException, org.apache.accumulo.core.tabletserver.thrift.TooManyFilesException {
+
Authorizations userauths = null;
if (!security.canScan(credentials, new String(textent.getTable()), range, columns, ssiList, ssio, authorizations))
throw new ThriftSecurityException(credentials.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
@@ -1195,6 +1196,7 @@ public class TabletServer extends AbstractMetricsImpl implements org.apache.accu
scanSession.ssio = ssio;
scanSession.auths = new Authorizations(authorizations);
scanSession.interruptFlag = new AtomicBoolean();
+ scanSession.readaheadThreshold = readaheadThreshold;
for (TColumn tcolumn : columns) {
scanSession.columnSet.add(new Column(tcolumn));
@@ -1277,7 +1279,7 @@ public class TabletServer extends AbstractMetricsImpl implements org.apache.accu
scanSession.batchCount++;
- if (scanResult.more && scanSession.batchCount > 3) {
+ if (scanResult.more && scanSession.batchCount > scanSession.readaheadThreshold) {
// start reading next batch while current batch is transmitted
// to client
scanSession.nextBatchTask = new NextBatchTask(scanID, scanSession.interruptFlag);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff95c714/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java b/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
index 9bb7604..f4eb234 100644
--- a/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
+++ b/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
@@ -138,7 +138,7 @@ public class NullTserver {
@Override
public InitialScan startScan(TInfo tinfo, TCredentials credentials, TKeyExtent extent, TRange range, List<TColumn> columns, int batchSize,
- List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated) {
+ List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated, long readaheadThreshold) {
return null;
}