@@ -0,0 +1,3130 @@
+package hex.deeplearning;%0A%0Aimport water.DRemoteTask;%0Aimport water.fvec.Frame;%0A%0A/**%0A * DRemoteTask-based Deep Learning.%0A * Every node has access to all the training data -%3E leads to optimal CPU utilization IF the data fits on every node.%0A */%0Apublic class DeepLearningTask2 extends DRemoteTask%3CDeepLearningTask2%3E %7B%0A  /**%0A   * Construct a DeepLearningTask2 where every node trains on the entire training dataset%0A   * @param train Frame containing training data%0A   * @param model_info Initial DeepLearningModelInfo (weights + biases)%0A   * @param sync_fraction Fraction of the training data to use for one SGD iteration%0A   */%0A  public DeepLearningTask2(Frame train, DeepLearningModel.DeepLearningModelInfo model_info, float sync_fraction) %7B%0A    assert(sync_fraction %3E 0 && sync_fraction %3C= 1.);%0A    _fr = train;%0A    _model_info = model_info;%0A    _sync_fraction = sync_fraction;%0A  %7D%0A%0A  /**%0A   * Returns the aggregated DeepLearning model that was trained by all nodes (over all the training data)%0A   * @return model_info object%0A   */%0A  public DeepLearningModel.DeepLearningModelInfo model_info() %7B%0A    return _res.model_info();%0A  %7D%0A%0A  final private Frame _fr;%0A  final private DeepLearningModel.DeepLearningModelInfo _model_info;%0A  final private float _sync_fraction;%0A  private DeepLearningTask _res;%0A%0A  /**%0A   * Do the local computation: Perform one DeepLearningTask (with run_local=true) iteration.%0A   * Pass over all the data (will be replicated in dfork() here), and use _sync_fraction random rows.%0A   * This calls DeepLearningTask's reduce() between worker threads that update the same local model_info via Hogwild!%0A   * Once the computation is done, reduce() will be called%0A   */%0A  @Override%0A  public void lcompute() %7B%0A    _res = new DeepLearningTask(_model_info, _sync_fraction);%0A    _res.setCompleter(this);%0A    _res.dfork(0, _fr, true /*run_local*/);%0A  %7D%0A%0A  /**%0A   * Reduce between worker ndoes, with network traffic (if %3E1 nodes)%0A   * After all reduce()'s are done, postGlobal() will be called%0A   * @param drt%0A   */%0A  @Override%0A  public void reduce(DeepLearningTask2 drt) %7B%0A    if (_res == null) %7B%0A      //in single-mode operation: assign result to me%0A      _res = drt._res;%0A    %7D else %7B%0A      //distributed reduction between nodes%0A      assert(drt._res._output != _res._output);%0A      _res._chunk_node_count += drt._res._chunk_node_count;%0A      _res._output.add(drt._res._output); //add models, but don't average yet%0A    %7D%0A    assert(_res._output.get_params().replicate_training_data);%0A  %7D%0A%0A  /**%0A   * Finish up the work after all nodes have reduced their models via the above reduce() method.%0A   * All we do is average the models and add to the global training sample counter.%0A   * After this returns, model_info() can be queried for the updated model.%0A   */%0A  @Override%0A  protected void postGlobal() %7B%0A    super.postGlobal();%0A    assert(_res._output.get_params().replicate_training_data);%0A    _res._output.div(_res._chunk_node_count); //model averaging%0A    _res._output.add_processed_global(_res._output.get_processed_local()); //switch from local counters to global counters%0A    _res._output.set_processed_local(0l);%0A  %7D%0A%0A%7D%0A

Add DeepLearningTask2 for distribute training on a replicated training dataset. This can improve the speed of DeepLearning training on small data.