| /******************************************************************** |
| * 2014 - |
| * open source under Apache License Version 2.0 |
| ********************************************************************/ |
| /** |
| * 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 "DateTime.h" |
| #include "Pipeline.h" |
| #include "Logger.h" |
| #include "Exception.h" |
| #include "ExceptionInternal.h" |
| #include "OutputStreamInter.h" |
| #include "FileSystemInter.h" |
| #include "DataTransferProtocolSender.h" |
| #include "datatransfer.pb.h" |
| |
| #include <inttypes.h> |
| |
| namespace Hdfs { |
| namespace Internal { |
| |
| PipelineImpl::PipelineImpl(bool append, const char * path, const SessionConfig & conf, |
| shared_ptr<FileSystemInter> filesystem, int checksumType, int chunkSize, |
| int replication, int64_t bytesSent, PacketPool & packetPool, shared_ptr<LocatedBlock> lastBlock) : |
| checksumType(checksumType), chunkSize(chunkSize), errorIndex(-1), replication(replication), bytesAcked( |
| bytesSent), bytesSent(bytesSent), packetPool(packetPool), filesystem(filesystem), lastBlock(lastBlock), path( |
| path) { |
| canAddDatanode = conf.canAddDatanode(); |
| blockWriteRetry = conf.getBlockWriteRetry(); |
| connectTimeout = conf.getOutputConnTimeout(); |
| readTimeout = conf.getOutputReadTimeout(); |
| writeTimeout = conf.getOutputWriteTimeout(); |
| clientName = filesystem->getClientName(); |
| |
| if (append) { |
| LOG(DEBUG2, "create pipeline for file %s to append to %s at position %" PRId64, |
| path, lastBlock->toString().c_str(), lastBlock->getNumBytes()); |
| stage = PIPELINE_SETUP_APPEND; |
| assert(lastBlock); |
| nodes = lastBlock->getLocations(); |
| storageIDs = lastBlock->getStorageIDs(); |
| buildForAppendOrRecovery(false); |
| stage = DATA_STREAMING; |
| } else { |
| LOG(DEBUG2, "create pipeline for file %s to write to a new block", path); |
| stage = PIPELINE_SETUP_CREATE; |
| buildForNewBlock(); |
| stage = DATA_STREAMING; |
| } |
| } |
| |
| int PipelineImpl::findNewDatanode(const std::vector<DatanodeInfo> & original) { |
| if (nodes.size() != original.size() + 1) { |
| THROW(HdfsIOException, "Failed to acquire a datanode for block %s from namenode.", |
| lastBlock->toString().c_str()); |
| } |
| |
| for (size_t i = 0; i < nodes.size(); i++) { |
| size_t j = 0; |
| |
| for (; j < original.size() && !(nodes[i] == original[j]); j++) |
| ; |
| |
| if (j == original.size()) { |
| return i; |
| } |
| } |
| |
| THROW(HdfsIOException, "Cannot add new datanode for block %s.", lastBlock->toString().c_str()); |
| } |
| |
| void PipelineImpl::transfer(const ExtendedBlock & blk, const DatanodeInfo & src, |
| const std::vector<DatanodeInfo> & targets, const Token & token) { |
| shared_ptr<Socket> so(new TcpSocketImpl); |
| shared_ptr<BufferedSocketReader> in(new BufferedSocketReaderImpl(*so)); |
| so->connect(src.getIpAddr().c_str(), src.getXferPort(), connectTimeout); |
| DataTransferProtocolSender sender(*so, writeTimeout, src.formatAddress()); |
| sender.transferBlock(blk, token, clientName.c_str(), targets); |
| int size; |
| size = in->readVarint32(readTimeout); |
| std::vector<char> buf(size); |
| in->readFully(&buf[0], size, readTimeout); |
| BlockOpResponseProto resp; |
| |
| if (!resp.ParseFromArray(&buf[0], size)) { |
| THROW(HdfsIOException, "cannot parse datanode response from %s fro block %s.", |
| src.formatAddress().c_str(), lastBlock->toString().c_str()); |
| } |
| |
| if (Status::DT_PROTO_SUCCESS != resp.status()) { |
| THROW(HdfsIOException, "Failed to transfer block to a new datanode %s for block %s.", |
| targets[0].formatAddress().c_str(), |
| lastBlock->toString().c_str()); |
| } |
| } |
| |
| bool PipelineImpl::addDatanodeToPipeline(const std::vector<DatanodeInfo> & excludedNodes) { |
| try { |
| /* |
| * get a new datanode |
| */ |
| std::vector<DatanodeInfo> original = nodes; |
| shared_ptr<LocatedBlock> lb; |
| lb = filesystem->getAdditionalDatanode(path, *lastBlock, nodes, storageIDs, |
| excludedNodes, 1); |
| nodes = lb->getLocations(); |
| storageIDs = lb->getStorageIDs(); |
| |
| /* |
| * failed to add new datanode into pipeline. |
| */ |
| if (original.size() == nodes.size()) { |
| LOG(LOG_ERROR, |
| "Failed to add new datanode into pipeline for block: %s file %s.", |
| lastBlock->toString().c_str(), path.c_str()); |
| } else { |
| /* |
| * find the new datanode |
| */ |
| int d = findNewDatanode(original); |
| /* |
| * in case transfer block fail. |
| */ |
| errorIndex = d; |
| /* |
| * transfer replica |
| */ |
| DatanodeInfo & src = d == 0 ? nodes[1] : nodes[d - 1]; |
| std::vector<DatanodeInfo> targets; |
| targets.push_back(nodes[d]); |
| LOG(INFO, "Replicate block %s from %s to %s for file %s.", lastBlock->toString().c_str(), |
| src.formatAddress().c_str(), targets[0].formatAddress().c_str(), path.c_str()); |
| transfer(*lastBlock, src, targets, lb->getToken()); |
| errorIndex = -1; |
| return true; |
| } |
| } catch (const HdfsCanceled & e) { |
| throw; |
| } catch (const HdfsFileSystemClosed & e) { |
| throw; |
| } catch (const SafeModeException & e) { |
| throw; |
| } catch (const HdfsException & e) { |
| std::string buffer; |
| LOG(LOG_ERROR, |
| "Failed to add a new datanode into pipeline for block: %s file %s.\n%s", |
| lastBlock->toString().c_str(), path.c_str(), GetExceptionDetail(e, buffer)); |
| } |
| |
| return false; |
| } |
| |
| void PipelineImpl::checkPipelineWithReplicas() { |
| if (static_cast<int>(nodes.size()) < replication) { |
| std::stringstream ss; |
| ss.imbue(std::locale::classic()); |
| int size = nodes.size(); |
| |
| for (int i = 0; i < size - 1; ++i) { |
| ss << nodes[i].formatAddress() << ", "; |
| } |
| |
| if (nodes.empty()) { |
| ss << "Empty"; |
| } else { |
| ss << nodes.back().formatAddress(); |
| } |
| |
| LOG(WARNING, |
| "the number of nodes in pipeline is %d [%s], is less than the expected number of replica %d for block %s file %s", |
| static_cast<int>(nodes.size()), ss.str().c_str(), replication, |
| lastBlock->toString().c_str(), path.c_str()); |
| } |
| } |
| |
| void PipelineImpl::buildForAppendOrRecovery(bool recovery) { |
| int64_t gs = 0; |
| int retry = blockWriteRetry; |
| exception_ptr lastException; |
| std::vector<DatanodeInfo> excludedNodes; |
| shared_ptr<LocatedBlock> lb; |
| std::string buffer; |
| |
| do { |
| /* |
| * Remove bad datanode from list of datanodes. |
| * If errorIndex was not set (i.e. appends), then do not remove |
| * any datanodes |
| */ |
| if (errorIndex >= 0) { |
| assert(lastBlock); |
| LOG(LOG_ERROR, "Pipeline: node %s is invalid and removed from pipeline when %s block %s for file %s, stage = %s.", |
| nodes[errorIndex].formatAddress().c_str(), |
| (recovery ? "recovery" : "append to"), lastBlock->toString().c_str(), |
| path.c_str(), StageToString(stage)); |
| excludedNodes.push_back(nodes[errorIndex]); |
| nodes.erase(nodes.begin() + errorIndex); |
| |
| if (!storageIDs.empty()) { |
| storageIDs.erase(storageIDs.begin() + errorIndex); |
| } |
| |
| if (nodes.empty()) { |
| THROW(HdfsIOException, |
| "Build pipeline to %s block %s failed: all datanodes are bad.", |
| (recovery ? "recovery" : "append to"), lastBlock->toString().c_str()); |
| } |
| |
| errorIndex = -1; |
| } |
| |
| try { |
| gs = 0; |
| |
| /* |
| * Check if the number of datanodes in pipeline satisfy the replication requirement, |
| * add new datanode if not |
| */ |
| if (stage != PIPELINE_SETUP_CREATE && stage != PIPELINE_CLOSE |
| && static_cast<int>(nodes.size()) < replication && canAddDatanode) { |
| if (!addDatanodeToPipeline(excludedNodes)) { |
| THROW(HdfsIOException, |
| "Failed to add new datanode into pipeline for block: %s file %s, " |
| "set \"output.replace-datanode-on-failure\" to \"false\" to disable this feature.", |
| lastBlock->toString().c_str(), path.c_str()); |
| } |
| } |
| |
| if (errorIndex >= 0) { |
| continue; |
| } |
| |
| checkPipelineWithReplicas(); |
| /* |
| * Update generation stamp and access token |
| */ |
| lb = filesystem->updateBlockForPipeline(*lastBlock); |
| gs = lb->getGenerationStamp(); |
| /* |
| * Try to build pipeline |
| */ |
| createBlockOutputStream(lb->getToken(), gs, recovery); |
| /* |
| * everything is ok, reset errorIndex. |
| */ |
| errorIndex = -1; |
| lastException = exception_ptr(); |
| break; //break on success |
| } catch (const HdfsInvalidBlockToken & e) { |
| lastException = current_exception(); |
| recovery = true; |
| LOG(LOG_ERROR, |
| "Pipeline: Failed to build pipeline for block %s file %s, new generation stamp is %" PRId64 ",\n%s", |
| lastBlock->toString().c_str(), path.c_str(), gs, GetExceptionDetail(e, buffer)); |
| LOG(INFO, "Try to recovery pipeline for block %s file %s.", |
| lastBlock->toString().c_str(), path.c_str()); |
| } catch (const HdfsTimeoutException & e) { |
| lastException = current_exception(); |
| recovery = true; |
| LOG(LOG_ERROR, |
| "Pipeline: Failed to build pipeline for block %s file %s, new generation stamp is %" PRId64 ",\n%s", |
| lastBlock->toString().c_str(), path.c_str(), gs, GetExceptionDetail(e, buffer)); |
| LOG(INFO, "Try to recovery pipeline for block %s file %s.", |
| lastBlock->toString().c_str(), path.c_str()); |
| } catch (const HdfsIOException & e) { |
| lastException = current_exception(); |
| /* |
| * Set recovery flag to true in case of failed to create a pipeline for appending. |
| */ |
| recovery = true; |
| LOG(LOG_ERROR, |
| "Pipeline: Failed to build pipeline for block %s file %s, new generation stamp is %" PRId64 ",\n%s", |
| lastBlock->toString().c_str(), path.c_str(), gs, GetExceptionDetail(e, buffer)); |
| LOG(INFO, "Try to recovery pipeline for block %s file %s.", lastBlock->toString().c_str(), path.c_str()); |
| } |
| |
| /* |
| * we don't known what happened, no datanode is reported failure, reduce retry count in case infinite loop. |
| * it may caused by rpc call throw HdfsIOException |
| */ |
| if (errorIndex < 0) { |
| --retry; |
| } |
| } while (retry > 0); |
| |
| if (lastException) { |
| rethrow_exception(lastException); |
| } |
| |
| /* |
| * Update pipeline at the namenode, non-idempotent RPC call. |
| */ |
| lb->setPoolId(lastBlock->getPoolId()); |
| lb->setBlockId(lastBlock->getBlockId()); |
| lb->setLocations(nodes); |
| lb->setStorageIDs(storageIDs); |
| lb->setNumBytes(lastBlock->getNumBytes()); |
| lb->setOffset(lastBlock->getOffset()); |
| filesystem->updatePipeline(*lastBlock, *lb, nodes, storageIDs); |
| lastBlock = lb; |
| } |
| |
| void PipelineImpl::locateNextBlock( |
| const std::vector<DatanodeInfo> & excludedNodes) { |
| milliseconds sleeptime(100); |
| milliseconds fiveSeconds(5000); |
| int retry = blockWriteRetry; |
| |
| while (true) { |
| try { |
| lastBlock = filesystem->addBlock(path, lastBlock.get(), |
| excludedNodes); |
| assert(lastBlock); |
| return; |
| } catch (const NotReplicatedYetException & e) { |
| LOG(DEBUG1, "Got NotReplicatedYetException when try to addBlock for block %s, " |
| "already retry %d times, max retry %d times", lastBlock->toString().c_str(), |
| blockWriteRetry - retry, blockWriteRetry); |
| |
| if (retry--) { |
| try { |
| sleep_for(sleeptime); |
| } catch (...) { |
| } |
| |
| sleeptime *= 2; |
| sleeptime = sleeptime < fiveSeconds ? sleeptime : fiveSeconds; |
| } else { |
| throw; |
| } |
| } |
| } |
| } |
| |
| static std::string FormatExcludedNodes( |
| const std::vector<DatanodeInfo> & excludedNodes) { |
| std::stringstream ss; |
| ss.imbue(std::locale::classic()); |
| ss << "["; |
| int size = excludedNodes.size(); |
| |
| for (int i = 0; i < size - 1; ++i) { |
| ss << excludedNodes[i].formatAddress() << ", "; |
| } |
| |
| if (excludedNodes.empty()) { |
| ss << "Empty"; |
| } else { |
| ss << excludedNodes.back().formatAddress(); |
| } |
| |
| ss << "]"; |
| return ss.str(); |
| } |
| |
| void PipelineImpl::buildForNewBlock() { |
| int retryAllocNewBlock = 0, retry = blockWriteRetry; |
| LocatedBlock lb; |
| std::vector<DatanodeInfo> excludedNodes; |
| shared_ptr<LocatedBlock> block = lastBlock; |
| std::string buffer; |
| |
| do { |
| errorIndex = -1; |
| lastBlock = block; |
| |
| try { |
| locateNextBlock(excludedNodes); |
| lastBlock->setNumBytes(0); |
| nodes = lastBlock->getLocations(); |
| storageIDs = lastBlock->getStorageIDs(); |
| } catch (const HdfsRpcException & e) { |
| const char * lastBlockName = lastBlock ? lastBlock->toString().c_str() : "Null"; |
| LOG(LOG_ERROR, |
| "Failed to allocate a new empty block for file %s, last block %s, excluded nodes %s.\n%s", |
| path.c_str(), lastBlockName, FormatExcludedNodes(excludedNodes).c_str(), GetExceptionDetail(e, buffer)); |
| |
| if (retryAllocNewBlock > blockWriteRetry) { |
| throw; |
| } |
| |
| LOG(INFO, "Retry to allocate a new empty block for file %s, last block %s, excluded nodes %s.", |
| path.c_str(), lastBlockName, FormatExcludedNodes(excludedNodes).c_str()); |
| ++retryAllocNewBlock; |
| continue; |
| } catch (const HdfsException & e) { |
| const char * lastBlockName = lastBlock ? lastBlock->toString().c_str() : "Null"; |
| LOG(LOG_ERROR, |
| "Failed to allocate a new empty block for file %s, last block %s, excluded nodes %s.\n%s", |
| path.c_str(), lastBlockName, FormatExcludedNodes(excludedNodes).c_str(), GetExceptionDetail(e, buffer)); |
| throw; |
| } |
| |
| retryAllocNewBlock = 0; |
| checkPipelineWithReplicas(); |
| |
| if (nodes.empty()) { |
| THROW(HdfsIOException, |
| "No datanode is available to create a pipeline for block %s file %s.", |
| lastBlock->toString().c_str(), path.c_str()); |
| } |
| |
| try { |
| createBlockOutputStream(lastBlock->getToken(), 0, false); |
| break; //break on success |
| } catch (const HdfsInvalidBlockToken & e) { |
| LOG(LOG_ERROR, |
| "Failed to setup the pipeline for new block %s file %s.\n%s", |
| lastBlock->toString().c_str(), path.c_str(), GetExceptionDetail(e, buffer)); |
| } catch (const HdfsTimeoutException & e) { |
| LOG(LOG_ERROR, |
| "Failed to setup the pipeline for new block %s file %s.\n%s", |
| lastBlock->toString().c_str(), path.c_str(), GetExceptionDetail(e, buffer)); |
| } catch (const HdfsIOException & e) { |
| LOG(LOG_ERROR, |
| "Failed to setup the pipeline for new block %s file %s.\n%s", |
| lastBlock->toString().c_str(), path.c_str(), GetExceptionDetail(e, buffer)); |
| } |
| |
| LOG(INFO, "Abandoning block: %s for file %s.", lastBlock->toString().c_str(), path.c_str()); |
| |
| try { |
| filesystem->abandonBlock(*lastBlock, path); |
| } catch (const HdfsException & e) { |
| LOG(LOG_ERROR, |
| "Failed to abandon useless block %s for file %s.\n%s", |
| lastBlock->toString().c_str(), path.c_str(), GetExceptionDetail(e, buffer)); |
| throw; |
| } |
| |
| if (errorIndex >= 0) { |
| LOG(INFO, "Excluding invalid datanode: %s for block %s for file %s", |
| nodes[errorIndex].formatAddress().c_str(), lastBlock->toString().c_str(), path.c_str()); |
| excludedNodes.push_back(nodes[errorIndex]); |
| } else { |
| /* |
| * we don't known what happened, no datanode is reported failure, reduce retry count in case of infinite loop. |
| */ |
| --retry; |
| } |
| } while (retry); |
| } |
| |
| /* |
| * bad link node must be either empty or a "IP:PORT" |
| */ |
| void PipelineImpl::checkBadLinkFormat(const std::string & n) { |
| std::string node = n; |
| |
| if (node.empty()) { |
| return; |
| } |
| |
| do { |
| const char * host = &node[0], *port; |
| size_t pos = node.find_last_of(":"); |
| |
| if (pos == node.npos || pos + 1 == node.length()) { |
| break; |
| } |
| |
| node[pos] = 0; |
| port = &node[pos + 1]; |
| struct addrinfo hints, *addrs; |
| memset(&hints, 0, sizeof(hints)); |
| hints.ai_family = PF_UNSPEC; |
| hints.ai_socktype = SOCK_STREAM; |
| hints.ai_flags = AI_NUMERICHOST | AI_NUMERICSERV; |
| int p; |
| char * end; |
| p = strtol(port, &end, 0); |
| |
| if (p >= 65536 || p <= 0 || end != port + strlen(port)) { |
| break; |
| } |
| |
| if (getaddrinfo(host, port, &hints, &addrs)) { |
| break; |
| } |
| |
| freeaddrinfo(addrs); |
| return; |
| } while (0); |
| |
| LOG(FATAL, "Cannot parser the firstBadLink string %s, it should be a bug or protocol incompatible.", |
| n.c_str()); |
| THROW(HdfsException, "Cannot parser the firstBadLink string %s, it should be a bug or protocol incompatible.", |
| n.c_str()); |
| } |
| |
| void PipelineImpl::createBlockOutputStream(const Token & token, int64_t gs, bool recovery) { |
| std::string firstBadLink; |
| exception_ptr lastError; |
| bool needWrapException = true; |
| |
| try { |
| sock = shared_ptr < Socket > (new TcpSocketImpl); |
| reader = shared_ptr<BufferedSocketReader>(new BufferedSocketReaderImpl(*sock)); |
| sock->connect(nodes[0].getIpAddr().c_str(), nodes[0].getXferPort(), |
| connectTimeout); |
| std::vector<DatanodeInfo> targets; |
| |
| for (size_t i = 1; i < nodes.size(); ++i) { |
| targets.push_back(nodes[i]); |
| } |
| |
| DataTransferProtocolSender sender(*sock, writeTimeout, |
| nodes[0].formatAddress()); |
| sender.writeBlock(*lastBlock, token, clientName.c_str(), targets, |
| (recovery ? (stage | 0x1) : stage), nodes.size(), |
| lastBlock->getNumBytes(), bytesSent, gs, checksumType, chunkSize); |
| int size; |
| size = reader->readVarint32(readTimeout); |
| std::vector<char> buf(size); |
| reader->readFully(&buf[0], size, readTimeout); |
| BlockOpResponseProto resp; |
| |
| if (!resp.ParseFromArray(&buf[0], size)) { |
| THROW(HdfsIOException, "cannot parse datanode response from %s for block %s.", |
| nodes[0].formatAddress().c_str(), lastBlock->toString().c_str()); |
| } |
| |
| Status pipelineStatus = resp.status(); |
| firstBadLink = resp.firstbadlink(); |
| |
| if (Status::DT_PROTO_SUCCESS != pipelineStatus) { |
| needWrapException = false; |
| |
| if (Status::DT_PROTO_ERROR_ACCESS_TOKEN == pipelineStatus) { |
| THROW(HdfsInvalidBlockToken, |
| "Got access token error for connect ack with firstBadLink as %s for block %s", |
| firstBadLink.c_str(), lastBlock->toString().c_str()); |
| } else { |
| THROW(HdfsIOException, "Bad connect ack with firstBadLink as %s for block %s", |
| firstBadLink.c_str(), lastBlock->toString().c_str()); |
| } |
| } |
| |
| return; |
| } catch (...) { |
| errorIndex = 0; |
| lastError = current_exception(); |
| } |
| |
| checkBadLinkFormat(firstBadLink); |
| |
| if (!firstBadLink.empty()) { |
| for (size_t i = 0; i < nodes.size(); ++i) { |
| if (nodes[i].getXferAddr() == firstBadLink) { |
| errorIndex = i; |
| break; |
| } |
| } |
| } |
| |
| assert(lastError); |
| |
| if (!needWrapException) { |
| rethrow_exception(lastError); |
| } |
| |
| try { |
| rethrow_exception(lastError); |
| } catch (const HdfsException & e) { |
| NESTED_THROW(HdfsIOException, |
| "Cannot create block output stream for block %s, " |
| "recovery flag: %s, with last generate stamp %" PRId64 ".", |
| lastBlock->toString().c_str(), (recovery ? "true" : "false"), gs); |
| } |
| } |
| |
| void PipelineImpl::resend() { |
| assert(stage != PIPELINE_CLOSE); |
| |
| for (size_t i = 0; i < packets.size(); ++i) { |
| ConstPacketBuffer b = packets[i]->getBuffer(); |
| sock->writeFully(b.getBuffer(), b.getSize(), writeTimeout); |
| int64_t tmp = packets[i]->getLastByteOffsetBlock(); |
| bytesSent = bytesSent > tmp ? bytesSent : tmp; |
| } |
| } |
| |
| void PipelineImpl::send(shared_ptr<Packet> packet) { |
| ConstPacketBuffer buffer = packet->getBuffer(); |
| |
| if (!packet->isHeartbeat()) { |
| packets.push_back(packet); |
| } |
| |
| /* |
| * too many packets pending on the ack. wait in case of consuming to much memory. |
| */ |
| if (static_cast<int>(packets.size()) > packetPool.getMaxSize()) { |
| waitForAcks(false); |
| } |
| |
| bool failover = false; |
| |
| do { |
| try { |
| if (failover) { |
| resend(); |
| } else { |
| assert(sock); |
| sock->writeFully(buffer.getBuffer(), buffer.getSize(), |
| writeTimeout); |
| int64_t tmp = packet->getLastByteOffsetBlock(); |
| bytesSent = bytesSent > tmp ? bytesSent : tmp; |
| } |
| |
| checkResponse(false); |
| return; |
| } catch (const HdfsIOException & e) { |
| if (errorIndex < 0) { |
| errorIndex = 0; |
| } |
| |
| sock.reset(); |
| } |
| |
| buildForAppendOrRecovery(true); |
| failover = true; |
| |
| if (stage == PIPELINE_CLOSE) { |
| assert(packets.size() == 1 && packets[0]->isLastPacketInBlock()); |
| packets.clear(); |
| break; |
| } |
| } while (true); |
| } |
| |
| void PipelineImpl::processAck(PipelineAck & ack) { |
| assert(!ack.isInvalid()); |
| int64_t seqno = ack.getSeqno(); |
| |
| if (HEART_BEAT_SEQNO == seqno) { |
| return; |
| } |
| |
| assert(!packets.empty()); |
| Packet & packet = *packets[0]; |
| |
| if (ack.isSuccess()) { |
| if (packet.getSeqno() != seqno) { |
| THROW(HdfsIOException, |
| "processAck: pipeline ack expecting seqno %" PRId64 " but received %" PRId64 " for block %s.", |
| packet.getSeqno(), seqno, lastBlock->toString().c_str()); |
| } |
| |
| int64_t tmp = packet.getLastByteOffsetBlock(); |
| bytesAcked = tmp > bytesAcked ? tmp : bytesAcked; |
| assert(lastBlock); |
| lastBlock->setNumBytes(bytesAcked); |
| |
| if (packet.isLastPacketInBlock()) { |
| sock.reset(); |
| } |
| |
| packetPool.relesePacket(packets[0]); |
| packets.pop_front(); |
| } else { |
| for (int i = ack.getNumOfReplies() - 1; i >= 0; --i) { |
| if (Status::DT_PROTO_SUCCESS != ack.getReply(i)) { |
| errorIndex = i; |
| /* |
| * handle block token expire as same as HdfsIOException. |
| */ |
| THROW(HdfsIOException, |
| "processAck: ack report error at node: %s for block %s.", |
| nodes[i].formatAddress().c_str(), lastBlock->toString().c_str()); |
| } |
| } |
| } |
| } |
| |
| void PipelineImpl::processResponse() { |
| PipelineAck ack; |
| std::vector<char> buf; |
| int size = reader->readVarint32(readTimeout); |
| ack.reset(); |
| buf.resize(size); |
| reader->readFully(&buf[0], size, readTimeout); |
| ack.readFrom(&buf[0], size); |
| |
| if (ack.isInvalid()) { |
| THROW(HdfsIOException, |
| "processAllAcks: get an invalid DataStreamer packet ack for block %s", |
| lastBlock->toString().c_str()); |
| } |
| |
| processAck(ack); |
| } |
| |
| void PipelineImpl::checkResponse(bool wait) { |
| int timeout = wait ? readTimeout : 0; |
| bool readable = reader->poll(timeout); |
| |
| if (readable) { |
| processResponse(); |
| } else if (wait) { |
| THROW(HdfsIOException, "Timeout when reading response for block %s, datanode %s do not response.", |
| lastBlock->toString().c_str(), |
| nodes[0].formatAddress().c_str()); |
| } |
| } |
| |
| void PipelineImpl::flush() { |
| waitForAcks(true); |
| } |
| |
| void PipelineImpl::waitForAcks(bool force) { |
| bool failover = false; |
| |
| while (!packets.empty()) { |
| /* |
| * just wait for some acks in case of consuming too much memory. |
| */ |
| if (!force && static_cast<int>(packets.size()) < packetPool.getMaxSize()) { |
| return; |
| } |
| |
| try { |
| if (failover) { |
| resend(); |
| } |
| |
| checkResponse(true); |
| failover = false; |
| } catch (const HdfsIOException & e) { |
| if (errorIndex < 0) { |
| errorIndex = 0; |
| } |
| |
| std::string buffer; |
| LOG(LOG_ERROR, |
| "Failed to flush pipeline on datanode %s for block %s file %s.\n%s", |
| nodes[errorIndex].formatAddress().c_str(), lastBlock->toString().c_str(), |
| path.c_str(), GetExceptionDetail(e, buffer)); |
| LOG(INFO, "Rebuild pipeline to flush for block %s file %s.", lastBlock->toString().c_str(), path.c_str()); |
| sock.reset(); |
| failover = true; |
| } |
| |
| if (failover) { |
| buildForAppendOrRecovery(true); |
| |
| if (stage == PIPELINE_CLOSE) { |
| assert(packets.size() == 1 && packets[0]->isLastPacketInBlock()); |
| packets.clear(); |
| break; |
| } |
| } |
| } |
| } |
| |
| shared_ptr<LocatedBlock> PipelineImpl::close(shared_ptr<Packet> lastPacket) { |
| waitForAcks(true); |
| lastPacket->setLastPacketInBlock(true); |
| stage = PIPELINE_CLOSE; |
| send(lastPacket); |
| waitForAcks(true); |
| sock.reset(); |
| lastBlock->setNumBytes(bytesAcked); |
| LOG(DEBUG2, "close pipeline for file %s, block %s with length %" PRId64, |
| path.c_str(), lastBlock->toString().c_str(), |
| lastBlock->getNumBytes()); |
| return lastBlock; |
| } |
| |
| } |
| } |
