2eb660d4b7
* Moving retrieve into multiple goroutines * Make sure we pass nil errors into err channel * restore tests * incorporate locks in retrieve.go * deserialize data only if we have something to deserealize when receiving bandwidth allocation in server store * Adding logic for retrieve to be more efficient * Add channel? * hmm * implement Throttle concurrency primitive * using throttle * Remove unused variables * Egon comments addressed * Get ba total correct * Consume without waiting * incrementally increase signing size * Get downloads working with throttle * Removed logging * Make sure we handle errors properly * Fix tests > > Co-authored-by: Kaloyan <kaloyan@storj.io> * Can't Fatalf in goroutine * Add missing returns to tests * add capacity to channel, smarter allocations * rename things and don't use size as limit * replace things with sync2.Throttle * fix compilation errors * add note about security * fix ordering * Max length is actually 64 bytes for piece ID * Max length is actually 64 bytes for piece ID * fix limit * error comes from pending allocs, so no need to relog * Optimize throughput * TODO * Deleted allocation manager * Return when someone sends a smaller bandwidth allocation than the previous message * review comments
81 lines
2.0 KiB
Go
81 lines
2.0 KiB
Go
// Copyright (C) 2018 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package server
|
|
|
|
import (
|
|
"github.com/gogo/protobuf/proto"
|
|
"storj.io/storj/pkg/utils"
|
|
pb "storj.io/storj/protos/piecestore"
|
|
)
|
|
|
|
// StreamWriter -- Struct for writing piece to server upload stream
|
|
type StreamWriter struct {
|
|
server *Server
|
|
stream pb.PieceStoreRoutes_RetrieveServer
|
|
}
|
|
|
|
// NewStreamWriter returns a new StreamWriter
|
|
func NewStreamWriter(s *Server, stream pb.PieceStoreRoutes_RetrieveServer) *StreamWriter {
|
|
return &StreamWriter{server: s, stream: stream}
|
|
}
|
|
|
|
// Write -- Write method for piece upload to stream for Server.Retrieve
|
|
func (s *StreamWriter) Write(b []byte) (int, error) {
|
|
// Write the buffer to the stream we opened earlier
|
|
if err := s.stream.Send(&pb.PieceRetrievalStream{Size: int64(len(b)), Content: b}); err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
return len(b), nil
|
|
}
|
|
|
|
// StreamReader is a struct for Retrieving data from server
|
|
type StreamReader struct {
|
|
src *utils.ReaderSource
|
|
bandwidthAllocation *pb.RenterBandwidthAllocation
|
|
currentTotal int64
|
|
}
|
|
|
|
// NewStreamReader returns a new StreamReader for Server.Store
|
|
func NewStreamReader(s *Server, stream pb.PieceStoreRoutes_StoreServer) *StreamReader {
|
|
sr := &StreamReader{}
|
|
sr.src = utils.NewReaderSource(func() ([]byte, error) {
|
|
|
|
recv, err := stream.Recv()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
pd := recv.GetPiecedata()
|
|
ba := recv.GetBandwidthallocation()
|
|
|
|
if ba != nil {
|
|
if err = s.verifySignature(stream.Context(), ba); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
deserializedData := &pb.RenterBandwidthAllocation_Data{}
|
|
err = proto.Unmarshal(ba.GetData(), deserializedData)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Update bandwidthallocation to be stored
|
|
if deserializedData.GetTotal() > sr.currentTotal {
|
|
sr.bandwidthAllocation = ba
|
|
sr.currentTotal = deserializedData.GetTotal()
|
|
}
|
|
}
|
|
|
|
return pd.GetContent(), nil
|
|
})
|
|
|
|
return sr
|
|
}
|
|
|
|
// Read -- Read method for piece download from stream
|
|
func (s *StreamReader) Read(b []byte) (int, error) {
|
|
return s.src.Read(b)
|
|
}
|