Updated Branches: refs/heads/trunk 28f4d1b09 -> 6b54593af
GIRAPH-821: proper handling of NegativeArraySizeException for all ByteArray backed messagestores (pavanka via majakabiljo) Project: http://git-wip-us.apache.org/repos/asf/giraph/repo Commit: http://git-wip-us.apache.org/repos/asf/giraph/commit/6b54593a Tree: http://git-wip-us.apache.org/repos/asf/giraph/tree/6b54593a Diff: http://git-wip-us.apache.org/repos/asf/giraph/diff/6b54593a Branch: refs/heads/trunk Commit: 6b54593af9156e346c40e6fafeb7668a7ac4ae6e Parents: 28f4d1b Author: Maja Kabiljo <[email protected]> Authored: Fri Jan 10 10:51:52 2014 -0800 Committer: Maja Kabiljo <[email protected]> Committed: Fri Jan 10 10:51:52 2014 -0800 ---------------------------------------------------------------------- CHANGELOG | 3 + .../ByteArrayMessagesPerVertexStore.java | 5 +- .../primitives/IntByteArrayMessageStore.java | 4 +- .../primitives/LongByteArrayMessageStore.java | 4 +- .../giraph/utils/ByteArrayVertexIdMessages.java | 8 +++ .../utils/VerboseByteArrayMessageWrite.java | 60 ++++++++++++++++++++ 6 files changed, 80 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/giraph/blob/6b54593a/CHANGELOG ---------------------------------------------------------------------- diff --git a/CHANGELOG b/CHANGELOG index 16a8777..a14c80b 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,6 +1,9 @@ Giraph Change Log Release 1.1.0 - unreleased + GIRAPH-821: proper handling of NegativeArraySizeException for all ByteArray backed + messagestores (pavanka via majakabiljo) + GIRAPH-820: add a configuration option to skip creating source vertices present only in edge input (pavanka via majakabiljo) http://git-wip-us.apache.org/repos/asf/giraph/blob/6b54593a/giraph-core/src/main/java/org/apache/giraph/comm/messages/ByteArrayMessagesPerVertexStore.java ---------------------------------------------------------------------- diff --git a/giraph-core/src/main/java/org/apache/giraph/comm/messages/ByteArrayMessagesPerVertexStore.java b/giraph-core/src/main/java/org/apache/giraph/comm/messages/ByteArrayMessagesPerVertexStore.java index d9ef449..2381078 100644 --- a/giraph-core/src/main/java/org/apache/giraph/comm/messages/ByteArrayMessagesPerVertexStore.java +++ b/giraph-core/src/main/java/org/apache/giraph/comm/messages/ByteArrayMessagesPerVertexStore.java @@ -25,6 +25,7 @@ import org.apache.giraph.utils.ByteArrayVertexIdMessages; import org.apache.giraph.utils.ExtendedDataInput; import org.apache.giraph.utils.RepresentativeByteArrayIterator; import org.apache.giraph.utils.VertexIdIterator; +import org.apache.giraph.utils.VerboseByteArrayMessageWrite; import org.apache.giraph.utils.io.DataInputOutput; import org.apache.hadoop.io.Writable; import org.apache.hadoop.io.WritableComparable; @@ -118,8 +119,8 @@ public class ByteArrayMessagesPerVertexStore<I extends WritableComparable, getDataInputOutput(partitionMap, vertexIdMessageIterator); synchronized (dataInputOutput) { - vertexIdMessageIterator.getCurrentMessage().write( - dataInputOutput.getDataOutput()); + VerboseByteArrayMessageWrite.verboseWriteCurrentMessage( + vertexIdMessageIterator, dataInputOutput.getDataOutput()); } } } http://git-wip-us.apache.org/repos/asf/giraph/blob/6b54593a/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/IntByteArrayMessageStore.java ---------------------------------------------------------------------- diff --git a/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/IntByteArrayMessageStore.java b/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/IntByteArrayMessageStore.java index c58868a..095c529 100644 --- a/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/IntByteArrayMessageStore.java +++ b/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/IntByteArrayMessageStore.java @@ -26,6 +26,7 @@ import org.apache.giraph.factories.MessageValueFactory; import org.apache.giraph.partition.Partition; import org.apache.giraph.utils.ByteArrayVertexIdMessages; import org.apache.giraph.utils.EmptyIterable; +import org.apache.giraph.utils.VerboseByteArrayMessageWrite; import org.apache.giraph.utils.io.DataInputOutput; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Writable; @@ -148,7 +149,8 @@ public class IntByteArrayMessageStore<M extends Writable> iterator.next(); DataInputOutput dataInputOutput = getDataInputOutput(partitionMap, iterator.getCurrentVertexId().get()); - iterator.getCurrentMessage().write(dataInputOutput.getDataOutput()); + VerboseByteArrayMessageWrite.verboseWriteCurrentMessage(iterator, + dataInputOutput.getDataOutput()); } } } http://git-wip-us.apache.org/repos/asf/giraph/blob/6b54593a/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/LongByteArrayMessageStore.java ---------------------------------------------------------------------- diff --git a/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/LongByteArrayMessageStore.java b/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/LongByteArrayMessageStore.java index b0a613b..a8aebe9 100644 --- a/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/LongByteArrayMessageStore.java +++ b/giraph-core/src/main/java/org/apache/giraph/comm/messages/primitives/LongByteArrayMessageStore.java @@ -26,6 +26,7 @@ import org.apache.giraph.factories.MessageValueFactory; import org.apache.giraph.partition.Partition; import org.apache.giraph.utils.ByteArrayVertexIdMessages; import org.apache.giraph.utils.EmptyIterable; +import org.apache.giraph.utils.VerboseByteArrayMessageWrite; import org.apache.giraph.utils.io.DataInputOutput; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Writable; @@ -149,7 +150,8 @@ public class LongByteArrayMessageStore<M extends Writable> iterator.next(); DataInputOutput dataInputOutput = getDataInputOutput(partitionMap, iterator.getCurrentVertexId().get()); - iterator.getCurrentMessage().write(dataInputOutput.getDataOutput()); + VerboseByteArrayMessageWrite.verboseWriteCurrentMessage(iterator, + dataInputOutput.getDataOutput()); } } } http://git-wip-us.apache.org/repos/asf/giraph/blob/6b54593a/giraph-core/src/main/java/org/apache/giraph/utils/ByteArrayVertexIdMessages.java ---------------------------------------------------------------------- diff --git a/giraph-core/src/main/java/org/apache/giraph/utils/ByteArrayVertexIdMessages.java b/giraph-core/src/main/java/org/apache/giraph/utils/ByteArrayVertexIdMessages.java index 4958ae3..0ac8fdf 100644 --- a/giraph-core/src/main/java/org/apache/giraph/utils/ByteArrayVertexIdMessages.java +++ b/giraph-core/src/main/java/org/apache/giraph/utils/ByteArrayVertexIdMessages.java @@ -189,6 +189,14 @@ public class ByteArrayVertexIdMessages<I extends WritableComparable, public void writeCurrentMessageBytes(DataOutput dataOutput) { try { dataOutput.write(getByteArray(), messageOffset, messageBytes); + } catch (NegativeArraySizeException e) { + throw new RuntimeException("The numbers of bytes sent to vertex " + + vertexId + " exceeded the max capacity of " + + "its ExtendedDataOutput. Please consider setting " + + "giraph.useBigDataIOForMessages=true. If there are super-vertices" + + " in the graph which receive a lot of messages (total serialized " + + "size of messages goes beyond the maximum size of a byte array), " + + "setting this option to true will remove that limit"); } catch (IOException e) { throw new IllegalStateException("writeCurrentMessageBytes: Got " + "IOException", e); http://git-wip-us.apache.org/repos/asf/giraph/blob/6b54593a/giraph-core/src/main/java/org/apache/giraph/utils/VerboseByteArrayMessageWrite.java ---------------------------------------------------------------------- diff --git a/giraph-core/src/main/java/org/apache/giraph/utils/VerboseByteArrayMessageWrite.java b/giraph-core/src/main/java/org/apache/giraph/utils/VerboseByteArrayMessageWrite.java new file mode 100644 index 0000000..8673732 --- /dev/null +++ b/giraph-core/src/main/java/org/apache/giraph/utils/VerboseByteArrayMessageWrite.java @@ -0,0 +1,60 @@ +/* + * 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.utils; + +import org.apache.hadoop.io.Writable; +import org.apache.hadoop.io.WritableComparable; + +import java.io.DataOutput; +import java.io.IOException; + +/** Verbose Error mesage for ByteArray based messages */ +public class VerboseByteArrayMessageWrite { + /** Do not construct */ + protected VerboseByteArrayMessageWrite() { + } + + /** + * verboseWriteCurrentMessage + * de-serialize, then write messages + * + * @param iterator iterator + * @param out DataOutput + * @param <I> vertexId + * @param <M> message + * @throws IOException + * @throws RuntimeException + */ + public static <I extends WritableComparable, M extends Writable> void + verboseWriteCurrentMessage( + ByteArrayVertexIdMessages<I, M>.VertexIdMessageIterator + iterator, DataOutput out) throws IOException { + try { + iterator.getCurrentMessage().write(out); + } catch (NegativeArraySizeException e) { + throw new RuntimeException("The numbers of bytes sent to vertex " + + iterator.getCurrentVertexId() + " exceeded the max capacity of " + + "its ExtendedDataOutput. Please consider setting " + + "giraph.useBigDataIOForMessages=true. If there are super-vertices" + + " in the graph which receive a lot of messages (total serialized " + + "size of messages goes beyond the maximum size of a byte array), " + + "setting this option to true will remove that limit"); + } + } +}
