| /******************************************************************** |
| * 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. |
| */ |
| #ifndef _HDFS_LIBHDFS3_CLIENT_PIPELINE_H_ |
| #define _HDFS_LIBHDFS3_CLIENT_PIPELINE_H_ |
| |
| #include "FileSystemInter.h" |
| #include "Memory.h" |
| #include "network/BufferedSocketReader.h" |
| #include "network/TcpSocket.h" |
| #include "Packet.h" |
| #include "PacketPool.h" |
| #include "PipelineAck.h" |
| #include "server/DatanodeInfo.h" |
| #include "server/LocatedBlock.h" |
| #include "server/Namenode.h" |
| #include "SessionConfig.h" |
| #include "Thread.h" |
| |
| #include <vector> |
| #include <deque> |
| |
| namespace Hdfs { |
| namespace Internal { |
| |
| enum BlockConstructionStage { |
| /** |
| * The enumerates are always listed as regular stage followed by the |
| * recovery stage. |
| * Changing this order will make getRecoveryStage not working. |
| */ |
| // pipeline set up for block append |
| PIPELINE_SETUP_APPEND = 0, |
| // pipeline set up for failed PIPELINE_SETUP_APPEND recovery |
| PIPELINE_SETUP_APPEND_RECOVERY = 1, |
| // data streaming |
| DATA_STREAMING = 2, |
| // pipeline setup for failed data streaming recovery |
| PIPELINE_SETUP_STREAMING_RECOVERY = 3, |
| // close the block and pipeline |
| PIPELINE_CLOSE = 4, |
| // Recover a failed PIPELINE_CLOSE |
| PIPELINE_CLOSE_RECOVERY = 5, |
| // pipeline set up for block creation |
| PIPELINE_SETUP_CREATE = 6 |
| }; |
| |
| static inline const char * StageToString(BlockConstructionStage stage) { |
| switch (stage) { |
| case PIPELINE_SETUP_APPEND: |
| return "PIPELINE_SETUP_APPEND"; |
| |
| case PIPELINE_SETUP_APPEND_RECOVERY: |
| return "PIPELINE_SETUP_APPEND_RECOVERY"; |
| |
| case DATA_STREAMING: |
| return "DATA_STREAMING"; |
| |
| case PIPELINE_SETUP_STREAMING_RECOVERY: |
| return "PIPELINE_SETUP_STREAMING_RECOVERY"; |
| |
| case PIPELINE_CLOSE: |
| return "PIPELINE_CLOSE"; |
| |
| case PIPELINE_CLOSE_RECOVERY: |
| return "PIPELINE_CLOSE_RECOVERY"; |
| |
| case PIPELINE_SETUP_CREATE: |
| return "PIPELINE_SETUP_CREATE"; |
| |
| default: |
| return "UNKNOWN STAGE"; |
| } |
| } |
| |
| class Packet; |
| class OutputStreamImpl; |
| |
| /** |
| * setup, data transfer, close, and failover. |
| */ |
| class Pipeline { |
| public: |
| |
| virtual ~Pipeline() {} |
| |
| /** |
| * send all data and wait for all ack. |
| */ |
| virtual void flush() = 0; |
| |
| /** |
| * send LastPacket and close the pipeline. |
| */ |
| virtual shared_ptr<LocatedBlock> close(shared_ptr<Packet> lastPacket) = 0; |
| |
| /** |
| * send a packet, retry on error until fatal. |
| * @param packet |
| */ |
| virtual void send(shared_ptr<Packet> packet) = 0; |
| }; |
| |
| class PipelineImpl : public Pipeline { |
| public: |
| /** |
| * construct and setup the pipeline for append. |
| */ |
| 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); |
| |
| /** |
| * send all data and wait for all ack. |
| */ |
| void flush(); |
| |
| /** |
| * send LastPacket and close the pipeline. |
| */ |
| shared_ptr<LocatedBlock> close(shared_ptr<Packet> lastPacket); |
| |
| /** |
| * send a packet, retry on error until fatal. |
| * @param packet |
| */ |
| void send(shared_ptr<Packet> packet); |
| |
| private: |
| bool addDatanodeToPipeline(const std::vector<DatanodeInfo> & excludedNodes); |
| void buildForAppendOrRecovery(bool recovery); |
| void buildForNewBlock(); |
| void checkPipelineWithReplicas(); |
| void checkResponse(bool wait); |
| void createBlockOutputStream(const Token & token, int64_t gs, bool recovery); |
| void locateNextBlock(const std::vector<DatanodeInfo> & excludedNodes); |
| void processAck(PipelineAck & ack); |
| void processResponse(); |
| void resend(); |
| void waitForAcks(bool force); |
| void transfer(const ExtendedBlock & blk, const DatanodeInfo & src, |
| const std::vector<DatanodeInfo> & targets, |
| const Token & token); |
| int findNewDatanode(const std::vector<DatanodeInfo> & original); |
| |
| private: |
| static void checkBadLinkFormat(const std::string & node); |
| |
| private: |
| BlockConstructionStage stage; |
| bool canAddDatanode; |
| int blockWriteRetry; |
| int checksumType; |
| int chunkSize; |
| int connectTimeout; |
| int errorIndex; |
| int readTimeout; |
| int replication; |
| int writeTimeout; |
| int64_t bytesAcked; //the size of bytes the ack received. |
| int64_t bytesSent; //the size of bytes has sent. |
| PacketPool & packetPool; |
| shared_ptr<BufferedSocketReader> reader; |
| shared_ptr<FileSystemInter> filesystem; |
| shared_ptr<LocatedBlock> lastBlock; |
| shared_ptr<Socket> sock; |
| std::deque<shared_ptr<Packet> > packets; |
| std::string clientName; |
| std::string path; |
| std::vector<DatanodeInfo> nodes; |
| std::vector<std::string> storageIDs; |
| |
| }; |
| |
| } |
| } |
| |
| #endif /* _HDFS_LIBHDFS3_CLIENT_PIPELINE_H_ */ |