You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by dv...@apache.org on 2013/02/28 19:19:55 UTC
svn commit: r1451273 - in /pig/branches/branch-0.11: CHANGES.txt
src/org/apache/pig/data/InternalSortedBag.java
Author: dvryaboy
Date: Thu Feb 28 18:19:55 2013
New Revision: 1451273
URL: http://svn.apache.org/r1451273
Log:
PIG-3212: Race Conditions in POSort and (Internal)SortedBag during Proactive Spill
Modified:
pig/branches/branch-0.11/CHANGES.txt
pig/branches/branch-0.11/src/org/apache/pig/data/InternalSortedBag.java
Modified: pig/branches/branch-0.11/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.11/CHANGES.txt?rev=1451273&r1=1451272&r2=1451273&view=diff
==============================================================================
--- pig/branches/branch-0.11/CHANGES.txt (original)
+++ pig/branches/branch-0.11/CHANGES.txt Thu Feb 28 18:19:55 2013
@@ -28,6 +28,8 @@ OPTIMIZATIONS
BUG FIXES
+PIG-3212: Race Conditions in POSort and (Internal)SortedBag during Proactive Spill (kadeng via dvryaboy)
+
PIG-3206: HBaseStorage does not work with Oozie pig action and secure HBase (rohini)
Modified: pig/branches/branch-0.11/src/org/apache/pig/data/InternalSortedBag.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.11/src/org/apache/pig/data/InternalSortedBag.java?rev=1451273&r1=1451272&r2=1451273&view=diff
==============================================================================
--- pig/branches/branch-0.11/src/org/apache/pig/data/InternalSortedBag.java (original)
+++ pig/branches/branch-0.11/src/org/apache/pig/data/InternalSortedBag.java Thu Feb 28 18:19:55 2013
@@ -110,25 +110,27 @@ public class InternalSortedBag extends S
@Override
public void add(Tuple t) {
- if(mReadStarted) {
- throw new IllegalStateException("InternalSortedBag is closed for adding new tuples");
- }
-
- if (mContents.size() > memLimit.getCacheLimit()) {
- proactive_spill(mComp);
+ synchronized(mContents) {
+ if(mReadStarted) {
+ throw new IllegalStateException("InternalSortedBag is closed for adding new tuples");
+ }
+
+ if (mContents.size() > memLimit.getCacheLimit()) {
+ proactive_spill(mComp);
+ }
+
+ mContents.add(t);
+
+ // check how many tuples memory can hold by getting average
+ // size of first 100 tuples
+ if(mSize < 100 && (mSpillFiles == null || mSpillFiles.isEmpty())&&t!=null)
+ {
+ memLimit.addNewObjSize(t.getMemorySize());
+ }
+
+ mSize++;
+ markSpillableIfNecessary();
}
-
- mContents.add(t);
-
- // check how many tuples memory can hold by getting average
- // size of first 100 tuples
- if(mSize < 100 && (mSpillFiles == null || mSpillFiles.isEmpty())&&t!=null)
- {
- memLimit.addNewObjSize(t.getMemorySize());
- }
-
- mSize++;
- markSpillableIfNecessary();
}
@Override
@@ -192,12 +194,14 @@ public class InternalSortedBag extends S
private int mCntr = 0;
SortedDataBagIterator() {
- // If this is the first read, we need to sort the data.
- if (!mReadStarted) {
- preMerge();
- Collections.sort((ArrayList<Tuple>)mContents, mComp);
- mReadStarted = true;
- }
+ // If this is the first read, we need to sort the data.
+ synchronized(mContents) {
+ if (!mReadStarted) {
+ preMerge();
+ Collections.sort((ArrayList<Tuple>)mContents, mComp);
+ mReadStarted = true;
+ }
+ }
}
@Override
@@ -436,7 +440,17 @@ public class InternalSortedBag extends S
@Override
public long spill(){
- return proactive_spill(mComp);
+ return proactive_spill(mComp);
}
+ @Override
+ public long proactive_spill(Comparator<Tuple> comp) {
+ synchronized(mContents) {
+ if (this.mReadStarted) {
+ return 0L;
+ }
+ return super.proactive_spill(comp);
+ }
+ }
+
}