<commit_before>/*
 * Copyright 2015 Google Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.google.android.agera;

import static com.google.android.agera.WorkerHandler.MSG_CALL_ACKNOWLEDGE_CANCEL;
import static com.google.android.agera.WorkerHandler.MSG_CALL_MAYBE_START_FLOW;
import static com.google.android.agera.WorkerHandler.workerHandler;
import static com.google.android.agera.Observables.compositeObservable;
import static com.google.android.agera.Observables.perMillisecondObservable;
import static com.google.android.agera.Preconditions.checkNotNull;
import static com.google.android.agera.Preconditions.checkState;
import static com.google.android.agera.RepositoryConfig.CANCEL_FLOW;
import static com.google.android.agera.RepositoryConfig.RESET_TO_INITIAL_VALUE;
import static com.google.android.agera.RepositoryConfig.SEND_INTERRUPT;
import static java.lang.Thread.currentThread;

import android.support.annotation.IntDef;
import android.support.annotation.NonNull;
import android.support.annotation.Nullable;

import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.util.List;
import java.util.concurrent.Executor;

@SuppressWarnings({"rawtypes", "unchecked"})
final class CompiledRepository extends BaseObservable
    implements Repository, Updatable, Runnable {

  @NonNull
  static Repository compiledRepository(
      @NonNull final Object initialValue,
      @NonNull final List<Observable> eventSources,
      final int frequency,
      @NonNull final List<Object> directives,
      @NonNull final Merger<Object, Object, Boolean> notifyChecker,
      @RepositoryConfig final int concurrentUpdateConfig,
      @RepositoryConfig final int deactivationConfig) {
    final Object[] directiveArray = directives.toArray();
    return new CompiledRepository(initialValue, compositeObservable(frequency,
        eventSources.toArray(new Observable[eventSources.size()])),
        directiveArray, notifyChecker, deactivationConfig, concurrentUpdateConfig);
  }

  //region Invariants

  @NonNull
  private final Object initialValue;
  @NonNull
  private final Observable eventSource;
  @NonNull
  private final Object[] directives;
  @NonNull
  private final Merger<Object, Object, Boolean> notifyChecker;
  @RepositoryConfig
  private final int deactivationConfig;
  @RepositoryConfig
  private final int concurrentUpdateConfig;
  @NonNull
  private final WorkerHandler workerHandler;

  CompiledRepository(
      @NonNull final Object initialValue,
      @NonNull final Observable eventSource,
      @NonNull final Object[] directives,
      @NonNull final Merger<Object, Object, Boolean> notifyChecker,
      @RepositoryConfig final int deactivationConfig,
      @RepositoryConfig final int concurrentUpdateConfig) {
    this.initialValue = initialValue;
    this.currentValue = initialValue;
    this.intermediateValue = initialValue; // non-final field but with @NonNull requirement
    this.eventSource = eventSource;
    this.directives = directives;
    this.notifyChecker = notifyChecker;
    this.deactivationConfig = deactivationConfig;
    this.concurrentUpdateConfig = concurrentUpdateConfig;
    this.workerHandler = workerHandler();
  }

  //endregion Invariants

  //region Data processing flow states

  @Retention(RetentionPolicy.SOURCE)
  @IntDef({IDLE, RUNNING, CANCEL_REQUESTED, PAUSED_AT_GO_TO, PAUSED_AT_GO_LAZY, RUNNING_LAZILY})
  private @interface RunState {}

  private static final int IDLE = 0;
  private static final int RUNNING = 1;
  private static final int CANCEL_REQUESTED = 2;
  private static final int PAUSED_AT_GO_TO = 3;
  private static final int PAUSED_AT_GO_LAZY = 4;
  private static final int RUNNING_LAZILY = 5;

  @RunState
  private int runState = IDLE;
  private boolean restartNeeded;
  /** Index of the last goTo()/goLazy() directive, for resuming, or -1 for other directives. */
  private int lastDirectiveIndex = -1;
  /** The current value to be exposed through the repository's get method. */
  @NonNull
  private Object currentValue;
  /** The intermediate value computed by the executed part of the flow. */
  @NonNull
  private Object intermediateValue;
  /** The thread currently running a directive that can be interrupted. */
  @Nullable
  private Thread currentThread;

  //endregion Data processing flow states

  //region Starting and requesting cancellation
  // - All methods in this region are called from the Worker Looper thread, but reading and writing
  //   states that might be accessed from a different thread are still synchronized.

  @Override
  protected void observableActivated() {
    eventSource.addUpdatable(this);
    maybeStartFlow();
  }

  @Override
  protected void observableDeactivated() {
    eventSource.removeUpdatable(this);
    maybeCancelFlow(deactivationConfig, false);
  }

  @Override
  public void update() {
    maybeCancelFlow(concurrentUpdateConfig, true);
    maybeStartFlow();
  }

  /**
   * Called on the worker looper thread. Starts the data processing flow if it's not running. This
   * also cancels the lazily-executed part of the flow if the run state is "paused at lazy".
   */
  void maybeStartFlow() {
    synchronized (this) {
      if (runState == IDLE || runState == PAUSED_AT_GO_LAZY) {
        runState = RUNNING;
        lastDirectiveIndex = -1; // this could be pointing at the goLazy directive
        restartNeeded = false;
      } else {
        return; // flow already running, do not continue.
      }
    }
    intermediateValue = currentValue;
    runFlowFrom(0, false);
  }

  /**
   * Called on the worker looper thread. Depending on the {@code config}, cancels the data
   * processing flow, resets the value, and/or sends the interrupt signal to the thread currently
   * processing a getFrom/mergeIn/transform instruction of the flow.
   *
   * @param scheduleRestart Whether to schedule a restart if a current flow is canceled.
   */
  private void maybeCancelFlow(@RepositoryConfig final int config, final boolean scheduleRestart) {
    synchronized (this) {
      if (runState == RUNNING || runState == PAUSED_AT_GO_TO) {
        restartNeeded = scheduleRestart;<commit_msg>Fixed restart racing condition in repo compiler (#73)<commit_after>/*
 * Copyright 2015 Google Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.google.android.agera;

import static com.google.android.agera.WorkerHandler.MSG_CALL_ACKNOWLEDGE_CANCEL;
import static com.google.android.agera.WorkerHandler.MSG_CALL_MAYBE_START_FLOW;
import static com.google.android.agera.WorkerHandler.workerHandler;
import static com.google.android.agera.Observables.compositeObservable;
import static com.google.android.agera.Observables.perMillisecondObservable;
import static com.google.android.agera.Preconditions.checkNotNull;
import static com.google.android.agera.Preconditions.checkState;
import static com.google.android.agera.RepositoryConfig.CANCEL_FLOW;
import static com.google.android.agera.RepositoryConfig.RESET_TO_INITIAL_VALUE;
import static com.google.android.agera.RepositoryConfig.SEND_INTERRUPT;
import static java.lang.Thread.currentThread;

import android.support.annotation.IntDef;
import android.support.annotation.NonNull;
import android.support.annotation.Nullable;

import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.util.List;
import java.util.concurrent.Executor;

@SuppressWarnings({"rawtypes", "unchecked"})
final class CompiledRepository extends BaseObservable
    implements Repository, Updatable, Runnable {

  @NonNull
  static Repository compiledRepository(
      @NonNull final Object initialValue,
      @NonNull final List<Observable> eventSources,
      final int frequency,
      @NonNull final List<Object> directives,
      @NonNull final Merger<Object, Object, Boolean> notifyChecker,
      @RepositoryConfig final int concurrentUpdateConfig,
      @RepositoryConfig final int deactivationConfig) {
    final Object[] directiveArray = directives.toArray();
    return new CompiledRepository(initialValue, compositeObservable(frequency,
        eventSources.toArray(new Observable[eventSources.size()])),
        directiveArray, notifyChecker, deactivationConfig, concurrentUpdateConfig);
  }

  //region Invariants

  @NonNull
  private final Object initialValue;
  @NonNull
  private final Observable eventSource;
  @NonNull
  private final Object[] directives;
  @NonNull
  private final Merger<Object, Object, Boolean> notifyChecker;
  @RepositoryConfig
  private final int deactivationConfig;
  @RepositoryConfig
  private final int concurrentUpdateConfig;
  @NonNull
  private final WorkerHandler workerHandler;

  CompiledRepository(
      @NonNull final Object initialValue,
      @NonNull final Observable eventSource,
      @NonNull final Object[] directives,
      @NonNull final Merger<Object, Object, Boolean> notifyChecker,
      @RepositoryConfig final int deactivationConfig,
      @RepositoryConfig final int concurrentUpdateConfig) {
    this.initialValue = initialValue;
    this.currentValue = initialValue;
    this.intermediateValue = initialValue; // non-final field but with @NonNull requirement
    this.eventSource = eventSource;
    this.directives = directives;
    this.notifyChecker = notifyChecker;
    this.deactivationConfig = deactivationConfig;
    this.concurrentUpdateConfig = concurrentUpdateConfig;
    this.workerHandler = workerHandler();
  }

  //endregion Invariants

  //region Data processing flow states

  @Retention(RetentionPolicy.SOURCE)
  @IntDef({IDLE, RUNNING, CANCEL_REQUESTED, PAUSED_AT_GO_TO, PAUSED_AT_GO_LAZY, RUNNING_LAZILY})
  private @interface RunState {}

  private static final int IDLE = 0;
  private static final int RUNNING = 1;
  private static final int CANCEL_REQUESTED = 2;
  private static final int PAUSED_AT_GO_TO = 3;
  private static final int PAUSED_AT_GO_LAZY = 4;
  private static final int RUNNING_LAZILY = 5;

  @RunState
  private int runState = IDLE;
  private boolean restartNeeded;
  /** Index of the last goTo()/goLazy() directive, for resuming, or -1 for other directives. */
  private int lastDirectiveIndex = -1;
  /** The current value to be exposed through the repository's get method. */
  @NonNull
  private Object currentValue;
  /** The intermediate value computed by the executed part of the flow. */
  @NonNull
  private Object intermediateValue;
  /** The thread currently running a directive that can be interrupted. */
  @Nullable
  private Thread currentThread;

  //endregion Data processing flow states

  //region Starting and requesting cancellation
  // - All methods in this region are called from the Worker Looper thread, but reading and writing
  //   states that might be accessed from a different thread are still synchronized.

  @Override
  protected void observableActivated() {
    eventSource.addUpdatable(this);
    maybeStartFlow();
  }

  @Override
  protected void observableDeactivated() {
    eventSource.removeUpdatable(this);
    maybeCancelFlow(deactivationConfig, false);
  }

  @Override
  public void update() {
    maybeCancelFlow(concurrentUpdateConfig, true);
    maybeStartFlow();
  }

  /**
   * Called on the worker looper thread. Starts the data processing flow if it's not running. This
   * also cancels the lazily-executed part of the flow if the run state is "paused at lazy".
   */
  void maybeStartFlow() {
    synchronized (this) {
      if (runState == IDLE || runState == PAUSED_AT_GO_LAZY) {
        runState = RUNNING;
        lastDirectiveIndex = -1; // this could be pointing at the goLazy directive
        restartNeeded = false;
      } else {
        if (runState == CANCEL_REQUESTED) {
          restartNeeded = true;
        }
        return; // flow already running, do not continue.
      }
    }
    intermediateValue = currentValue;
    runFlowFrom(0, false);
  }

  /**
   * Called on the worker looper thread. Depending on the {@code config}, cancels the data
   * processing flow, resets the value, and/or sends the interrupt signal to the thread currently
   * processing a getFrom/mergeIn/transform instruction of the flow.
   *
   * @param scheduleRestart Whether to schedule a restart if a current flow is canceled.
   */
  private void maybeCancelFlow(@RepositoryConfig final int config, final boolean scheduleRestart) {
    synchronized (this) {
      if (runState == RUNNING || runState == PAUSED_AT_GO_TO) {
        restartNeeded = scheduleRestart;
{'fraction_non_alphanumeric': 0.0751918536009445, 'fraction_numerical': 0.002804014167650531, 'mean_word_length': 4.166984369043081, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6440595', 'n_tokens_mistral': 3702, 'n_tokens_neox': 3632, 'n_words': 1203}