mirrors/ctrl
1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2025-01-18 13:49:29 +00:00
Code Activity

added seprata timwe for MaxAllowedCopy time which can be specified in the original message

Browse source
This commit is contained in:
postmannen 2022-06-16 07:12:03 +02:00
parent 1bdee3872b
commit f3f5e64c45
1 changed files with 51 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
requests_copy.go
View file

@ -10,6 +10,7 @@ import (
"os"
"path/filepath"
"strconv"
"time"
"github.com/fxamacker/cbor/v2"
"github.com/google/uuid"
@ -25,6 +26,7 @@ type copyInitialData struct {
DstDir string
DstFile string
SplitChunkSize int
MaxTotalCopyTime int
FileMode fs.FileMode
}
@ -94,7 +96,10 @@ func (m methodREQCopySrc) handler(proc process, message Message, node string) ([
// Set default split chunk size, will be replaced with value from
// methodArgs[3] if defined.
splitChunkSize := 2
splitChunkSize := 100000
// Set default max total copy time, will be replaced with value from
// methodArgs[4] if defined.
maxTotalCopyTime := message.MethodTimeout
// Verify and check the methodArgs
switch {
@ -108,7 +113,16 @@ func (m methodREQCopySrc) handler(proc process, message Message, node string) ([
var err error
splitChunkSize, err = strconv.Atoi(message.MethodArgs[3])
if err != nil {
er := fmt.Errorf("error: methodREQCopySrc: ch")
er := fmt.Errorf("error: methodREQCopySrc: unble to convert splitChunkSize into int value: %v", err)
proc.errorKernel.errSend(proc, message, er)
}
case len(message.MethodArgs) > 4:
// Check if split chunk size was set, if not set default.
var err error
maxTotalCopyTime, err = strconv.Atoi(message.MethodArgs[3])
if err != nil {
er := fmt.Errorf("error: methodREQCopySrc: unble to convert maxTotalCopyTime into int value: %v", err)
proc.errorKernel.errSend(proc, message, er)
}
}
@ -119,11 +133,13 @@ func (m methodREQCopySrc) handler(proc process, message Message, node string) ([
DstNode := message.MethodArgs[1]
DstFilePath := message.MethodArgs[2]
// Get a context with the timeout specified in message.MethodTimeout.
// Since the subProc spawned will outlive this method here we do not
// want to cancel this method. We care about the methodTimeout, but
// we ignore the CancelFunc.
ctx, cancel := getContextForMethodTimeout(proc.ctx, message)
// Create a child context to use with the procFunc with timeout set to the max allowed total copy time
// specified in the message.
var ctx context.Context
var cancel context.CancelFunc
func() {
ctx, cancel = context.WithTimeout(proc.ctx, time.Second*time.Duration(maxTotalCopyTime))
}()
// Create a subject for one copy request
uid := uuid.New()
@ -159,6 +175,7 @@ func (m methodREQCopySrc) handler(proc process, message Message, node string) ([
DstDir: dstDir,
DstFile: dstFile,
SplitChunkSize: splitChunkSize,
MaxTotalCopyTime: maxTotalCopyTime,
FileMode: fileMode,
}
@ -245,11 +262,13 @@ func (m methodREQCopyDst) handler(proc process, message Message, node string) ([
return
}
// Get a context with the timeout specified in message.MethodTimeout.
// Since the subProc spawned will outlive this method here we do not
// want to cancel this method. We care about the methodTimeout, but
// we ignore the CancelFunc.
ctx, cancel := getContextForMethodTimeout(proc.ctx, message)
// Create a child context to use with the procFunc with timeout set to the max allowed total copy time
// specified in the message.
var ctx context.Context
var cancel context.CancelFunc
func() {
ctx, cancel = context.WithTimeout(proc.ctx, time.Second*time.Duration(cia.MaxTotalCopyTime))
}()
// Create a subject for one copy request
sub := newSubjectNoVerifyHandler(cia.DstMethod, node)