utils

Documentation

Index

• Constants
• func Chunk2ChunkProto(chunk *Chunk) (chunkPb *pb.ChunkProto)
• func ParseInstr(instr string) (instrs []string)
• func ReadConfig(cfgPath string, cfgName string) (cfg *viper.Viper)
• type Chunk
  • func NewChunk(content []byte) (*Chunk, error)
• type ChunkPb
  • func NewChunkPb(content []byte) (*ChunkPb, error)
• type HDFSOutputStream
  • func (ops *HDFSOutputStream) Write(localPath string)
• type Packet
  • func NewNullPacket(offsetInBlock uint64, seqNo uint32, lastPacketInBlock bool, dataLen uint32) (pkt *Packet, err error)
• type PacketCtx
  • func NewPacketCtx(rw io.ReadWriter, OffsetInBlock uint64, SeqNo uint32, LastPacketInBlock bool, ...) (ctx *PacketCtx)
  • func (ctx *PacketCtx) Read(pkts []Packet) (n int, err error)
  • func (ctx *PacketCtx) Write(pkts []Packet) (n int, err error)
• type PacketQueue
  • func NewPacketQueue() *PacketQueue
  • func (pq *PacketQueue) GetLength() int
  • func (pq *PacketQueue) Pop() (out *Packet, err error)
  • func (pq *PacketQueue) Push(elem Packet) error
• type PacketReadWriter

Constants

const (
	// BUFCHUNKNUM is num of chunk in buffer
	// Why it's 9? I think 126 % 9 = 0
	// (In fact, 126 % 7 also equal 0)
	// BUF write to Packet will process well
	BUFCHUNKNUM = 9
	// BUFSIZE is size of buffer used in write process
	BUFSIZE = CHUNKCONTENTSIZE * BUFCHUNKNUM

	// CHUNKCONTENTSIZE is size of chunk's content
	CHUNKCONTENTSIZE = 512
	// CHUNKCHECKSUMSIZE is size of chunk's checksum
	CHUNKCHECKSUMSIZE = 4
	// CHUNKTOTALSIZE equal size(checksum) + size(content)
	CHUNKTOTALSIZE = CHUNKCONTENTSIZE + CHUNKCHECKSUMSIZE

	// PACKETCHUNKNUM is num of chunk in packet
	PACKETCHUNKNUM = 126
	// PACKETCHUNKSIZE is increasing offset in file when writing & reading
	PACKETCHUNKSIZE = PACKETCHUNKNUM * CHUNKCONTENTSIZE
	// PACKETTOTALSIZE is size of packet
	PACKETTOTALSIZE = PACKETCHUNKNUM * CHUNKTOTALSIZE
)

Functions

func Chunk2ChunkProto

func Chunk2ChunkProto(chunk *Chunk) (chunkPb *pb.ChunkProto)

Chunk2ChunkProto is a translator

func ParseInstr

func ParseInstr(instr string) (instrs []string)

ParseInstr is a function to process instrucion

func ReadConfig

func ReadConfig(cfgPath string, cfgName string) (cfg *viper.Viper)

ReadConfig is a method to read yaml

Types

type Chunk

type Chunk struct {
	Checksum uint32
	Content  []byte
}

Chunk is a base data structure in hdfs

func NewChunk

func NewChunk(content []byte) (*Chunk, error)

NewChunk is a constructor of Chunk

type ChunkPb

type ChunkPb struct {
	// contains filtered or unexported fields
}

ChunkPb is a data structure defined by protobuf

func NewChunkPb

func NewChunkPb(content []byte) (*ChunkPb, error)

NewChunkPb is a constructor of Chunk defined by protobuf

type HDFSOutputStream

type HDFSOutputStream struct {
	Target net.TCPAddr
	// contains filtered or unexported fields
}

HDFSOutputStream can write

func (*HDFSOutputStream) Write

func (ops *HDFSOutputStream) Write(localPath string)

Write is user interface a function, use a stream-object to upload file to hdfs

1. communicate with namenode build a conn between name node and client, send localPath, query blocks and get metadata
2. build a conn between data node and client
3. for { get NewPacket as "pkt" from file descriptor. conn.Write(pkt) }

type Packet

type Packet struct {
	Header       *pb.PacketHeaderProto
	PacketData   []Chunk
	PacketOffset int
	// contains filtered or unexported fields
}

Packet is base data structure in hdfs PacketData should be separate into checksumField and DataField, but I do not choose to implement that kind of packet format

PacketOffset means the number of which Write function has appended, if PacketOffset equal PACKETCHUNKNUM(126), listen fiber will flush it.

func NewNullPacket

func NewNullPacket(offsetInBlock uint64, seqNo uint32, lastPacketInBlock bool, dataLen uint32) (pkt *Packet, err error)

NewNullPacket is a default construct

type PacketCtx

type PacketCtx struct {
	OffsetInBlock     uint64
	SeqNo             uint32
	LastPacketInBlock bool
	DataLen           uint32
	Buf               []byte
	EmptyBuf          []byte
	// contains filtered or unexported fields
}

PacketCtx is a user interface rw(io.ReadWriter) is usually file descriptor Buf's length equal CHUNKCONTENTSIZE

func NewPacketCtx

func NewPacketCtx(rw io.ReadWriter, OffsetInBlock uint64, SeqNo uint32, LastPacketInBlock bool, DataLen uint32) (ctx *PacketCtx)

NewPacketCtx is a constructor

func (*PacketCtx) Read

func (ctx *PacketCtx) Read(pkts []Packet) (n int, err error)

n means success number

func (*PacketCtx) Write

func (ctx *PacketCtx) Write(pkts []Packet) (n int, err error)

n means the number of write packets

type PacketQueue

type PacketQueue struct {
	// contains filtered or unexported fields
}

PacketQueue is a base data structure in hdfs

func NewPacketQueue

func NewPacketQueue() *PacketQueue

NewPacketQueue is a constructor

func (*PacketQueue) GetLength

func (pq *PacketQueue) GetLength() int

GetLength is a method to return the num of elements

func (*PacketQueue) Pop

func (pq *PacketQueue) Pop() (out *Packet, err error)

Pop is to pop a elem from queue

func (*PacketQueue) Push

func (pq *PacketQueue) Push(elem Packet) error

Push is to push a elem into queue

type PacketReadWriter

type PacketReadWriter interface {
	Read(pktBuf []Packet)
	Write(pktBuf []Packet)
}

PacketReadWriter is an interface