hadoop-hdfs-project/hadoop-hdfs/src/contrib/libhdfs3/src/client/LocalBlockReader.cc - hadoop - 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.
  */

 #include "BigEndian.h"
 #include "datatransfer.pb.h"
 #include "Exception.h"
 #include "ExceptionInternal.h"
 #include "HWCrc32c.h"
 #include "LocalBlockReader.h"
 #include "SWCrc32c.h"

 #include "hdfs.pb.h"

 #include <inttypes.h>
 #include <limits>

 #define BMVERSION 1
 #define BMVERSION_SIZE 2

 #define HEADER_SIZE (BMVERSION_SIZE + \
       CHECKSUM_TYPE_SIZE + CHECKSUM_BYTES_PER_CHECKSUM_SIZE)

 using hadoop::hdfs::ChecksumTypeProto;

 namespace hdfs {
 namespace internal {

 LocalBlockReader::LocalBlockReader(const BlockLocalPathInfo &info,
         const ExtendedBlock &block, int64_t offset, bool verify,
         SessionConfig &conf, std::vector<char> &buffer) :
     verify(verify), pbuffer(NULL), pMetaBuffer(NULL), block(block),
       checksumSize(0), chunkSize(0), position(0), size(0), cursor(
         0), length(block.getNumBytes()),
           dataFilePath(info.getLocalBlockPath()), metaFilePath(
             info.getLocalMetaPath()), buffer(buffer) {
     exception_ptr lastError;

     try {
         if (conf.doUseMappedFile()) {
             metaFd = shared_ptr<MappedFileWrapper>(new MappedFileWrapper);
             dataFd = shared_ptr<MappedFileWrapper>(new MappedFileWrapper);
         } else {
             metaFd = shared_ptr<CFileWrapper>(new CFileWrapper);
             dataFd = shared_ptr<CFileWrapper>(new CFileWrapper);
         }

         if (!metaFd->open(metaFilePath)) {
             THROW(HdfsIOException,
                   "LocalBlockReader cannot open metadata file \"%s\", %s",
                   metaFilePath.c_str(), GetSystemErrorInfo(errno));
         }

         std::vector<char> header;
         pMetaBuffer = metaFd->read(header, HEADER_SIZE);
         int16_t version = ReadBigEndian16FromArray(&pMetaBuffer[0]);

         if (BMVERSION != version) {
             THROW(HdfsIOException,
                   "LocalBlockReader get an unmatched block, expected block "
                   "version %d, real version is %d",
                   BMVERSION, static_cast<int>(version));
         }

         switch (pMetaBuffer[BMVERSION_SIZE]) {
         case ChecksumTypeProto::CHECKSUM_NULL:
             this->verify = false;
             checksumSize = 0;
             metaFd.reset();
             break;

         case ChecksumTypeProto::CHECKSUM_CRC32:
             THROW(HdfsIOException,
                   "LocalBlockReader does not support CRC32 checksum.");
             break;

         case ChecksumTypeProto::CHECKSUM_CRC32C:
             if (HWCrc32c::available()) {
                 checksum = shared_ptr<Checksum>(new HWCrc32c());
             } else {
                 checksum = shared_ptr<Checksum>(new SWCrc32c());
             }

             chunkSize = ReadBigEndian32FromArray(
                             &pMetaBuffer[BMVERSION_SIZE + CHECKSUM_TYPE_SIZE]);
             checksumSize = sizeof(int32_t);
             break;

         default:
             THROW(HdfsIOException,
                   "LocalBlockReader cannot recognize checksum type: %d.",
                   static_cast<int>(pMetaBuffer[BMVERSION_SIZE]));
         }

         if (verify && chunkSize <= 0) {
             THROW(HdfsIOException,
                   "LocalBlockReader get an invalid checksum parameter, "
                   "bytes per chunk: %d.",
                   chunkSize);
         }

         if (!dataFd->open(dataFilePath)) {
             THROW(HdfsIOException,
                   "LocalBlockReader cannot open data file \"%s\", %s",
                   dataFilePath.c_str(), GetSystemErrorInfo(errno));
         }

         localBufferSize = conf.getLocalReadBufferSize();

         if (verify) {
             localBufferSize = (localBufferSize + chunkSize - 1) /
                 (chunkSize * chunkSize);
         }

         if (offset > 0) {
             skip(offset);
         }
     } catch (...) {
         if (metaFd) {
             metaFd->close();
         }

         if (dataFd) {
             dataFd->close();
         }

         lastError = current_exception();
     }

     try {
         if (lastError != exception_ptr()) {
             rethrow_exception(lastError);
         }
     } catch (const HdfsCanceled &e) {
         throw;
     } catch (const HdfsException &e) {
         NESTED_THROW(HdfsIOException,
                  "Failed to construct LocalBlockReader for block: %s.",
                  block.toString().c_str());
     }
 }

 LocalBlockReader::~LocalBlockReader() {
 }

 void LocalBlockReader::readAndVerify(int32_t bufferSize) {
     assert(true == verify);
     assert(cursor % chunkSize == 0);
     int chunks = (bufferSize + chunkSize - 1) / chunkSize;
     pbuffer = dataFd->read(buffer, bufferSize);
     pMetaBuffer = metaFd->read(metaBuffer, chunks * checksumSize);

     for (int i = 0; i < chunks; ++i) {
         checksum->reset();
         int chunk = chunkSize;

         if (chunkSize * (i + 1) > bufferSize) {
             chunk = bufferSize % chunkSize;
         }

         checksum->update(&pbuffer[i * chunkSize], chunk);
         uint32_t target = ReadBigEndian32FromArray(
                               &pMetaBuffer[i * checksumSize]);

         if (target != checksum->getValue()) {
             THROW(ChecksumException,
                   "LocalBlockReader checksum not match for block file: %s",
                   dataFilePath.c_str());
         }
     }
 }

 int32_t LocalBlockReader::readInternal(char * buf, int32_t len) {
     int32_t todo = len;

     /*
      * read from buffer.
      */
     if (position < size) {
         todo = todo < size - position ? todo : size - position;
         memcpy(buf, &pbuffer[position], todo);
         position += todo;
         cursor += todo;
         return todo;
     }

     /*
      * end of block
      */
     todo = todo < length - cursor ? todo : length - cursor;

     if (0 == todo) {
         return 0;
     }

     /*
      * bypass the buffer
      */
     if (!verify
             && (todo > localBufferSize || todo == length - cursor)) {
         dataFd->copy(buf, todo);
         cursor += todo;
         return todo;
     }

     /*
      * fill buffer.
      */
     int bufferSize = localBufferSize;
     bufferSize = bufferSize < length - cursor ? bufferSize : length - cursor;
     assert(bufferSize > 0);

     if (verify) {
         readAndVerify(bufferSize);
     } else {
         pbuffer = dataFd->read(buffer, bufferSize);
     }

     position = 0;
     size = bufferSize;
     assert(position < size);
     return readInternal(buf, todo);
 }

 int32_t LocalBlockReader::read(char *buf, int32_t size) {
     try {
         return readInternal(buf, size);
     } catch (const HdfsCanceled &e) {
         throw;
     } catch (const HdfsException &e) {
         NESTED_THROW(HdfsIOException,
                      "LocalBlockReader failed to read from position: %" PRId64
                      ", length: %d, block: %s.",
                      cursor, size, block.toString().c_str());
     }

     assert(!"cannot reach here");
     return 0;
 }

 void LocalBlockReader::skip(int64_t len) {
     assert(len < length - cursor);

     try {
         int64_t todo = len;

         while (todo > 0) {
             /*
              * skip the data in buffer.
              */
             if (size - position > 0) {
                 int batch = todo < size - position ? todo : size - position;
                 position += batch;
                 todo -= batch;
                 cursor += batch;
                 continue;
             }

             if (verify) {
                 int64_t lastChunkSize = (cursor + todo) % chunkSize;
                 cursor = (cursor + todo) / chunkSize * chunkSize;
                 int64_t metaCursor = HEADER_SIZE
                                      + checksumSize * (cursor / chunkSize);
                 metaFd->seek(metaCursor);
                 todo = lastChunkSize;
             } else {
                 cursor += todo;
                 todo = 0;
             }

             if (cursor > 0) {
                 dataFd->seek(cursor);
             }

             /*
              * fill buffer again and verify checksum
              */
             if (todo > 0) {
                 assert(true == verify);
                 int bufferSize = localBufferSize;
                 bufferSize =
                     bufferSize < length - cursor ?
                     bufferSize : length - cursor;
                 readAndVerify(bufferSize);
                 position = 0;
                 size = bufferSize;
             }
         }
     } catch (const HdfsCanceled &e) {
         throw;
     } catch (const HdfsException &e) {
         NESTED_THROW(HdfsIOException,
                      "LocalBlockReader failed to skip from position: %" PRId64
                      ", length: %d, block: %s.",
                      cursor, size, block.toString().c_str());
     }
 }

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

	#include "BigEndian.h"
	#include "datatransfer.pb.h"
	#include "Exception.h"
	#include "ExceptionInternal.h"
	#include "HWCrc32c.h"
	#include "LocalBlockReader.h"
	#include "SWCrc32c.h"

	#include "hdfs.pb.h"

	#include <inttypes.h>
	#include <limits>

	#define BMVERSION 1
	#define BMVERSION_SIZE 2

	#define HEADER_SIZE (BMVERSION_SIZE + \
	CHECKSUM_TYPE_SIZE + CHECKSUM_BYTES_PER_CHECKSUM_SIZE)

	using hadoop::hdfs::ChecksumTypeProto;

	namespace hdfs {
	namespace internal {

	LocalBlockReader::LocalBlockReader(const BlockLocalPathInfo &info,
	const ExtendedBlock &block, int64_t offset, bool verify,
	SessionConfig &conf, std::vector<char> &buffer) :
	verify(verify), pbuffer(NULL), pMetaBuffer(NULL), block(block),
	checksumSize(0), chunkSize(0), position(0), size(0), cursor(
	0), length(block.getNumBytes()),
	dataFilePath(info.getLocalBlockPath()), metaFilePath(
	info.getLocalMetaPath()), buffer(buffer) {
	exception_ptr lastError;

	try {
	if (conf.doUseMappedFile()) {
	metaFd = shared_ptr<MappedFileWrapper>(new MappedFileWrapper);
	dataFd = shared_ptr<MappedFileWrapper>(new MappedFileWrapper);
	} else {
	metaFd = shared_ptr<CFileWrapper>(new CFileWrapper);
	dataFd = shared_ptr<CFileWrapper>(new CFileWrapper);
	}

	if (!metaFd->open(metaFilePath)) {
	THROW(HdfsIOException,
	"LocalBlockReader cannot open metadata file \"%s\", %s",
	metaFilePath.c_str(), GetSystemErrorInfo(errno));
	}

	std::vector<char> header;
	pMetaBuffer = metaFd->read(header, HEADER_SIZE);
	int16_t version = ReadBigEndian16FromArray(&pMetaBuffer[0]);

	if (BMVERSION != version) {
	THROW(HdfsIOException,
	"LocalBlockReader get an unmatched block, expected block "
	"version %d, real version is %d",
	BMVERSION, static_cast<int>(version));
	}

	switch (pMetaBuffer[BMVERSION_SIZE]) {
	case ChecksumTypeProto::CHECKSUM_NULL:
	this->verify = false;
	checksumSize = 0;
	metaFd.reset();
	break;

	case ChecksumTypeProto::CHECKSUM_CRC32:
	THROW(HdfsIOException,
	"LocalBlockReader does not support CRC32 checksum.");
	break;

	case ChecksumTypeProto::CHECKSUM_CRC32C:
	if (HWCrc32c::available()) {
	checksum = shared_ptr<Checksum>(new HWCrc32c());
	} else {
	checksum = shared_ptr<Checksum>(new SWCrc32c());
	}

	chunkSize = ReadBigEndian32FromArray(
	&pMetaBuffer[BMVERSION_SIZE + CHECKSUM_TYPE_SIZE]);
	checksumSize = sizeof(int32_t);
	break;

	default:
	THROW(HdfsIOException,
	"LocalBlockReader cannot recognize checksum type: %d.",
	static_cast<int>(pMetaBuffer[BMVERSION_SIZE]));
	}

	if (verify && chunkSize <= 0) {
	THROW(HdfsIOException,
	"LocalBlockReader get an invalid checksum parameter, "
	"bytes per chunk: %d.",
	chunkSize);
	}

	if (!dataFd->open(dataFilePath)) {
	THROW(HdfsIOException,
	"LocalBlockReader cannot open data file \"%s\", %s",
	dataFilePath.c_str(), GetSystemErrorInfo(errno));
	}

	localBufferSize = conf.getLocalReadBufferSize();

	if (verify) {
	localBufferSize = (localBufferSize + chunkSize - 1) /
	(chunkSize * chunkSize);
	}

	if (offset > 0) {
	skip(offset);
	}
	} catch (...) {
	if (metaFd) {
	metaFd->close();
	}

	if (dataFd) {
	dataFd->close();
	}

	lastError = current_exception();
	}

	try {
	if (lastError != exception_ptr()) {
	rethrow_exception(lastError);
	}
	} catch (const HdfsCanceled &e) {
	throw;
	} catch (const HdfsException &e) {
	NESTED_THROW(HdfsIOException,
	"Failed to construct LocalBlockReader for block: %s.",
	block.toString().c_str());
	}
	}

	LocalBlockReader::~LocalBlockReader() {
	}

	void LocalBlockReader::readAndVerify(int32_t bufferSize) {
	assert(true == verify);
	assert(cursor % chunkSize == 0);
	int chunks = (bufferSize + chunkSize - 1) / chunkSize;
	pbuffer = dataFd->read(buffer, bufferSize);
	pMetaBuffer = metaFd->read(metaBuffer, chunks * checksumSize);

	for (int i = 0; i < chunks; ++i) {
	checksum->reset();
	int chunk = chunkSize;

	if (chunkSize * (i + 1) > bufferSize) {
	chunk = bufferSize % chunkSize;
	}

	checksum->update(&pbuffer[i * chunkSize], chunk);
	uint32_t target = ReadBigEndian32FromArray(
	&pMetaBuffer[i * checksumSize]);

	if (target != checksum->getValue()) {
	THROW(ChecksumException,
	"LocalBlockReader checksum not match for block file: %s",
	dataFilePath.c_str());
	}
	}
	}

	int32_t LocalBlockReader::readInternal(char * buf, int32_t len) {
	int32_t todo = len;

	/*
	* read from buffer.
	*/
	if (position < size) {
	todo = todo < size - position ? todo : size - position;
	memcpy(buf, &pbuffer[position], todo);
	position += todo;
	cursor += todo;
	return todo;
	}

	/*
	* end of block
	*/
	todo = todo < length - cursor ? todo : length - cursor;

	if (0 == todo) {
	return 0;
	}

	/*
	* bypass the buffer
	*/
	if (!verify
	&& (todo > localBufferSize \|\| todo == length - cursor)) {
	dataFd->copy(buf, todo);
	cursor += todo;
	return todo;
	}

	/*
	* fill buffer.
	*/
	int bufferSize = localBufferSize;
	bufferSize = bufferSize < length - cursor ? bufferSize : length - cursor;
	assert(bufferSize > 0);

	if (verify) {
	readAndVerify(bufferSize);
	} else {
	pbuffer = dataFd->read(buffer, bufferSize);
	}

	position = 0;
	size = bufferSize;
	assert(position < size);
	return readInternal(buf, todo);
	}

	int32_t LocalBlockReader::read(char *buf, int32_t size) {
	try {
	return readInternal(buf, size);
	} catch (const HdfsCanceled &e) {
	throw;
	} catch (const HdfsException &e) {
	NESTED_THROW(HdfsIOException,
	"LocalBlockReader failed to read from position: %" PRId64
	", length: %d, block: %s.",
	cursor, size, block.toString().c_str());
	}

	assert(!"cannot reach here");
	return 0;
	}

	void LocalBlockReader::skip(int64_t len) {
	assert(len < length - cursor);

	try {
	int64_t todo = len;

	while (todo > 0) {
	/*
	* skip the data in buffer.
	*/
	if (size - position > 0) {
	int batch = todo < size - position ? todo : size - position;
	position += batch;
	todo -= batch;
	cursor += batch;
	continue;
	}

	if (verify) {
	int64_t lastChunkSize = (cursor + todo) % chunkSize;
	cursor = (cursor + todo) / chunkSize * chunkSize;
	int64_t metaCursor = HEADER_SIZE
	+ checksumSize * (cursor / chunkSize);
	metaFd->seek(metaCursor);
	todo = lastChunkSize;
	} else {
	cursor += todo;
	todo = 0;
	}

	if (cursor > 0) {
	dataFd->seek(cursor);
	}

	/*
	* fill buffer again and verify checksum
	*/
	if (todo > 0) {
	assert(true == verify);
	int bufferSize = localBufferSize;
	bufferSize =
	bufferSize < length - cursor ?
	bufferSize : length - cursor;
	readAndVerify(bufferSize);
	position = 0;
	size = bufferSize;
	}
	}
	} catch (const HdfsCanceled &e) {
	throw;
	} catch (const HdfsException &e) {
	NESTED_THROW(HdfsIOException,
	"LocalBlockReader failed to skip from position: %" PRId64
	", length: %d, block: %s.",
	cursor, size, block.toString().c_str());
	}
	}

	}
	}