<commit_before>// Copyright 2011 Google Inc. All Rights Reserved.
package com.google.appengine.tools.mapreduce;

import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.concurrent.Executors.callable;

import com.google.appengine.api.datastore.CommittedButStillApplyingException;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.DatastoreTimeoutException;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityTranslator;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.tools.cloudstorage.ExceptionHandler;
import com.google.appengine.tools.cloudstorage.RetryHelper;
import com.google.appengine.tools.cloudstorage.RetryParams;
import com.google.apphosting.api.ApiProxy.ApiProxyException;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;

import java.io.Serializable;
import java.util.Collection;
import java.util.ConcurrentModificationException;

/**
 * DatastoreMutationPool allows you to pool datastore operations such that they
 * are applied in batches, requiring fewer datastore API calls.  Mutations are
 * accumulated until they reach a count limit on the number of unflushed
 * mutations, until they reach a size limit on the byte count of unflushed
 * mutations, or until a manual flush is requested.
 *
 */
public class DatastoreMutationPool {

  public static final int DEFAULT_COUNT_LIMIT = 100;
  public static final int DEFAULT_BYTES_LIMIT = 256 * 1024;
  private static final ExceptionHandler EXCEPTION_HANDLER = new ExceptionHandler.Builder()
      .retryOn(ApiProxyException.class, ConcurrentModificationException.class,
          CommittedButStillApplyingException.class, DatastoreTimeoutException.class)
      .build();
  public static final Params DEFAULT_PARAMS = new Params.Builder().build();

  private final Params params;
  private final DatastoreService ds;
  private final Collection<Entity> puts = Lists.newArrayList();
  private int putsBytes;
  private final Collection<Key> deletes = Lists.newArrayList();
  private int deletesBytes;

  public static class Params implements Serializable {

    private static final long serialVersionUID = -4072996713626072011L;

    private final RetryParams retryParams;
    private final int countLimit;
    private final int bytesLimit;

    private Params(Builder builder) {
      retryParams = builder.retryParams;
      countLimit = builder.countLimit;
      bytesLimit = builder.bytesLimit;
    }

    public RetryParams getRetryParams() {
      return retryParams;
    }

    public int getCountLimit() {
      return countLimit;
    }

    public int getBytesLimit() {
      return bytesLimit;
    }

    public static class Builder {

      private int bytesLimit = DEFAULT_BYTES_LIMIT;
      private int countLimit = DEFAULT_COUNT_LIMIT;
      private RetryParams retryParams = RetryParams.getDefaultInstance();

      public Builder bytesLimit(int bytesLimit) {
        this.bytesLimit = bytesLimit;
        return this;
      }

      public Builder countLimit(int countLimit) {
        this.countLimit = countLimit;
        return this;
      }

      public Builder retryParams(RetryParams retryParams) {
        this.retryParams = checkNotNull(retryParams);
        return this;
      }

      public Params build() {
        return new Params(this);
      }
    }
  }

  private DatastoreMutationPool(DatastoreService ds, Params params) {
    this.ds = ds;
    this.params = params;
  }

  private static class Listener extends LifecycleListener {
    // Will never actually be serialized since it removes itself from the
    // registry on endSlice().  Declaring serialVersionUID only to avoid
    // compiler warning.
    private static final long serialVersionUID = 121172329066475329L;

    private DatastoreMutationPool pool;
    private final LifecycleListenerRegistry registry;

    Listener(DatastoreMutationPool pool, LifecycleListenerRegistry registry) {
      this.pool = checkNotNull(pool, "Null pool");
      this.registry = checkNotNull(registry, "Null registry");
    }

    @Override public void endSlice() {
      Preconditions.checkState(pool != null, "%s: endSlice() called twice?", this);
      registry.removeListener(this);
      try {
        pool.flush();
      } finally {
        pool = null;
      }
    }
  }

  public static DatastoreMutationPool forManualFlushing(DatastoreService ds, Params params) {
    return new DatastoreMutationPool(ds, params);
  }

  public static DatastoreMutationPool forManualFlushing(
      DatastoreService ds, int countLimit, int bytesLimit) {
    Params.Builder paramBuilder = new Params.Builder();
    paramBuilder.countLimit(countLimit);
    paramBuilder.bytesLimit(bytesLimit);
    return forManualFlushing(ds, paramBuilder.build());
  }

  public static DatastoreMutationPool forManualFlushing() {
    return forManualFlushing(
        DatastoreServiceFactory.getDatastoreService(), DEFAULT_COUNT_LIMIT, DEFAULT_BYTES_LIMIT);
  }

  public static DatastoreMutationPool forRegistry(
      LifecycleListenerRegistry registry, DatastoreService ds, int countLimit, int bytesLimit) {
    DatastoreMutationPool pool = forManualFlushing(ds, countLimit, bytesLimit);
    registry.addListener(new Listener(pool, registry));
    return pool;
  }

  public static DatastoreMutationPool forRegistry(LifecycleListenerRegistry registry) {
    return forRegistry(registry,
        DatastoreServiceFactory.getDatastoreService(), DEFAULT_COUNT_LIMIT, DEFAULT_BYTES_LIMIT);
  }

  public static DatastoreMutationPool forWorker(Worker<?> worker,
      DatastoreService ds, int countLimit, int bytesLimit) {
    return forRegistry(worker.getLifecycleListenerRegistry(), ds, countLimit, bytesLimit);
  }

  public static DatastoreMutationPool forWorker(Worker<?> worker) {
    return forRegistry(worker.getLifecycleListenerRegistry());
  }

  /**
   * Adds a mutation to put the given entity to the datastore.
   */
  public void delete(Key key) {
    // This is probably a serious overestimation, but I can't see a good
    // way to find the size in the public API.
    int bytesHere = KeyFactory.keyToString(key).length();

    // Do this before the add so that we guarantee that size is never > sizeLimit
    if (deletesBytes + bytesHere >= params.getBytesLimit()) {
      flushDeletes();
    }

    deletesBytes += bytesHere;
    deletes.add(key);
<commit_msg>Java: Remove DatastoreMutationPool#forWorker and #forRegistry methods.<commit_after>// Copyright 2011 Google Inc. All Rights Reserved.
package com.google.appengine.tools.mapreduce;

import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.concurrent.Executors.callable;

import com.google.appengine.api.datastore.CommittedButStillApplyingException;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.DatastoreTimeoutException;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityTranslator;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.tools.cloudstorage.ExceptionHandler;
import com.google.appengine.tools.cloudstorage.RetryHelper;
import com.google.appengine.tools.cloudstorage.RetryParams;
import com.google.apphosting.api.ApiProxy.ApiProxyException;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;

import java.io.Serializable;
import java.util.Collection;
import java.util.ConcurrentModificationException;

/**
 * DatastoreMutationPool allows you to pool datastore operations such that they
 * are applied in batches, requiring fewer datastore API calls.  Mutations are
 * accumulated until they reach a count limit on the number of unflushed
 * mutations, until they reach a size limit on the byte count of unflushed
 * mutations, or until a manual flush is requested.
 *
 */
public class DatastoreMutationPool {

  public static final int DEFAULT_COUNT_LIMIT = 100;
  public static final int DEFAULT_BYTES_LIMIT = 256 * 1024;
  private static final ExceptionHandler EXCEPTION_HANDLER = new ExceptionHandler.Builder()
      .retryOn(ApiProxyException.class, ConcurrentModificationException.class,
          CommittedButStillApplyingException.class, DatastoreTimeoutException.class)
      .build();
  public static final Params DEFAULT_PARAMS = new Params.Builder().build();

  private final Params params;
  private final DatastoreService ds;
  private final Collection<Entity> puts = Lists.newArrayList();
  private int putsBytes;
  private final Collection<Key> deletes = Lists.newArrayList();
  private int deletesBytes;

  public static class Params implements Serializable {

    private static final long serialVersionUID = -4072996713626072011L;

    private final RetryParams retryParams;
    private final int countLimit;
    private final int bytesLimit;

    private Params(Builder builder) {
      retryParams = builder.retryParams;
      countLimit = builder.countLimit;
      bytesLimit = builder.bytesLimit;
    }

    public RetryParams getRetryParams() {
      return retryParams;
    }

    public int getCountLimit() {
      return countLimit;
    }

    public int getBytesLimit() {
      return bytesLimit;
    }

    public static class Builder {

      private int bytesLimit = DEFAULT_BYTES_LIMIT;
      private int countLimit = DEFAULT_COUNT_LIMIT;
      private RetryParams retryParams = RetryParams.getDefaultInstance();

      public Builder bytesLimit(int bytesLimit) {
        this.bytesLimit = bytesLimit;
        return this;
      }

      public Builder countLimit(int countLimit) {
        this.countLimit = countLimit;
        return this;
      }

      public Builder retryParams(RetryParams retryParams) {
        this.retryParams = checkNotNull(retryParams);
        return this;
      }

      public Params build() {
        return new Params(this);
      }
    }
  }

  private DatastoreMutationPool(DatastoreService ds, Params params) {
    this.ds = ds;
    this.params = params;
  }

  private static class Listener extends LifecycleListener {
    // Will never actually be serialized since it removes itself from the
    // registry on endSlice().  Declaring serialVersionUID only to avoid
    // compiler warning.
    private static final long serialVersionUID = 121172329066475329L;

    private DatastoreMutationPool pool;
    private final LifecycleListenerRegistry registry;

    Listener(DatastoreMutationPool pool, LifecycleListenerRegistry registry) {
      this.pool = checkNotNull(pool, "Null pool");
      this.registry = checkNotNull(registry, "Null registry");
    }

    @Override public void endSlice() {
      Preconditions.checkState(pool != null, "%s: endSlice() called twice?", this);
      registry.removeListener(this);
      try {
        pool.flush();
      } finally {
        pool = null;
      }
    }
  }

  public static DatastoreMutationPool forManualFlushing(DatastoreService ds, Params params) {
    return new DatastoreMutationPool(ds, params);
  }

  public static DatastoreMutationPool forManualFlushing(
      DatastoreService ds, int countLimit, int bytesLimit) {
    Params.Builder paramBuilder = new Params.Builder();
    paramBuilder.countLimit(countLimit);
    paramBuilder.bytesLimit(bytesLimit);
    return forManualFlushing(ds, paramBuilder.build());
  }

  public static DatastoreMutationPool forManualFlushing() {
    return forManualFlushing(
        DatastoreServiceFactory.getDatastoreService(), DEFAULT_COUNT_LIMIT, DEFAULT_BYTES_LIMIT);
  }

  /**
   * Adds a mutation to put the given entity to the datastore.
   */
  public void delete(Key key) {
    // This is probably a serious overestimation, but I can't see a good
    // way to find the size in the public API.
    int bytesHere = KeyFactory.keyToString(key).length();

    // Do this before the add so that we guarantee that size is never > sizeLimit
    if (deletesBytes + bytesHere >= params.getBytesLimit()) {
      flushDeletes();
    }

    deletesBytes += bytesHere;
    deletes.add(key);

{'fraction_non_alphanumeric': 0.07641357027463651, 'fraction_numerical': 0.008239095315024232, 'mean_word_length': 3.9385719984044676, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, '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': '7324521', 'n_tokens_mistral': 3517, 'n_tokens_neox': 3383, 'n_words': 930}