hdfs-bulk-ops.h

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed 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 IMPALA_UTIL_HDFS_BULK_OPS_H
#define IMPALA_UTIL_HDFS_BULK_OPS_H

#include <string>
#include <boost/thread.hpp>

#include "common/hdfs.h"
#include "common/atomic.h"
#include "common/status.h"
#include "util/thread-pool.h"

namespace impala {

enum HdfsOpType {
  DELETE,
  CREATE_DIR,
  RENAME,
  DELETE_THEN_CREATE,
  CHMOD
};

class HdfsOperationSet;

class HdfsOp {
 public:
  HdfsOp(HdfsOpType op, const std::string& src, HdfsOperationSet* op_set);

  HdfsOp(HdfsOpType op, const std::string& src, const std::string& dst,
      HdfsOperationSet* op_set);

  HdfsOp(HdfsOpType op, const std::string& src, short permissions,
      HdfsOperationSet* op_set);

  HdfsOp();

  HdfsOpType op() const { return op_; }
  const std::string& src() const { return src_; }
  const std::string& dst() const { return dst_; }

  void Execute() const;

 private:
  HdfsOpType op_;

  std::string src_;

  std::string dst_;

  short permissions_;

  HdfsOperationSet* op_set_;
};

typedef ThreadPool<HdfsOp> HdfsOpThreadPool;

HdfsOpThreadPool* CreateHdfsOpThreadPool(const std::string& name, uint32_t num_threads,
    uint32_t max_queue_length);

class HdfsOperationSet {
 public:
  HdfsOperationSet(hdfsFS* hdfs_connection);

  void Add(HdfsOpType op, const std::string& src);

  void Add(HdfsOpType op, const std::string& src, const std::string& dst);

  void Add(HdfsOpType op, const std::string& src, short permissions);

  bool Execute(HdfsOpThreadPool* pool, bool abort_on_error);

  typedef std::pair<const HdfsOp*, std::string> Error;
  typedef std::vector<Error> Errors;

  const Errors& errors() { return errors_; }

  hdfsFS* hdfs_connection() const { return hdfs_connection_; }

 private:
  std::vector<HdfsOp> ops_;

  Promise<bool> promise_;

  AtomicInt<int64_t> num_ops_;

  hdfsFS* hdfs_connection_;

  boost::mutex errors_lock_;

  Errors errors_;

  bool abort_on_error_;

  friend class HdfsOp;

  void MarkOneOpDone();

  void AddError(const std::string& err, const HdfsOp* op);

  bool ShouldAbort();
};

}

#endif // IMPALA_UTIL_HDFS_BULK_OPS_H