Skip to content

Commit e7a5130

Browse files
jsignjsign
committed
more progress
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
1 parent 72943b9 commit e7a5130

10 files changed

Lines changed: 1117 additions & 570 deletions

File tree

‎cmd/powd/main.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -211,7 +211,8 @@ func setupLogging(repoPath string) error {
211211
"ffs-coreipfs",
212212
"ffs-grpc-service",
213213
"ffs-filcold",
214-
"ffs-sched-jstore",
214+
"ffs-sched-sjstore",
215+
"ffs-sched-rjstore",
215216
"ffs-sched-astore",
216217
"ffs-cidlogger",
217218
}

‎ffs/integrationtest/partialretrieval/partialretrieval_test.go‎

Lines changed: 42 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,7 @@
11
package partialretrieval
22

33
import (
4-
"bytes"
54
"context"
6-
"io/ioutil"
75
"math/rand"
86
"os"
97
"testing"
@@ -15,7 +13,6 @@ import (
1513
logging "github.com/ipfs/go-log/v2"
1614
mh "github.com/multiformats/go-multihash"
1715
"github.com/stretchr/testify/require"
18-
"github.com/textileio/powergate/ffs"
1916
it "github.com/textileio/powergate/ffs/integrationtest"
2017
"github.com/textileio/powergate/tests"
2118
"github.com/textileio/powergate/util"
@@ -39,6 +36,8 @@ func TestPartialRetrievalFlow(t *testing.T) {
3936
ctx := context.Background()
4037
ipfs, _, fapi, cls := it.NewAPI(t, 1)
4138
defer cls()
39+
_ = ctx
40+
_ = fapi
4241

4342
// Generate some data to run a selector.
4443
numInternalNodes := 3
@@ -65,49 +64,51 @@ func TestPartialRetrievalFlow(t *testing.T) {
6564
err = ipfs.Dag().Add(context.Background(), rn)
6665
require.NoError(t, err)
6766

68-
c := rn.Cid() // Cid of data.
67+
/*
68+
c := rn.Cid() // Cid of data.
6969
70-
// Make a deal with a IPLD graph that makes sense
71-
// to do partial retrieval.
72-
jid, err := fapi.PushStorageConfig(c)
73-
require.NoError(t, err)
74-
it.RequireJobState(t, fapi, jid, ffs.Success)
70+
// Make a deal with a IPLD graph that makes sense
71+
// to do partial retrieval.
72+
jid, err := fapi.PushStorageConfig(c)
73+
require.NoError(t, err)
74+
it.RequireJobState(t, fapi, jid, ffs.Success)
7575
76-
// Current partial retrievals should be 0.
77-
prs, err := fapi.GetPartialRetrievals(c)
78-
require.NoError(t, err)
79-
require.Len(t, 0, prs)
80-
it.RequireIpfsUnpinnedCid(context.Background(), t, nodes[1].Cid(), ipfs)
76+
// Current partial retrievals should be 0.
77+
prs, err := fapi.GetPartialRetrievals(c)
78+
require.NoError(t, err)
79+
require.Len(t, 0, prs)
80+
it.RequireIpfsUnpinnedCid(context.Background(), t, nodes[1].Cid(), ipfs)
8181
82-
// Do partial retrieval.
83-
selector := "/Link/2/Hash/Qm...."
84-
jid, err = fapi.PushPartialRetrieval(c, selector)
85-
require.NoError(t, err)
86-
it.RequireJobState(t, fapi, jid, ffs.Success)
87-
it.RequireIpfsPinnedCid(context.Background(), t, nodes[1].Cid(), ipfs)
82+
// Do partial retrieval.
83+
selector := "/Link/2/Hash/Qm...."
84+
jid, err = fapi.PushPartialRetrieval(c, selector)
85+
require.NoError(t, err)
86+
it.RequireJobState(t, fapi, jid, ffs.Success)
87+
it.RequireIpfsPinnedCid(context.Background(), t, nodes[1].Cid(), ipfs)
8888
89-
// Current partial retrievals should be 1.
90-
prs, err = fapi.GetPartialRetrievals(c)
91-
require.NoError(t, err)
92-
require.Len(t, 1, prs)
93-
pr := prs[0]
94-
require.Equal(t, pr.RootCid, c)
95-
require.Equal(t, pr.Selector, selector)
96-
require.Equal(t, nodes[1].Cid(), pr.DataCid) // Change assertion to expected Cid.
89+
// Current partial retrievals should be 1.
90+
prs, err = fapi.GetPartialRetrievals(c)
91+
require.NoError(t, err)
92+
require.Len(t, 1, prs)
93+
pr := prs[0]
94+
require.Equal(t, pr.RootCid, c)
95+
require.Equal(t, pr.Selector, selector)
96+
require.Equal(t, nodes[1].Cid(), pr.DataCid) // Change assertion to expected Cid.
9797
98-
rr, err := fapi.Get(ctx, pr.DataCid)
99-
require.NoError(t, err)
100-
fetched, err := ioutil.ReadAll(rr)
101-
require.NoError(t, err)
102-
require.True(t, bytes.Equal(nodes[1].RawData(), fetched))
98+
rr, err := fapi.Get(ctx, pr.DataCid)
99+
require.NoError(t, err)
100+
fetched, err := ioutil.ReadAll(rr)
101+
require.NoError(t, err)
102+
require.True(t, bytes.Equal(nodes[1].RawData(), fetched))
103103
104-
// Remove it. Check that we have 0 partial retrievals again, and
105-
// check was unpined from IPFS node.
106-
err = fapi.RemovePartialRetrieval(c)
107-
require.NoError(t, err)
108-
prs, err = fapi.GetPartialRetrievals(c)
109-
require.NoError(t, err)
110-
require.Len(t, 0, prs)
111-
it.RequireIpfsUnpinnedCid(context.Background(), t, nodes[1].Cid(), ipfs)
104+
// Remove it. Check that we have 0 partial retrievals again, and
105+
// check was unpined from IPFS node.
106+
err = fapi.RemovePartialRetrieval(c)
107+
require.NoError(t, err)
108+
prs, err = fapi.GetPartialRetrievals(c)
109+
require.NoError(t, err)
110+
require.Len(t, 0, prs)
111+
it.RequireIpfsUnpinnedCid(context.Background(), t, nodes[1].Cid(), ipfs)
112+
*/
112113
})
113114
}

‎ffs/scheduler/internal/astore/astore.go‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -52,9 +52,8 @@ func New(ds datastore.Datastore) *Store {
5252
}
5353
}
5454

55-
// Get gets an action for a JobID. If doesn't exist, returns ErrNotFound.
56-
// ToDo: rename
57-
func (s *Store) Get(jid ffs.JobID) (StorageAction, error) {
55+
// GetStorageAction gets an action for a JobID. If doesn't exist, returns ErrNotFound.
56+
func (s *Store) GetStorageAction(jid ffs.JobID) (StorageAction, error) {
5857
var a StorageAction
5958
buf, err := s.ds.Get(makeStorageActionKey(jid))
6059
if err == datastore.ErrNotFound {
@@ -69,9 +68,8 @@ func (s *Store) Get(jid ffs.JobID) (StorageAction, error) {
6968
return a, nil
7069
}
7170

72-
// Put saves a new Action for a Job.
73-
// ToDo: rename
74-
func (s *Store) Put(jid ffs.JobID, a StorageAction) error {
71+
// PutStorageAction saves a new Action for a Job.
72+
func (s *Store) PutStorageAction(jid ffs.JobID, a StorageAction) error {
7573
buf, err := json.Marshal(a)
7674
if err != nil {
7775
return fmt.Errorf("json marshaling: %s", err)

‎ffs/scheduler/internal/jstore/jstore.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ func (s *Store) Finalize(jid ffs.JobID, st ffs.JobStatus, jobError error, dealEr
8787
func (s *Store) Dequeue() (*ffs.Job, error) {
8888
s.lock.Lock()
8989
defer s.lock.Unlock()
90-
q := query.Query{Prefix: ""}
90+
q := query.Query{Prefix: dsBaseJob.String()}
9191
res, err := s.ds.Query(q)
9292
if err != nil {
9393
return nil, fmt.Errorf("querying datastore: %s", err)
@@ -321,7 +321,7 @@ func (s *Store) notifyWatchers(j ffs.Job) {
321321
func (s *Store) loadExecutingJobs() error {
322322
s.lock.Lock()
323323
defer s.lock.Unlock()
324-
q := query.Query{Prefix: ""}
324+
q := query.Query{Prefix: dsBaseStartedDeals.String()}
325325
res, err := s.ds.Query(q)
326326
if err != nil {
327327
return fmt.Errorf("querying executing jobs in datastore: %s", err)

‎ffs/scheduler/internal/rjstore/rjstore.go‎

Lines changed: 206 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,206 @@
1+
package rjstore
2+
3+
import (
4+
"context"
5+
"encoding/json"
6+
"errors"
7+
"fmt"
8+
"sync"
9+
10+
datastore "github.com/ipfs/go-datastore"
11+
"github.com/ipfs/go-datastore/query"
12+
logging "github.com/ipfs/go-log/v2"
13+
"github.com/textileio/powergate/ffs"
14+
)
15+
16+
var (
17+
log = logging.Logger("ffs-sched-rjstore")
18+
19+
// ErrNotFound indicates the job doesn't exists.
20+
ErrNotFound = errors.New("retrieval job not found")
21+
dsBaseJob = datastore.NewKey("job")
22+
)
23+
24+
// Store is a persistent store for Retrieval Jobs.
25+
type Store struct {
26+
lock sync.Mutex
27+
ds datastore.Datastore
28+
watchers []watcher
29+
}
30+
31+
// watcher represents an API instance who is watching for
32+
// Job updates.
33+
type watcher struct {
34+
iid ffs.APIID
35+
C chan ffs.Job
36+
}
37+
38+
// New returns a new retrieval Job store.
39+
func New(ds datastore.Datastore) (*Store, error) {
40+
s := &Store{ds: ds}
41+
return s, nil
42+
}
43+
44+
// Finalize sets a Job status to a final state.
45+
func (s *Store) Finalize(jid ffs.JobID, st ffs.JobStatus, jobError error) error {
46+
s.lock.Lock()
47+
defer s.lock.Unlock()
48+
49+
j, err := s.get(jid)
50+
if err != nil {
51+
return err
52+
}
53+
switch st {
54+
case ffs.Success, ffs.Failed, ffs.Canceled:
55+
// Success: Job executed within expected behavior.
56+
// Failed: Job executed with expected failure scenario.
57+
// Canceled: Job was canceled by the client.
58+
default:
59+
return fmt.Errorf("can't finalize job with status %s", ffs.JobStatusStr[st])
60+
}
61+
j.Status = st
62+
if jobError != nil {
63+
j.ErrCause = jobError.Error()
64+
}
65+
if err := s.put(j); err != nil {
66+
return fmt.Errorf("saving in datastore: %s", err)
67+
}
68+
return nil
69+
}
70+
71+
// Dequeue dequeues a ready to be executed Job.
72+
func (s *Store) Dequeue() (*ffs.Job, error) {
73+
s.lock.Lock()
74+
defer s.lock.Unlock()
75+
q := query.Query{Prefix: dsBaseJob.String()}
76+
res, err := s.ds.Query(q)
77+
if err != nil {
78+
return nil, fmt.Errorf("querying datastore: %s", err)
79+
}
80+
defer func() {
81+
if err := res.Close(); err != nil {
82+
log.Errorf("closing dequeue query result: %s", err)
83+
}
84+
}()
85+
for r := range res.Next() {
86+
var j ffs.Job
87+
if err := json.Unmarshal(r.Value, &j); err != nil {
88+
return nil, fmt.Errorf("unmarshalling job: %s", err)
89+
}
90+
if j.Status == ffs.Queued {
91+
j.Status = ffs.Executing
92+
if err := s.put(j); err != nil {
93+
return nil, err
94+
}
95+
return &j, nil
96+
}
97+
}
98+
return nil, nil
99+
}
100+
101+
// Enqueue queues a new Job.
102+
func (s *Store) Enqueue(j ffs.Job) error {
103+
s.lock.Lock()
104+
defer s.lock.Unlock()
105+
j.Status = ffs.Queued
106+
if err := s.put(j); err != nil {
107+
return fmt.Errorf("saving to datastore: %s", err)
108+
}
109+
return nil
110+
}
111+
112+
// Get returns the current state of Job. If doesn't exist, returns
113+
// ErrNotFound.
114+
func (s *Store) Get(jid ffs.JobID) (ffs.Job, error) {
115+
s.lock.Lock()
116+
defer s.lock.Unlock()
117+
118+
return s.get(jid)
119+
}
120+
121+
// Watch subscribes to Job changes from a specified Api instance.
122+
func (s *Store) Watch(ctx context.Context, c chan<- ffs.Job, iid ffs.APIID) error {
123+
s.lock.Lock()
124+
ic := make(chan ffs.Job, 1)
125+
s.watchers = append(s.watchers, watcher{iid: iid, C: ic})
126+
s.lock.Unlock()
127+
128+
stop := false
129+
for !stop {
130+
select {
131+
case <-ctx.Done():
132+
stop = true
133+
case l, ok := <-ic:
134+
if !ok {
135+
return fmt.Errorf("jobstore was closed with a listening client")
136+
}
137+
c <- l
138+
}
139+
}
140+
141+
s.lock.Lock()
142+
defer s.lock.Unlock()
143+
for i := range s.watchers {
144+
if s.watchers[i].C == ic {
145+
s.watchers = append(s.watchers[:i], s.watchers[i+1:]...)
146+
break
147+
}
148+
}
149+
return nil
150+
}
151+
152+
// Close closes the Store, unregistering any subscribed watchers.
153+
func (s *Store) Close() error {
154+
s.lock.Lock()
155+
defer s.lock.Unlock()
156+
for i := range s.watchers {
157+
close(s.watchers[i].C)
158+
}
159+
s.watchers = nil
160+
return nil
161+
}
162+
163+
func (s *Store) put(j ffs.Job) error {
164+
buf, err := json.Marshal(j)
165+
if err != nil {
166+
return fmt.Errorf("marshaling for datastore: %s", err)
167+
}
168+
if err := s.ds.Put(makeKey(j.ID), buf); err != nil {
169+
return fmt.Errorf("saving to datastore: %s", err)
170+
}
171+
s.notifyWatchers(j)
172+
return nil
173+
}
174+
175+
func (s *Store) get(jid ffs.JobID) (ffs.Job, error) {
176+
buf, err := s.ds.Get(makeKey(jid))
177+
if err == datastore.ErrNotFound {
178+
return ffs.Job{}, ErrNotFound
179+
}
180+
if err != nil {
181+
return ffs.Job{}, fmt.Errorf("getting job from datastore: %s", err)
182+
}
183+
var job ffs.Job
184+
if err := json.Unmarshal(buf, &job); err != nil {
185+
return job, fmt.Errorf("unmarshaling job from datastore: %s", err)
186+
}
187+
return job, nil
188+
}
189+
190+
func (s *Store) notifyWatchers(j ffs.Job) {
191+
for _, w := range s.watchers {
192+
if w.iid != j.APIID {
193+
continue
194+
}
195+
select {
196+
case w.C <- j:
197+
log.Infof("notifying watcher")
198+
default:
199+
log.Warnf("slow watcher skipped job %s", j.ID)
200+
}
201+
}
202+
}
203+
204+
func makeKey(jid ffs.JobID) datastore.Key {
205+
return dsBaseJob.ChildString(jid.String())
206+
}

0 commit comments

Comments
 (0)