zk_dn_client.h

// Copyright 2017 Rice University, COMP 413 2017

#ifndef ZOOKEEPER_INCLUDE_ZK_DN_CLIENT_H_
#define ZOOKEEPER_INCLUDE_ZK_DN_CLIENT_H_

#include "zk_client_common.h"
#include <google/protobuf/message.h>
#include <atomic>
#include <string>
#include "hdfs.pb.h"

class TransferServer;

namespace zkclient {

typedef struct {
  std::string ip;
  uint32_t ipcPort;
} DataNodeId;

// TODO(2016): Store hostname in payload as a vararg?
typedef struct {
  uint32_t ipcPort;
  uint32_t xferPort;
  uint64_t disk_bytes;    // total space on disk
  uint64_t free_bytes;    // free space on disk
  uint32_t xmits;            // current number of xmits
} DataNodePayload;

typedef struct {
  char ipPort[256];
} DataNodeZNode;

typedef struct {
  uint64_t block_size;  // size of the block in bytes
} BlockZNode;

class ZkClientDn : public ZkClientCommon {
 public:
  ZkClientDn(const std::string &ip,
             const std::string &zkIpAndAddress,
             uint64_t total_disk_space,
             const uint32_t ipcPort = 50020,
             const uint32_t xferPort = 50010);

  ZkClientDn(const std::string &ip,
             std::shared_ptr<ZKWrapper>,
             uint64_t total_disk_space,
             const uint32_t ipcPort = 50020,
             const uint32_t xferPort = 50010);
  ~ZkClientDn();

  void registerDataNode(const std::string &ip,
                        uint64_t total_disk_space,
                        const uint32_t ipcPort,
                        const uint32_t xferPort);

  bool blockReceived(uint64_t uuid, uint64_t size_bytes);

  bool blockSizeUpdated(uint64_t uuid, uint64_t size_bytes);

  bool sendStats(uint64_t free_space, uint32_t xmits);

  void setTransferServer(std::shared_ptr<TransferServer> &server);

  bool push_dn_on_repq(std::string dn_name, uint64_t blockid);

  bool poll_replication_queue();

  bool poll_delete_queue();

  bool poll_reconstruct_queue();

  std::string get_datanode_id();

 private:
  std::shared_ptr<TransferServer> server;

  void processDeleteQueue();

  std::string build_datanode_id(DataNodeId data_node_id);

  DataNodeId data_node_id;
  DataNodePayload data_node_payload;

  static const std::string CLASS_NAME;

  void initWorkQueue(std::string queueName, void (*watchFuncPtr)(zhandle_t *,
                                                                 int,
                                                                 int,
                                                                 const char *,
                                                                 void *));

  void handleReplicateCmds(const std::string &path);

  bool handleReconstructCmds(const std::string &path);

  static void thisDNDeleteQueueWatcher(zhandle_t *zzh,
                                       int type,
                                       int state,
                                       const char *path,
                                       void *watcherCtx);

  bool find_datanode_with_block(uint64_t &block_uuid,
                                std::string &datanode,
                                int &error_code);

  bool buildExtendedBlockProto(hadoop::hdfs::ExtendedBlockProto *eb,
                               const std::uint64_t &block_id,
                               const uint64_t &block_size);
};
}  // namespace zkclient

#endif   // ZOOKEEPER_INCLUDE_ZK_DN_CLIENT_H_