-
Notifications
You must be signed in to change notification settings - Fork 918
/
service.go
140 lines (119 loc) · 4.01 KB
/
service.go
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
package service
import (
"context"
"fmt"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-cid"
"golang.org/x/sync/errgroup"
"github.com/celestiaorg/celestia-node/share"
"github.com/celestiaorg/celestia-node/share/eds"
"github.com/celestiaorg/celestia-node/share/ipld"
"github.com/celestiaorg/nmt"
"github.com/celestiaorg/nmt/namespace"
)
// TODO(@Wondertan): Simple thread safety for Start and Stop would not hurt.
type ShareService struct {
share.Availability
rtrv *eds.Retriever
bServ blockservice.BlockService
// session is blockservice sub-session that applies optimization for fetching/loading related
// nodes, like shares prefer session over blockservice for fetching nodes.
session blockservice.BlockGetter
cancel context.CancelFunc
}
// NewService creates a new basic share.Module.
func NewShareService(bServ blockservice.BlockService, avail share.Availability) *ShareService {
return &ShareService{
rtrv: eds.NewRetriever(bServ),
Availability: avail,
bServ: bServ,
}
}
func (s *ShareService) Start(context.Context) error {
if s.session != nil || s.cancel != nil {
return fmt.Errorf("share: service already started")
}
// NOTE: The ctx given as param is used to control Start flow and only needed when Start is
// blocking, but this one is not.
//
// The newer context here is created to control lifecycle of the session and peer discovery.
ctx, cancel := context.WithCancel(context.Background())
s.cancel = cancel
s.session = blockservice.NewSession(ctx, s.bServ)
return nil
}
func (s *ShareService) Stop(context.Context) error {
if s.session == nil || s.cancel == nil {
return fmt.Errorf("share: service already stopped")
}
s.cancel()
s.cancel = nil
s.session = nil
return nil
}
func (s *ShareService) GetShare(ctx context.Context, dah *share.Root, row, col int) (share.Share, error) {
root, leaf := ipld.Translate(dah, row, col)
nd, err := share.GetShare(ctx, s.bServ, root, leaf, len(dah.RowsRoots))
if err != nil {
return nil, err
}
return nd, nil
}
func (s *ShareService) GetShares(ctx context.Context, root *share.Root) ([][]share.Share, error) {
eds, err := s.rtrv.Retrieve(ctx, root)
if err != nil {
return nil, err
}
origWidth := int(eds.Width() / 2)
shares := make([][]share.Share, origWidth)
for i := 0; i < origWidth; i++ {
row := eds.Row(uint(i))
shares[i] = make([]share.Share, origWidth)
for j := 0; j < origWidth; j++ {
shares[i][j] = row[j]
}
}
return shares, nil
}
// GetSharesByNamespace iterates over a square's row roots and accumulates the found shares in the
// given namespace.ID.
func (s *ShareService) GetSharesByNamespace(
ctx context.Context,
root *share.Root,
nID namespace.ID,
) ([]share.Share, error) {
if len(nID) != share.NamespaceSize {
return nil, fmt.Errorf("expected namespace ID of size %d, got %d", share.NamespaceSize, len(nID))
}
rowRootCIDs := make([]cid.Cid, 0)
for _, row := range root.RowsRoots {
if !nID.Less(nmt.MinNamespace(row, nID.Size())) && nID.LessOrEqual(nmt.MaxNamespace(row, nID.Size())) {
rowRootCIDs = append(rowRootCIDs, ipld.MustCidFromNamespacedSha256(row))
}
}
if len(rowRootCIDs) == 0 {
return nil, nil
}
errGroup, ctx := errgroup.WithContext(ctx)
shares := make([][]share.Share, len(rowRootCIDs))
for i, rootCID := range rowRootCIDs {
// shadow loop variables, to ensure correct values are captured
i, rootCID := i, rootCID
errGroup.Go(func() (err error) {
shares[i], err = share.GetSharesByNamespace(ctx, s.bServ, rootCID, nID, len(root.RowsRoots), nil)
return
})
}
if err := errGroup.Wait(); err != nil {
return nil, err
}
// we don't know the amount of shares in the namespace, so we cannot preallocate properly
// TODO(@Wondertan): Consider improving encoding schema for data in the shares that will also
// include metadata with the amount of shares. If we are talking about plenty of data here, proper
// preallocation would make a difference
var out []share.Share
for i := 0; i < len(rowRootCIDs); i++ {
out = append(out, shares[i]...)
}
return out, nil
}