From 36a2a3ddbc5580b4745f1ea84009533cce982793 Mon Sep 17 00:00:00 2001 From: David Turanski Date: Tue, 22 Jan 2013 20:11:45 -0500 Subject: [PATCH] SGF-149 implement chunking in PojoFunctionWrapper --- .../function/BatchingResultSender.java | 171 ++++++++++++++++++ .../gemfire/function/PojoFunctionWrapper.java | 57 +++--- .../function/config/GemfireFunction.java | 9 + .../function/config/GemfireFunctionUtils.java | 19 +- .../execution/AbstractFunctionExecution.java | 4 +- .../GemfireFunctionProxyFactoryBean.java | 8 +- .../function/BatchingResultSenderTest.java | 139 ++++++++++++++ .../execution/FunctionIntegrationTests.java | 15 +- .../GemfireFunctionProxyFactoryBeanTests.java | 18 +- 9 files changed, 390 insertions(+), 50 deletions(-) create mode 100644 src/main/java/org/springframework/data/gemfire/function/BatchingResultSender.java create mode 100644 src/test/java/org/springframework/data/gemfire/function/BatchingResultSenderTest.java diff --git a/src/main/java/org/springframework/data/gemfire/function/BatchingResultSender.java b/src/main/java/org/springframework/data/gemfire/function/BatchingResultSender.java new file mode 100644 index 00000000..b0ef73bf --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/function/BatchingResultSender.java @@ -0,0 +1,171 @@ +/* + * Copyright 2002-2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package org.springframework.data.gemfire.function; + +import java.lang.reflect.Array; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; + +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +import com.gemstone.gemfire.cache.execute.ResultSender; + +/** + * Sends collection results using a {@link ResultSender} in chunks determined by batchSize + * + * @author David Turanski + * + */ +public class BatchingResultSender { + private final int batchSize; + private ResultSender resultSender; + + public BatchingResultSender(int batchSize, ResultSender resultSender) { + Assert.notNull(resultSender, "resultSender cannot be null"); + Assert.isTrue(batchSize >= 0, "batchSize must be >= 0"); + this.batchSize = batchSize; + this.resultSender = resultSender; + } + + + public void sendResults(Iterable result) { + if (batchSize == 0) { + resultSender.lastResult(result); + return; + } + + List chunk = new ArrayList(batchSize); + + for (Iterator it = result.iterator(); it.hasNext();) { + if (chunk.size() < batchSize) { + chunk.add(it.next()); + } + + if (chunk.size() == batchSize || !it.hasNext()) { + if (it.hasNext()) { + resultSender.sendResult(chunk); + } else { + resultSender.lastResult(chunk); + } + chunk.clear(); + } + } + } + + + public void sendArrayResults(Object result) { + + if (batchSize == 0) { + resultSender.lastResult(result); + return; + } + + Assert.isTrue(ObjectUtils.isArray(result)); + + int length = Array.getLength(result); + + for (int from =0; from < length; from += batchSize) { + int to = Math.min(length,from + batchSize); + Object chunk = copyOfRange(result,from, to); + + if (to == length -1) { + resultSender.lastResult(chunk); + } else { + resultSender.sendResult(chunk); + } + } + } + + + /** + * @param result + * @param from + * @param to + * @return + */ + private Object copyOfRange(Object result, int from, int to) { + + Class arrayClass = result.getClass(); + int size = to - from; + + if (int[].class.isAssignableFrom(arrayClass)) { + int[] array = new int[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getInt(result, from + i); + } + return array; + } + + if (float[].class.isAssignableFrom(arrayClass)) { + float[] array = new float[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getFloat(result, from + i); + } + return array; + } + + if (double[].class.isAssignableFrom(arrayClass)) { + double[] array = new double[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getDouble(result, from + i); + } + return array; + } + + if (boolean[].class.isAssignableFrom(arrayClass)) { + boolean[] array = new boolean[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getBoolean(result, from + i); + } + return array; + } + + if (byte[].class.isAssignableFrom(arrayClass)) { + byte[] array = new byte[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getByte(result, from + i); + } + return array; + } + + if (short[].class.isAssignableFrom(arrayClass)) { + short[] array = new short[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getShort(result, from + i); + } + return array; + } + + if (long[].class.isAssignableFrom(arrayClass)) { + long[] array = new long[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getLong(result, from + i); + } + return array; + } + + if (char[].class.isAssignableFrom(arrayClass)) { + char[] array = new char[size]; + for(int i = 0; i < size ; ++i){ + array[i] = Array.getChar(result, from + i); + } + return array; + } + + return Arrays.copyOfRange((Object[])result, from, to); + + } +} diff --git a/src/main/java/org/springframework/data/gemfire/function/PojoFunctionWrapper.java b/src/main/java/org/springframework/data/gemfire/function/PojoFunctionWrapper.java index 31307c2b..a4e37b50 100644 --- a/src/main/java/org/springframework/data/gemfire/function/PojoFunctionWrapper.java +++ b/src/main/java/org/springframework/data/gemfire/function/PojoFunctionWrapper.java @@ -12,9 +12,11 @@ */ package org.springframework.data.gemfire.function; - import java.lang.reflect.Method; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; +import java.util.Iterator; import java.util.List; import org.apache.commons.logging.Log; @@ -22,7 +24,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; - import com.gemstone.gemfire.cache.execute.Function; import com.gemstone.gemfire.cache.execute.FunctionContext; import com.gemstone.gemfire.cache.execute.ResultSender; @@ -37,7 +38,7 @@ import com.gemstone.gemfire.cache.execute.ResultSender; * @author David Turanski * */ - + @SuppressWarnings("serial") public class PojoFunctionWrapper implements Function { @@ -45,25 +46,26 @@ public class PojoFunctionWrapper implements Function { private volatile boolean HA; private volatile boolean optimizeForWrite; - private final boolean hasResult; + private volatile boolean hasResult; private final Object target; private final Method method; private final String id; + private volatile int batchSize; private final FunctionArgumentResolver functionArgumentResolver; public PojoFunctionWrapper(Object target, Method method, String id) { - + this.functionArgumentResolver = new FunctionContextInjectingArgumentResolver(method); - + this.id = StringUtils.hasText(id) ? id : method.getName(); this.target = target; this.method = method; this.HA = false; - + this.hasResult = !(method.getReturnType().equals(void.class)); - + this.optimizeForWrite = false; } @@ -94,23 +96,29 @@ public class PojoFunctionWrapper implements Function { public void setOptimizeForWrite(boolean optimizeForWrite) { this.optimizeForWrite = optimizeForWrite; } - + + public void setBatchSize(int batchSize) { + this.batchSize = batchSize; + } + + public void setHasResult(boolean hasResult) { + this.hasResult = hasResult; + } + //@Override public void execute(FunctionContext functionContext) { - + Object[] args = this.functionArgumentResolver.resolveFunctionArguments(functionContext); - Object result = null; result = invokeTargetMethod(args); - + if (hasResult()) { sendResults(functionContext.getResultSender(), result); } } - protected final Object invokeTargetMethod(Object[] args) { if (logger.isDebugEnabled()) { @@ -126,31 +134,16 @@ public class PojoFunctionWrapper implements Function { return (Object) ReflectionUtils.invokeMethod(method, target, (Object[]) args); } - @SuppressWarnings("unchecked") private void sendResults(ResultSender resultSender, Object result) { if (result == null) { resultSender.lastResult(null); return; } - - List results = null; - - if (ObjectUtils.isArray(result)) { - results = Arrays.asList((Object[]) result); - - } else if (List.class.isAssignableFrom(result.getClass())) { - results = (List) result; - } - if (results != null) { - int i = 0; - for (Object item : results) { - if (i++ < results.size() - 1) { - resultSender.sendResult(item); - } else { - resultSender.lastResult(item); - } - } + if (ObjectUtils.isArray(result)) { + new BatchingResultSender(batchSize, resultSender).sendArrayResults(result); + } else if (Iterable.class.isAssignableFrom(result.getClass())) { + new BatchingResultSender(batchSize, resultSender).sendResults((Iterable) result); } else { resultSender.lastResult(result); } diff --git a/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunction.java b/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunction.java index 984beddf..d0a25c03 100644 --- a/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunction.java +++ b/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunction.java @@ -40,4 +40,13 @@ public @interface GemfireFunction { * is the function optimized for write operations */ boolean optimizeForWrite() default false; + /** + * controls the maximum number of results sent at one time + */ + int batchSize() default 0; + /** + * normally follows the method return type, i.e., false if void, true otherwise. This allows overriding + * a void method which uses the resultSender directly. + */ + boolean hasResult() default false; } diff --git a/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunctionUtils.java b/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunctionUtils.java index 9fa3e73e..3f4a8a4c 100644 --- a/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunctionUtils.java +++ b/src/main/java/org/springframework/data/gemfire/function/config/GemfireFunctionUtils.java @@ -51,9 +51,25 @@ public abstract class GemfireFunctionUtils { if (attributes.containsKey("HA")) { function.setHA((Boolean) attributes.get("HA")); } + if (attributes.containsKey("optimizeForWrite")) { function.setOptimizeForWrite((Boolean) attributes.get("optimizeForWrite")); } + + if (attributes.containsKey("batchSize")) { + int batchSize = (Integer) attributes.get("batchSize"); + Assert.isTrue(batchSize >= 0, String.format("batchSize must be a non-negative value %s.%s", + target.getClass().getName(), method.getName())); + function.setBatchSize(batchSize); + } + + if (attributes.containsKey("hasResult")) { + boolean hasResult = (Boolean)attributes.get("hasResult"); + //Only set if true + if (hasResult) { + function.setHasResult(hasResult); + } + } if (FunctionService.isRegistered(function.getId())) { if (overwrite) { @@ -63,6 +79,7 @@ public abstract class GemfireFunctionUtils { FunctionService.unregisterFunction(function.getId()); } } + if (!FunctionService.isRegistered(function.getId())) { FunctionService.registerFunction(function); if (log.isDebugEnabled()) { @@ -70,7 +87,7 @@ public abstract class GemfireFunctionUtils { } } else { if (log.isDebugEnabled()) { - log.debug("function already registered " + function.getId()); + log.debug("function " + function.getId()+ "is already registered"); } } } diff --git a/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java b/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java index 42362277..53e93aec 100644 --- a/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java +++ b/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java @@ -100,7 +100,9 @@ abstract class AbstractFunctionExecution { resultCollector = (ResultCollector) execution.execute(function); } - logger.debug("ResultsCollector:" + resultCollector.getClass().getName()); + if (logger.isDebugEnabled()) { + logger.debug("using ResultsCollector:" + resultCollector.getClass().getName()); + } Iterable results = null; diff --git a/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java b/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java index 3a5fa824..a4e0786e 100644 --- a/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java @@ -68,9 +68,9 @@ public class GemfireFunctionProxyFactoryBean implements FactoryBean, Met - protected Iterable invokeFunction(Method method, Object[] args) { + protected Object invokeFunction(Method method, Object[] args) { MethodMetadata mmd = this.methodMetadata.getMethodMetadata(method); - return this.gemfireFunctionOperations.execute(mmd.getFunctionId(), args); + return this.gemfireFunctionOperations.executeAndExtract(mmd.getFunctionId(), args); } @Override @@ -89,9 +89,7 @@ public class GemfireFunctionProxyFactoryBean implements FactoryBean, Met logger.debug("invoking method " + invocation.getMethod().getName()); } - Iterable results = invokeFunction(invocation.getMethod(), invocation.getArguments()); - - return extractResult(results, invocation.getMethod().getReturnType()); + return invokeFunction(invocation.getMethod(), invocation.getArguments()); } diff --git a/src/test/java/org/springframework/data/gemfire/function/BatchingResultSenderTest.java b/src/test/java/org/springframework/data/gemfire/function/BatchingResultSenderTest.java new file mode 100644 index 00000000..0d26d479 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/function/BatchingResultSenderTest.java @@ -0,0 +1,139 @@ +/* + * Copyright 2002-2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package org.springframework.data.gemfire.function; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.junit.Test; + +import com.gemstone.gemfire.cache.execute.ResultSender; + +/** + * @author David Turanski + * + */ +public class BatchingResultSenderTest { + + @Test + public void testArrayChunking() { + + testBatchingResultSender(new TestArrayResultSender(),1); + testBatchingResultSender(new TestArrayResultSender(),0); + testBatchingResultSender(new TestArrayResultSender(),9); + testBatchingResultSender(new TestArrayResultSender(),10); + testBatchingResultSender(new TestArrayResultSender(),1000); + } + + + @Test + public void testListChunking() { + testBatchingResultSender(new TestListResultSender(),1); + testBatchingResultSender(new TestListResultSender(),0); + testBatchingResultSender(new TestListResultSender(),9); + testBatchingResultSender(new TestListResultSender(),10); + testBatchingResultSender(new TestListResultSender(),1000); + } + + private void testBatchingResultSender(AbstractTestResultSender resultSender, int batchSize){ + BatchingResultSender brs = new BatchingResultSender(batchSize, resultSender); + + List result = new ArrayList(); + for (int i = 0; i< 100; i++) { + result.add(i); + } + //TODO: Clean this up. Ok for test code + if (resultSender instanceof TestArrayResultSender) { + brs.sendArrayResults(result.toArray(new Integer[100])); + } else { + brs.sendResults(result); + } + + assertEquals(100,resultSender.getResults().size()); + + for(int i=0; i< 100; i++) { + assertEquals(i,resultSender.getResults().get(i)); + } + + } + + public static abstract class AbstractTestResultSender implements ResultSender { + private List results = new ArrayList(); + + /* (non-Javadoc) + * @see com.gemstone.gemfire.cache.execute.ResultSender#lastResult(java.lang.Object) + */ + @Override + public void lastResult(Object arg0) { + if (arg0 == null) { + return; + } + addResults(arg0, results); + } + + /* (non-Javadoc) + * @see com.gemstone.gemfire.cache.execute.ResultSender#sendException(java.lang.Throwable) + */ + @Override + public void sendException(Throwable arg0) { + fail(); + + } + + /* (non-Javadoc) + * @see com.gemstone.gemfire.cache.execute.ResultSender#sendResult(java.lang.Object) + */ + @Override + public void sendResult(Object arg0) { + if (arg0 == null) { + return; + } + addResults(arg0, results); + } + + protected abstract void addResults(Object item, List results); + + public List getResults() { + return this.results; + } + + + } + + public static class TestArrayResultSender extends AbstractTestResultSender { + + protected void addResults(Object arg0, List results) { + assertTrue(arg0.getClass().isArray()); + Object[] array = (Object[]) arg0; + for (Object obj: array) { + results.add(obj); + } + } + } + + public static class TestListResultSender extends AbstractTestResultSender { + protected void addResults(Object arg0, List results) { + if (arg0 == null) { + return; + } + assertTrue(arg0 instanceof Collection); + Collection list = (Collection) arg0; + results.addAll(list); + } + } +} diff --git a/src/test/java/org/springframework/data/gemfire/function/execution/FunctionIntegrationTests.java b/src/test/java/org/springframework/data/gemfire/function/execution/FunctionIntegrationTests.java index 35766d83..9659a8be 100644 --- a/src/test/java/org/springframework/data/gemfire/function/execution/FunctionIntegrationTests.java +++ b/src/test/java/org/springframework/data/gemfire/function/execution/FunctionIntegrationTests.java @@ -102,7 +102,7 @@ public class FunctionIntegrationTests { assertEquals(2,map.get("two").intValue()); assertEquals(3,map.get("three").intValue()); - result = template.execute("collections",Arrays.asList(new Integer[]{1,2,3,4,5})); + result = template.executeAndExtract("collections",Arrays.asList(new Integer[]{1,2,3,4,5})); assertTrue(result.getClass().getName(),result instanceof List); List list = (List)result; @@ -111,7 +111,13 @@ public class FunctionIntegrationTests { assertEquals(i,list.get(i-1)); } } - + + @Test + public void testArrayReturnTypes() { + GemfireOnRegionOperations template = new GemfireOnRegionFunctionTemplate(region); + Object result = template.executeAndExtract("arrays",new int[]{1,2,3,4,5}); + assertTrue(result.getClass().getName(),result instanceof int[]); + } /* * This gets wrapped in a GemFire Function and registered on the forked server. */ @@ -143,5 +149,10 @@ public class FunctionIntegrationTests { } return new HashMap(dataSet); } + + @GemfireFunction(id="arrays",batchSize=2) + public int[] collections(int[] args) { + return args; + } } } diff --git a/src/test/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBeanTests.java b/src/test/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBeanTests.java index 887ebf57..577ac5a6 100644 --- a/src/test/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBeanTests.java +++ b/src/test/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBeanTests.java @@ -21,6 +21,7 @@ package org.springframework.data.gemfire.function.execution; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.lang.reflect.AccessibleObject; @@ -49,7 +50,6 @@ public class GemfireFunctionProxyFactoryBeanTests { functionOperations = mock(GemfireFunctionOperations.class); } - @SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testInvokeAndExtractWithAnnotatedFunctionId() throws Throwable { @@ -58,17 +58,17 @@ public class GemfireFunctionProxyFactoryBeanTests { MethodInvocation invocation = new TestInvocation(IFoo.class).withMethodNameAndArgTypes("oneArg",String.class); - List results = Arrays.asList(new Integer[]{1}); + int results = 1; - when(functionOperations.execute("oneArg",invocation.getArguments())).thenReturn(results); - + when(functionOperations.executeAndExtract("oneArg",invocation.getArguments())).thenReturn(results); Object result = proxy.invoke(invocation); - assertTrue(result instanceof Integer); - assertEquals(new Integer(1),result); + verify(functionOperations).executeAndExtract("oneArg",invocation.getArguments()); + assertTrue(result.getClass().getName(), result instanceof Integer); + assertEquals(1,result); } - @SuppressWarnings({ "unchecked", "rawtypes" }) + @SuppressWarnings({ "rawtypes" }) @Test public void testInvoke() throws Throwable { @@ -79,9 +79,9 @@ public class GemfireFunctionProxyFactoryBeanTests { List results = Arrays.asList(new Integer[]{1,2,3}); - when(functionOperations.execute("collections",invocation.getArguments())).thenReturn(results); - + when(functionOperations.executeAndExtract("collections",invocation.getArguments())).thenReturn(results); Object result = proxy.invoke(invocation); + verify(functionOperations).executeAndExtract("collections",invocation.getArguments()); ; assertTrue(result instanceof List); assertEquals(3,((List)result).size()); }