View Javadoc

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.messages;
20  
21  import java.io.IOException;
22  import java.util.Iterator;
23  
24  import junit.framework.Assert;
25  
26  import org.apache.giraph.bsp.CentralizedServiceWorker;
27  import org.apache.giraph.combiner.DoubleSumMessageCombiner;
28  import org.apache.giraph.comm.messages.primitives.long_id.LongByteArrayMessageStore;
29  import org.apache.giraph.comm.messages.primitives.LongDoubleMessageStore;
30  import org.apache.giraph.conf.GiraphConfiguration;
31  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
32  import org.apache.giraph.factories.TestMessageValueFactory;
33  import org.apache.giraph.graph.BasicComputation;
34  import org.apache.giraph.graph.Vertex;
35  import org.apache.giraph.partition.Partition;
36  import org.apache.giraph.partition.PartitionStore;
37  import org.apache.giraph.utils.ByteArrayVertexIdMessages;
38  import org.apache.hadoop.io.DoubleWritable;
39  import org.apache.hadoop.io.LongWritable;
40  import org.apache.hadoop.io.NullWritable;
41  import org.apache.hadoop.io.Writable;
42  import org.junit.Before;
43  import org.junit.Test;
44  import org.mockito.Mockito;
45  import org.mockito.invocation.InvocationOnMock;
46  import org.mockito.stubbing.Answer;
47  
48  import com.google.common.collect.Iterables;
49  import com.google.common.collect.Lists;
50  
51  public class TestLongDoublePrimitiveMessageStores {
52    private static final int NUM_PARTITIONS = 2;
53    private static CentralizedServiceWorker<LongWritable, Writable, Writable>
54      service;
55  
56    @Before
57    public void prepare() {
58      service = Mockito.mock(CentralizedServiceWorker.class);
59      Mockito.when(
60          service.getPartitionId(Mockito.any(LongWritable.class))).thenAnswer(
61          new Answer<Integer>() {
62            @Override
63            public Integer answer(InvocationOnMock invocation) {
64              LongWritable vertexId = (LongWritable) invocation.getArguments()[0];
65              return (int) (vertexId.get() % NUM_PARTITIONS);
66            }
67          }
68      );
69      PartitionStore partitionStore = Mockito.mock(PartitionStore.class);
70      Mockito.when(service.getPartitionStore()).thenReturn(partitionStore);
71      Mockito.when(partitionStore.getPartitionIds()).thenReturn(
72          Lists.newArrayList(0, 1));
73      Partition partition = Mockito.mock(Partition.class);
74      Mockito.when(partition.getVertexCount()).thenReturn(Long.valueOf(1));
75      Mockito.when(partitionStore.getNextPartition()).thenReturn(partition);
76      Mockito.when(partitionStore.getNextPartition()).thenReturn(partition);
77    }
78  
79    private static class LongDoubleNoOpComputation extends
80        BasicComputation<LongWritable, NullWritable, NullWritable,
81            DoubleWritable> {
82      @Override
83      public void compute(Vertex<LongWritable, NullWritable, NullWritable> vertex,
84          Iterable<DoubleWritable> messages) throws IOException {
85      }
86    }
87  
88    private static ImmutableClassesGiraphConfiguration<LongWritable, Writable,
89      Writable> createLongDoubleConf() {
90  
91      GiraphConfiguration initConf = new GiraphConfiguration();
92      initConf.setComputationClass(LongDoubleNoOpComputation.class);
93      return new ImmutableClassesGiraphConfiguration(initConf);
94    }
95  
96    private static ByteArrayVertexIdMessages<LongWritable, DoubleWritable>
97    createLongDoubleMessages() {
98      ByteArrayVertexIdMessages<LongWritable, DoubleWritable> messages =
99          new ByteArrayVertexIdMessages<LongWritable, DoubleWritable>(
100             new TestMessageValueFactory<DoubleWritable>(DoubleWritable.class));
101     messages.setConf(createLongDoubleConf());
102     messages.initialize();
103     return messages;
104   }
105 
106   private static void insertLongDoubleMessages(
107       MessageStore<LongWritable, DoubleWritable> messageStore) {
108     ByteArrayVertexIdMessages<LongWritable, DoubleWritable> messages =
109         createLongDoubleMessages();
110     messages.add(new LongWritable(0), new DoubleWritable(1));
111     messages.add(new LongWritable(2), new DoubleWritable(3));
112     messages.add(new LongWritable(0), new DoubleWritable(4));
113     messageStore.addPartitionMessages(0, messages);
114     messages = createLongDoubleMessages();
115     messages.add(new LongWritable(1), new DoubleWritable(1));
116     messages.add(new LongWritable(1), new DoubleWritable(3));
117     messages.add(new LongWritable(1), new DoubleWritable(4));
118     messageStore.addPartitionMessages(1, messages);
119     messages = createLongDoubleMessages();
120     messages.add(new LongWritable(0), new DoubleWritable(5));
121     messageStore.addPartitionMessages(0, messages);
122   }
123 
124   @Test
125   public void testLongDoubleMessageStore() {
126     LongDoubleMessageStore messageStore =
127         new LongDoubleMessageStore(service, new DoubleSumMessageCombiner());
128     insertLongDoubleMessages(messageStore);
129 
130     Iterable<DoubleWritable> m0 =
131         messageStore.getVertexMessages(new LongWritable(0));
132     Assert.assertEquals(1, Iterables.size(m0));
133     Assert.assertEquals(10.0, m0.iterator().next().get());
134     Iterable<DoubleWritable> m1 =
135         messageStore.getVertexMessages(new LongWritable(1));
136     Assert.assertEquals(1, Iterables.size(m1));
137     Assert.assertEquals(8.0, m1.iterator().next().get());
138     Iterable<DoubleWritable> m2 =
139         messageStore.getVertexMessages(new LongWritable(2));
140     Assert.assertEquals(1, Iterables.size(m2));
141     Assert.assertEquals(3.0, m2.iterator().next().get());
142     Assert.assertTrue(
143         Iterables.isEmpty(messageStore.getVertexMessages(new LongWritable(3))));
144   }
145 
146   @Test
147   public void testLongByteArrayMessageStore() {
148     LongByteArrayMessageStore<DoubleWritable> messageStore =
149         new LongByteArrayMessageStore<DoubleWritable>(
150             new TestMessageValueFactory<DoubleWritable>(DoubleWritable.class),
151             service, createLongDoubleConf());
152     insertLongDoubleMessages(messageStore);
153 
154     Iterable<DoubleWritable> m0 =
155         messageStore.getVertexMessages(new LongWritable(0));
156     Assert.assertEquals(3, Iterables.size(m0));
157     Iterator<DoubleWritable> i0 = m0.iterator();
158     Assert.assertEquals(1.0, i0.next().get());
159     Assert.assertEquals(4.0, i0.next().get());
160     Assert.assertEquals(5.0, i0.next().get());
161     Iterable<DoubleWritable> m1 =
162         messageStore.getVertexMessages(new LongWritable(1));
163     Assert.assertEquals(3, Iterables.size(m1));
164     Iterator<DoubleWritable> i1 = m1.iterator();
165     Assert.assertEquals(1.0, i1.next().get());
166     Assert.assertEquals(3.0, i1.next().get());
167     Assert.assertEquals(4.0, i1.next().get());
168     Iterable<DoubleWritable> m2 =
169         messageStore.getVertexMessages(new LongWritable(2));
170     Assert.assertEquals(1, Iterables.size(m2));
171     Assert.assertEquals(3.0, m2.iterator().next().get());
172     Assert.assertTrue(
173         Iterables.isEmpty(messageStore.getVertexMessages(new LongWritable(3))));
174   }
175 }