<commit_before>/**
 * Copyright (C) 2014 - present by OpenGamma Inc. and the OpenGamma group of companies
 *
 * Please see distribution for license.
 */
package com.opengamma.sesame.server;

import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.threeten.bp.Duration;

import com.opengamma.engine.marketdata.spec.MarketDataSpecification;
import com.opengamma.id.ExternalIdBundle;
import com.opengamma.id.VersionCorrection;
import com.opengamma.master.security.ManageableSecurity;
import com.opengamma.sesame.config.EmptyFunctionArguments;
import com.opengamma.sesame.config.FunctionArguments;
import com.opengamma.sesame.engine.CycleArguments;
import com.opengamma.sesame.engine.Results;
import com.opengamma.sesame.engine.View;
import com.opengamma.sesame.marketdata.CycleMarketDataFactory;
import com.opengamma.sesame.marketdata.FieldName;
import com.opengamma.sesame.marketdata.MarketDataFactory;
import com.opengamma.sesame.marketdata.StrategyAwareMarketDataSource;
import com.opengamma.util.result.Result;
import com.opengamma.util.tuple.Pair;
import com.opengamma.util.tuple.Pairs;

/**
 * Responsible for running a cycles of the engine.
 * <p>
 * This involves checking the source of market data and ensuring that
 * all required data is retrieved. To achieve this for data sources which
 * do not eagerly retrieve data (e.g. most live data providers), a cycle may
 * be run multiple times until all its market data requirements are fulfilled.
 */
public class CycleRunner {

  /**
   * A market data source used for setting up cycles.
   * It is not expected to be used to access market data but merely
   * provides a suitable initial value.
   */
  public static final StrategyAwareMarketDataSource INITIAL_MARKET_DATA_SOURCE = new InitialMarketDataSource();

  /**
   * The view to be executed, not null.
   */
  private final View _view;
  /**
   * The manager of the market data sources, not null.
   */
  private final MarketDataFactory _marketDataFactory;
  /**
   * The cycle options determining how the cycles of the view
   * should be executed, not null.
   */
  private final CycleOptions _cycleOptions;
  /**
   * The trades/securities to execute the cycles with, not null
   * but may be empty.
   */
  private final List<ManageableSecurity> _inputs;
  /**
   * Handles the results produced by each cycle of the engine, not null.
   */
  private final CycleResultsHandler _handler;
  /**
   * Determines whether the execution of the cycles should be terminated, not null.
   * This is generally used when an infinite set of cycles has been requested
   * and we want to stop processing (e.g. a UI using streaming data, which the
   * user then decides they have finished with).
   */
  private final CycleTerminator _cycleTerminator;
  /**
   * Responsible for executing the next cycle after a the minimum period
   * has passed.
   */
  private final ScheduledExecutorService _scheduler = Executors.newScheduledThreadPool(1);
  /**
   * Minimum time period between cycles, not null. If a series of cycles are to
   * be run and the first starts at time <code>t</code>, then the next cycle
   * will not be started until at least time <code>t + minimumTimeBetweenCycles</code>.
   * This prevents the server "spinning" for cases where the cycles execute quickly.
   */
  private final Duration _minimumTimeBetweenCycles;

  /**
   * Creates the new cycle runner.
   *
   * @param view  the view to be executed
   * @param marketDataFactory  the factory for market data sources
   * @param cycleOptions  the cycle options determining how the cycles of the view should be executed
   * @param inputs  the trades/securities to execute the cycles with, may be empty
   * @param handler  handler for the results produced by each cycle of the engine
   * @param cycleTerminator  determines whether the execution of the cycles should be terminated
   * @param minimumTimeBetweenCycles  minimum duration between cycles, not null
   */
  public CycleRunner(View view,
                     MarketDataFactory marketDataFactory,
                     CycleOptions cycleOptions,
                     List<ManageableSecurity> inputs,
                     CycleResultsHandler handler,
                     CycleTerminator cycleTerminator,
                     Duration minimumTimeBetweenCycles) {

    _view = view;
    _marketDataFactory = marketDataFactory;
    _cycleOptions = cycleOptions;
    _inputs = inputs;
    _handler = handler;
    _cycleTerminator = cycleTerminator;
    _minimumTimeBetweenCycles = minimumTimeBetweenCycles;
  }

  //-------------------------------------------------------------------------
  /**
   * Execute the view with each of the cycle options, checking if
   * early termination is required. We keep track of the market data
   * source used as this may be used to help setup the next cycle.
   */
  public void execute() {

    // We keep track of the market data being used so that when required we can
    // track the changes in market data between cycles
    final CycleMarketDataFactory cycleMarketDataFactory =
        new DefaultCycleMarketDataFactory(_marketDataFactory, INITIAL_MARKET_DATA_SOURCE);

    // Iterate over the cycle options. As they may be infinite (e.g. streaming),
    // we need to check the terminator to see if external events mean we should stop.
    final Iterator<IndividualCycleOptions> cycleOptions = _cycleOptions.iterator();

    // We create an initial task and execute it immediately. The task
    // itself will then schedule the next task, which in turn creates
    // the next etc. There are alternative implementations that could
    // be used and these should be reassessed:
    //   - create a simple timed task that periodically enables a
    //     semaphore/latch. Loop over cycle options and await the
    //     semaphore. When acquired, execute the cycle
    //   - use the scheduler to periodically schedule a job. Job
    //     picks up the next cycle option and executes
    _scheduler.execute(createNextCycleTask(cycleMarketDataFactory, cycleOptions, _handler, _scheduler));

    // Wait for all work to be completed before exiting
    try {
      _scheduler.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);<commit_msg>PLT-397 - Add throttle to function server<commit_after>/**
 * Copyright (C) 2014 - present by OpenGamma Inc. and the OpenGamma group of companies
 *
 * Please see distribution for license.
 */
package com.opengamma.sesame.server;

import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.threeten.bp.Duration;

import com.opengamma.engine.marketdata.spec.MarketDataSpecification;
import com.opengamma.id.ExternalIdBundle;
import com.opengamma.id.VersionCorrection;
import com.opengamma.master.security.ManageableSecurity;
import com.opengamma.sesame.config.EmptyFunctionArguments;
import com.opengamma.sesame.config.FunctionArguments;
import com.opengamma.sesame.engine.CycleArguments;
import com.opengamma.sesame.engine.Results;
import com.opengamma.sesame.engine.View;
import com.opengamma.sesame.marketdata.CycleMarketDataFactory;
import com.opengamma.sesame.marketdata.FieldName;
import com.opengamma.sesame.marketdata.MarketDataFactory;
import com.opengamma.sesame.marketdata.StrategyAwareMarketDataSource;
import com.opengamma.util.ArgumentChecker;
import com.opengamma.util.result.Result;
import com.opengamma.util.tuple.Pair;
import com.opengamma.util.tuple.Pairs;

/**
 * Responsible for running a cycles of the engine.
 * <p>
 * This involves checking the source of market data and ensuring that
 * all required data is retrieved. To achieve this for data sources which
 * do not eagerly retrieve data (e.g. most live data providers), a cycle may
 * be run multiple times until all its market data requirements are fulfilled.
 */
public class CycleRunner {

  /**
   * A market data source used for setting up cycles.
   * It is not expected to be used to access market data but merely
   * provides a suitable initial value.
   */
  public static final StrategyAwareMarketDataSource INITIAL_MARKET_DATA_SOURCE = new InitialMarketDataSource();

  /**
   * The view to be executed, not null.
   */
  private final View _view;
  /**
   * The manager of the market data sources, not null.
   */
  private final MarketDataFactory _marketDataFactory;
  /**
   * The cycle options determining how the cycles of the view
   * should be executed, not null.
   */
  private final CycleOptions _cycleOptions;
  /**
   * The trades/securities to execute the cycles with, not null
   * but may be empty.
   */
  private final List<ManageableSecurity> _inputs;
  /**
   * Handles the results produced by each cycle of the engine, not null.
   */
  private final CycleResultsHandler _handler;
  /**
   * Determines whether the execution of the cycles should be terminated, not null.
   * This is generally used when an infinite set of cycles has been requested
   * and we want to stop processing (e.g. a UI using streaming data, which the
   * user then decides they have finished with).
   */
  private final CycleTerminator _cycleTerminator;
  /**
   * Minimum time period between cycles, not null. If a series of cycles are to
   * be run and the first starts at time <code>t</code>, then the next cycle
   * will not be started until at least time <code>t + minimumTimeBetweenCycles</code>.
   * This prevents the server "spinning" for cases where the cycles execute quickly.
   */
  private final Duration _minimumTimeBetweenCycles;

  /**
   * Creates the new cycle runner.
   *
   * @param view  the view to be executed, not null
   * @param marketDataFactory  the factory for market data sources, not null
   * @param cycleOptions  the cycle options determining how the cycles of the view should be executed, not null
   * @param inputs  the trades/securities to execute the cycles with, not null but may be empty
   * @param handler  handler for the results produced by each cycle of the engine, not null
   * @param cycleTerminator  determines whether the execution of the cycles should be terminated, not null
   * @param minimumTimeBetweenCycles  minimum duration between cycles, not null
   */
  public CycleRunner(View view,
                     MarketDataFactory marketDataFactory,
                     CycleOptions cycleOptions,
                     List<ManageableSecurity> inputs,
                     CycleResultsHandler handler,
                     CycleTerminator cycleTerminator,
                     Duration minimumTimeBetweenCycles) {

    _view = ArgumentChecker.notNull(view, "view");
    _marketDataFactory = ArgumentChecker.notNull(marketDataFactory, "marketDataFactory");
    _cycleOptions = ArgumentChecker.notNull(cycleOptions, "cycleOptions");
    _inputs = ArgumentChecker.notNull(inputs, "inputs");
    _handler = ArgumentChecker.notNull(handler, "handler");
    _cycleTerminator = ArgumentChecker.notNull(cycleTerminator, "cycleTerminator");
    _minimumTimeBetweenCycles = ArgumentChecker.notNull(minimumTimeBetweenCycles, "minimumTimeBetweenCycles");
  }

  //-------------------------------------------------------------------------
  /**
   * Execute the view with each of the cycle options, checking if
   * early termination is required. We keep track of the market data
   * source used as this may be used to help setup the next cycle.
   */
  public void execute() {

    // We keep track of the market data being used so that when required we can
    // track the changes in market data between cycles
    final CycleMarketDataFactory cycleMarketDataFactory =
        new DefaultCycleMarketDataFactory(_marketDataFactory, INITIAL_MARKET_DATA_SOURCE);

    // Iterate over the cycle options. As they may be infinite (e.g. streaming),
    // we need to check the terminator to see if external events mean we should stop.
    final Iterator<IndividualCycleOptions> cycleOptions = _cycleOptions.iterator();

    // Scheduler to be used for executing the next cycle after
    // the minimum period has passed.
    ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    // We create an initial task and execute it immediately. The task
    // itself will then schedule the next task, which in turn creates
    // the next etc. There are alternative implementations that could
    // be used and these should be reassessed:
    //   - create a simple timed task that periodically enables a
    //     semaphore/latch. Loop over cycle options and await the
    //     semaphore. When acquired, execute the cycle
    //   - use the scheduler to periodically schedule a job. Job
    //     picks up the next cycle option and executes
    scheduler.execute(createNextCycleTask(cycleMarketDataFactory, cycleOptions, _handler, scheduler));

    // Wait for all work to be completed before exiting
    try {
      scheduler.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
{'fraction_non_alphanumeric': 0.07739399894187893, 'fraction_numerical': 0.0009069609251001436, 'mean_word_length': 3.3858137222406364, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5833393', 'n_tokens_mistral': 3540, 'n_tokens_neox': 3397, 'n_words': 1348}