| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You 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.apache.geode.cache.client.internal; |
| |
| import static org.apache.geode.internal.cache.execute.AbstractExecution.DEFAULT_CLIENT_FUNCTION_TIMEOUT; |
| |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.function.Supplier; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.InternalGemFireError; |
| import org.apache.geode.cache.client.PoolFactory; |
| import org.apache.geode.cache.client.ServerConnectivityException; |
| import org.apache.geode.cache.client.ServerOperationException; |
| import org.apache.geode.cache.execute.Function; |
| import org.apache.geode.cache.execute.FunctionException; |
| import org.apache.geode.cache.execute.ResultCollector; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.internal.ServerLocation; |
| import org.apache.geode.internal.Version; |
| import org.apache.geode.internal.cache.execute.AbstractExecution; |
| import org.apache.geode.internal.cache.execute.FunctionStats; |
| import org.apache.geode.internal.cache.execute.InternalFunctionException; |
| import org.apache.geode.internal.cache.execute.InternalFunctionInvocationTargetException; |
| import org.apache.geode.internal.cache.execute.MemberMappedArgument; |
| import org.apache.geode.internal.cache.tier.MessageType; |
| import org.apache.geode.internal.cache.tier.sockets.ChunkedMessage; |
| import org.apache.geode.internal.cache.tier.sockets.Message; |
| import org.apache.geode.internal.cache.tier.sockets.Part; |
| import org.apache.geode.internal.logging.LogService; |
| |
| /** |
| * Executes the function on server (possibly without region/cache).<br> |
| * Also gets the result back from the server |
| * |
| * @since GemFire 5.8 |
| */ |
| |
| public class ExecuteFunctionOp { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| /** index of allMembers in flags[] */ |
| public static final int ALL_MEMBERS_INDEX = 0; |
| /** index of ignoreFailedMembers in flags[] */ |
| public static final int IGNORE_FAILED_MEMBERS_INDEX = 1; |
| |
| private ExecuteFunctionOp() { |
| // no instances allowed |
| } |
| |
| public static void execute(final PoolImpl pool, |
| boolean allServers, ResultCollector rc, |
| final boolean isHA, UserAttributes attributes, String[] groups, |
| final ExecuteFunctionOpImpl executeFunctionOp, |
| final Supplier<ExecuteFunctionOpImpl> executeFunctionOpSupplier, |
| final Supplier<ExecuteFunctionOpImpl> reExecuteFunctionOpSupplier) { |
| |
| final AbstractOp op = executeFunctionOp; |
| |
| if (allServers && groups.length == 0) { |
| |
| List callableTasks = constructAndGetFunctionTasks(pool, |
| attributes, executeFunctionOpSupplier); |
| |
| SingleHopClientExecutor.submitAll(callableTasks); |
| |
| } else { |
| |
| boolean reexecute = false; |
| |
| int maxRetryAttempts = pool.getRetryAttempts(); |
| if (!isHA) { |
| maxRetryAttempts = 0; |
| } |
| |
| do { |
| try { |
| if (reexecute) { |
| pool.execute(reExecuteFunctionOpSupplier.get(), 0); |
| } else { |
| pool.execute(op, 0); |
| } |
| reexecute = false; |
| } catch (InternalFunctionInvocationTargetException e) { |
| if (isHA) { |
| reexecute = true; |
| } |
| rc.clearResults(); |
| } catch (ServerOperationException serverOperationException) { |
| throw serverOperationException; |
| |
| } catch (ServerConnectivityException se) { |
| |
| if (maxRetryAttempts == PoolFactory.DEFAULT_RETRY_ATTEMPTS) { |
| // If the retryAttempt is set to default(-1). Try it on all servers once. |
| // Calculating number of servers when function is re-executed as it involves |
| // messaging locator. |
| maxRetryAttempts = pool.getConnectionSource().getAllServers().size() - 1; |
| } |
| |
| if ((maxRetryAttempts--) < 1) { |
| throw se; |
| } |
| |
| reexecute = true; |
| rc.clearResults(); |
| } |
| } while (reexecute); |
| } |
| } |
| |
| private static List constructAndGetFunctionTasks(final PoolImpl pool, |
| UserAttributes attributes, |
| final Supplier<ExecuteFunctionOpImpl> executeFunctionOpSupplier) { |
| final List<SingleHopOperationCallable> tasks = new ArrayList<>(); |
| List<ServerLocation> servers = pool.getConnectionSource().getAllServers(); |
| for (ServerLocation server : servers) { |
| final AbstractOp op = executeFunctionOpSupplier.get(); |
| SingleHopOperationCallable task = |
| new SingleHopOperationCallable(server, pool, op, attributes); |
| tasks.add(task); |
| } |
| return tasks; |
| } |
| |
| static byte[] getByteArrayForFlags(boolean... flags) { |
| byte[] retVal = null; |
| if (flags.length > 0) { |
| retVal = new byte[flags.length]; |
| for (int i = 0; i < flags.length; i++) { |
| if (flags[i]) { |
| retVal[i] = 1; |
| } else { |
| retVal[i] = 0; |
| } |
| } |
| } |
| return retVal; |
| } |
| |
| public static class ExecuteFunctionOpImpl extends AbstractOpWithTimeout { |
| |
| private ResultCollector resultCollector; |
| |
| // To get the instance of the Function Statistics we need the function name or instance |
| private String functionId; |
| |
| private Function function; |
| |
| private Object args; |
| |
| private MemberMappedArgument memberMappedArg; |
| |
| private byte hasResult; |
| |
| private boolean isFnSerializationReqd; |
| |
| private String[] groups; |
| |
| /** |
| * [0] = allMembers [1] = ignoreFailedMembers |
| */ |
| private byte[] flags; |
| |
| /** |
| * number of parts in the request message |
| */ |
| private static final int MSG_PARTS = 6; |
| |
| /** |
| * @throws org.apache.geode.SerializationException if serialization fails |
| */ |
| public ExecuteFunctionOpImpl(Function function, Object args, |
| MemberMappedArgument memberMappedArg, |
| ResultCollector rc, boolean isFnSerializationReqd, |
| byte isReexecute, |
| String[] groups, boolean allMembers, boolean ignoreFailedMembers, |
| final int timeoutMs) { |
| super(MessageType.EXECUTE_FUNCTION, MSG_PARTS, timeoutMs); |
| byte fnState = AbstractExecution.getFunctionState(function.isHA(), function.hasResult(), |
| function.optimizeForWrite()); |
| |
| addBytes(isReexecute, fnState); |
| if (isFnSerializationReqd) { |
| getMessage().addStringOrObjPart(function); |
| } else { |
| getMessage().addStringOrObjPart(function.getId()); |
| } |
| getMessage().addObjPart(args); |
| getMessage().addObjPart(memberMappedArg); |
| getMessage().addObjPart(groups); |
| flags = getByteArrayForFlags(allMembers, ignoreFailedMembers); |
| getMessage().addBytesPart(flags); |
| resultCollector = rc; |
| if (isReexecute == 1) { |
| resultCollector.clearResults(); |
| } |
| functionId = function.getId(); |
| this.function = function; |
| this.args = args; |
| this.memberMappedArg = memberMappedArg; |
| this.hasResult = fnState; |
| this.isFnSerializationReqd = isFnSerializationReqd; |
| this.groups = groups; |
| } |
| |
| public ExecuteFunctionOpImpl(String functionId, Object args2, |
| MemberMappedArgument memberMappedArg, byte hasResult, ResultCollector rc, |
| boolean isFnSerializationReqd, boolean isHA, boolean optimizeForWrite, byte isReexecute, |
| String[] groups, boolean allMembers, boolean ignoreFailedMembers, final int timeoutMs) { |
| super(MessageType.EXECUTE_FUNCTION, MSG_PARTS, timeoutMs); |
| byte fnState = AbstractExecution.getFunctionState(isHA, hasResult == (byte) 1, |
| optimizeForWrite); |
| |
| addBytes(isReexecute, fnState); |
| getMessage().addStringOrObjPart(functionId); |
| getMessage().addObjPart(args2); |
| getMessage().addObjPart(memberMappedArg); |
| getMessage().addObjPart(groups); |
| flags = getByteArrayForFlags(allMembers, ignoreFailedMembers); |
| getMessage().addBytesPart(flags); |
| resultCollector = rc; |
| if (isReexecute == 1) { |
| resultCollector.clearResults(); |
| } |
| this.functionId = functionId; |
| args = args2; |
| this.memberMappedArg = memberMappedArg; |
| this.hasResult = fnState; |
| this.isFnSerializationReqd = isFnSerializationReqd; |
| this.groups = groups; |
| } |
| |
| ExecuteFunctionOpImpl(ExecuteFunctionOpImpl op, byte isReexecute) { |
| super(MessageType.EXECUTE_FUNCTION, MSG_PARTS, op.getTimeoutMs()); |
| resultCollector = op.resultCollector; |
| function = op.function; |
| functionId = op.functionId; |
| hasResult = op.hasResult; |
| args = op.args; |
| memberMappedArg = op.memberMappedArg; |
| isFnSerializationReqd = op.isFnSerializationReqd; |
| groups = op.groups; |
| flags = op.flags; |
| |
| addBytes(isReexecute, hasResult); |
| if (isFnSerializationReqd) { |
| getMessage().addStringOrObjPart(function); |
| } else { |
| getMessage().addStringOrObjPart(function.getId()); |
| } |
| getMessage().addObjPart(args); |
| getMessage().addObjPart(memberMappedArg); |
| getMessage().addObjPart(groups); |
| getMessage().addBytesPart(flags); |
| if (isReexecute == 1) { |
| resultCollector.clearResults(); |
| } |
| |
| } |
| |
| private void addBytes(byte isReexecute, byte fnStateOrHasResult) { |
| if (getTimeoutMs() == DEFAULT_CLIENT_FUNCTION_TIMEOUT) { |
| if (isReexecute == 1) { |
| getMessage().addBytesPart( |
| new byte[] {AbstractExecution.getReexecuteFunctionState(fnStateOrHasResult)}); |
| } else { |
| getMessage().addBytesPart(new byte[] {fnStateOrHasResult}); |
| } |
| } else { |
| byte[] bytes = new byte[5]; |
| if (isReexecute == 1) { |
| bytes[0] = AbstractExecution.getReexecuteFunctionState(fnStateOrHasResult); |
| } else { |
| bytes[0] = fnStateOrHasResult; |
| } |
| Part.encodeInt(getTimeoutMs(), bytes, 1); |
| getMessage().addBytesPart(bytes); |
| } |
| } |
| |
| /** |
| * ignoreFaileMember flag is at index 1 |
| */ |
| private boolean getIgnoreFailedMembers() { |
| boolean ignoreFailedMembers = false; |
| if (flags != null && flags.length > 1) { |
| if (flags[IGNORE_FAILED_MEMBERS_INDEX] == 1) { |
| ignoreFailedMembers = true; |
| } |
| } |
| return ignoreFailedMembers; |
| } |
| |
| @Override |
| protected Object processResponse(Message msg) throws Exception { |
| ChunkedMessage executeFunctionResponseMsg = (ChunkedMessage) msg; |
| try { |
| // Read the header which describes the type of message following |
| executeFunctionResponseMsg.readHeader(); |
| switch (executeFunctionResponseMsg.getMessageType()) { |
| case MessageType.EXECUTE_FUNCTION_RESULT: |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "ExecuteFunctionOpImpl#processResponse: received message of type EXECUTE_FUNCTION_RESULT."); |
| } |
| Exception exception = null; |
| // Read the chunk |
| do { |
| executeFunctionResponseMsg.receiveChunk(); |
| Object resultResponse = executeFunctionResponseMsg.getPart(0).getObject(); |
| Object result; |
| if (resultResponse instanceof ArrayList) { |
| result = ((ArrayList) resultResponse).get(0); |
| } else { |
| result = resultResponse; |
| } |
| if (result instanceof FunctionException) { |
| FunctionException ex = ((FunctionException) result); |
| if (ex instanceof InternalFunctionException || getIgnoreFailedMembers()) { |
| Throwable cause = ex.getCause() == null ? ex : ex.getCause(); |
| DistributedMember memberID = |
| (DistributedMember) ((ArrayList) resultResponse).get(1); |
| resultCollector.addResult(memberID, cause); |
| FunctionStats.getFunctionStats(functionId).incResultsReceived(); |
| } else { |
| exception = ex; |
| } |
| } else if (result instanceof Throwable) { |
| String s = "While performing a remote " + getOpName(); |
| exception = new ServerOperationException(s, (Throwable) result); |
| // Get the exception toString part. |
| // This was added for c++ thin client and not used in java |
| } else { |
| DistributedMember memberID = |
| (DistributedMember) ((ArrayList) resultResponse).get(1); |
| resultCollector.addResult(memberID, result); |
| FunctionStats.getFunctionStats(functionId).incResultsReceived(); |
| } |
| } while (!executeFunctionResponseMsg.isLastChunk()); |
| |
| if (exception != null) { |
| throw exception; |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "ExecuteFunctionOpImpl#processResponse: received all the results from server successfully."); |
| } |
| return null; |
| case MessageType.EXCEPTION: |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "ExecuteFunctionOpImpl#processResponse: received message of type EXCEPTION"); |
| } |
| // Read the chunk |
| executeFunctionResponseMsg.receiveChunk(); |
| Part part0 = executeFunctionResponseMsg.getPart(0); |
| Object obj = part0.getObject(); |
| if (obj instanceof FunctionException) { |
| throw ((FunctionException) obj); |
| } else { |
| final Throwable t = (Throwable) obj; |
| throw new ServerOperationException( |
| ": While performing a remote execute Function" + t.getMessage(), t); |
| } |
| case MessageType.EXECUTE_FUNCTION_ERROR: |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "ExecuteFunctionOpImpl#processResponse: received message of type EXECUTE_FUNCTION_ERROR"); |
| } |
| // Read the chunk |
| executeFunctionResponseMsg.receiveChunk(); |
| String errorMessage = executeFunctionResponseMsg.getPart(0).getString(); |
| throw new ServerOperationException(errorMessage); |
| default: |
| throw new InternalGemFireError(String.format("Unknown message type %s", |
| executeFunctionResponseMsg.getMessageType())); |
| } |
| } finally { |
| executeFunctionResponseMsg.clear(); |
| } |
| } |
| |
| @Override |
| protected boolean isErrorResponse(int msgType) { |
| return msgType == MessageType.EXECUTE_FUNCTION_ERROR; |
| } |
| |
| @Override |
| protected long startAttempt(ConnectionStats stats) { |
| return stats.startExecuteFunction(); |
| } |
| |
| protected String getOpName() { |
| return "executeFunction"; |
| } |
| |
| @Override |
| protected void endSendAttempt(ConnectionStats stats, long start) { |
| stats.endExecuteFunctionSend(start, hasFailed()); |
| } |
| |
| @Override |
| protected void endAttempt(ConnectionStats stats, long start) { |
| stats.endExecuteFunction(start, hasTimedOut(), hasFailed()); |
| } |
| |
| @Override |
| protected Message createResponseMessage() { |
| return new ChunkedMessage(1, Version.CURRENT); |
| } |
| } |
| } |
