1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// 
type ExtentClient struct {
    streamers       map[uint64]*Streamer
    dataWrapper     *Wrapper
    //...
}

type Streamer struct {
    extents     *ExtentCache
    handler     *ExtentHandler
    dirtylist   *DirtyExtentList
}

type ExtentHandler struct {

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

// sdk 写数据入口
func (ec *ExtentClient) Write(inode uint64, offset int, data []byte, direct bool) {
    s := client.GetStreamer(inode)                //
    s.once.Do(func(){ s.GetExtents(); })
    s.IssueWriteRequest(offset, data, direct)    //streamer处理写请求, 获取一个写请求，然后将请求放入s.request 通道
}

// 
func (s *Streamer)write(data []byte, offset, size int, direct bool) {
    // 准备写请求
    requests := s.extents.PrepareWriteRequests(offset, size, data)
    //...
    // 
    for _, req := range requests {
        s.doWrite()
    }
}

// streamer 通过extenthandler 将数据写入到datanode中
func (s *Streamer) doWrite(data []byte, offset, size int, direct bool) {
    // set storeMode
    storeMode := func(tailOff, tinySizeLimit) {
        if tailOff > tinySizeLimit {
            return proto.NormalExtentType
        } else {
            return proto.TinyExtentType
        }
    }(offset+size, s.tinySizeLimit)

    //重试写入
    for i := 0; i < MaxNewHandlerRetry; i++ {
        // 
        ek, err = s.handler.write(data, offset, size, direct)
        // 写成功
        if err == nil && ek != nil {
            break
        }
    }
}

// 写
func (eh *ExtentHandler) write() {
    for total < size {

        eh.flushPacket() // eh.request <- packet
    }

}

// 发送协程,
func (eh *ExtentHandler) sender() {
    for {
        select {
            case packet := <- eh.request:

        }
    }
}

// 接收协程
func (eh *ExtentHandler) receiver() {
    for {
        select {
            case reply := <- eh.re
        }
    }

}