modules/core/src/main/java/org/apache/ignite/internal/processors/query/stat/StatisticsProcessor.java - ignite - Git at Google

 /*
  * 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.query.stat;

 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
 import java.util.function.Supplier;

 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.internal.NodeStoppingException;
 import org.apache.ignite.internal.processors.query.stat.config.StatisticsColumnConfiguration;
 import org.apache.ignite.internal.processors.query.stat.config.StatisticsObjectConfiguration;
 import org.apache.ignite.internal.processors.query.stat.task.GatherPartitionStatistics;
 import org.apache.ignite.thread.IgniteThreadPoolExecutor;

 /**
  * Process all tasks, related to statistics repository. Mostly - statistics collection,
  * invalidation (due to configuration, topology or obsolescence issues) and loads.
  * Input tasks should be scheduled throug management pool while gathering pool used to process heavy
  * operations in parallel.
  *
  * Manage gathering pool and it's jobs. To guarantee gracefull shutdown:
  * 1) Any job can be added into gatheringInProgress only in active state (check after adding)
  * 2) State can be disactivated only after cancelling all jobs and getting busyLock block
  * 3) Each job should do it's work in busyLock with periodically checking of it's cancellation status.
  */
 public class StatisticsProcessor {
     /** Logger. */
     private final IgniteLogger log;

     /** Ignite statistics repository. */
     private final IgniteStatisticsRepository statRepo;

     /** Ignite Thread pool executor to do statistics collection tasks. */
     private final BusyExecutor gatheringBusyExecutor;

     /** (cacheGroupId -> gather context) */
     private final ConcurrentMap<StatisticsKey, LocalStatisticsGatheringContext> gatheringInProgress =
         new ConcurrentHashMap<>();

     /**
      * Constructor.
      *
      * @param repo IgniteStatisticsRepository.
      * @param gatherPool Thread pool to gather statistics in.
      * @param stopping External stopping state supplier.
      * @param logSupplier Log supplier function.
      */
     public StatisticsProcessor(
         IgniteStatisticsRepository repo,
         IgniteThreadPoolExecutor gatherPool,
         Supplier<Boolean> stopping,
         Function<Class<?>, IgniteLogger> logSupplier
     ) {
         this.statRepo = repo;
         this.gatheringBusyExecutor = new BusyExecutor("gathering", gatherPool, stopping, logSupplier);
         this.log = logSupplier.apply(StatisticsProcessor.class);
     }

     /**
      * Update statistics for the given key to actual state.
      * If byObsolescence and tbl is not {@code null} - does not clear any other partitions.
      *
      * 1) Replace previous gathering context if exist and needed (replace byObsolescence gathering with new one or
      * replace gathering with older configuration or topology version with new one).
      * 2) If byObsolescence and no table awailable - clean obsolescence and partition statistics for the given key.
      * 3) Submit tasks for each specified partition.
      * 4) after last task finish gathering - it starts aggregation.
      * 5) read all partitions & obsolescence from repo and
      * if byObsolescence = {@code true} - remove unnecessary one and aggregate by specified list
      * if byObsolescence = {@code false} - aggregate all presented in store (because it should contains only actual ones)
      * 5) save aggregated local statistics
      *
      * @param ctx Statistics Gathering context.
      */
     public void updateLocalStatistics(LocalStatisticsGatheringContext ctx) {
         if (log.isDebugEnabled()) {
             log.debug(String.format(
                 "Start statistics processing: forceRecollect=%b, cfg=%s, partToProcess = %s, topVer=%s",
                 ctx.forceRecollect(), ctx.configuration(), ctx.allParts(), ctx.topologyVersion()));
         }

         if (registerNewTask(ctx)) {
             try {
                 if (ctx.forceRecollect())
                     // To save all obsolescence info even if there is no partitions to recollect.
                     statRepo.saveObsolescenceInfo(ctx.configuration().key());

                 if (ctx.table() == null || ctx.configuration() == null || ctx.configuration().columns().isEmpty()) {
                     statRepo.clearLocalPartitionsStatistics(ctx.configuration().key(), null);
                     ctx.future().complete(null);

                     return;
                 }

                 if (ctx.remainingParts().isEmpty())
                     ctx.future().complete(null);
                 else
                     submitTasks(ctx);
             }
             catch (Throwable t) {
                 // Submit tasks can't fire an error, so no need to cancel and wait for tasks finished here.
                 ctx.future().completeExceptionally(t);
             }
         }
         else {
             if (log.isDebugEnabled())
                 log.debug("Gathered by key " + ctx.configuration().key() + " were skipped due to previous one.");
         }
     }

     /**
      * Try to register new task. Returned task will remove itself from gatheringInProgress after completion.
      * If there are some other task for the given key - operation will be scheduled right after it if necessary
      * (current task have newer configuration or topology).
      *
      * @param ctx Task to register.
      * @return {@code true} if task was actually pushed as Gathering in progress task, {@code false} - otherwise.
      */
     private boolean registerNewTask(LocalStatisticsGatheringContext ctx) {
         boolean res[] = new boolean[1];

         gatheringInProgress.compute(ctx.configuration().key(), (k, v) -> {
             if (v == null) {
                 // No old context - start new
                 res[0] = true;

                 ctx.future().whenComplete((r, t) -> {
                     if (t != null) {
                         if (t instanceof CancellationException || t instanceof NodeStoppingException) {
                             if (log.isDebugEnabled())
                                 log.debug("Got " + t.getClass() + " exception during statistics collection by key "
                                     + ctx.configuration().key() + "." );
                         }
                         else
                             log.warning("Unexpected error during statistics collection by key " +
                                 ctx.configuration().key() + " . " + t.getMessage(), t);
                     }

                     // Anyway - try to remove itself from gathering map.
                     gatheringInProgress.remove(ctx.configuration().key(), ctx);
                 });

                 return ctx;
             }

             // If there are key - check if we can cancel it.
             if (v.topologyVersion() == null ||
                 (ctx.topologyVersion() != null && v.topologyVersion().compareTo(ctx.topologyVersion()) < 0) ||
                 v.configuration().compareTo(ctx.configuration()) < 0) {
                 // Old context for older topology or config - cancel and start new
                 v.cancel();

                 v.future().whenComplete((r, t) -> {
                     // Will be executed before original, so have to try to cancel previous context to add new one.
                     gatheringInProgress.remove(ctx.configuration().key(), v);

                     boolean rescheduled = gatheringBusyExecutor.busyRun(() -> updateLocalStatistics(ctx));

                     if (!rescheduled && log.isDebugEnabled()) {
                         log.debug("Unable to reschedule statistics task by key " + ctx.configuration().key()
                             + " due to inactive state.");
                     }

                 });

                 res[0] = false;

                 return v;
             }

             res[0] = false;

             return v;
         });

         return res[0];
     }

     /**
      * Generate and subtim tasks into gathering pool for existing gathering context.
      *
      * @param ctx Context to generate tasks by.
      */
     private void submitTasks(LocalStatisticsGatheringContext ctx) {
         for (int part : ctx.remainingParts()) {
             final GatherPartitionStatistics task = new GatherPartitionStatistics(
                 statRepo,
                 ctx,
                 part,
                 log
             );

             gatheringBusyExecutor.submit(new CancellableTask() {
                     /** {@inheritDoc} */
                     @Override public void run() {
                         processPartitionTask(task);
                     }

                     /** {@inheritDoc} */
                     @Override public void cancel() {
                         task.context().cancel();
                     }
                 })
                 .thenAccept(success -> {
                     if (!success) {
                         if (log.isDebugEnabled()) {
                             log.debug(String.format("Gathering failed for key %s.%d ", ctx.configuration().key(),
                                 task.partition()));
                         }

                         ctx.partitionNotAvailable(task.partition());
                     }
                 });
         }
     }

     /**
      * Aggregate partition statistics to local one.
      * @param ctx Context to use in aggregation.
      */
     private void aggregateStatistics(LocalStatisticsGatheringContext ctx) {
         if (ctx.cancelled())
             return;

         StatisticsKey key = ctx.configuration().key();
         Collection<ObjectPartitionStatisticsImpl> allParts = statRepo.getLocalPartitionsStatistics(key);

         if (ctx.forceRecollect())
             statRepo.aggregatedLocalStatistics(allParts, ctx.configuration(), ctx.topologyVersion());
         else {
             Set<Integer> partsToRemove = new HashSet<>();
             Collection<ObjectPartitionStatisticsImpl> partsToAggregate = new ArrayList<>();

             for (ObjectPartitionStatisticsImpl partStat : allParts) {
                 if (ctx.allParts() == null || !ctx.allParts().contains(partStat.partId()))
                     partsToRemove.add(partStat.partId());
                 else
                     partsToAggregate.add(partStat);
             }

             if (!partsToRemove.isEmpty())
                 statRepo.clearLocalPartitionsStatistics(ctx.configuration().key(), partsToRemove);

             if (!partsToAggregate.isEmpty())
                 statRepo.aggregatedLocalStatistics(partsToAggregate, ctx.configuration(), ctx.topologyVersion());
         }
     }

     /**
      * Process partition: call gather task, mark partition done, aggregate and complete future if needed.
      * @param task Partition gathering task to process.
      */
     private void processPartitionTask(final GatherPartitionStatistics task) {
         LocalStatisticsGatheringContext ctx = task.context();

         try {
             task.call();

             if (ctx.partitionDone(task.partition())) {
                 if (log.isDebugEnabled())
                     log.debug("Local partitions statistics successfully gathered by key " +
                         ctx.configuration().key());

                 aggregateStatistics(ctx);

                 ctx.future().complete(null);
             }
         }
         catch (Throwable t) {
             ctx.partitionNotAvailable(task.partition());

             if (t instanceof GatherStatisticCancelException) {
                 if (log.isDebugEnabled()) {
                     log.debug("Collect statistics task was cancelled " +
                         "[key=" + ctx.configuration().key() + ", part=" + task.partition() + ']');
                 }
             }
             else if (t.getCause() instanceof NodeStoppingException) {
                 if (log.isDebugEnabled())
                     log.debug("Node stopping during statistics collection on " +
                         "[key=" + ctx.configuration().key() + ", part=" + task.partition() + ']');
             }
             else
                 log.warning("Unexpected error on statistic gathering", t);
         }
     }

     /**
      * Check if specified object stistic fully meet specified statistics object configuration.
      *
      * @param stat Object statistics to test.
      * @param cfg Statistics object configuration to compare with.
      * @return {@code true} if specified statistics fully meet to specified configuration requiremenrs,
      *         {@code false} - otherwise.
      */
     private boolean partStatSuitToConfiguration(ObjectStatisticsImpl stat, StatisticsObjectConfiguration cfg) {
         if (stat == null)
             return false;

         if (stat.columnsStatistics().size() != cfg.columns().size())
             return false;

         for (StatisticsColumnConfiguration colCfg : cfg.columns().values()) {
             ColumnStatistics colStat = stat.columnStatistics(colCfg.name());

             if (colStat == null || colCfg.version() > colStat.version())
                 return false;
         }

         return true;
     }

     /**
      * Start gathering.
      */
     public void start() {
         if (log.isDebugEnabled())
             log.debug("Statistics gathering started.");

         gatheringBusyExecutor.activate();
     }

     /**
      * Stop gathering.
      */
     public void stop() {
         if (log.isTraceEnabled())
             log.trace(String.format("Statistics gathering stopping %d task...", gatheringInProgress.size()));

         // Can skip waiting for each task finished because of global busyLock.
         gatheringBusyExecutor.deactivate();

         if (log.isDebugEnabled())
             log.debug("Statistics gathering stopped.");
     }
 }
	/*
	* 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.query.stat;

	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.HashSet;
	import java.util.Set;
	import java.util.concurrent.CancellationException;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.ConcurrentMap;
	import java.util.function.Function;
	import java.util.function.Supplier;

	import org.apache.ignite.IgniteLogger;
	import org.apache.ignite.internal.NodeStoppingException;
	import org.apache.ignite.internal.processors.query.stat.config.StatisticsColumnConfiguration;
	import org.apache.ignite.internal.processors.query.stat.config.StatisticsObjectConfiguration;
	import org.apache.ignite.internal.processors.query.stat.task.GatherPartitionStatistics;
	import org.apache.ignite.thread.IgniteThreadPoolExecutor;

	/**
	* Process all tasks, related to statistics repository. Mostly - statistics collection,
	* invalidation (due to configuration, topology or obsolescence issues) and loads.
	* Input tasks should be scheduled throug management pool while gathering pool used to process heavy
	* operations in parallel.
	*
	* Manage gathering pool and it's jobs. To guarantee gracefull shutdown:
	* 1) Any job can be added into gatheringInProgress only in active state (check after adding)
	* 2) State can be disactivated only after cancelling all jobs and getting busyLock block
	* 3) Each job should do it's work in busyLock with periodically checking of it's cancellation status.
	*/
	public class StatisticsProcessor {
	/** Logger. */
	private final IgniteLogger log;

	/** Ignite statistics repository. */
	private final IgniteStatisticsRepository statRepo;

	/** Ignite Thread pool executor to do statistics collection tasks. */
	private final BusyExecutor gatheringBusyExecutor;

	/** (cacheGroupId -> gather context) */
	private final ConcurrentMap<StatisticsKey, LocalStatisticsGatheringContext> gatheringInProgress =
	new ConcurrentHashMap<>();

	/**
	* Constructor.
	*
	* @param repo IgniteStatisticsRepository.
	* @param gatherPool Thread pool to gather statistics in.
	* @param stopping External stopping state supplier.
	* @param logSupplier Log supplier function.
	*/
	public StatisticsProcessor(
	IgniteStatisticsRepository repo,
	IgniteThreadPoolExecutor gatherPool,
	Supplier<Boolean> stopping,
	Function<Class<?>, IgniteLogger> logSupplier
	) {
	this.statRepo = repo;
	this.gatheringBusyExecutor = new BusyExecutor("gathering", gatherPool, stopping, logSupplier);
	this.log = logSupplier.apply(StatisticsProcessor.class);
	}

	/**
	* Update statistics for the given key to actual state.
	* If byObsolescence and tbl is not {@code null} - does not clear any other partitions.
	*
	* 1) Replace previous gathering context if exist and needed (replace byObsolescence gathering with new one or
	* replace gathering with older configuration or topology version with new one).
	* 2) If byObsolescence and no table awailable - clean obsolescence and partition statistics for the given key.
	* 3) Submit tasks for each specified partition.
	* 4) after last task finish gathering - it starts aggregation.
	* 5) read all partitions & obsolescence from repo and
	* if byObsolescence = {@code true} - remove unnecessary one and aggregate by specified list
	* if byObsolescence = {@code false} - aggregate all presented in store (because it should contains only actual ones)
	* 5) save aggregated local statistics
	*
	* @param ctx Statistics Gathering context.
	*/
	public void updateLocalStatistics(LocalStatisticsGatheringContext ctx) {
	if (log.isDebugEnabled()) {
	log.debug(String.format(
	"Start statistics processing: forceRecollect=%b, cfg=%s, partToProcess = %s, topVer=%s",
	ctx.forceRecollect(), ctx.configuration(), ctx.allParts(), ctx.topologyVersion()));
	}

	if (registerNewTask(ctx)) {
	try {
	if (ctx.forceRecollect())
	// To save all obsolescence info even if there is no partitions to recollect.
	statRepo.saveObsolescenceInfo(ctx.configuration().key());

	if (ctx.table() == null \|\| ctx.configuration() == null \|\| ctx.configuration().columns().isEmpty()) {
	statRepo.clearLocalPartitionsStatistics(ctx.configuration().key(), null);
	ctx.future().complete(null);

	return;
	}

	if (ctx.remainingParts().isEmpty())
	ctx.future().complete(null);
	else
	submitTasks(ctx);
	}
	catch (Throwable t) {
	// Submit tasks can't fire an error, so no need to cancel and wait for tasks finished here.
	ctx.future().completeExceptionally(t);
	}
	}
	else {
	if (log.isDebugEnabled())
	log.debug("Gathered by key " + ctx.configuration().key() + " were skipped due to previous one.");
	}
	}

	/**
	* Try to register new task. Returned task will remove itself from gatheringInProgress after completion.
	* If there are some other task for the given key - operation will be scheduled right after it if necessary
	* (current task have newer configuration or topology).
	*
	* @param ctx Task to register.
	* @return {@code true} if task was actually pushed as Gathering in progress task, {@code false} - otherwise.
	*/
	private boolean registerNewTask(LocalStatisticsGatheringContext ctx) {
	boolean res[] = new boolean[1];

	gatheringInProgress.compute(ctx.configuration().key(), (k, v) -> {
	if (v == null) {
	// No old context - start new
	res[0] = true;

	ctx.future().whenComplete((r, t) -> {
	if (t != null) {
	if (t instanceof CancellationException \|\| t instanceof NodeStoppingException) {
	if (log.isDebugEnabled())
	log.debug("Got " + t.getClass() + " exception during statistics collection by key "
	+ ctx.configuration().key() + "." );
	}
	else
	log.warning("Unexpected error during statistics collection by key " +
	ctx.configuration().key() + " . " + t.getMessage(), t);
	}

	// Anyway - try to remove itself from gathering map.
	gatheringInProgress.remove(ctx.configuration().key(), ctx);
	});

	return ctx;
	}

	// If there are key - check if we can cancel it.
	if (v.topologyVersion() == null \|\|
	(ctx.topologyVersion() != null && v.topologyVersion().compareTo(ctx.topologyVersion()) < 0) \|\|
	v.configuration().compareTo(ctx.configuration()) < 0) {
	// Old context for older topology or config - cancel and start new
	v.cancel();

	v.future().whenComplete((r, t) -> {
	// Will be executed before original, so have to try to cancel previous context to add new one.
	gatheringInProgress.remove(ctx.configuration().key(), v);

	boolean rescheduled = gatheringBusyExecutor.busyRun(() -> updateLocalStatistics(ctx));

	if (!rescheduled && log.isDebugEnabled()) {
	log.debug("Unable to reschedule statistics task by key " + ctx.configuration().key()
	+ " due to inactive state.");
	}

	});

	res[0] = false;

	return v;
	}

	res[0] = false;

	return v;
	});

	return res[0];
	}

	/**
	* Generate and subtim tasks into gathering pool for existing gathering context.
	*
	* @param ctx Context to generate tasks by.
	*/
	private void submitTasks(LocalStatisticsGatheringContext ctx) {
	for (int part : ctx.remainingParts()) {
	final GatherPartitionStatistics task = new GatherPartitionStatistics(
	statRepo,
	ctx,
	part,
	log
	);

	gatheringBusyExecutor.submit(new CancellableTask() {
	/** {@inheritDoc} */
	@Override public void run() {
	processPartitionTask(task);
	}

	/** {@inheritDoc} */
	@Override public void cancel() {
	task.context().cancel();
	}
	})
	.thenAccept(success -> {
	if (!success) {
	if (log.isDebugEnabled()) {
	log.debug(String.format("Gathering failed for key %s.%d ", ctx.configuration().key(),
	task.partition()));
	}

	ctx.partitionNotAvailable(task.partition());
	}
	});
	}
	}

	/**
	* Aggregate partition statistics to local one.
	* @param ctx Context to use in aggregation.
	*/
	private void aggregateStatistics(LocalStatisticsGatheringContext ctx) {
	if (ctx.cancelled())
	return;

	StatisticsKey key = ctx.configuration().key();
	Collection<ObjectPartitionStatisticsImpl> allParts = statRepo.getLocalPartitionsStatistics(key);

	if (ctx.forceRecollect())
	statRepo.aggregatedLocalStatistics(allParts, ctx.configuration(), ctx.topologyVersion());
	else {
	Set<Integer> partsToRemove = new HashSet<>();
	Collection<ObjectPartitionStatisticsImpl> partsToAggregate = new ArrayList<>();

	for (ObjectPartitionStatisticsImpl partStat : allParts) {
	if (ctx.allParts() == null \|\| !ctx.allParts().contains(partStat.partId()))
	partsToRemove.add(partStat.partId());
	else
	partsToAggregate.add(partStat);
	}

	if (!partsToRemove.isEmpty())
	statRepo.clearLocalPartitionsStatistics(ctx.configuration().key(), partsToRemove);

	if (!partsToAggregate.isEmpty())
	statRepo.aggregatedLocalStatistics(partsToAggregate, ctx.configuration(), ctx.topologyVersion());
	}
	}

	/**
	* Process partition: call gather task, mark partition done, aggregate and complete future if needed.
	* @param task Partition gathering task to process.
	*/
	private void processPartitionTask(final GatherPartitionStatistics task) {
	LocalStatisticsGatheringContext ctx = task.context();

	try {
	task.call();

	if (ctx.partitionDone(task.partition())) {
	if (log.isDebugEnabled())
	log.debug("Local partitions statistics successfully gathered by key " +
	ctx.configuration().key());

	aggregateStatistics(ctx);

	ctx.future().complete(null);
	}
	}
	catch (Throwable t) {
	ctx.partitionNotAvailable(task.partition());

	if (t instanceof GatherStatisticCancelException) {
	if (log.isDebugEnabled()) {
	log.debug("Collect statistics task was cancelled " +
	"[key=" + ctx.configuration().key() + ", part=" + task.partition() + ']');
	}
	}
	else if (t.getCause() instanceof NodeStoppingException) {
	if (log.isDebugEnabled())
	log.debug("Node stopping during statistics collection on " +
	"[key=" + ctx.configuration().key() + ", part=" + task.partition() + ']');
	}
	else
	log.warning("Unexpected error on statistic gathering", t);
	}
	}

	/**
	* Check if specified object stistic fully meet specified statistics object configuration.
	*
	* @param stat Object statistics to test.
	* @param cfg Statistics object configuration to compare with.
	* @return {@code true} if specified statistics fully meet to specified configuration requiremenrs,
	* {@code false} - otherwise.
	*/
	private boolean partStatSuitToConfiguration(ObjectStatisticsImpl stat, StatisticsObjectConfiguration cfg) {
	if (stat == null)
	return false;

	if (stat.columnsStatistics().size() != cfg.columns().size())
	return false;

	for (StatisticsColumnConfiguration colCfg : cfg.columns().values()) {
	ColumnStatistics colStat = stat.columnStatistics(colCfg.name());

	if (colStat == null \|\| colCfg.version() > colStat.version())
	return false;
	}

	return true;
	}

	/**
	* Start gathering.
	*/
	public void start() {
	if (log.isDebugEnabled())
	log.debug("Statistics gathering started.");

	gatheringBusyExecutor.activate();
	}

	/**
	* Stop gathering.
	*/
	public void stop() {
	if (log.isTraceEnabled())
	log.trace(String.format("Statistics gathering stopping %d task...", gatheringInProgress.size()));

	// Can skip waiting for each task finished because of global busyLock.
	gatheringBusyExecutor.deactivate();

	if (log.isDebugEnabled())
	log.debug("Statistics gathering stopped.");
	}
	}