Skip to content

Commit e33d2c6

Browse files
dabasovdabasov
authored
Merge pull request #1437 from 0chain/feat/connection-hash
Add connection hash and optimize change creation
2 parents d2ced01 + 454a3eb commit e33d2c6

File tree

9 files changed

+52
-23
lines changed

9 files changed

+52
-23
lines changed

‎code/go/0chain.net/blobbercore/allocation/allocationchange.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ type AllocationChange struct {
7373
Connection AllocationChangeCollector `gorm:"foreignKey:ConnectionID"` // References allocation_connections(id)
7474
Input string `gorm:"column:input"`
7575
FilePath string `gorm:"-"`
76+
LookupHash string `gorm:"column:lookup_hash;size:64"`
7677
datastore.ModelWithTS
7778
}
7879

@@ -97,6 +98,11 @@ func (change *AllocationChange) Save(ctx context.Context) error {
9798
return db.Save(change).Error
9899
}
99100

101+
func (change *AllocationChange) Create(ctx context.Context) error {
102+
db := datastore.GetStore().GetTransaction(ctx)
103+
return db.Create(change).Error
104+
}
105+
100106
func ParseAffectedFilePath(input string) (string, error) {
101107
inputMap := make(map[string]interface{})
102108
err := json.Unmarshal([]byte(input), &inputMap)

‎code/go/0chain.net/blobbercore/allocation/connection.go‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -182,11 +182,7 @@ func SaveFileChange(ctx context.Context, connectionID, pathHash, fileName string
182182
change.lock.Lock()
183183
defer change.lock.Unlock()
184184
connectionObj.lock.Unlock()
185-
dbConnectionObj, err := GetAllocationChanges(ctx, connectionID, connectionObj.AllocationID, connectionObj.ClientID)
186-
if err != nil {
187-
return saveChange, err
188-
}
189-
err = cmd.UpdateChange(ctx, dbConnectionObj)
185+
err := cmd.AddChange(ctx)
190186
if err != nil {
191187
return saveChange, err
192188
}

‎code/go/0chain.net/blobbercore/allocation/file_changer_base.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,9 @@ type FileCommand interface {
9696
// UpdateChange update AllocationChangeProcessor. It will be president in db for committing transcation
9797
UpdateChange(ctx context.Context, connectionObj *AllocationChangeCollector) error
9898

99+
// AddChange add Allocation change to db
100+
AddChange(ctx context.Context) error
101+
99102
//NumBlocks return number of blocks uploaded by the client
100103
GetNumBlocks() int64
101104
}

‎code/go/0chain.net/blobbercore/handler/file_command_delete.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@ func (cmd *DeleteFileCommand) IsValidated(ctx context.Context, req *http.Request
5757
}
5858
return common.NewError("bad_db_operation", err.Error())
5959
}
60+
cmd.existingFileRef.LookupHash = lookUpHash
6061
return nil
6162
}
6263

@@ -67,6 +68,12 @@ func (cmd *DeleteFileCommand) UpdateChange(ctx context.Context, connectionObj *a
6768
return connectionObj.Save(ctx)
6869
}
6970

71+
func (cmd *DeleteFileCommand) AddChange(ctx context.Context) error {
72+
connectionInput, _ := cmd.changeProcessor.Marshal()
73+
cmd.allocationChange.Input = connectionInput
74+
return cmd.allocationChange.Create(ctx)
75+
}
76+
7077
// ProcessContent flush file to FileStorage
7178
func (cmd *DeleteFileCommand) ProcessContent(_ context.Context, allocationObj *allocation.Allocation) (allocation.UploadResult, error) {
7279
deleteSize := cmd.existingFileRef.Size
@@ -86,6 +93,7 @@ func (cmd *DeleteFileCommand) ProcessContent(_ context.Context, allocationObj *a
8693
cmd.allocationChange.ConnectionID = connectionID
8794
cmd.allocationChange.Size = 0 - deleteSize
8895
cmd.allocationChange.Operation = constants.FileOperationDelete
96+
cmd.allocationChange.LookupHash = cmd.existingFileRef.LookupHash
8997

9098
allocation.UpdateConnectionObjSize(connectionID, cmd.allocationChange.Size)
9199

‎code/go/0chain.net/blobbercore/handler/file_command_update.go‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -231,6 +231,13 @@ func (cmd *UpdateFileCommand) UpdateChange(ctx context.Context, connectionObj *a
231231
return connectionObj.Save(ctx)
232232
}
233233

234+
func (cmd *UpdateFileCommand) AddChange(ctx context.Context) error {
235+
connectionInput, _ := cmd.fileChanger.Marshal()
236+
cmd.allocationChange.LookupHash = cmd.existingFileRef.LookupHash
237+
cmd.allocationChange.Input = connectionInput
238+
return cmd.allocationChange.Create(ctx)
239+
}
240+
234241
func (cmd *UpdateFileCommand) GetNumBlocks() int64 {
235242
if cmd.fileChanger.IsFinal {
236243
return int64(math.Ceil(float64(cmd.fileChanger.Size*1.0) / float64(cmd.fileChanger.ChunkSize)))

‎code/go/0chain.net/blobbercore/handler/file_command_upload.go‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,6 @@ func (cmd *UploadFileCommand) IsValidated(ctx context.Context, req *http.Request
123123

124124
// ProcessContent flush file to FileStorage
125125
func (cmd *UploadFileCommand) ProcessContent(ctx context.Context, allocationObj *allocation.Allocation) (allocation.UploadResult, error) {
126-
logging.Logger.Info("UploadFileCommand.ProcessContent", zap.Any("fileChanger", cmd.fileChanger.Path), zap.Any("uploadOffset", cmd.fileChanger.UploadOffset), zap.Any("isFinal", cmd.fileChanger.IsFinal))
127126
result := allocation.UploadResult{}
128127
defer cmd.contentFile.Close()
129128

@@ -144,13 +143,11 @@ func (cmd *UploadFileCommand) ProcessContent(ctx context.Context, allocationObj
144143
logging.Logger.Error("UploadFileCommand.ProcessContent", zap.Error(err))
145144
return result, common.NewError("upload_error", "Failed to write file. "+err.Error())
146145
}
147-
148146
result.Filename = cmd.fileChanger.Filename
149147
result.ValidationRoot = fileOutputData.ValidationRoot
150148
result.Size = fileOutputData.Size
151149

152150
allocationSize := allocation.GetConnectionObjSize(connectionID)
153-
154151
cmd.fileChanger.AllocationID = allocationObj.ID
155152

156153
cmd.allocationChange = &allocation.AllocationChange{}
@@ -233,7 +230,7 @@ func (cmd *UploadFileCommand) UpdateChange(ctx context.Context, connectionObj *a
233230
if c.Operation != constants.FileOperationInsert || cmd.fileChanger.Path != filePath {
234231
continue
235232
}
236-
c.Size = connectionObj.Size
233+
c.Size = cmd.fileChanger.Size
237234
c.Input, _ = cmd.fileChanger.Marshal()
238235

239236
//c.ModelWithTS.UpdatedAt = time.Now()
@@ -251,6 +248,13 @@ func (cmd *UploadFileCommand) UpdateChange(ctx context.Context, connectionObj *a
251248
return connectionObj.Save(ctx)
252249
}
253250

251+
func (cmd *UploadFileCommand) AddChange(ctx context.Context) error {
252+
connectionInput, _ := cmd.fileChanger.Marshal()
253+
cmd.allocationChange.LookupHash = reference.GetReferenceLookup(cmd.fileChanger.AllocationID, cmd.fileChanger.Path)
254+
cmd.allocationChange.Input = connectionInput
255+
return cmd.allocationChange.Create(ctx)
256+
}
257+
254258
func (cmd *UploadFileCommand) GetNumBlocks() int64 {
255259
if cmd.fileChanger.IsFinal {
256260
return int64(math.Ceil(float64(cmd.fileChanger.Size*1.0) / float64(cmd.fileChanger.ChunkSize)))

‎code/go/0chain.net/blobbercore/handler/handler_test.go‎

Lines changed: 4 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -277,7 +277,7 @@ func TestHandlers_Requiring_Signature(t *testing.T) {
277277
WithArgs(aa, aa, aa, aa, aa, aa, aa).
278278
WillReturnResult(sqlmock.NewResult(0, 0))
279279
mock.ExpectQuery(regexp.QuoteMeta(`INSERT INTO "allocation_changes"`)).
280-
WithArgs(aa, aa, aa, aa, aa, aa).
280+
WithArgs(aa, aa, aa, aa, aa, aa, aa).
281281
WillReturnRows(
282282
sqlmock.NewRows([]string{}),
283283
)
@@ -657,7 +657,7 @@ func TestHandlers_Requiring_Signature(t *testing.T) {
657657
WillReturnResult(sqlmock.NewResult(0, 0))
658658

659659
mock.ExpectQuery(regexp.QuoteMeta(`INSERT INTO "allocation_changes"`)).
660-
WithArgs(aa, aa, aa, aa, aa, aa).
660+
WithArgs(aa, aa, aa, aa, aa, aa, aa).
661661
WillReturnRows(
662662
sqlmock.NewRows([]string{}),
663663
)
@@ -743,7 +743,7 @@ func TestHandlers_Requiring_Signature(t *testing.T) {
743743
WillReturnResult(sqlmock.NewResult(0, 0))
744744

745745
mock.ExpectQuery(regexp.QuoteMeta(`INSERT INTO "allocation_changes"`)).
746-
WithArgs(aa, aa, aa, aa, aa, aa).
746+
WithArgs(aa, aa, aa, aa, aa, aa, aa).
747747
WillReturnRows(
748748
sqlmock.NewRows([]string{}),
749749
)
@@ -847,17 +847,8 @@ func TestHandlers_Requiring_Signature(t *testing.T) {
847847
sqlmock.NewRows([]string{"found"}).
848848
AddRow(false),
849849
)
850-
mock.ExpectQuery(regexp.QuoteMeta(`SELECT * FROM "allocation_connections" WHERE`)).
851-
WithArgs(connectionID, alloc.ID, alloc.OwnerID, allocation.DeletedConnection).
852-
WillReturnRows(
853-
sqlmock.NewRows([]string{}).
854-
AddRow(),
855-
)
856-
mock.ExpectExec(`INSERT INTO "allocation_connections"`).
857-
WithArgs(aa, aa, aa, aa, aa, aa, aa).
858-
WillReturnResult(sqlmock.NewResult(0, 0))
859850
mock.ExpectQuery(regexp.QuoteMeta(`INSERT INTO "allocation_changes"`)).
860-
WithArgs(aa, aa, aa, aa, aa, aa).
851+
WithArgs(aa, aa, aa, aa, aa, aa, aa).
861852
WillReturnRows(
862853
sqlmock.NewRows([]string{}),
863854
)

‎code/go/0chain.net/blobbercore/handler/object_operation_handler.go‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -842,6 +842,7 @@ func (fsh *StorageHandler) RenameObject(ctx context.Context, r *http.Request) (i
842842
allocationChange := &allocation.AllocationChange{}
843843
allocationChange.ConnectionID = connectionObj.ID
844844
allocationChange.Size = 0
845+
allocationChange.LookupHash = pathHash
845846
allocationChange.Operation = constants.FileOperationRename
846847
dfc := &allocation.RenameFileChange{ConnectionID: connectionObj.ID,
847848
AllocationID: connectionObj.AllocationID, Path: objectRef.Path, Type: objectRef.Type}
@@ -951,6 +952,7 @@ func (fsh *StorageHandler) CopyObject(ctx context.Context, r *http.Request) (int
951952
allocationChange := &allocation.AllocationChange{}
952953
allocationChange.ConnectionID = connectionObj.ID
953954
allocationChange.Size = objectRef.Size
955+
allocationChange.LookupHash = pathHash
954956
allocationChange.Operation = constants.FileOperationCopy
955957
dfc := &allocation.CopyFileChange{ConnectionID: connectionObj.ID,
956958
AllocationID: connectionObj.AllocationID, DestPath: destPath}
@@ -1062,6 +1064,7 @@ func (fsh *StorageHandler) MoveObject(ctx context.Context, r *http.Request) (int
10621064
allocationChange := &allocation.AllocationChange{}
10631065
allocationChange.ConnectionID = connectionObj.ID
10641066
allocationChange.Size = 0
1067+
allocationChange.LookupHash = pathHash
10651068
allocationChange.Operation = constants.FileOperationMove
10661069
dfc := &allocation.MoveFileChange{
10671070
ConnectionID: connectionObj.ID,
@@ -1278,12 +1281,15 @@ func (fsh *StorageHandler) WriteFile(ctx context.Context, r *http.Request) (*all
12781281
}
12791282

12801283
elapsedVerifySig := time.Since(st)
1284+
st = time.Now()
12811285

12821286
cmd := createFileCommand(r)
12831287
err = cmd.IsValidated(ctx, r, allocationObj, clientID)
12841288
if err != nil {
12851289
return nil, err
12861290
}
1291+
elapsedIsValidated := time.Since(st)
1292+
st = time.Now()
12871293
// call process content, which writes to file checks if conn obj needs to be updated and if commit hasher needs to be called
12881294
res, err := cmd.ProcessContent(ctx, allocationObj)
12891295
if err != nil {
@@ -1305,14 +1311,16 @@ func (fsh *StorageHandler) WriteFile(ctx context.Context, r *http.Request) (*all
13051311
if blocks > 0 {
13061312
go AddUploadedData(clientID, blocks)
13071313
}
1314+
elapsedProcessContent := time.Since(st)
13081315
Logger.Info("[upload]elapsed",
13091316
zap.String("alloc_id", allocationID),
13101317
zap.String("file", cmd.GetPath()),
13111318
zap.Duration("parse_form", elapsedParseForm),
13121319
zap.Duration("get_processor", elapsedGetConnectionProcessor),
13131320
zap.Duration("get_alloc", elapsedAllocation),
13141321
zap.Duration("sig", elapsedVerifySig),
1315-
zap.Duration("validate", time.Since(st)),
1322+
zap.Duration("validate", elapsedIsValidated),
1323+
zap.Duration("process_content", elapsedProcessContent),
13161324
zap.Duration("total", time.Since(startTime)))
13171325
return &res, nil
13181326
}

‎goose/migrations/1717416291_change_lookuphash.sql‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
-- +goose Up
2+
-- +goose StatementBegin
3+
ALTER TABLE allocation_changes ADD COLUMN lookup_hash character varying(64);
4+
5+
-- CREATE UNIQUE INDEX idx_allocation_changes_lookup_hash ON allocation_changes USING HASH(lookup_hash,connection_id);
6+
-- +goose StatementEnd

0 commit comments

Comments
 (0)