http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1cc3d528/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/IgniteStreamerImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/IgniteStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/IgniteStreamerImpl.java deleted file mode 100644 index 4610204..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/IgniteStreamerImpl.java +++ /dev/null @@ -1,1375 +0,0 @@ -/* - * 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.ignite.internal.processors.streamer; - -import org.apache.ignite.*; -import org.apache.ignite.events.*; -import org.apache.ignite.internal.*; -import org.apache.ignite.internal.cluster.*; -import org.apache.ignite.internal.managers.communication.*; -import org.apache.ignite.internal.managers.deployment.*; -import org.apache.ignite.internal.managers.eventstorage.*; -import org.apache.ignite.internal.util.*; -import org.apache.ignite.internal.util.future.*; -import org.apache.ignite.internal.util.lang.*; -import org.apache.ignite.internal.util.tostring.*; -import org.apache.ignite.internal.util.typedef.*; -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.internal.util.worker.*; -import org.apache.ignite.lang.*; -import org.apache.ignite.plugin.extensions.communication.*; -import org.apache.ignite.streamer.*; -import org.apache.ignite.streamer.router.*; -import org.apache.ignite.thread.*; -import org.jdk8.backport.*; -import org.jetbrains.annotations.*; - -import java.io.*; -import java.util.*; -import java.util.concurrent.*; - -import static org.apache.ignite.events.EventType.*; -import static org.apache.ignite.internal.GridTopic.*; - -/** - * - */ -public class IgniteStreamerImpl implements IgniteStreamerEx, Externalizable { - /** */ - private static final long serialVersionUID = 0L; - - /** Number of message send retries. */ - private static final int SEND_RETRY_COUNT = 3; - - /** Retry delay. */ - private static final int SEND_RETRY_DELAY = 1000; - - /** How many cancelled future IDs to keep in history. */ - private static final int CANCELLED_FUTS_HISTORY_SIZE = 4096; - - /** Log. */ - private IgniteLogger log; - - /** Context. */ - private GridKernalContext ctx; - - /** */ - private StreamerContext streamerCtx; - - /** Read-write lock. */ - private GridSpinReadWriteLock lock; - - /** Stopping flag. */ - private boolean stopping; - - /** Streamer configuration. */ - private StreamerConfiguration c; - - /** Name. */ - private String name; - - /** Stages. */ - @GridToStringInclude - private Map<String, StreamerStageWrapper> stages; - - /** Windows. */ - @GridToStringInclude - private Map<String, StreamerWindow> winMap; - - /** Default streamer window. */ - private StreamerWindow dfltWin; - - /** */ - private String firstStage; - - /** Router. */ - private StreamerEventRouter router; - - /** At least once. */ - private boolean atLeastOnce; - - /** Streamer metrics. */ - private volatile StreamerMetricsHolder streamerMetrics; - - /** Topic. */ - private Object topic; - - /** Stage execution futures. */ - private ConcurrentMap<IgniteUuid, GridStreamerStageExecutionFuture> stageFuts; - - /** Batch execution futures. */ - private ConcurrentMap<IgniteUuid, BatchExecutionFuture> batchFuts; - - /** Streamer executor service. */ - private ExecutorService execSvc; - - /** Failure listeners. */ - private Collection<StreamerFailureListener> failureLsnrs = new ConcurrentLinkedQueue<>(); - - /** Cancelled */ - private Collection<IgniteUuid> cancelledFutIds = - new GridBoundedConcurrentLinkedHashSet<>(CANCELLED_FUTS_HISTORY_SIZE); - - /** Load control semaphore. */ - private Semaphore sem; - - /** Deploy class. */ - private Class<?> depCls; - - /** Executor service capacity. */ - private int execSvcCap; - - /** Window lock. */ - private final GridSpinReadWriteLock winLock = new GridSpinReadWriteLock(); - - /** - * Empty constructor required by {@link Externalizable}. - */ - public IgniteStreamerImpl() { - // No-op. - } - - /** - * @param ctx Kernal context. - * @param c Configuration. - */ - public IgniteStreamerImpl(GridKernalContext ctx, StreamerConfiguration c) { - this.ctx = ctx; - - log = ctx.log(IgniteStreamerImpl.class); - - atLeastOnce = c.isAtLeastOnce(); - name = c.getName(); - router = c.getRouter(); - this.c = c; - - if (atLeastOnce) { - if (c.getMaximumConcurrentSessions() > 0) - sem = new Semaphore(c.getMaximumConcurrentSessions()); - } - - topic = name == null ? TOPIC_STREAM : TOPIC_STREAM.topic(name); - lock = new GridSpinReadWriteLock(); - stageFuts = new ConcurrentHashMap8<>(); - batchFuts = new ConcurrentHashMap8<>(); - - streamerCtx = new GridStreamerContextImpl(ctx, c, this); - } - - /** - * @throws IgniteCheckedException If failed. - */ - @SuppressWarnings("unchecked") - public void start() throws IgniteCheckedException { - if (log.isDebugEnabled()) - log.debug("Starting streamer: " + name); - - if (F.isEmpty(c.getStages())) - throw new IgniteCheckedException("Streamer should have at least one stage configured " + - "(fix configuration and restart): " + name); - - if (F.isEmpty(c.getWindows())) - throw new IgniteCheckedException("Streamer should have at least one window configured " + - "(fix configuration and restart): " + name); - - prepareResources(); - - U.startLifecycleAware(lifecycleAwares()); - - stages = U.newLinkedHashMap(c.getStages().size()); - - int stageIdx = 0; - - StreamerStageWrapper prev = null; - - for (StreamerStage s : c.getStages()) { - String sName = s.name(); - - if (F.isEmpty(sName)) - throw new IgniteCheckedException("Streamer stage should have non-empty name [streamerName=" + name + - ", stage=" + s + ']'); - - if (stages.containsKey(sName)) - throw new IgniteCheckedException("Streamer stages have duplicate names (all names should be unique) " + - "[streamerName=" + name + ", stage=" + s + ", stageName=" + sName + ']'); - - if (firstStage == null) - firstStage = sName; - - StreamerStageWrapper wrapper = new StreamerStageWrapper(s, stageIdx); - - stages.put(sName, wrapper); - - if (prev != null) - prev.nextStageName(s.name()); - - prev = wrapper; - - stageIdx++; - } - - winMap = new LinkedHashMap<>(); - - for (StreamerWindow w : c.getWindows()) { - String wName = w.name(); - - if (F.isEmpty(wName)) - throw new IgniteCheckedException("Streamer window should have non-empty name [streamerName=" + name + - ", window=" + w + ']'); - - if (winMap.containsKey(wName)) - throw new IgniteCheckedException("Streamer windows have duplicate names (all names should be unique). " + - "If you use two or more windows of the same type you need to assign their names explicitly " + - "[streamer=" + name + ", windowName=" + wName + ']'); - - winMap.put(wName, w); - - if (dfltWin == null) - dfltWin = w; - } - - execSvc = new IgniteThreadPoolExecutor( - ctx.gridName(), - c.getThreadPoolSize(), - c.getThreadPoolSize(), - 0, - new LinkedBlockingQueue<Runnable>()); - - execSvcCap = c.getThreadPoolSize(); - - resetMetrics(); - - if (router == null) - router = new StreamerLocalEventRouter(); - - ctx.io().addMessageListener(topic, new GridMessageListener() { - @Override public void onMessage(UUID nodeId, Object msg) { - if (log.isDebugEnabled()) - log.debug("Received message [nodeId=" + nodeId + ", msg=" + msg + ']'); - - processStreamerMessage(nodeId, msg); - } - }); - - ctx.event().addLocalEventListener(new GridLocalEventListener() { - @Override public void onEvent(Event evt) { - DiscoveryEvent discoEvt = (DiscoveryEvent)evt; - - for (GridStreamerStageExecutionFuture fut : stageFuts.values()) - fut.onNodeLeft(discoEvt.eventNode().id()); - } - }, EVT_NODE_LEFT, EVT_NODE_FAILED); - } - - /** - * Injects resources into streamer components. - * - * @throws IgniteCheckedException If failed. - */ - private void prepareResources() throws IgniteCheckedException { - for (StreamerStage s : c.getStages()) - ctx.resource().injectGeneric(s); - - if (router == null) - router = new StreamerLocalEventRouter(); - - ctx.resource().injectGeneric(router); - - for (StreamerWindow w : c.getWindows()) - ctx.resource().injectGeneric(w); - } - - /** - * On kernal stop callback. - * - * @param cancel Cancel. - */ - public void onKernalStop(boolean cancel) { - // No further requests will be processed neither locally nor remotely. - lock.writeLock(); - - try { - stopping = true; - } - finally { - lock.writeUnlock(); - } - - if (cancel) { - for (BatchExecutionFuture execFut : batchFuts.values()) { - try { - execFut.cancel(); - } - catch (IgniteCheckedException e) { - U.warn(log, "Failed to cancel batch execution future on node stop (will ignore) " + - "[execFut=" + execFut + ", err=" + e + ']'); - } - } - } - else { - // We need to wait for all locally scheduled stage futures to complete. - for (GridStreamerStageExecutionFuture fut : stageFuts.values()) { - try { - if (fut.rootExecution()) { - if (log.isDebugEnabled()) - log.debug("Waiting root execution future on kernal stop: " + fut); - - fut.get(); - } - } - catch (IgniteCheckedException ignore) { - // For failed futures callback will be executed, no need to care about this exception here. - } - } - - for (BatchExecutionFuture execFut : batchFuts.values()) { - try { - execFut.get(); - } - catch (IgniteCheckedException e) { - if (!e.hasCause(IgniteInterruptedCheckedException.class)) - U.warn(log, "Failed to wait for batch execution future completion (will ignore) " + - "[execFut=" + execFut + ", e=" + e + ']'); - } - } - } - - for (StreamerStageWrapper stage : stages.values()) { - try { - ctx.resource().cleanupGeneric(stage.unwrap()); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to cleanup stage [stage=" + stage + ", streamer=" + this + ']', e); - } - } - } - - /** - * @param cancel Whether currently running tasks should be cancelled. - */ - public void stop(boolean cancel) { - ctx.io().removeMessageListener(topic); - - // There is no point to wait for tasks execution since it was already handled by flags. - execSvc.shutdownNow(); - - U.stopLifecycleAware(log, lifecycleAwares()); - } - - /** - * @return Streamer components which can implement {@link org.apache.ignite.lifecycle.LifecycleAware} interface. - */ - private Iterable<Object> lifecycleAwares() { - Collection<Object> objs = new ArrayList<>(); - - objs.addAll(configuration().getStages()); - objs.addAll(configuration().getWindows()); - objs.add(router); - - return objs; - } - - /** {@inheritDoc} */ - @Nullable @Override public String name() { - return name; - } - - /** {@inheritDoc} */ - @Override public StreamerConfiguration configuration() { - return c; - } - - /** {@inheritDoc} */ - @Override public void addEvent(Object evt, Object... evts) { - A.notNull(evt, "evt"); - - if (!F.isEmpty(evts)) - addEvents(F.concat(false, evt, Arrays.asList(evts))); - else - addEvents(Collections.singleton(evt)); - } - - /** {@inheritDoc} */ - @Override public void addEventToStage(String stageName, Object evt, Object... evts) { - A.notNull(stageName, "stageName"); - A.notNull(evt, "evt"); - - if (!F.isEmpty(evts)) - addEventsToStage(stageName, F.concat(false, evt, Arrays.asList(evts))); - else - addEventsToStage(stageName, Collections.singleton(evt)); - } - - /** {@inheritDoc} */ - @Override public void addEvents(Collection<?> evts) { - A.ensure(!F.isEmpty(evts), "evts cannot be null or empty"); - - addEventsToStage(firstStage, evts); - } - - /** {@inheritDoc} */ - @Override public void addEventsToStage(String stageName, Collection<?> evts) { - A.notNull(stageName, "stageName"); - A.ensure(!F.isEmpty(evts), "evts cannot be empty or null"); - - ctx.gateway().readLock(); - - try { - addEvents0(null, 0, U.currentTimeMillis(), null, Collections.singleton(ctx.localNodeId()), stageName, evts); - } - catch (IgniteCheckedException e) { - throw U.convertException(e); - } - finally { - ctx.gateway().readUnlock(); - } - } - - /** {@inheritDoc} */ - @Override public StreamerContext context() { - return streamerCtx; - } - - /** {@inheritDoc} */ - @Override public void addStreamerFailureListener(StreamerFailureListener lsnr) { - failureLsnrs.add(lsnr); - } - - /** {@inheritDoc} */ - @Override public void removeStreamerFailureListener(StreamerFailureListener lsnr) { - failureLsnrs.remove(lsnr); - } - - /** {@inheritDoc} */ - @Override public StreamerMetrics metrics() { - StreamerMetrics ret = new StreamerMetricsAdapter(streamerMetrics); - - streamerMetrics.sampleCurrentStages(); - - return ret; - } - - /** {@inheritDoc} */ - @Override public void reset() { - winLock.writeLock(); - - try { - for (StreamerWindow win : winMap.values()) - win.reset(); - - streamerCtx.localSpace().clear(); - } - finally { - winLock.writeUnlock(); - } - } - - /** {@inheritDoc} */ - @Override public void resetMetrics() { - StreamerStageMetricsHolder[] stageHolders = new StreamerStageMetricsHolder[c.getStages().size()]; - - int idx = 0; - - for (StreamerStage stage : c.getStages()) { - stageHolders[idx] = new StreamerStageMetricsHolder(stage.name()); - - idx++; - } - - StreamerWindowMetricsHolder[] windowHolders = new StreamerWindowMetricsHolder[c.getWindows().size()]; - - idx = 0; - - for (StreamerWindow w : c.getWindows()) { - windowHolders[idx] = new StreamerWindowMetricsHolder(w); - - idx++; - } - - streamerMetrics = new StreamerMetricsHolder(stageHolders, windowHolders, execSvcCap); - } - - /** {@inheritDoc} */ - @Override public void deployClass(Class<?> depCls) { - this.depCls = depCls; - } - - /** {@inheritDoc} */ - @Override public <E> StreamerWindow<E> window() { - return (StreamerWindow<E>)dfltWin; - } - - /** {@inheritDoc} */ - @Override public <E> StreamerWindow<E> window(String windowName) { - return (StreamerWindow<E>)winMap.get(windowName); - } - - /** - * @return {@code atLeastOnce} configuration property. - */ - boolean atLeastOnce() { - return atLeastOnce; - } - - /** - * @return Stage future map size. - */ - int stageFutureMapSize() { - return stageFuts.size(); - } - - /** - * @return Batch future map size. - */ - int batchFutureMapSize() { - return batchFuts.size(); - } - - /** - * @param execId Execution ID, {@code null} if root execution. - * @param failoverAttempt Attempt count. - * @param execStartTs Execution start timestamp, ignored if root execution. - * @param parentFutId Parent future ID. - * @param execNodeIds Execution node IDs. - * @param stageName Stage name. - * @param evts Events. - * @return Future. - * @throws IgniteInterruptedCheckedException If failed. - */ - private GridStreamerStageExecutionFuture addEvents0( - @Nullable IgniteUuid execId, - int failoverAttempt, - long execStartTs, - @Nullable IgniteUuid parentFutId, - @Nullable Collection<UUID> execNodeIds, - String stageName, - Collection<?> evts - ) throws IgniteInterruptedCheckedException { - assert !F.isEmpty(evts); - assert !F.isEmpty(stageName); - - GridStreamerStageExecutionFuture fut = new GridStreamerStageExecutionFuture( - this, - execId, - failoverAttempt, - execStartTs, - parentFutId, - execNodeIds, - stageName, - evts); - - if (atLeastOnce && fut.rootExecution()) { - StreamerMetricsHolder metrics0 = streamerMetrics; - - metrics0.onSessionStarted(); - - fut.metrics(metrics0); - } - - // Acquire semaphore on first future submit. - if (atLeastOnce && fut.rootExecution() && fut.failoverAttemptCount() == 0) { - try { - if (sem != null) - sem.acquire(); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - - throw new IgniteInterruptedCheckedException(e); - } - } - - fut.map(); - - if (!atLeastOnce && fut.isFailed()) - notifyFailure(fut.stageName(), fut.events(), fut.error()); - - // Mind the gap. - for (UUID nodeId : fut.executionNodeIds()) { - if (!ctx.discovery().alive(nodeId)) - fut.onNodeLeft(nodeId); - } - - return fut; - } - - /** {@inheritDoc} */ - @Override public GridKernalContext kernalContext() { - return ctx; - } - - /** {@inheritDoc} */ - @Override public void onFutureMapped(GridStreamerStageExecutionFuture fut) { - if (atLeastOnce) { - GridStreamerStageExecutionFuture old = stageFuts.putIfAbsent(fut.id(), fut); - - assert old == null : "Streamer execution future should be mapped only once: " + old; - } - } - - /** {@inheritDoc} */ - @Override public void onFutureCompleted(GridStreamerStageExecutionFuture fut) { - if (atLeastOnce) { - if (fut.rootExecution() && !fut.isFailed() && sem != null) - sem.release(); - - GridStreamerStageExecutionFuture old = stageFuts.remove(fut.id()); - - assert fut == old : "Invalid future in map [fut=" + fut + ", old=" + old + ']'; - - if (fut.isFailed() || fut.isCancelled()) - cancelChildStages(fut); - - if (fut.rootExecution() && fut.isFailed()) - failover(fut); - } - } - - /** {@inheritDoc} */ - @Override public StreamerEventRouter eventRouter() { - return router; - } - - /** {@inheritDoc} */ - @Override public void scheduleExecutions(GridStreamerStageExecutionFuture fut, - Map<UUID, GridStreamerExecutionBatch> execs) throws IgniteCheckedException { - for (Map.Entry<UUID, GridStreamerExecutionBatch> entry : execs.entrySet()) { - UUID nodeId = entry.getKey(); - GridStreamerExecutionBatch batch = entry.getValue(); - - if (ctx.localNodeId().equals(nodeId)) - scheduleLocal(batch); - else { - if (log.isDebugEnabled()) - log.debug("Sending batch execution request to remote node [nodeId=" + nodeId + - ", futId=" + batch.futureId() + ", stageName=" + batch.stageName() + ']'); - - sendWithRetries(nodeId, createExecutionRequest(batch)); - - if (!ctx.discovery().alive(nodeId)) - fut.onNodeLeft(nodeId); - } - } - } - - /** {@inheritDoc} */ - @Override public void onUndeploy(ClassLoader undeployedLdr) { - if (log.isDebugEnabled()) - log.debug("Processing undeployment event undeployedLdr=" + undeployedLdr + ']'); - - unwindUndeploys(undeployedLdr, true); - } - - /** {@inheritDoc} */ - @Override public void onQueryCompleted(long time, int nodes) { - streamerMetrics.onQueryCompleted(time, nodes); - } - - /** - * Schedules batch execution locally. - * - * @param batch Batch to execute. - * @throws IgniteCheckedException If schedule was attempted on stopping grid. - */ - private void scheduleLocal(final GridStreamerExecutionBatch batch) throws IgniteCheckedException { - final IgniteUuid futId = batch.futureId(); - - lock.readLock(); - - try { - if (stopping) - throw new IgniteCheckedException("Failed to schedule local batch execution (grid is stopping): " + batch); - - if (log.isDebugEnabled()) - log.debug("Scheduling local batch execution [futId=" + futId + ", stageName=" + batch.stageName() + ']'); - - StreamerStageWrapper wrapper = stages.get(batch.stageName()); - - if (wrapper == null) { - completeParentStage(ctx.localNodeId(), batch.futureId(), - new IgniteCheckedException("Failed to process streamer batch (stage was not found): " + - batch.stageName() + ']')); - - return; - } - - // Capture metrics holders for batch execution. - StreamerMetricsHolder streamerMetrics0 = streamerMetrics; - - BatchWorker worker = new BatchWorker(batch, wrapper, streamerMetrics0); - - BatchExecutionFuture batchFut = worker.completionFuture(); - - BatchExecutionFuture old = batchFuts.putIfAbsent(futId, batchFut); - - assert old == null : "Duplicate batch execution future [old=" + old + ", batchFut=" + batchFut + ']'; - - if (cancelled(futId)) { - // Batch was cancelled before execution started, remove future and return. - batchFuts.remove(futId, batchFut); - - return; - } - - streamerMetrics0.onStageScheduled(); - - execSvc.submit(worker); - - batchFut.listen(new CI1<IgniteInternalFuture<Object>>() { - @Override public void apply(IgniteInternalFuture<Object> t) { - BatchExecutionFuture fut = (BatchExecutionFuture)t; - - if (log.isDebugEnabled()) - log.debug("Completed batch execution future: " + fut); - - batchFuts.remove(futId, fut); - - if (!fut.isCancelled() && atLeastOnce) - completeParentStage(ctx.localNodeId(), batch.futureId(), fut.error()); - } - }); - } - finally { - lock.readUnlock(); - } - } - - /** - * Completes parent execution future. - * - * @param completeNodeId Node ID completed batch. - * @param futId Future ID to complete. - * @param err Error, if any. - */ - private void completeParentStage(UUID completeNodeId, IgniteUuid futId, @Nullable Throwable err) { - lock.readLock(); - - try { - if (stopping && !atLeastOnce) { - if (log.isDebugEnabled()) - log.debug("Failed to notify parent stage completion (node is stopping) [futId=" + futId + - ", err=" + err + ']'); - - return; - } - - UUID dstNodeId = futId.globalId(); - - if (ctx.localNodeId().equals(dstNodeId)) { - GridStreamerStageExecutionFuture stageFut = stageFuts.get(futId); - - if (log.isDebugEnabled()) - log.debug("Notifying local execution future [completeNodeId=" + completeNodeId + - ", stageFut=" + stageFut + ", err=" + err + ']'); - - if (stageFut != null) - stageFut.onExecutionCompleted(completeNodeId, err); - } - else { - try { - if (log.isDebugEnabled()) - log.debug("Sending completion response to remote node [nodeId=" + dstNodeId + - ", futId=" + futId + ", err=" + err + ']'); - - byte[] errBytes = err != null ? ctx.config().getMarshaller().marshal(err) : null; - - sendWithRetries(dstNodeId, new GridStreamerResponse(futId, errBytes)); - } - catch (IgniteCheckedException e) { - if (!e.hasCause(ClusterTopologyCheckedException.class)) - log.error("Failed to complete parent stage [futId=" + futId + ", err=" + e + ']'); - } - } - } - finally { - lock.readUnlock(); - } - } - - /** - * Cancels child batches of execution future. - * - * @param fut Future to cancel child batches. - */ - private void cancelChildStages(GridStreamerStageExecutionFuture fut) { - for (UUID nodeId : fut.childExecutions().keySet()) - cancelChildStage(nodeId, fut.id()); - } - - /** - * @param nodeId Node ID to cancel future on. - * @param cancelledFutId Future ID to cancel. - */ - private void cancelChildStage(UUID nodeId, IgniteUuid cancelledFutId) { - assert atLeastOnce; - - if (nodeId.equals(ctx.localNodeId())) { - cancelledFutIds.add(cancelledFutId); - - Iterator<Map.Entry<IgniteUuid, BatchExecutionFuture>> it = batchFuts.entrySet().iterator(); - - while (it.hasNext()) { - Map.Entry<IgniteUuid, BatchExecutionFuture> entry = it.next(); - - if (entry.getKey().equals(cancelledFutId)) { - BatchExecutionFuture batchFut = entry.getValue(); - - try { - batchFut.cancel(); - } - catch (IgniteCheckedException e) { - log.warning("Failed to cancel batch execution future [cancelledFutId=" + cancelledFutId + - ", batchFut=" + batchFut + ']', e); - } - - it.remove(); - } - } - } - else { - try { - sendWithRetries(nodeId, new GridStreamerCancelRequest(cancelledFutId)); - } - catch (IgniteCheckedException e) { - if (!e.hasCause(ClusterTopologyCheckedException.class)) - log.error("Failed to send streamer cancel request to remote node [nodeId=" + nodeId + - ", cancelledFutId=" + cancelledFutId + ']', e); - } - } - } - - /** - * Attempts to failover pipeline execution. - * - * @param fut Future. - */ - private void failover(GridStreamerStageExecutionFuture fut) { - assert fut.rootExecution(); - assert fut.error() != null; - - if (fut.failoverAttemptCount() >= c.getMaximumFailoverAttempts() || stopping) { - // Release semaphore when no failover will be attempted anymore. - if (sem != null) - sem.release(); - - notifyFailure(fut.stageName(), fut.events(), fut.error()); - } - else { - try { - addEvents0(null, fut.failoverAttemptCount() + 1, 0, null, Collections.singleton(ctx.localNodeId()), - fut.stageName(), fut.events()); - } - catch (IgniteInterruptedCheckedException e) { - e.printStackTrace(); - - assert false : "Failover submit should never attempt to acquire semaphore: " + fut + ']'; - } - } - } - - /** - * Notifies failure listeners. - * - * @param stageName Stage name. - * @param evts Events. - * @param err Error cause. - */ - private void notifyFailure(String stageName, Collection<Object> evts, Throwable err) { - for (StreamerFailureListener lsnr : failureLsnrs) - lsnr.onFailure(stageName, evts, err); - } - - /** - * Checks if cancel request was received for this future ID. - * - * @param futId Future ID. - * @return {@code True} if future was cancelled, {@code false} otherwise. - */ - public boolean cancelled(IgniteUuid futId) { - return cancelledFutIds.contains(futId); - } - - /** - * Processes streamer message. - * - * @param sndNodeId Sender node ID. - * @param msg Message. - */ - private void processStreamerMessage(UUID sndNodeId, Object msg) { - if (msg instanceof GridStreamerExecutionRequest) { - GridStreamerExecutionRequest req = (GridStreamerExecutionRequest)msg; - - GridStreamerExecutionBatch batch; - - try { - batch = executionBatch(sndNodeId, req); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to unmarshal execution batch (was class undeployed?) " + - "[sndNodeId=" + sndNodeId + ", msg=" + msg + ']', e); - - return; - } - - try { - scheduleLocal(batch); - } - catch (IgniteCheckedException e) { - // Notify parent in case of error. - completeParentStage(ctx.localNodeId(), batch.futureId(), e); - } - } - else if (msg instanceof GridStreamerCancelRequest) - // This will be a local call. - cancelChildStage(ctx.localNodeId(), ((GridStreamerCancelRequest)msg).cancelledFutureId()); - else if (msg instanceof GridStreamerResponse) { - GridStreamerResponse res = (GridStreamerResponse)msg; - - assert res.futureId().globalId().equals(ctx.localNodeId()) : - "Wrong message received [res=" + res + ", sndNodeId=" + sndNodeId + - ", locNodeId=" + ctx.localNodeId() + ']'; - - Throwable err = null; - - if (res.errorBytes() != null) { - try { - err = ctx.config().getMarshaller().unmarshal(res.errorBytes(), null); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to unmarshal response.", e); - } - } - - // This will complete stage locally. - completeParentStage(sndNodeId, res.futureId(), err); - } - } - - /** - * Clear entries of removed classloader classes. - * - * @param undeployedClsLdr Classloader. - * @param info Whether to print info to log. - */ - private void unwindUndeploys(ClassLoader undeployedClsLdr, boolean info) { - assert undeployedClsLdr != null; - - int undeployWindowCnt = 0; - - Iterator<Object> it = streamerCtx.window().iterator(); - - while (it.hasNext()) { - Object evt = it.next(); - - if (undeployedClsLdr.equals(evt.getClass().getClassLoader())) { - it.remove(); - - undeployWindowCnt++; - } - } - - int undeploySpaceCnt = 0; - - Iterator<Map.Entry<Object, Object>> entryIt = streamerCtx.localSpace().entrySet().iterator(); - - while (entryIt.hasNext()) { - Map.Entry<Object, Object> entry = entryIt.next(); - - if (undeployedClsLdr.equals(entry.getKey().getClass().getClassLoader()) || - undeployedClsLdr.equals(entry.getValue().getClass().getClassLoader())) { - entryIt.remove(); - - undeploySpaceCnt++; - } - } - - if (info) { - if (log.isInfoEnabled() && (undeployWindowCnt > 0 || undeploySpaceCnt > 0)) - log.info("Undeployed all streamer events (if any) for obsolete class loader " + - "[undeployedClsLdr=" + undeployedClsLdr + ", undeployWindowCnt=" + undeployWindowCnt + - ", undeploySpaceCnt=" + undeploySpaceCnt + ']'); - } - } - - /** - * Creates execution request from batch. Will include deployment information if P2P class loading - * is enabled. - * - * @param batch Execution batch. - * @return Execution request. - * @throws IgniteCheckedException If failed. - */ - private Message createExecutionRequest(GridStreamerExecutionBatch batch) - throws IgniteCheckedException { - boolean depEnabled = ctx.deploy().enabled(); - - byte[] batchBytes = ctx.config().getMarshaller().marshal(batch); - - if (!depEnabled) - return new GridStreamerExecutionRequest(true, batchBytes, null, null, null, null, null); - else { - GridPeerDeployAware pda = new StreamerPda(batch.events()); - - GridDeployment dep = ctx.deploy().deploy(pda.deployClass(), pda.classLoader()); - - if (dep == null) - throw new IgniteCheckedException("Failed to get deployment for batch request [batch=" + batch + - ", pda=" + pda + ']'); - - return new GridStreamerExecutionRequest( - false, - batchBytes, - dep.deployMode(), - dep.sampleClassName(), - dep.userVersion(), - dep.participants(), - dep.classLoaderId() - ); - } - } - - /** - * Gets execution batch from execution request. Will explicitly unmarshal batch if P2P class loading - * is enabled. - * - * @param nodeId Node ID. - * @param req Execution request. - * @return Execution batch. - * @throws IgniteCheckedException If unmarshalling failed. - */ - private GridStreamerExecutionBatch executionBatch(UUID nodeId, GridStreamerExecutionRequest req) - throws IgniteCheckedException { - GridDeployment dep = null; - - if (!req.forceLocalDeployment()) { - dep = ctx.deploy().getGlobalDeployment( - req.deploymentMode(), - req.sampleClassName(), - req.sampleClassName(), - req.userVersion(), - nodeId, - req.classLoaderId(), - req.loaderParticipants(), - null); - - if (dep == null) - throw new IgniteCheckedException("Failed to obtain global deployment based on deployment metadata " + - "[nodeId=" + nodeId + ", req=" + req + ']'); - } - - GridStreamerExecutionBatch batch = ctx.config().getMarshaller().unmarshal(req.batchBytes(), - dep != null ? dep.classLoader() : null); - - // Set deployment to check for undeployment after stage execution. - batch.deployment(dep); - - return batch; - } - - /** - * Tries to send message to remote node. - * - * @param dstNodeId Destination node ID. - * @param msg Message to send. - * @throws IgniteCheckedException If failed. - */ - private void sendWithRetries(UUID dstNodeId, Message msg) throws IgniteCheckedException { - for (int i = 0; i < SEND_RETRY_COUNT; i++) { - try { - ctx.io().send(dstNodeId, topic, msg, GridIoPolicy.SYSTEM_POOL); - - return; - } - catch (IgniteCheckedException e) { - if (log.isDebugEnabled()) - log.debug("Failed to send message to remote node (will retry) [dstNodeId=" + dstNodeId + - ", msg=" + msg + ", err=" + e + ']'); - - if (!ctx.discovery().alive(dstNodeId)) - throw new ClusterTopologyCheckedException("Failed to send message (destination node left grid): " + - dstNodeId); - - if (i == SEND_RETRY_COUNT - 1) - throw e; - - U.sleep(SEND_RETRY_DELAY); - } - } - } - - /** {@inheritDoc} */ - @Override public void writeExternal(ObjectOutput out) throws IOException { - out.writeObject(ctx); - U.writeString(out, name); - } - - /** {@inheritDoc} */ - @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - ctx = (GridKernalContext)in.readObject(); - name = U.readString(in); - } - - /** - * Reconstructs object on unmarshalling. - * - * @return Reconstructed object. - */ - protected Object readResolve() { - return ctx.stream().streamer(name); - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(IgniteStreamerImpl.class, this); - } - - /** - * Data streamer peer-deploy aware. - */ - private class StreamerPda implements GridPeerDeployAware { - /** */ - private static final long serialVersionUID = 0L; - - /** Deploy class. */ - private Class<?> cls; - - /** Class loader. */ - private ClassLoader ldr; - - /** Collection of objects to detect deploy class and class loader. */ - private Collection<Object> objs; - - /** - * Constructs data streamer peer-deploy aware. - * - * @param objs Collection of objects to detect deploy class and class loader. - */ - private StreamerPda(Collection<Object> objs) { - this.objs = objs; - } - - /** {@inheritDoc} */ - @Override public Class<?> deployClass() { - if (cls == null) { - Class<?> cls0 = null; - - if (depCls != null) - cls0 = depCls; - else { - for (Iterator<Object> it = objs.iterator(); (cls0 == null || U.isJdk(cls0)) && it.hasNext();) - cls0 = U.detectClass(it.next()); - - if (cls0 == null || U.isJdk(cls0)) - cls0 = IgniteStreamerImpl.class; - } - - assert cls0 != null : "Failed to detect deploy class [objs=" + objs + ']'; - - cls = cls0; - } - - return cls; - } - - /** {@inheritDoc} */ - @Override public ClassLoader classLoader() { - if (ldr == null) { - ClassLoader ldr0 = deployClass().getClassLoader(); - - // Safety. - if (ldr0 == null) - ldr0 = U.gridClassLoader(); - - assert ldr0 != null : "Failed to detect classloader [objs=" + objs + ']'; - - ldr = ldr0; - } - - return ldr; - } - } - - /** - * Stage batch worker. - */ - private class BatchWorker extends GridWorker { - /** Batch */ - private GridStreamerExecutionBatch batch; - - /** */ - private StreamerStageWrapper stageWrapper; - - /** Streamer metrics holder. */ - private StreamerMetricsHolder streamerHolder; - - /** Schedule timestamp. */ - private long schedTs; - - /** Stage completion future. */ - private BatchExecutionFuture fut = new BatchExecutionFuture(); - - /** - * Creates worker. - * - * @param batch Batch. - * @param stageWrapper Wrapper. - * @param streamerHolder Stream holder. - */ - private BatchWorker( - GridStreamerExecutionBatch batch, - StreamerStageWrapper stageWrapper, - StreamerMetricsHolder streamerHolder - ) { - super(ctx.gridName(), "streamer-batch-worker-" + batch.stageName(), log); - - assert stageWrapper != null; - - this.batch = batch; - this.stageWrapper = stageWrapper; - this.streamerHolder = streamerHolder; - - schedTs = U.currentTimeMillis(); - - fut.setWorker(this); - } - - /** - * @return Completion future. - */ - public BatchExecutionFuture completionFuture() { - return fut; - } - - /** {@inheritDoc} */ - @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException { - try { - long start = U.currentTimeMillis(); - - streamerHolder.onStageExecutionStarted(stageWrapper.index(), start - schedTs); - - long end = 0; - - try { - if (log.isDebugEnabled()) - log.debug("Running streamer stage [stage=" + stageWrapper.name() + - ", futId=" + batch.futureId() + ']'); - - StreamerContext ctxDelegate = new GridStreamerContextDelegate(context(), - stageWrapper.nextStageName()); - - winLock.readLock(); - - Map<String, Collection<?>> res; - - try { - res = stageWrapper.run(ctxDelegate, batch.events()); - } - finally { - winLock.readUnlock(); - } - - // Close window for undeploy event. - GridDeployment dep = batch.deployment(); - - if (dep != null && dep.obsolete()) - unwindUndeploys(dep.classLoader(), false); - - if (res != null) { - for (Map.Entry<String, Collection<?>> entry : res.entrySet()) { - if (entry.getKey() == null) - throw new IgniteCheckedException("Failed to pass events to next stage " + - "(stage name cannot be null)."); - - GridStreamerStageExecutionFuture part = addEvents0( - batch.executionId(), - 0, - batch.executionStartTimeStamp(), - batch.futureId(), - batch.executionNodeIds(), - entry.getKey(), - entry.getValue()); - - if (atLeastOnce) - fut.add(part); - } - } - else { - if (log.isDebugEnabled()) - log.debug("Finished pipeline execution [stage=" + stageWrapper.name() + - ", futId=" + batch.futureId() + ']'); - - end = U.currentTimeMillis(); - - streamerHolder.onPipelineCompleted(end - batch.executionStartTimeStamp(), - batch.executionNodeIds().size()); - } - } - catch (IgniteCheckedException e) { - if (!atLeastOnce) { - notifyFailure(batch.stageName(), batch.events(), e); - - streamerHolder.onStageFailure(stageWrapper.index()); - } - - fut.onDone(e); - } - finally { - if (end == 0) - end = U.currentTimeMillis(); - - streamerHolder.onStageExecutionFinished(stageWrapper.index(), end - start); - } - } - finally { - fut.markInitialized(); - } - } - } - - /** - * Batch execution future. - */ - private static class BatchExecutionFuture extends GridCompoundFuture<Object, Object> { - /** */ - private static final long serialVersionUID = 0L; - - /** */ - private BatchWorker w; - - /** {@inheritDoc} */ - @Override public boolean cancel() throws IgniteCheckedException { - assert w != null; - - if (!super.cancel()) - return false; - - w.cancel(); - - return true; - } - - /** - * @param w Worker. - */ - public void setWorker(BatchWorker w) { - assert w != null; - - this.w = w; - } - } -}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1cc3d528/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMBeanAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMBeanAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMBeanAdapter.java deleted file mode 100644 index 652e0b8..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMBeanAdapter.java +++ /dev/null @@ -1,116 +0,0 @@ -/* - * 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.ignite.internal.processors.streamer; - -import org.apache.ignite.streamer.*; -import org.jetbrains.annotations.*; - -/** - * Streamer MBean implementation. - */ -public class StreamerMBeanAdapter implements StreamerMBean { - /** Streamer. */ - private IgniteStreamerImpl streamer; - - /** - * @param streamer Streamer. - */ - public StreamerMBeanAdapter(IgniteStreamerImpl streamer) { - this.streamer = streamer; - } - - /** {@inheritDoc} */ - @Nullable @Override public String getName() { - return streamer.name(); - } - - /** {@inheritDoc} */ - @Override public boolean isAtLeastOnce() { - return streamer.atLeastOnce(); - } - - /** {@inheritDoc} */ - @Override public int getStageFutureMapSize() { - return streamer.stageFutureMapSize(); - } - - /** {@inheritDoc} */ - @Override public int getBatchFutureMapSize() { - return streamer.batchFutureMapSize(); - } - - /** {@inheritDoc} */ - @Override public int getStageActiveExecutionCount() { - return streamer.metrics().stageActiveExecutionCount(); - } - - /** {@inheritDoc} */ - @Override public int getStageWaitingExecutionCount() { - return streamer.metrics().stageWaitingExecutionCount(); - } - - /** {@inheritDoc} */ - @Override public long getStageTotalExecutionCount() { - return streamer.metrics().stageTotalExecutionCount(); - } - - /** {@inheritDoc} */ - @Override public long getPipelineMaximumExecutionTime() { - return streamer.metrics().pipelineMaximumExecutionTime(); - } - - /** {@inheritDoc} */ - @Override public long getPipelineMinimumExecutionTime() { - return streamer.metrics().pipelineMinimumExecutionTime(); - } - - /** {@inheritDoc} */ - @Override public long getPipelineAverageExecutionTime() { - return streamer.metrics().pipelineAverageExecutionTime(); - } - - /** {@inheritDoc} */ - @Override public int getPipelineMaximumExecutionNodes() { - return streamer.metrics().pipelineMaximumExecutionNodes(); - } - - /** {@inheritDoc} */ - @Override public int getPipelineMinimumExecutionNodes() { - return streamer.metrics().pipelineMinimumExecutionNodes(); - } - - /** {@inheritDoc} */ - @Override public int getPipelineAverageExecutionNodes() { - return streamer.metrics().pipelineAverageExecutionNodes(); - } - - /** {@inheritDoc} */ - @Override public int getCurrentActiveSessions() { - return streamer.metrics().currentActiveSessions(); - } - - /** {@inheritDoc} */ - @Override public int getMaximumActiveSessions() { - return streamer.metrics().maximumActiveSessions(); - } - - /** {@inheritDoc} */ - @Override public int getFailuresCount() { - return streamer.metrics().failuresCount(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1cc3d528/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMetricsAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMetricsAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMetricsAdapter.java deleted file mode 100644 index d1e37ba..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMetricsAdapter.java +++ /dev/null @@ -1,272 +0,0 @@ -/* - * 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.ignite.internal.processors.streamer; - -import org.apache.ignite.internal.util.tostring.*; -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.streamer.*; - -import java.util.*; - -/** - * Streamer metrics adapter. - */ -public class StreamerMetricsAdapter implements StreamerMetrics { - /** */ - private int stageActiveExecCnt; - - /** */ - private int stageWaitingExecCnt; - - /** */ - private long stageTotalExecCnt; - - /** */ - private long pipelineMaxExecTime; - - /** */ - private long pipelineMinExecTime; - - /** */ - private long pipelineAvgExecTime; - - /** */ - private int pipelineMaxExecNodes; - - /** */ - private int pipelineMinExecNodes; - - /** */ - private int pipelineAvgExecNodes; - - /** */ - private long qryMaxExecTime; - - /** */ - private long qryMinExecTime; - - /** */ - private long qryAvgExecTime; - - /** */ - private int qryMaxExecNodes; - - /** */ - private int qryMinExecNodes; - - /** */ - private int qryAvgExecNodes; - - /** */ - private int curActiveSes; - - /** */ - private int maxActiveSes; - - /** */ - private int failuresCnt; - - /** */ - private int execSvcCap; - - /** */ - @GridToStringInclude - private Map<String, StreamerStageMetrics> stageMetrics; - - /** */ - @GridToStringInclude - private Map<String, StreamerWindowMetrics> windowMetrics; - - /** - * Empty constructor. - */ - public StreamerMetricsAdapter() { - // No-op. - } - - /** - * @param metrics Metrics. - */ - public StreamerMetricsAdapter(StreamerMetrics metrics) { - // Preserve alphabetic order for maintenance. - curActiveSes = metrics.currentActiveSessions(); - execSvcCap = metrics.executorServiceCapacity(); - failuresCnt = metrics.failuresCount(); - maxActiveSes = metrics.maximumActiveSessions(); - pipelineAvgExecNodes = metrics.pipelineAverageExecutionNodes(); - pipelineAvgExecTime = metrics.pipelineAverageExecutionTime(); - pipelineMaxExecNodes = metrics.pipelineMaximumExecutionNodes(); - pipelineMaxExecTime = metrics.pipelineMaximumExecutionTime(); - pipelineMinExecNodes = metrics.pipelineMinimumExecutionNodes(); - pipelineMinExecTime = metrics.pipelineMinimumExecutionTime(); - qryAvgExecNodes = metrics.queryAverageExecutionNodes(); - qryAvgExecTime = metrics.queryAverageExecutionTime(); - qryMaxExecNodes = metrics.queryMaximumExecutionNodes(); - qryMaxExecTime = metrics.queryMaximumExecutionTime(); - qryMinExecNodes = metrics.queryMinimumExecutionNodes(); - qryMinExecTime = metrics.queryMinimumExecutionTime(); - stageActiveExecCnt = metrics.stageActiveExecutionCount(); - stageTotalExecCnt = metrics.stageTotalExecutionCount(); - stageWaitingExecCnt = metrics.stageWaitingExecutionCount(); - - // Stage metrics. - Map<String, StreamerStageMetrics> map = U.newLinkedHashMap(metrics.stageMetrics().size()); - - for (StreamerStageMetrics m : metrics.stageMetrics()) - map.put(m.name(), new StreamerStageMetricsAdapter(m)); - - stageMetrics = Collections.unmodifiableMap(map); - - Map<String, StreamerWindowMetrics> map0 = U.newLinkedHashMap(metrics.windowMetrics().size()); - - for (StreamerWindowMetrics m : metrics.windowMetrics()) - map0.put(m.name(), new StreamerWindowMetricsAdapter(m)); - - windowMetrics = Collections.unmodifiableMap(map0); - } - - /** {@inheritDoc} */ - @Override public int stageActiveExecutionCount() { - return stageActiveExecCnt; - } - - /** {@inheritDoc} */ - @Override public int stageWaitingExecutionCount() { - return stageWaitingExecCnt; - } - - /** {@inheritDoc} */ - @Override public long stageTotalExecutionCount() { - return stageTotalExecCnt; - } - - /** {@inheritDoc} */ - @Override public long pipelineMaximumExecutionTime() { - return pipelineMaxExecTime; - } - - /** {@inheritDoc} */ - @Override public long pipelineMinimumExecutionTime() { - return pipelineMinExecTime; - } - - /** {@inheritDoc} */ - @Override public long pipelineAverageExecutionTime() { - return pipelineAvgExecTime; - } - - /** {@inheritDoc} */ - @Override public int pipelineMaximumExecutionNodes() { - return pipelineMaxExecNodes; - } - - /** {@inheritDoc} */ - @Override public int pipelineMinimumExecutionNodes() { - return pipelineMinExecNodes; - } - - /** {@inheritDoc} */ - @Override public int pipelineAverageExecutionNodes() { - return pipelineAvgExecNodes; - } - - /** {@inheritDoc} */ - @Override public long queryMaximumExecutionTime() { - return qryMaxExecTime; - } - - /** {@inheritDoc} */ - @Override public long queryMinimumExecutionTime() { - return qryMinExecTime; - } - - /** {@inheritDoc} */ - @Override public long queryAverageExecutionTime() { - return qryAvgExecTime; - } - - /** {@inheritDoc} */ - @Override public int queryMaximumExecutionNodes() { - return qryMaxExecNodes; - } - - /** {@inheritDoc} */ - @Override public int queryMinimumExecutionNodes() { - return qryMinExecNodes; - } - - /** {@inheritDoc} */ - @Override public int queryAverageExecutionNodes() { - return qryAvgExecNodes; - } - - /** {@inheritDoc} */ - @Override public int currentActiveSessions() { - return curActiveSes; - } - - /** {@inheritDoc} */ - @Override public int maximumActiveSessions() { - return maxActiveSes; - } - - /** {@inheritDoc} */ - @Override public int failuresCount() { - return failuresCnt; - } - - /** {@inheritDoc} */ - @Override public int executorServiceCapacity() { - return execSvcCap; - } - - /** {@inheritDoc} */ - @Override public StreamerStageMetrics stageMetrics(String stageName) { - StreamerStageMetrics metrics = stageMetrics.get(stageName); - - if (metrics == null) - throw new IllegalArgumentException("Streamer stage is not configured: " + stageName); - - return metrics; - } - - /** {@inheritDoc} */ - @Override public Collection<StreamerStageMetrics> stageMetrics() { - return stageMetrics.values(); - } - - /** {@inheritDoc} */ - @Override public StreamerWindowMetrics windowMetrics(String winName) { - StreamerWindowMetrics metrics = windowMetrics.get(winName); - - if (metrics == null) - throw new IllegalArgumentException("Streamer window is not configured: " + winName); - - return metrics; - } - - /** {@inheritDoc} */ - @Override public Collection<StreamerWindowMetrics> windowMetrics() { - return windowMetrics.values(); - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(StreamerMetricsAdapter.class, this); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1cc3d528/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMetricsHolder.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMetricsHolder.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMetricsHolder.java deleted file mode 100644 index c130510..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerMetricsHolder.java +++ /dev/null @@ -1,424 +0,0 @@ -/* - * 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.ignite.internal.processors.streamer; - -import org.apache.ignite.internal.util.*; -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.streamer.*; -import org.jdk8.backport.*; - -import java.util.*; - -/** - * Metrics holder. - * - * Note that for current active stages we use maximum active stages over last second. - */ -public class StreamerMetricsHolder implements StreamerMetrics { - /** Max active stages over last minute. */ - private GridAtomicInteger stageActiveMaxLastSec = new GridAtomicInteger(); - - /** Last stage max update ts. */ - private volatile long lastStageSampleTs; - - /** Number of running stages. */ - private LongAdder stageActiveCnt = new LongAdder(); - - /** Number of waiting stages. */ - private LongAdder stageWaitingCnt = new LongAdder(); - - /** Total number of stages executed. */ - private LongAdder stageTotalCnt = new LongAdder(); - - /** Max exec time. */ - private GridAtomicLong pipelineMaxExecTime = new GridAtomicLong(); - - /** Min exec time. */ - private GridAtomicLong pipelineMinExecTime = new GridAtomicLong(Long.MAX_VALUE); - - /** Pipeline average exec time sampler. */ - private LongAdder pipelineSumExecTime = new LongAdder(); - - /** Max exec nodes. */ - private GridAtomicInteger pipelineMaxExecNodes = new GridAtomicInteger(); - - /** Min exec nodes. */ - private GridAtomicInteger pipelineMinExecNodes = new GridAtomicInteger(Integer.MAX_VALUE); - - /** Avg exec nodes. */ - private LongAdder pipelineSumExecNodes = new LongAdder(); - - /** Total number of pipelines finished on this node. */ - private LongAdder pipelineTotalCnt = new LongAdder(); - - /** Query max exec time. */ - private GridAtomicLong qryMaxExecTime = new GridAtomicLong(); - - /** Query min exec time. */ - private GridAtomicLong qryMinExecTime = new GridAtomicLong(Long.MAX_VALUE); - - /** Query average exec time sampler. */ - private LongAdder qrySumExecTime = new LongAdder(); - - /** Query max exec nodes. */ - private GridAtomicInteger qryMaxExecNodes = new GridAtomicInteger(); - - /** Query min exec nodes. */ - private GridAtomicInteger qryMinExecNodes = new GridAtomicInteger(Integer.MAX_VALUE); - - /** Query avg exec nodes. */ - private LongAdder qrySumExecNodes = new LongAdder(); - - /** Total number of queries finished on this node. */ - private LongAdder qryTotalCnt = new LongAdder(); - - /** Current active sessions. */ - private LongAdder curActiveSessions = new LongAdder(); - - /** Max active sessions. */ - private GridAtomicInteger maxActiveSessions = new GridAtomicInteger(); - - /** Failures count. */ - private LongAdder failuresCnt = new LongAdder(); - - /** Stages metrics. */ - private final StreamerStageMetricsHolder[] stageMetrics; - - /** Stage metrics map. */ - private final Map<String, StreamerStageMetrics> stageMetricsMap; - - /** Window metrics map. */ - private final Map<String, StreamerWindowMetrics> windowMetricsMap; - - /** Executor service capacity. */ - private final int execSvcCap; - - /** - * @param stageMetrics Array of stage metrics holders. - * @param windowMetrics Array of window metrics holders. - * @param execSvcCap Executor service capacity. - */ - public StreamerMetricsHolder( - StreamerStageMetricsHolder[] stageMetrics, - StreamerWindowMetricsHolder[] windowMetrics, - int execSvcCap - ) { - this.execSvcCap = execSvcCap; - this.stageMetrics = stageMetrics; - - Map<String, StreamerStageMetrics> map = new LinkedHashMap<>(); - - for (StreamerStageMetricsHolder holder : stageMetrics) - map.put(holder.name(), holder); - - stageMetricsMap = Collections.unmodifiableMap(map); - - Map<String, StreamerWindowMetrics> map0 = new LinkedHashMap<>(); - - for (StreamerWindowMetricsHolder holder : windowMetrics) - map0.put(holder.name(), holder); - - windowMetricsMap = Collections.unmodifiableMap(map0); - } - - /** {@inheritDoc} */ - @Override public int stageActiveExecutionCount() { - return stageActiveMaxLastSec.get(); - } - - /** {@inheritDoc} */ - @Override public int stageWaitingExecutionCount() { - return stageWaitingCnt.intValue(); - } - - /** {@inheritDoc} */ - @Override public long stageTotalExecutionCount() { - return stageTotalCnt.longValue(); - } - - /** {@inheritDoc} */ - @Override public long pipelineMaximumExecutionTime() { - return pipelineMaxExecTime.get(); - } - - /** {@inheritDoc} */ - @Override public long pipelineMinimumExecutionTime() { - long min = pipelineMinExecTime.get(); - - return min == Long.MAX_VALUE ? 0 : min; - } - - /** {@inheritDoc} */ - @Override public long pipelineAverageExecutionTime() { - long totalTime = pipelineSumExecTime.sum(); - - long execs = pipelineTotalCnt.sum(); - - return execs == 0 ? 0 : totalTime / execs; - } - - /** {@inheritDoc} */ - @Override public int pipelineMaximumExecutionNodes() { - return pipelineMaxExecNodes.get(); - } - - /** {@inheritDoc} */ - @Override public int pipelineMinimumExecutionNodes() { - int min = pipelineMinExecNodes.get(); - - return min == Integer.MAX_VALUE ? 0 : min; - } - - /** {@inheritDoc} */ - @Override public int pipelineAverageExecutionNodes() { - long totalNodes = pipelineSumExecNodes.sum(); - - long execs = pipelineTotalCnt.sum(); - - return execs == 0 ? 0 : (int)(totalNodes / execs); - } - - /** {@inheritDoc} */ - @Override public long queryMaximumExecutionTime() { - return qryMaxExecTime.get(); - } - - /** {@inheritDoc} */ - @Override public long queryMinimumExecutionTime() { - long min = qryMinExecTime.get(); - - return min == Long.MAX_VALUE ? 0 : min; - } - - /** {@inheritDoc} */ - @Override public long queryAverageExecutionTime() { - long totalTime = qrySumExecTime.sum(); - - long execs = qryTotalCnt.sum(); - - return execs == 0 ? 0 : totalTime / execs; - } - - /** {@inheritDoc} */ - @Override public int queryMaximumExecutionNodes() { - return qryMaxExecNodes.get(); - } - - /** {@inheritDoc} */ - @Override public int queryMinimumExecutionNodes() { - int min = qryMinExecNodes.get(); - - return min == Integer.MAX_VALUE ? 0 : min; - } - - /** {@inheritDoc} */ - @Override public int queryAverageExecutionNodes() { - long totalNodes = qrySumExecNodes.sum(); - - long execs = qryTotalCnt.sum(); - - return execs == 0 ? 0 : (int)(totalNodes / execs); - } - - /** {@inheritDoc} */ - @Override public int currentActiveSessions() { - return curActiveSessions.intValue(); - } - - /** {@inheritDoc} */ - @Override public int maximumActiveSessions() { - return maxActiveSessions.get(); - } - - /** {@inheritDoc} */ - @Override public int failuresCount() { - return failuresCnt.intValue(); - } - - /** {@inheritDoc} */ - @Override public int executorServiceCapacity() { - return execSvcCap; - } - - /** {@inheritDoc} */ - @Override public StreamerStageMetrics stageMetrics(String stageName) { - StreamerStageMetrics metrics = stageMetricsMap.get(stageName); - - if (metrics == null) - throw new IllegalArgumentException("Streamer stage is not configured: " + stageName); - - return metrics; - } - - /** {@inheritDoc} */ - @Override public Collection<StreamerStageMetrics> stageMetrics() { - return stageMetricsMap.values(); - } - - /** {@inheritDoc} */ - @Override public StreamerWindowMetrics windowMetrics(String winName) { - StreamerWindowMetrics metrics = windowMetricsMap.get(winName); - - if (metrics == null) - throw new IllegalArgumentException("Streamer window is not configured: " + winName); - - return metrics; - } - - /** {@inheritDoc} */ - @Override public Collection<StreamerWindowMetrics> windowMetrics() { - return windowMetricsMap.values(); - } - - /** - * Stage scheduled callback. - */ - public void onStageScheduled() { - stageWaitingCnt.increment(); - } - - /** - * Stage execution started callback. - * - * @param idx Stage index. - * @param waitTime Stage wait time. - */ - public void onStageExecutionStarted(int idx, long waitTime) { - if (waitTime < 0) - waitTime = 0; - - stageActiveCnt.increment(); - stageWaitingCnt.decrement(); - - stageMetrics[idx].onExecutionStarted(waitTime); - - sampleCurrentStages(); - } - - /** - * Stage execution finished callback. - * - * @param idx Stage index. - * @param execTime Stage execution time. - */ - public void onStageExecutionFinished(int idx, long execTime) { - if (execTime < 0) - execTime = 0; - - stageActiveCnt.decrement(); - - stageTotalCnt.increment(); - - stageMetrics[idx].onExecutionFinished(execTime); - - sampleCurrentStages(); - } - - /** - * Pipeline completed callback. - * - * @param execTime Pipeline execution time. - * @param execNodes Pipeline execution nodes. - */ - public void onPipelineCompleted(long execTime, int execNodes) { - if (execTime < 0) - execTime = 0; - - pipelineMaxExecTime.setIfGreater(execTime); - pipelineMinExecTime.setIfLess(execTime); - pipelineSumExecTime.add(execTime); - - pipelineMaxExecNodes.setIfGreater(execNodes); - pipelineMinExecNodes.setIfLess(execNodes); - pipelineSumExecNodes.add(execNodes); - - pipelineTotalCnt.increment(); - } - - /** - * Query completed callback. - * - * @param execTime Query execution time. - * @param execNodes Query execution nodes. - */ - public void onQueryCompleted(long execTime, int execNodes) { - if (execTime < 0) - execTime = 0; - - qryMaxExecTime.setIfGreater(execTime); - qryMinExecTime.setIfLess(execTime); - qrySumExecTime.add(execTime); - - qryMaxExecNodes.setIfGreater(execNodes); - qryMinExecNodes.setIfLess(execNodes); - qrySumExecNodes.add(execNodes); - - qryTotalCnt.increment(); - } - - /** - * Session started callback. - */ - public void onSessionStarted() { - curActiveSessions.increment(); - - maxActiveSessions.setIfGreater(curActiveSessions.intValue()); - } - - /** - * Session finished callback. - */ - public void onSessionFinished() { - curActiveSessions.decrement(); - } - - /** - * Session failed callback. - */ - public void onSessionFailed() { - curActiveSessions.decrement(); - - failuresCnt.increment(); - } - - /** - * Stage failure callback. - * - * @param idx Stage index. - */ - public void onStageFailure(int idx) { - stageMetrics[idx].onFailure(); - } - - /** - * Samples current sessions. - */ - public void sampleCurrentStages() { - long now = U.currentTimeMillis(); - - int cur = (int)stageActiveCnt.sum(); - - if (now - lastStageSampleTs > 1000) { - stageActiveMaxLastSec.set(cur); - - lastStageSampleTs = now; - } - else - stageActiveMaxLastSec.setIfGreater(cur); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1cc3d528/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMBeanAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMBeanAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMBeanAdapter.java deleted file mode 100644 index 8c2251b..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMBeanAdapter.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * 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.ignite.internal.processors.streamer; - -import org.apache.ignite.streamer.*; - -/** - * Streamer stage MBean adapter. - */ -@SuppressWarnings("ConstantConditions") -public class StreamerStageMBeanAdapter implements StreamerStageMBean { - /** Stage name. */ - private String stageName; - - /** Stage class name. */ - private String stageClsName; - - /** */ - private IgniteStreamerImpl streamer; - - /** - * @param stageName Stage name. - * @param stageClsName Stage class name. - * @param streamer Streamer implementation. - */ - public StreamerStageMBeanAdapter(String stageName, String stageClsName, IgniteStreamerImpl streamer) { - this.stageName = stageName; - this.stageClsName = stageClsName; - this.streamer = streamer; - } - - /** {@inheritDoc} */ - @Override public String getName() { - return stageName; - } - - /** {@inheritDoc} */ - @Override public String getStageClassName() { - return stageClsName; - } - - /** {@inheritDoc} */ - @Override public long getMinimumExecutionTime() { - return streamer.metrics().stageMetrics(stageName).minimumExecutionTime(); - } - - /** {@inheritDoc} */ - @Override public long getMaximumExecutionTime() { - return streamer.metrics().stageMetrics(stageName).maximumExecutionTime(); - } - - /** {@inheritDoc} */ - @Override public long getAverageExecutionTime() { - return streamer.metrics().stageMetrics(stageName).averageExecutionTime(); - } - - /** {@inheritDoc} */ - @Override public long getMinimumWaitingTime() { - return streamer.metrics().stageMetrics(stageName).minimumWaitingTime(); - } - - /** {@inheritDoc} */ - @Override public long getMaximumWaitingTime() { - return streamer.metrics().stageMetrics(stageName).maximumWaitingTime(); - } - - /** {@inheritDoc} */ - @Override public long getAverageWaitingTime() { - return streamer.metrics().stageMetrics(stageName).averageWaitingTime(); - } - - /** {@inheritDoc} */ - @Override public long getTotalExecutionCount() { - return streamer.metrics().stageMetrics(stageName).totalExecutionCount(); - } - - /** {@inheritDoc} */ - @Override public int getFailuresCount() { - return streamer.metrics().stageMetrics(stageName).failuresCount(); - } - - /** {@inheritDoc} */ - @Override public boolean isExecuting() { - return streamer.metrics().stageMetrics(stageName).executing(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1cc3d528/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMetricsAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMetricsAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMetricsAdapter.java deleted file mode 100644 index c5f7a1b..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMetricsAdapter.java +++ /dev/null @@ -1,135 +0,0 @@ -/* - * 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.ignite.internal.processors.streamer; - -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.streamer.*; - -/** - * Streamer stage metrics adapter. - */ -public class StreamerStageMetricsAdapter implements StreamerStageMetrics { - /** */ - private String name; - - /** */ - private long minExecTime; - - /** */ - private long maxExecTime; - - /** */ - private long avgExecTime; - - /** */ - private long minWaitTime; - - /** */ - private long maxWaitTime; - - /** */ - private long avgWaitTime; - - /** */ - private long totalExecCnt; - - /** */ - private int failuresCnt; - - /** */ - private boolean executing; - - /** - * Empty constructor. - */ - public StreamerStageMetricsAdapter() { - // No-op. - } - - /** - * @param metrics Metrics. - */ - public StreamerStageMetricsAdapter(StreamerStageMetrics metrics) { - // Preserve alphabetic order for maintenance. - avgExecTime = metrics.averageExecutionTime(); - avgWaitTime = metrics.averageWaitingTime(); - executing = metrics.executing(); - failuresCnt = metrics.failuresCount(); - maxExecTime = metrics.maximumExecutionTime(); - maxWaitTime = metrics.maximumWaitingTime(); - minExecTime = metrics.minimumExecutionTime(); - minWaitTime = metrics.minimumWaitingTime(); - name = metrics.name(); - totalExecCnt = metrics.totalExecutionCount(); - } - - /** {@inheritDoc} */ - @Override public String name() { - return name; - } - - /** {@inheritDoc} */ - @Override public long minimumExecutionTime() { - return minExecTime; - } - - /** {@inheritDoc} */ - @Override public long maximumExecutionTime() { - return maxExecTime; - } - - /** {@inheritDoc} */ - @Override public long averageExecutionTime() { - return avgExecTime; - } - - /** {@inheritDoc} */ - @Override public long totalExecutionCount() { - return totalExecCnt; - } - - /** {@inheritDoc} */ - @Override public long minimumWaitingTime() { - return minWaitTime; - } - - /** {@inheritDoc} */ - @Override public long maximumWaitingTime() { - return maxWaitTime; - } - - /** {@inheritDoc} */ - @Override public long averageWaitingTime() { - return avgWaitTime; - } - - /** {@inheritDoc} */ - @Override public int failuresCount() { - return failuresCnt; - } - - /** {@inheritDoc} */ - @Override public boolean executing() { - return executing; - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(StreamerStageMetricsAdapter.class, this); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1cc3d528/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMetricsHolder.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMetricsHolder.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMetricsHolder.java deleted file mode 100644 index ed95628..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageMetricsHolder.java +++ /dev/null @@ -1,167 +0,0 @@ -/* - * 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.ignite.internal.processors.streamer; - -import org.apache.ignite.internal.util.*; -import org.apache.ignite.streamer.*; -import org.jdk8.backport.*; - -/** - * Streamer stage metrics holder. - */ -public class StreamerStageMetricsHolder implements StreamerStageMetrics { - /** Stage name. */ - private String name; - - /** Minimum execution time. */ - private GridAtomicLong minExecTime = new GridAtomicLong(Long.MAX_VALUE); - - /** Maximum execution time. */ - private GridAtomicLong maxExecTime = new GridAtomicLong(); - - /** Stage execution time sum. */ - private LongAdder sumExecTime = new LongAdder(); - - /** Stage minimum waiting time. */ - private GridAtomicLong minWaitTime = new GridAtomicLong(Long.MAX_VALUE); - - /** Stage maximum waiting time. */ - private GridAtomicLong maxWaitTime = new GridAtomicLong(); - - /** Stage average waiting time sum. */ - private LongAdder sumWaitTime = new LongAdder(); - - /** Total number of times this stage was executed. */ - private LongAdder totalExecCnt = new LongAdder(); - - /** Failures count. */ - private LongAdder failuresCnt = new LongAdder(); - - /** Number of threads executing this stage. */ - private LongAdder curActive = new LongAdder(); - - /** - * @param name Stage name. - */ - public StreamerStageMetricsHolder(String name) { - this.name = name; - } - - /** {@inheritDoc} */ - @Override public String name() { - return name; - } - - /** {@inheritDoc} */ - @Override public long minimumExecutionTime() { - long min = minExecTime.get(); - - return min == Long.MAX_VALUE ? 0 : min; - } - - /** {@inheritDoc} */ - @Override public long maximumExecutionTime() { - return maxExecTime.get(); - } - - /** {@inheritDoc} */ - @Override public long averageExecutionTime() { - long execTime = sumExecTime.sum(); - - long execs = totalExecCnt.sum(); - - return execs == 0 ? 0 : execTime / execs; - } - - /** {@inheritDoc} */ - @Override public long minimumWaitingTime() { - long min = minWaitTime.get(); - - return min == Long.MAX_VALUE ? 0 : min; - } - - /** {@inheritDoc} */ - @Override public long maximumWaitingTime() { - return maxWaitTime.get(); - } - - /** {@inheritDoc} */ - @Override public long averageWaitingTime() { - long waitTime = sumWaitTime.sum(); - - long execs = totalExecCnt.sum(); - - return execs == 0 ? 0 : waitTime / execs; - } - - /** {@inheritDoc} */ - @Override public long totalExecutionCount() { - return totalExecCnt.longValue(); - } - - /** {@inheritDoc} */ - @Override public int failuresCount() { - return failuresCnt.intValue(); - } - - /** {@inheritDoc} */ - @Override public boolean executing() { - return curActive.intValue() > 0; - } - - /** - * Execution started callback. - * - * @param waitTime Wait time. - */ - public void onExecutionStarted(long waitTime) { - if (waitTime < 0) - waitTime = 0; - - curActive.increment(); - - maxWaitTime.setIfGreater(waitTime); - minWaitTime.setIfLess(waitTime); - sumWaitTime.add(waitTime); - } - - /** - * Execution finished callback. - * - * @param execTime Stage execution time. - */ - public void onExecutionFinished(long execTime) { - if (execTime < 0) - execTime = 0; - - curActive.decrement(); - - maxExecTime.setIfGreater(execTime); - minExecTime.setIfLess(execTime); - sumExecTime.add(execTime); - - totalExecCnt.increment(); - } - - /** - * Failure callback. - */ - public void onFailure() { - failuresCnt.increment(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1cc3d528/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageWrapper.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageWrapper.java deleted file mode 100644 index 3efb1b0..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/streamer/StreamerStageWrapper.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * 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.ignite.internal.processors.streamer; - -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.streamer.*; -import org.jetbrains.annotations.*; - -import java.util.*; - -/** - * Stage wrapper that handles metrics calculation and time measurement. - */ -public class StreamerStageWrapper implements StreamerStage<Object> { - /** Stage delegate. */ - private StreamerStage<Object> delegate; - - /** Stage index. */ - private int idx; - - /** Next stage name. Set after creation. */ - private String nextStageName; - - /** - * @param delegate Delegate stage. - * @param idx Index. - */ - public StreamerStageWrapper(StreamerStage<Object> delegate, int idx) { - this.delegate = delegate; - this.idx = idx; - } - - /** - * @return Stage index. - */ - public int index() { - return idx; - } - - /** - * @return Next stage name in pipeline or {@code null} if this is the last stage. - */ - @Nullable public String nextStageName() { - return nextStageName; - } - - /** - * @param nextStageName Next stage name in pipeline. - */ - public void nextStageName(String nextStageName) { - this.nextStageName = nextStageName; - } - - /** {@inheritDoc} */ - @Override public String name() { - return delegate.name(); - } - - /** {@inheritDoc} */ - @Override public Map<String, Collection<?>> run(StreamerContext ctx, Collection<Object> evts) { - return delegate.run(ctx, evts); - } - - /** - * @return Delegate. - */ - public StreamerStage unwrap() { - return delegate; - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(StreamerStageWrapper.class, this); - } -}
