flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From uce <...@git.apache.org>
Subject [GitHub] incubator-flink pull request: [FLINK-987] Add seeking to DataOutpu...
Date Tue, 22 Jul 2014 13:12:36 GMT
Github user uce commented on a diff in the pull request:

    https://github.com/apache/incubator-flink/pull/77#discussion_r15226179
  
    --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedOutputView.java
---
    @@ -20,397 +20,233 @@
     package org.apache.flink.runtime.memorymanager;
     
     import java.io.IOException;
    -import java.io.UTFDataFormatException;
    +import java.util.ArrayList;
    +import java.util.List;
    +
    +import com.google.common.base.Preconditions;
     
    -import org.apache.flink.core.memory.DataInputView;
    -import org.apache.flink.core.memory.DataOutputView;
     import org.apache.flink.core.memory.MemorySegment;
     
     
     /**
    - * The base class for all output views that are backed by multiple memory pages. This
base class contains all
    - * encoding methods to write data to a page and detect page boundary crossing. The concrete
sub classes must
    - * implement the methods to collect the current page and provide the next memory page
once the boundary is crossed.
    - * <p>
    - * The paging assumes that all memory segments are of the same size.
    + * The base class for all output views that are backed by multiple memory pages which
are not always kept in memory
    + * but which are supposed to be spilled after being written to. The concrete sub classes
must
    + * implement the methods to collect the current page and provide the next memory page
once the boundary is crossed or
    + * once the underlying buffers are unlocked by a call to {@link #unlock()}.
      */
    -public abstract class AbstractPagedOutputView implements DataOutputView {
    -	
    -	private MemorySegment currentSegment;			// the current memory segment to write to
    -	
    -	protected final int segmentSize;				// the size of the memory segments
    -	
    -	protected final int headerLength;				// the number of bytes to skip at the beginning
of each segment
    -	
    -	private int positionInSegment;					// the offset in the current segment
    -	
    -	private byte[] utfBuffer;						// the reusable array for UTF encodings
    -	
    -	
    +public abstract class AbstractPagedOutputView extends AbstractMemorySegmentOutputView
{
    +
    +	// Can we just return the current segment when we get the next one or do we have to
keep
    +	// them because {@link lock} was called. We need a counter because of nested locking.
    +	protected int lockCount;
    +
    +	// The segments that are currently locked. When {@link lock} is called we lock the current
    +	// segment and all the segments that we "advance" to. Only when {@link unlock} is called
can we return
    +	// all of the segments but the most recent one.
    +	private List<MemorySegment> lockedSegments;
    +
    +	// Index of current segment in list of locked segments. When we are not locked this
is set to 0, because
    +	// then we always only have one segment.
    +	private int currentSegmentIndex;
    +
     	// --------------------------------------------------------------------------------------------
     	//                                    Constructors
     	// --------------------------------------------------------------------------------------------
    -	
    +
     	/**
     	 * Creates a new output view that writes initially to the given initial segment. All
segments in the
     	 * view have to be of the given {@code segmentSize}. A header of length {@code headerLength}
is left
     	 * at the beginning of each segment.
    -	 * 
    +	 *
     	 * @param initialSegment The segment that the view starts writing to.
     	 * @param segmentSize The size of the memory segments.
     	 * @param headerLength The number of bytes to skip at the beginning of each segment
for the header.
     	 */
     	protected AbstractPagedOutputView(MemorySegment initialSegment, int segmentSize, int
headerLength) {
    +		super(initialSegment, segmentSize, headerLength);
     		if (initialSegment == null) {
     			throw new NullPointerException("Initial Segment may not be null");
     		}
    -		this.segmentSize = segmentSize;
    -		this.headerLength = headerLength;
     		this.currentSegment = initialSegment;
     		this.positionInSegment = headerLength;
    +
    +		lockedSegments = new ArrayList<MemorySegment>();
    +
    +		// at the beginning we only have one segment
    +		currentSegmentIndex = 0;
    +		lockCount = 0;
     	}
    -	
    +
     	/**
     	 * @param segmentSize The size of the memory segments.
     	 * @param headerLength The number of bytes to skip at the beginning of each segment
for the header.
     	 */
    -	protected AbstractPagedOutputView(int segmentSize, int headerLength)
    -	{
    -		this.segmentSize = segmentSize;
    -		this.headerLength = headerLength;
    +	protected AbstractPagedOutputView(int segmentSize, int headerLength) {
    +		super(segmentSize, headerLength);
    +
    +		lockedSegments = new ArrayList<MemorySegment>();
    +		currentSegmentIndex = 0;
    +		lockCount = 0;
     	}
    -	
    +
     
     	// --------------------------------------------------------------------------------------------
     	//                                  Page Management
     	// --------------------------------------------------------------------------------------------
    -	
    +
     	/**
    -	 * 
    +	 *
     	 * This method must return a segment. If no more segments are available, it must throw
an
    --- End diff --
    
    I think you formulated the comments like this for subclasses who implement the methods,
but it might be confusing for someone who just reads the (inherited) comment later.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

Mime
View raw message