/*
    * 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.giraph.comm;
  
  import org.apache.giraph.bsp.CentralizedServiceWorker;
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.utils.VertexIdData;
  import org.apache.giraph.worker.WorkerInfo;
  import org.apache.hadoop.io.WritableComparable;
  
  import javax.annotation.concurrent.NotThreadSafe;
  
  /**
   * An abstract structure for caching data indexed by vertex id,
   * to be sent to workers in bulk. Not thread-safe.
   *
   * @param <I> Vertex id
   * @param <T> Data
   * @param <B> Specialization of {@link VertexIdData} for T
   */
  @NotThreadSafe
  @SuppressWarnings("unchecked")
  public abstract class SendVertexIdDataCache<I extends WritableComparable, T,
      B extends VertexIdData<I, T>> extends SendDataCache<B> {
    /**
     * Constructor.
     *
     * @param conf Giraph configuration
     * @param serviceWorker Service worker
     * @param maxRequestSize Maximum request size (in bytes)
     * @param additionalRequestSize Additional request size (expressed as a
     *                              ratio of the average request size)
     */
    public SendVertexIdDataCache(ImmutableClassesGiraphConfiguration conf,
                                 CentralizedServiceWorker<?, ?, ?> serviceWorker,
                                 int maxRequestSize,
                                 float additionalRequestSize) {
      super(conf, serviceWorker, maxRequestSize, additionalRequestSize);
    }
  
    /**
     * Create a new {@link VertexIdData} specialized for the use case.
     *
     * @return A new instance of {@link VertexIdData}
     */
    public abstract B createVertexIdData();
  
    /**
     * Add data to the cache.
     *
     * @param workerInfo the remote worker destination
     * @param partitionId the remote Partition this message belongs to
     * @param destVertexId vertex id that is ultimate destination
     * @param data Data to send to remote worker
     * @return Size of messages for the worker.
     */
    public int addData(WorkerInfo workerInfo,
                       int partitionId, I destVertexId, T data) {
      // Get the data collection
      VertexIdData<I, T> partitionData =
          getPartitionData(workerInfo, partitionId);
      int originalSize = partitionData.getSize();
      partitionData.add(destVertexId, data);
      // Update the size of cached, outgoing data per worker
      return incrDataSize(workerInfo.getTaskId(),
          partitionData.getSize() - originalSize);
    }
  
    /**
     * This method is similar to the method above,
     * but use a serialized id to replace original I type
     * destVertexId.
     *
     * @param workerInfo The remote worker destination
     * @param partitionId The remote Partition this message belongs to
     * @param serializedId The byte array to store the serialized target vertex id
     * @param idPos The length of bytes of serialized id in the byte array above
     * @param data Data to send to remote worker
     * @return The number of bytes added to the target worker
     */
    public int addData(WorkerInfo workerInfo, int partitionId,
                       byte[] serializedId, int idPos, T data) {
      // Get the data collection
     VertexIdData<I, T> partitionData =
         getPartitionData(workerInfo, partitionId);
     int originalSize = partitionData.getSize();
     partitionData.add(serializedId, idPos, data);
     // Update the size of cached, outgoing data per worker
     return incrDataSize(workerInfo.getTaskId(),
         partitionData.getSize() - originalSize);
   }
 
   /**
    * This method tries to get a partition data from the data cache.
    * If null, it will create one.
    *
    * @param workerInfo The remote worker destination
    * @param partitionId The remote Partition this message belongs to
    * @return The partition data in data cache
    */
   private VertexIdData<I, T> getPartitionData(WorkerInfo workerInfo,
                                                        int partitionId) {
     // Get the data collection
     B partitionData = getData(partitionId);
     if (partitionData == null) {
       partitionData = createVertexIdData();
       partitionData.setConf(getConf());
       partitionData.initialize(getInitialBufferSize(workerInfo.getTaskId()));
       setData(partitionId, partitionData);
     }
 
     return partitionData;
   }
 }