Start uploading buffered files immediately

Author: mike76-dev

modules/manager/contractor/contractor.go

@@ -49,6 +49,7 @@ type Contractor struct {
 	interruptMaintenance   chan struct{}
 	maintenanceLock        siasync.TryMutex
 	uploadingBufferedFiles bool
+	runningUploads         map[string]func()
 
 	blockHeight uint64
 	synced      chan struct{}
@@ -336,6 +337,7 @@ func contractorBlockingStartup(db *sql.DB, cs modules.ConsensusSet, m modules.Ma
 
 		interruptMaintenance: make(chan struct{}),
 		synced:               make(chan struct{}),
+		runningUploads:       make(map[string]func()),
 
 		renters: make(map[types.PublicKey]modules.Renter),
 
@@ -690,3 +692,29 @@ func (c *Contractor) DownloadObject(w io.Writer, rpk types.PublicKey, bucket, pa
 
 	return err
 }
+
+// StartUploading initiates uploading buffered files.
+func (c *Contractor) StartUploading() {
+	if err := c.tg.Add(); err != nil {
+		return
+	}
+	defer c.tg.Done()
+	c.managedUploadBufferedFiles()
+}
+
+// CancelUpload cancels a running upload.
+func (c *Contractor) CancelUpload(rpk types.PublicKey, bucket, path []byte) {
+	pk := make([]byte, 32)
+	copy(pk, rpk[:])
+	mapKey := string(pk) + string(bucket) + ":" + string(path)
+	c.mu.Lock()
+	cancelFunc, ok := c.runningUploads[mapKey]
+	if ok {
+		delete(c.runningUploads, mapKey)
+		cancelFunc()
+		if len(c.runningUploads) == 0 {
+			c.uploadingBufferedFiles = false
+		}
+	}
+	c.mu.Unlock()
+}

modules/manager/contractor/database.go

@@ -1522,6 +1522,7 @@ func (c *Contractor) managedUploadBufferedFiles() {
 	c.mu.Lock()
 	if c.uploadingBufferedFiles {
 		c.mu.Unlock()
+		c.log.Println("INFO: skipping file uploads since another thread is running already")
 		return
 	}
 	c.uploadingBufferedFiles = true
@@ -1548,6 +1549,11 @@ func (c *Contractor) managedUploadBufferedFiles() {
 	defer rows.Close()
 
 	for rows.Next() {
+		select {
+		case <-c.tg.StopChan():
+			return
+		default:
+		}
 		var n, encrypted string
 		pk := make([]byte, 32)
 		var bucket, path, mimeType []byte

modules/manager/contractor/upload.go

@@ -230,7 +230,18 @@ func (mgr *uploadManager) migrate(ctx context.Context, rpk types.PublicKey, shar
 func (c *Contractor) managedUploadObject(r io.Reader, rpk types.PublicKey, bucket, path, mimeType []byte, encrypted string) (fm modules.FileMetadata, err error) {
 	// Create the context and setup its cancelling.
 	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	pk := make([]byte, 32)
+	copy(pk, rpk[:])
+	mapKey := string(pk) + string(bucket) + ":" + string(path)
+	c.mu.Lock()
+	c.runningUploads[mapKey] = cancel
+	c.mu.Unlock()
+	defer func() {
+		c.mu.Lock()
+		delete(c.runningUploads, mapKey)
+		cancel()
+		c.mu.Unlock()
+	}()
 
 	// Fetch necessary params.
 	c.mu.RLock()

modules/manager/database.go

@@ -642,7 +642,14 @@ func (m *Manager) RegisterUpload(pk types.PublicKey, bucket, path, mimeType []by
 		complete,
 		encText,
 	)
-	return err
+	if err != nil {
+		return err
+	}
+
+	// Initiate the upload.
+	go m.hostContractor.StartUploading()
+
+	return nil
 }
 
 // GetBufferSize returns the total size of the temporary files.
@@ -677,20 +684,27 @@ func (m *Manager) GetBufferSize() (total uint64, err error) {
 // DeleteBufferedFiles deletes the files waiting to be uploaded.
 func (m *Manager) DeleteBufferedFiles(pk types.PublicKey) error {
 	// Make a list of file names.
-	rows, err := m.db.Query("SELECT filename FROM ctr_uploads WHERE renter_pk = ?", pk[:])
+	rows, err := m.db.Query(`
+		SELECT filename, bucket, filepath
+		FROM ctr_uploads
+		WHERE renter_pk = ?
+	`, pk[:])
 	if err != nil {
 		m.log.Println("ERROR: unable to query files:", err)
 		return modules.AddContext(err, "unable to query files")
 	}
 
+	// Make a list of filenames and also cancel any running uploads.
 	var names []string
 	for rows.Next() {
 		var name string
-		if err := rows.Scan(&name); err != nil {
+		var bucket, path []byte
+		if err := rows.Scan(&name, &bucket, &path); err != nil {
 			rows.Close()
 			m.log.Println("ERROR: unable to retrieve filename:", err)
 			return modules.AddContext(err, "unable to retrieve filename")
 		}
+		m.hostContractor.CancelUpload(pk, bucket, path)
 		names = append(names, name)
 	}
 	rows.Close()
@@ -723,6 +737,9 @@ func (m *Manager) DeleteBufferedFiles(pk types.PublicKey) error {
 // DeleteBufferedFile deletes the specified file and the associated
 // database record.
 func (m *Manager) DeleteBufferedFile(pk types.PublicKey, bucket, path []byte) error {
+	// Cancel the upload.
+	m.hostContractor.CancelUpload(pk, bucket, path)
+
 	var name string
 	err := m.db.QueryRow(`
 		SELECT filename
@@ -1203,5 +1220,8 @@ func (m *Manager) AssembleParts(pk types.PublicKey, id types.Hash256) (err error
 		return modules.AddContext(err, "couldn't register new upload")
 	}
 
+	// Initiate the upload.
+	go m.hostContractor.StartUploading()
+
 	return
 }

modules/manager/manager.go

@@ -56,6 +56,9 @@ type hostContractor interface {
 	// CancelContract cancels the renter's contract.
 	CancelContract(types.FileContractID) error
 
+	// CancelUpload cancels a running upload.
+	CancelUpload(types.PublicKey, []byte, []byte)
+
 	// Contract returns the contract with the given ID.
 	Contract(types.FileContractID) (modules.RenterContract, bool)
 
@@ -136,6 +139,9 @@ type hostContractor interface {
 	// RetrieveMetadata retrieves the file metadata from the database.
 	RetrieveMetadata(types.PublicKey, []modules.BucketFiles) ([]modules.FileMetadata, error)
 
+	// StartUploading initiates uploading buffered files.
+	StartUploading()
+
 	// Synced returns a channel that is closed when the contractor is fully
 	// synced with the peer-to-peer network.
 	Synced() <-chan struct{}