client/src/main/java/org/apache/crail/CrailBufferedOutputStream.java - incubator-crail - Git at Google

 /*
  *
  * 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.crail;

 import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.util.LinkedList;
 import java.util.concurrent.Future;

 import org.apache.crail.conf.CrailConstants;
 import org.apache.crail.utils.CrailImmediateOperation;
 import org.apache.crail.utils.CrailUtils;
 import org.apache.crail.utils.RingBuffer;
 import org.slf4j.Logger;


 public class CrailBufferedOutputStream extends OutputStream {
 	public static final Logger LOG = CrailUtils.getLogger();

 	private CrailStore crailFS;
 	private CrailFile file;
 	private long writeHint;
 	private CrailOutputStream outputStream;
 	private LinkedList<CrailBuffer> originalBuffers;
 	private RingBuffer<CrailBuffer> readySlices;
 	private RingBuffer<CrailBuffer> pendingSlices;
 	private RingBuffer<Future<CrailResult>> pendingFutures;
 	private long position;
 	private boolean open;
 	private CrailBufferedStatistics statistics;
 	private int actualSliceSize;

 	private CrailImmediateOperation noOp;
 	private ByteBuffer tmpBoundaryBuffer;
 	private byte[] tmpByteBuf;

 	CrailBufferedOutputStream(CrailFile file, long writeHint) throws Exception {
 		this.crailFS = file.getFileSystem();
 		this.file = file;
 		this.writeHint = writeHint;
 		this.outputStream = null;
 		this.statistics = new CrailBufferedStatistics("buffered/out");

 		int allocationSize = Math.max(CrailConstants.BUFFER_SIZE, CrailConstants.SLICE_SIZE);
 		this.actualSliceSize = Math.min(CrailConstants.BUFFER_SIZE, CrailConstants.SLICE_SIZE);
 		int sliceCount = allocationSize / actualSliceSize;
 		this.originalBuffers = new LinkedList<>();
 		this.readySlices = new RingBuffer<>(sliceCount);
 		this.pendingSlices = new RingBuffer<>(sliceCount);
 		this.pendingFutures = new RingBuffer<>(sliceCount);

 		for (int currentSize = 0; currentSize < allocationSize; currentSize += CrailConstants.BUFFER_SIZE){
 			CrailBuffer buffer = crailFS.allocateBuffer();
 			originalBuffers.add(buffer);
 		}
 		for (CrailBuffer buffer : originalBuffers){
 			while(buffer.hasRemaining()){
 				buffer.limit(buffer.position() + actualSliceSize);
 				CrailBuffer slice = buffer.slice();
 				slice.clear();
 				readySlices.add(slice);

 				int newpos = buffer.position() + actualSliceSize;
 				buffer.clear();
 				buffer.position(newpos);
 			}
 		}
 		/* adjust first slice remaining to align position to slice size */
 		CrailBuffer firstSlice = readySlices.peek();
 		long streamPosition = outputStream().position();
 		firstSlice.limit(firstSlice.remaining() - (int)(streamPosition % firstSlice.remaining()));

 		this.tmpByteBuf = new byte[1];
 		this.tmpBoundaryBuffer = ByteBuffer.allocate(8);
 		this.noOp = new CrailImmediateOperation(0);
 		this.position = 0;
 		this.open = true;
 	}

 	public final void write(int dataBuf) throws IOException {
 		tmpByteBuf[0] = (byte) dataBuf;
 		this.write(tmpByteBuf);
 	}

 	public final void write(byte[] dataBuf) throws IOException {
 		this.write(dataBuf, 0, dataBuf.length);
 	}

 	public final void write(byte[] dataBuf, int off, int len) throws IOException {
 		try {
 			if (dataBuf == null) {
 				throw new NullPointerException();
 			} else if ((off < 0) || (off > dataBuf.length) || (len < 0) || ((off + len) > dataBuf.length) || ((off + len) < 0)) {
 				throw new IndexOutOfBoundsException();
 			} else if (!open) {
 				throw new IOException("stream closed");
 			} else if (len == 0) {
 				return;
 			}

 			while (len > 0){
 				CrailBuffer slice = getSlice();
 				int bufferRemaining = Math.min(len, slice.remaining());
 				slice.put(dataBuf, off, bufferRemaining);
 				off += bufferRemaining;
 				len -= bufferRemaining;
 				position += bufferRemaining;
 				syncSlice();
 			}
 		} catch (Exception e) {
 			throw new IOException(e);
 		}
 	}

 	public final void write(ByteBuffer dataBuf) throws IOException {
 		try {
 			if (dataBuf == null) {
 				throw new NullPointerException();
 			} else if (!open) {
 				throw new IOException("stream closed");
 			} else if (dataBuf.remaining() == 0) {
 				return;
 			}

 			int len = dataBuf.remaining();
 			while (len > 0) {
 				CrailBuffer slice = getSlice();
 				int bufferRemaining = Math.min(len, slice.remaining());
 				int oldLimit = dataBuf.limit();
 				dataBuf.limit(dataBuf.position() + bufferRemaining);
 				slice.put(dataBuf);
 				dataBuf.limit(oldLimit);
 				len -= bufferRemaining;
 				position += bufferRemaining;
 				syncSlice();
 			}
 		} catch (Exception e) {
 			throw new IOException(e);
 		}
 	}

 	public final void writeDouble(double value) throws Exception {
 		CrailBuffer slice = getSlice();
 		if (slice.remaining() >= Double.BYTES){
 			slice.putDouble(value);
 			syncSlice();
 			position += Double.BYTES;
 		} else {
 			tmpBoundaryBuffer.clear();
 			tmpBoundaryBuffer.putDouble(value);
 			tmpBoundaryBuffer.flip();
 			write(tmpBoundaryBuffer);
 		}
 	}

 	public final void writeFloat(float value) throws Exception {
 		CrailBuffer slice = getSlice();
 		if (slice.remaining() >= Float.BYTES){
 			slice.putFloat(value);
 			syncSlice();
 			position += Float.BYTES;
 		} else {
 			tmpBoundaryBuffer.clear();
 			tmpBoundaryBuffer.putFloat(value);
 			tmpBoundaryBuffer.flip();
 			write(tmpBoundaryBuffer);
 		}
 	}

 	public final void writeInt(int value) throws Exception {
 		CrailBuffer slice = getSlice();
 		if (slice.remaining() >= Integer.BYTES){
 			slice.putInt(value);
 			syncSlice();
 			position += Integer.BYTES;
 		} else {
 			tmpBoundaryBuffer.clear();
 			tmpBoundaryBuffer.putInt(value);
 			tmpBoundaryBuffer.flip();
 			write(tmpBoundaryBuffer);
 		}
 	}

 	public final void writeLong(long value) throws Exception {
 		CrailBuffer slice = getSlice();
 		if (slice.remaining() >= Long.BYTES){
 			slice.putLong(value);
 			syncSlice();
 			position += Long.BYTES;
 		} else {
 			tmpBoundaryBuffer.clear();
 			tmpBoundaryBuffer.putLong(value);
 			tmpBoundaryBuffer.flip();
 			write(tmpBoundaryBuffer);
 		}
 	}

 	public final void writeShort(short value) throws Exception {
 		CrailBuffer slice = getSlice();
 		if (slice.remaining() >= Short.BYTES){
 			slice.putShort(value);
 			syncSlice();
 			position += Short.BYTES;
 		} else {
 			tmpBoundaryBuffer.clear();
 			tmpBoundaryBuffer.putShort(value);
 			tmpBoundaryBuffer.flip();
 			write(tmpBoundaryBuffer);
 		}
 	}

 	public Future<CrailResult> purge() throws IOException {
 		if (!open) {
 			throw new IOException("stream closed");
 		}

 		try {
 			while(!readySlices.isEmpty()){
 				CrailBuffer slice = readySlices.poll();
 				if (slice.position() > 0){
 					slice.flip();
 					Future<CrailResult> future = outputStream().write(slice);
 					pendingSlices.add(slice);
 					pendingFutures.add(future);
 				}
 			}

 			if (pendingFutures.isEmpty()){
 				return noOp;
 			} else {
 				CrailPurgeOperation purgeOp = new CrailPurgeOperation();
 				while(!pendingFutures.isEmpty()){
 					Future<CrailResult> future = pendingFutures.poll();
 					purgeOp.add(future);
 				}
 				return purgeOp;
 			}
 		} catch (Exception e) {
 			throw new IOException(e);
 		}
 	}

 	public Future<Void> sync() throws IOException {
 		Future<Void> future = outputStream().sync();
 		return future;
 	}

 	public void close() throws IOException {
 		try {
 			if (!open){
 				return;
 			}

 			while(!readySlices.isEmpty()){
 				CrailBuffer slice = readySlices.poll();
 				if (slice.position() > 0){
 					slice.flip();
 					Future<CrailResult> future = outputStream().write(slice);
 					pendingSlices.add(slice);
 					pendingFutures.add(future);
 				}
 			}

 			while(!pendingFutures.isEmpty()){
 				Future<CrailResult> future = pendingFutures.poll();
 				future.get();
 			}

 			while(!originalBuffers.isEmpty()){
 				CrailBuffer buffer = originalBuffers.remove();
 				crailFS.freeBuffer(buffer);
 			}

 			outputStream().close();
 			this.crailFS.getStatistics().addProvider(statistics);
 			open = false;
 		} catch (Exception e) {
 			throw new IOException(e);
 		}
 	}

 	public long position() {
 		return position;
 	}

 	@Override
 	public void flush() throws IOException {
 		//flush should not be called, use purge instead!
 	}

 	public CrailNode getFile() throws IOException {
 		return outputStream().getFile();
 	}

 	private CrailBuffer getSlice() throws Exception {
 		CrailBuffer slice = readySlices.peek();
 		if (slice == null){
 			Future<CrailResult> future = pendingFutures.poll();
 			statistics.incTotalOps();
 			if (future.isDone()){
 				statistics.incNonBlockingOps();
 			} else {
 				statistics.incBlockingOps();
 			}
 			future.get();
 			slice = pendingSlices.poll();
 			slice.clear();
 			readySlices.add(slice);
 		}
 		return slice;
 	}

 	private void syncSlice() throws Exception {
 		CrailBuffer slice = readySlices.peek();
 		if (slice != null && slice.remaining() == 0){
 			slice = readySlices.poll();
 			slice.flip();
 			Future<CrailResult> future = outputStream().write(slice);
 			pendingSlices.add(slice);
 			pendingFutures.add(future);
 		}
 	}

 	final CrailOutputStream outputStream() throws IOException {
 		if (outputStream == null){
 			try {
 				outputStream = file.getDirectOutputStream(writeHint);
 			} catch(Exception e){
 				throw new IOException(e);
 			}
 		}
 		return outputStream;
 	}

 }
	/*
	*
	* 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.crail;

	import java.io.IOException;
	import java.io.OutputStream;
	import java.nio.ByteBuffer;
	import java.util.LinkedList;
	import java.util.concurrent.Future;

	import org.apache.crail.conf.CrailConstants;
	import org.apache.crail.utils.CrailImmediateOperation;
	import org.apache.crail.utils.CrailUtils;
	import org.apache.crail.utils.RingBuffer;
	import org.slf4j.Logger;


	public class CrailBufferedOutputStream extends OutputStream {
	public static final Logger LOG = CrailUtils.getLogger();

	private CrailStore crailFS;
	private CrailFile file;
	private long writeHint;
	private CrailOutputStream outputStream;
	private LinkedList<CrailBuffer> originalBuffers;
	private RingBuffer<CrailBuffer> readySlices;
	private RingBuffer<CrailBuffer> pendingSlices;
	private RingBuffer<Future<CrailResult>> pendingFutures;
	private long position;
	private boolean open;
	private CrailBufferedStatistics statistics;
	private int actualSliceSize;

	private CrailImmediateOperation noOp;
	private ByteBuffer tmpBoundaryBuffer;
	private byte[] tmpByteBuf;

	CrailBufferedOutputStream(CrailFile file, long writeHint) throws Exception {
	this.crailFS = file.getFileSystem();
	this.file = file;
	this.writeHint = writeHint;
	this.outputStream = null;
	this.statistics = new CrailBufferedStatistics("buffered/out");

	int allocationSize = Math.max(CrailConstants.BUFFER_SIZE, CrailConstants.SLICE_SIZE);
	this.actualSliceSize = Math.min(CrailConstants.BUFFER_SIZE, CrailConstants.SLICE_SIZE);
	int sliceCount = allocationSize / actualSliceSize;
	this.originalBuffers = new LinkedList<>();
	this.readySlices = new RingBuffer<>(sliceCount);
	this.pendingSlices = new RingBuffer<>(sliceCount);
	this.pendingFutures = new RingBuffer<>(sliceCount);

	for (int currentSize = 0; currentSize < allocationSize; currentSize += CrailConstants.BUFFER_SIZE){
	CrailBuffer buffer = crailFS.allocateBuffer();
	originalBuffers.add(buffer);
	}
	for (CrailBuffer buffer : originalBuffers){
	while(buffer.hasRemaining()){
	buffer.limit(buffer.position() + actualSliceSize);
	CrailBuffer slice = buffer.slice();
	slice.clear();
	readySlices.add(slice);

	int newpos = buffer.position() + actualSliceSize;
	buffer.clear();
	buffer.position(newpos);
	}
	}
	/* adjust first slice remaining to align position to slice size */
	CrailBuffer firstSlice = readySlices.peek();
	long streamPosition = outputStream().position();
	firstSlice.limit(firstSlice.remaining() - (int)(streamPosition % firstSlice.remaining()));

	this.tmpByteBuf = new byte[1];
	this.tmpBoundaryBuffer = ByteBuffer.allocate(8);
	this.noOp = new CrailImmediateOperation(0);
	this.position = 0;
	this.open = true;
	}

	public final void write(int dataBuf) throws IOException {
	tmpByteBuf[0] = (byte) dataBuf;
	this.write(tmpByteBuf);
	}

	public final void write(byte[] dataBuf) throws IOException {
	this.write(dataBuf, 0, dataBuf.length);
	}

	public final void write(byte[] dataBuf, int off, int len) throws IOException {
	try {
	if (dataBuf == null) {
	throw new NullPointerException();
	} else if ((off < 0) \|\| (off > dataBuf.length) \|\| (len < 0) \|\| ((off + len) > dataBuf.length) \|\| ((off + len) < 0)) {
	throw new IndexOutOfBoundsException();
	} else if (!open) {
	throw new IOException("stream closed");
	} else if (len == 0) {
	return;
	}

	while (len > 0){
	CrailBuffer slice = getSlice();
	int bufferRemaining = Math.min(len, slice.remaining());
	slice.put(dataBuf, off, bufferRemaining);
	off += bufferRemaining;
	len -= bufferRemaining;
	position += bufferRemaining;
	syncSlice();
	}
	} catch (Exception e) {
	throw new IOException(e);
	}
	}

	public final void write(ByteBuffer dataBuf) throws IOException {
	try {
	if (dataBuf == null) {
	throw new NullPointerException();
	} else if (!open) {
	throw new IOException("stream closed");
	} else if (dataBuf.remaining() == 0) {
	return;
	}

	int len = dataBuf.remaining();
	while (len > 0) {
	CrailBuffer slice = getSlice();
	int bufferRemaining = Math.min(len, slice.remaining());
	int oldLimit = dataBuf.limit();
	dataBuf.limit(dataBuf.position() + bufferRemaining);
	slice.put(dataBuf);
	dataBuf.limit(oldLimit);
	len -= bufferRemaining;
	position += bufferRemaining;
	syncSlice();
	}
	} catch (Exception e) {
	throw new IOException(e);
	}
	}

	public final void writeDouble(double value) throws Exception {
	CrailBuffer slice = getSlice();
	if (slice.remaining() >= Double.BYTES){
	slice.putDouble(value);
	syncSlice();
	position += Double.BYTES;
	} else {
	tmpBoundaryBuffer.clear();
	tmpBoundaryBuffer.putDouble(value);
	tmpBoundaryBuffer.flip();
	write(tmpBoundaryBuffer);
	}
	}

	public final void writeFloat(float value) throws Exception {
	CrailBuffer slice = getSlice();
	if (slice.remaining() >= Float.BYTES){
	slice.putFloat(value);
	syncSlice();
	position += Float.BYTES;
	} else {
	tmpBoundaryBuffer.clear();
	tmpBoundaryBuffer.putFloat(value);
	tmpBoundaryBuffer.flip();
	write(tmpBoundaryBuffer);
	}
	}

	public final void writeInt(int value) throws Exception {
	CrailBuffer slice = getSlice();
	if (slice.remaining() >= Integer.BYTES){
	slice.putInt(value);
	syncSlice();
	position += Integer.BYTES;
	} else {
	tmpBoundaryBuffer.clear();
	tmpBoundaryBuffer.putInt(value);
	tmpBoundaryBuffer.flip();
	write(tmpBoundaryBuffer);
	}
	}

	public final void writeLong(long value) throws Exception {
	CrailBuffer slice = getSlice();
	if (slice.remaining() >= Long.BYTES){
	slice.putLong(value);
	syncSlice();
	position += Long.BYTES;
	} else {
	tmpBoundaryBuffer.clear();
	tmpBoundaryBuffer.putLong(value);
	tmpBoundaryBuffer.flip();
	write(tmpBoundaryBuffer);
	}
	}

	public final void writeShort(short value) throws Exception {
	CrailBuffer slice = getSlice();
	if (slice.remaining() >= Short.BYTES){
	slice.putShort(value);
	syncSlice();
	position += Short.BYTES;
	} else {
	tmpBoundaryBuffer.clear();
	tmpBoundaryBuffer.putShort(value);
	tmpBoundaryBuffer.flip();
	write(tmpBoundaryBuffer);
	}
	}

	public Future<CrailResult> purge() throws IOException {
	if (!open) {
	throw new IOException("stream closed");
	}

	try {
	while(!readySlices.isEmpty()){
	CrailBuffer slice = readySlices.poll();
	if (slice.position() > 0){
	slice.flip();
	Future<CrailResult> future = outputStream().write(slice);
	pendingSlices.add(slice);
	pendingFutures.add(future);
	}
	}

	if (pendingFutures.isEmpty()){
	return noOp;
	} else {
	CrailPurgeOperation purgeOp = new CrailPurgeOperation();
	while(!pendingFutures.isEmpty()){
	Future<CrailResult> future = pendingFutures.poll();
	purgeOp.add(future);
	}
	return purgeOp;
	}
	} catch (Exception e) {
	throw new IOException(e);
	}
	}

	public Future<Void> sync() throws IOException {
	Future<Void> future = outputStream().sync();
	return future;
	}

	public void close() throws IOException {
	try {
	if (!open){
	return;
	}

	while(!readySlices.isEmpty()){
	CrailBuffer slice = readySlices.poll();
	if (slice.position() > 0){
	slice.flip();
	Future<CrailResult> future = outputStream().write(slice);
	pendingSlices.add(slice);
	pendingFutures.add(future);
	}
	}

	while(!pendingFutures.isEmpty()){
	Future<CrailResult> future = pendingFutures.poll();
	future.get();
	}

	while(!originalBuffers.isEmpty()){
	CrailBuffer buffer = originalBuffers.remove();
	crailFS.freeBuffer(buffer);
	}

	outputStream().close();
	this.crailFS.getStatistics().addProvider(statistics);
	open = false;
	} catch (Exception e) {
	throw new IOException(e);
	}
	}

	public long position() {
	return position;
	}

	@Override
	public void flush() throws IOException {
	//flush should not be called, use purge instead!
	}

	public CrailNode getFile() throws IOException {
	return outputStream().getFile();
	}

	private CrailBuffer getSlice() throws Exception {
	CrailBuffer slice = readySlices.peek();
	if (slice == null){
	Future<CrailResult> future = pendingFutures.poll();
	statistics.incTotalOps();
	if (future.isDone()){
	statistics.incNonBlockingOps();
	} else {
	statistics.incBlockingOps();
	}
	future.get();
	slice = pendingSlices.poll();
	slice.clear();
	readySlices.add(slice);
	}
	return slice;
	}

	private void syncSlice() throws Exception {
	CrailBuffer slice = readySlices.peek();
	if (slice != null && slice.remaining() == 0){
	slice = readySlices.poll();
	slice.flip();
	Future<CrailResult> future = outputStream().write(slice);
	pendingSlices.add(slice);
	pendingFutures.add(future);
	}
	}

	final CrailOutputStream outputStream() throws IOException {
	if (outputStream == null){
	try {
	outputStream = file.getDirectOutputStream(writeHint);
	} catch(Exception e){
	throw new IOException(e);
	}
	}
	return outputStream;
	}

	}