Coverage Report - org.apache.giraph.comm.requests.SendPartitionCurrentMessagesRequest
 
Classes in this File Line Coverage Branch Coverage Complexity
SendPartitionCurrentMessagesRequest
0%
0/20
N/A
0
 
 1  
 /*
 2  
  * Licensed to the Apache Software Foundation (ASF) under one
 3  
  * or more contributor license agreements.  See the NOTICE file
 4  
  * distributed with this work for additional information
 5  
  * regarding copyright ownership.  The ASF licenses this file
 6  
  * to you under the Apache License, Version 2.0 (the
 7  
  * "License"); you may not use this file except in compliance
 8  
  * with the License.  You may obtain a copy of the License at
 9  
  *
 10  
  *     http://www.apache.org/licenses/LICENSE-2.0
 11  
  *
 12  
  * Unless required by applicable law or agreed to in writing, software
 13  
  * distributed under the License is distributed on an "AS IS" BASIS,
 14  
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 15  
  * See the License for the specific language governing permissions and
 16  
  * limitations under the License.
 17  
  */
 18  
 
 19  
 package org.apache.giraph.comm.requests;
 20  
 
 21  
 import java.io.DataInput;
 22  
 import java.io.DataOutput;
 23  
 import java.io.IOException;
 24  
 
 25  
 import org.apache.giraph.comm.ServerData;
 26  
 import org.apache.giraph.utils.ByteArrayVertexIdMessages;
 27  
 import org.apache.hadoop.io.Writable;
 28  
 import org.apache.hadoop.io.WritableComparable;
 29  
 
 30  
 /**
 31  
  * Send a collection of vertex messages for a partition. It adds messages to
 32  
  * current message store and it should be used only during partition exchange.
 33  
  *
 34  
  * @param <I> Vertex id
 35  
  * @param <V> Vertex data
 36  
  * @param <E> Edge data
 37  
  * @param <M> Message data
 38  
  */
 39  
 public class SendPartitionCurrentMessagesRequest<I extends WritableComparable,
 40  
   V extends Writable, E extends Writable, M extends Writable> extends
 41  
   WritableRequest<I, V, E> implements WorkerRequest<I, V, E> {
 42  
   /** Destination partition for these vertices' messages*/
 43  
   private int partitionId;
 44  
   /** Map of destination vertex ID's to message lists */
 45  
   private ByteArrayVertexIdMessages<I, M> vertexIdMessageMap;
 46  
 
 47  
   /** Constructor used for reflection only */
 48  0
   public SendPartitionCurrentMessagesRequest() { }
 49  
 
 50  
   /**
 51  
    * Constructor used to send request.
 52  
    *
 53  
    * @param partitionId Partition to send the request to
 54  
    * @param vertexIdMessages Map of messages to send
 55  
    */
 56  
   public SendPartitionCurrentMessagesRequest(int partitionId,
 57  
     ByteArrayVertexIdMessages<I, M> vertexIdMessages) {
 58  0
     super();
 59  0
     this.partitionId = partitionId;
 60  0
     this.vertexIdMessageMap = vertexIdMessages;
 61  0
   }
 62  
 
 63  
   @Override
 64  
   public RequestType getType() {
 65  0
     return RequestType.SEND_PARTITION_CURRENT_MESSAGES_REQUEST;
 66  
   }
 67  
 
 68  
   @Override
 69  
   public void readFieldsRequest(DataInput input) throws IOException {
 70  0
     partitionId = input.readInt();
 71  
     // At this moment the Computation class have already been replaced with
 72  
     // the new one, and we deal with messages from previous superstep
 73  0
     vertexIdMessageMap = new ByteArrayVertexIdMessages<>(
 74  0
         getConf().<M>createIncomingMessageValueFactory());
 75  0
     vertexIdMessageMap.setConf(getConf());
 76  0
     vertexIdMessageMap.initialize();
 77  0
     vertexIdMessageMap.readFields(input);
 78  0
   }
 79  
 
 80  
   @Override
 81  
   public void writeRequest(DataOutput output) throws IOException {
 82  0
     output.writeInt(partitionId);
 83  0
     vertexIdMessageMap.write(output);
 84  0
   }
 85  
 
 86  
   @Override
 87  
   public void doRequest(ServerData<I, V, E> serverData) {
 88  0
     serverData.<M>getCurrentMessageStore().addPartitionMessages(partitionId,
 89  
         vertexIdMessageMap);
 90  0
   }
 91  
 
 92  
   @Override
 93  
   public int getSerializedSize() {
 94  0
     return super.getSerializedSize() + 4 +
 95  0
         vertexIdMessageMap.getSerializedSize();
 96  
   }
 97  
 }