<commit_before>package mountlib

import (
	"io"
	"sync"

	"github.com/ncw/rclone/fs"
	"github.com/pkg/errors"
)

// WriteFileHandle is an open for write handle on a File
type WriteFileHandle struct {
	mu          sync.Mutex
	closed      bool // set if handle has been closed
	remote      string
	pipeReader  *io.PipeReader
	pipeWriter  *io.PipeWriter
	o           fs.Object
	result      chan error
	file        *File
	writeCalled bool // set the first time Write() is called
	offset      int64
	hash        *fs.MultiHasher
}

func newWriteFileHandle(d *Dir, f *File, src fs.ObjectInfo) (*WriteFileHandle, error) {
	var hash *fs.MultiHasher
	if !f.d.fsys.noChecksum {
		var err error
		hash, err = fs.NewMultiHasherTypes(src.Fs().Hashes())
		if err != nil {
			fs.Errorf(src.Fs(), "newWriteFileHandle hash error: %v", err)
		}
	}

	fh := &WriteFileHandle{
		remote: src.Remote(),
		result: make(chan error, 1),
		file:   f,
		hash:   hash,
	}
	fh.pipeReader, fh.pipeWriter = io.Pipe()
	r := fs.NewAccountSizeName(fh.pipeReader, 0, src.Remote()).WithBuffer() // account the transfer
	go func() {
		o, err := d.f.Put(r, src)
		fh.o = o
		fh.result <- err
	}()
	fh.file.addWriters(1)
	fh.file.setSize(0)
	fs.Stats.Transferring(fh.remote)
	return fh, nil
}

// String converts it to printable
func (fh *WriteFileHandle) String() string {
	if fh == nil {
		return "<nil *WriteFileHandle>"
	}
	if fh.file == nil {
		return "<nil *WriteFileHandle.file>"
	}
	return fh.file.String() + " (w)"
}

// Node returns the Node assocuated with this - satisfies Noder interface
func (fh *WriteFileHandle) Node() Node {
	return fh.file
}

// Write data to the file handle
func (fh *WriteFileHandle) Write(data []byte, offset int64) (written int64, err error) {
	// fs.Debugf(fh.remote, "WriteFileHandle.Write len=%d", len(data))
	fh.mu.Lock()
	defer fh.mu.Unlock()
	if fh.offset != offset {
		fs.Errorf(fh.remote, "WriteFileHandle.Write can't seek in file")
		return 0, ESPIPE
	}
	if fh.closed {
		fs.Errorf(fh.remote, "WriteFileHandle.Write error: %v", EBADF)
		return 0, EBADF
	}
	fh.writeCalled = true
	// FIXME should probably check the file isn't being seeked?
	n, err := fh.pipeWriter.Write(data)
	written = int64(n)
	fh.offset += written
	fh.file.setSize(fh.offset)
	if err != nil {
		fs.Errorf(fh.remote, "WriteFileHandle.Write error: %v", err)
		return 0, err
	}
	// fs.Debugf(fh.remote, "WriteFileHandle.Write OK (%d bytes written)", n)
	if fh.hash != nil {
		_, err = fh.hash.Write(data[:n])
		if err != nil {
			fs.Errorf(fh.remote, "WriteFileHandle.Write HashError: %v", err)
			return written, err
		}
	}
	return written, nil
}

// Offset returns the offset of the file pointer
func (fh *WriteFileHandle) Offset() (offset int64) {
	return fh.offset
}

// close the file handle returning EBADF if it has been
// closed already.
//
// Must be called with fh.mu held
func (fh *WriteFileHandle) close() error {
	if fh.closed {
		return EBADF
	}
	fh.closed = true
	fs.Stats.DoneTransferring(fh.remote, true)
	fh.file.addWriters(-1)
	writeCloseErr := fh.pipeWriter.Close()
	err := <-fh.result
	readCloseErr := fh.pipeReader.Close()
	if err == nil {
		fh.file.setObject(fh.o)
		err = writeCloseErr
	}
	if err == nil {
		err = readCloseErr
	}
	if err == nil && fh.hash != nil {
		for hashType, srcSum := range fh.hash.Sums() {
			dstSum, err := fh.o.Hash(hashType)
			if err != nil {
				return err
			}
			if !fs.HashEquals(srcSum, dstSum) {
				return errors.Errorf("corrupted on transfer: %v hash differ %q vs %q", hashType, srcSum, dstSum)
			}
		}
	}
	return err
}

// Flush is called on each close() of a file descriptor. So if a
// filesystem wants to return write errors in close() and the file has<commit_msg>mount: fix hang on errored upload<commit_after>package mountlib

import (
	"io"
	"sync"

	"github.com/ncw/rclone/fs"
	"github.com/pkg/errors"
)

// WriteFileHandle is an open for write handle on a File
type WriteFileHandle struct {
	mu          sync.Mutex
	closed      bool // set if handle has been closed
	remote      string
	pipeWriter  *io.PipeWriter
	o           fs.Object
	result      chan error
	file        *File
	writeCalled bool // set the first time Write() is called
	offset      int64
	hash        *fs.MultiHasher
}

func newWriteFileHandle(d *Dir, f *File, src fs.ObjectInfo) (*WriteFileHandle, error) {
	var hash *fs.MultiHasher
	if !f.d.fsys.noChecksum {
		var err error
		hash, err = fs.NewMultiHasherTypes(src.Fs().Hashes())
		if err != nil {
			fs.Errorf(src.Fs(), "newWriteFileHandle hash error: %v", err)
		}
	}

	fh := &WriteFileHandle{
		remote: src.Remote(),
		result: make(chan error, 1),
		file:   f,
		hash:   hash,
	}
	var pipeReader *io.PipeReader
	pipeReader, fh.pipeWriter = io.Pipe()
	go func() {
		r := fs.NewAccountSizeName(pipeReader, 0, src.Remote()).WithBuffer() // account the transfer
		o, err := d.f.Put(r, src)
		if err != nil {
			fs.Errorf(fh.remote, "WriteFileHandle.New Put failed: %v", err)
		}
		// Close the Account and thus the pipeReader so the pipeWriter fails with ErrClosedPipe
		_ = r.Close()
		fh.o = o
		fh.result <- err
	}()
	fh.file.addWriters(1)
	fh.file.setSize(0)
	fs.Stats.Transferring(fh.remote)
	return fh, nil
}

// String converts it to printable
func (fh *WriteFileHandle) String() string {
	if fh == nil {
		return "<nil *WriteFileHandle>"
	}
	if fh.file == nil {
		return "<nil *WriteFileHandle.file>"
	}
	return fh.file.String() + " (w)"
}

// Node returns the Node assocuated with this - satisfies Noder interface
func (fh *WriteFileHandle) Node() Node {
	return fh.file
}

// Write data to the file handle
func (fh *WriteFileHandle) Write(data []byte, offset int64) (written int64, err error) {
	// fs.Debugf(fh.remote, "WriteFileHandle.Write len=%d", len(data))
	fh.mu.Lock()
	defer fh.mu.Unlock()
	if fh.offset != offset {
		fs.Errorf(fh.remote, "WriteFileHandle.Write can't seek in file")
		return 0, ESPIPE
	}
	if fh.closed {
		fs.Errorf(fh.remote, "WriteFileHandle.Write error: %v", EBADF)
		return 0, EBADF
	}
	fh.writeCalled = true
	// FIXME should probably check the file isn't being seeked?
	n, err := fh.pipeWriter.Write(data)
	written = int64(n)
	fh.offset += written
	fh.file.setSize(fh.offset)
	if err != nil {
		fs.Errorf(fh.remote, "WriteFileHandle.Write error: %v", err)
		return 0, err
	}
	// fs.Debugf(fh.remote, "WriteFileHandle.Write OK (%d bytes written)", n)
	if fh.hash != nil {
		_, err = fh.hash.Write(data[:n])
		if err != nil {
			fs.Errorf(fh.remote, "WriteFileHandle.Write HashError: %v", err)
			return written, err
		}
	}
	return written, nil
}

// Offset returns the offset of the file pointer
func (fh *WriteFileHandle) Offset() (offset int64) {
	return fh.offset
}

// close the file handle returning EBADF if it has been
// closed already.
//
// Must be called with fh.mu held
func (fh *WriteFileHandle) close() error {
	if fh.closed {
		return EBADF
	}
	fh.closed = true
	fs.Stats.DoneTransferring(fh.remote, true)
	fh.file.addWriters(-1)
	writeCloseErr := fh.pipeWriter.Close()
	err := <-fh.result
	if err == nil {
		fh.file.setObject(fh.o)
		err = writeCloseErr
	}
	if err == nil && fh.hash != nil {
		for hashType, srcSum := range fh.hash.Sums() {
			dstSum, err := fh.o.Hash(hashType)
			if err != nil {
				return err
			}
			if !fs.HashEquals(srcSum, dstSum) {
				return errors.Errorf("corrupted on transfer: %v hash differ %q vs %q", hashType, srcSum, dstSum)
			}
		}
	}
	return err
}

// Flush is called on each close() of a file descriptor. So if a
// filesystem wants to return write errors in close() and the file has
{'fraction_non_alphanumeric': 0.1332361129524062, 'fraction_numerical': 0.004772636881877237, 'mean_word_length': 3.266968325791855, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3998170', 'n_tokens_mistral': 2941, 'n_tokens_neox': 2737, 'n_words': 807}