Time out in-progress transfers

If an entire part is a single libcurl operation, a client that starts
writing - but then leaves for an extended period of time - will leave
dangling references inside libcurl (eventually exhausting the number
of allowable transfers).

This commit adds a background thread for S3File that will go through
all pending uploads and check to see if they are still live; if not,
then it'll timeout the operation.

Author: bbockelm

src/HTTPCommands.cc

@@ -43,6 +43,7 @@ std::shared_ptr<HandlerQueue> HTTPRequest::m_queue =
 	std::make_unique<HandlerQueue>();
 bool HTTPRequest::m_workers_initialized = false;
 std::vector<CurlWorker *> HTTPRequest::m_workers;
+std::chrono::steady_clock::duration HTTPRequest::m_timeout_duration = std::chrono::seconds(10);
 
 namespace {
 
@@ -235,6 +236,12 @@ size_t HTTPRequest::ReadCallback(char *buffer, size_t size, size_t n, void *v) {
 	// been sent.
 	HTTPRequest::Payload *payload = (HTTPRequest::Payload *)v;
 
+	if (payload->m_parent.Timeout()) {
+		payload->m_parent.errorCode = "E_TIMEOUT";
+		payload->m_parent.errorMessage = "Upload operation timed out";
+		return CURL_READFUNC_ABORT;
+	}
+
 	if (payload->sentSoFar == static_cast<off_t>(payload->data.size())) {
 		payload->sentSoFar = 0;
 		if (payload->final) {
@@ -269,6 +276,16 @@ bool HTTPRequest::sendPreparedRequest(const std::string &uri,
 	if (!m_is_streaming && !final) {
 		m_is_streaming = true;
 	}
+	if (m_timeout) {
+		errorCode = "E_TIMEOUT";
+		errorMessage = "Transfer has timed out due to inactivity.";
+		return false;
+	}
+	if (!errorCode.empty()) {
+		return false;
+	}
+
+	m_last_request = std::chrono::steady_clock::now();
 	m_final = final;
 	// Detect whether we were given an undersized buffer in non-streaming mode
 	if (!m_is_streaming && payload_size &&
@@ -294,6 +311,27 @@ bool HTTPRequest::sendPreparedRequest(const std::string &uri,
 	return errorCode.empty();
 }
 
+void HTTPRequest::Tick(std::chrono::steady_clock::time_point now) {
+	if (!m_is_streaming) {
+		return;
+	}
+	if (now - m_last_request <= m_timeout_duration) {
+		return;
+	}
+
+	if (m_timeout) {
+		return;
+	}
+	m_timeout = true;
+
+	if (m_unpause_queue) {
+		std::unique_lock<std::mutex> lk(m_mtx);
+		m_result_ready = false;
+		m_unpause_queue->Produce(this);
+		m_cv.wait(lk, [&]{return m_result_ready;});
+	}
+}
+
 bool HTTPRequest::ReleaseHandle(CURL *curl) {
 	m_curl_handle = nullptr;
 
@@ -604,11 +642,13 @@ HTTPRequest::CurlResult HTTPRequest::ProcessCurlResult(CURL *curl,
 	auto unique = std::unique_ptr<void, decltype(cleaner)>((void *)1, cleaner);
 
 	if (rv != 0) {
-		errorCode = "E_CURL_IO";
-		std::ostringstream error;
-		error << "curl failed (" << rv << "): '" << curl_easy_strerror(rv)
-			  << "'.";
-		errorMessage = error.str();
+		if (errorCode.empty()) {
+			errorCode = "E_CURL_IO";
+			std::ostringstream error;
+			error << "curl failed (" << rv << "): '" << curl_easy_strerror(rv)
+				<< "'.";
+			errorMessage = error.str();
+		}
 
 		return CurlResult::Fail;
 	}

src/HTTPCommands.hh

@@ -81,6 +81,20 @@ class HTTPRequest {
 	// context.
 	static void Init(XrdSysError &);
 
+	// Perform maintenance of the request.
+	void Tick(std::chrono::steady_clock::time_point);
+
+	// Sets the duration after which an in-progress operation may be considered
+	// stalled and hence timeout.
+	static void SetStallTimeout(std::chrono::steady_clock::duration timeout) {
+		m_timeout_duration = timeout;
+	}
+
+	// Return the stall timeout duration currently in use.
+	static std::chrono::steady_clock::duration GetStallTimeout() {
+		return m_timeout_duration;
+	}
+
   protected:
 	// Send the request to the HTTP server.
 	// Blocks until the request has completed.
@@ -118,6 +132,10 @@ class HTTPRequest {
 		m_unpause_queue = queue;
 	}
 
+	// Return whether or not the request has timed out since the last
+	// call to send more data.
+	bool Timeout() const {return m_timeout;}
+
 	typedef std::map<std::string, std::string> AttributeValueMap;
 	AttributeValueMap query_parameters;
 	AttributeValueMap headers;
@@ -190,6 +208,7 @@ class HTTPRequest {
 						 // call of the overall HTTPRequest
 	bool m_is_streaming{
 		false}; // Flag indicating this command is a streaming request.
+	bool m_timeout{false}; // Flag indicating the request has timed out.
 	bool m_result_ready{false}; // Flag indicating the results data is ready.
 	off_t m_payload_size{0}; // Size of the entire upload payload; 0 if unknown.
 	std::string m_protocol;
@@ -198,6 +217,14 @@ class HTTPRequest {
 	CURL *m_curl_handle{nullptr}; // The curl handle for the ongoing request
 	char m_errorBuffer[CURL_ERROR_SIZE]; // Static error buffer for libcurl
 	unsigned m_retry_count{0};
+
+	// Time when the last request was sent on this object; used to determine
+	// whether the operation has timed out.
+	std::chrono::steady_clock::time_point m_last_request{std::chrono::steady_clock::now()};
+
+	// Duration after which a partially-completed request will timeout if
+	// no progress has been made.
+	static std::chrono::steady_clock::duration m_timeout_duration;
 };
 
 class HTTPUpload : public HTTPRequest {

src/S3File.cc

@@ -32,7 +32,9 @@
 
 #include <curl/curl.h>
 
+#include <algorithm>
 #include <charconv>
+#include <iostream>
 #include <filesystem>
 #include <map>
 #include <memory>
@@ -41,6 +43,7 @@
 #include <stdlib.h>
 #include <string>
 #include <string_view>
+#include <thread>
 #include <vector>
 
 using namespace XrdHTTPServer;
@@ -49,6 +52,10 @@ S3FileSystem *g_s3_oss = nullptr;
 
 XrdVERSIONINFO(XrdOssGetFileSystem, S3);
 
+std::vector<std::pair<std::weak_ptr<std::mutex>, std::weak_ptr<AmazonS3SendMultipartPart>>> S3File::m_pending_ops;
+std::mutex S3File::m_pending_lk;
+std::once_flag S3File::m_monitor_launch;
+
 S3File::S3File(XrdSysError &log, S3FileSystem *oss)
 	: m_log(log), m_oss(oss), content_length(0), last_modified(0),
 	  partNumber(1) {}
@@ -61,6 +68,9 @@ int S3File::Open(const char *path, int Oflag, mode_t Mode, XrdOucEnv &env) {
 	if (Oflag & O_APPEND) {
 		m_log.Log(LogMask::Info, "Open", "File opened for append:", path);
 	}
+	if (Oflag & (O_RDWR | O_WRONLY)) {
+		m_write_lk.reset(new std::mutex);
+	}
 
 	char *asize_char;
 	if ((asize_char = env.Get("oss.asize"))) {
@@ -206,16 +216,28 @@ int S3File::Fstat(struct stat *buff) {
 }
 
 ssize_t S3File::Write(const void *buffer, off_t offset, size_t size) {
+	auto write_mutex = m_write_lk;
+	if (!write_mutex) {
+		return -EBADF;
+	}
+	std::lock_guard lk(*write_mutex);
+
 	if (offset != m_write_offset) {
 		m_log.Emsg(
 			"Write",
 			"Out-of-order write detected; S3 requires writes to be in order");
+		m_write_offset = -1;
+		return -EIO;
+	}
+	if (m_write_offset == -1) {
+		// Previous I/O error has occurred.  File is in bad state, immediately fail.
 		return -EIO;
 	}
 	if (uploadId == "") {
 		AmazonS3CreateMultipartUpload startUpload(m_ai, m_object, m_log);
 		if (!startUpload.SendRequest()) {
 			m_log.Emsg("Write", "S3 multipart request failed");
+			m_write_offset = -1;
 			return -ENOENT;
 		}
 		std::string errMsg;
@@ -240,6 +262,10 @@ ssize_t S3File::Write(const void *buffer, off_t offset, size_t size) {
 		}
 
 		m_write_op.reset(new AmazonS3SendMultipartPart(m_ai, m_object, m_log));
+		{
+			std::lock_guard lk(m_pending_lk);
+			m_pending_ops.emplace_back(m_write_lk, m_write_op);
+		}
 
 		// Calculate the size of the current chunk, if it's known.
 		m_part_size = m_s3_part_size;
@@ -271,8 +297,15 @@ ssize_t S3File::ContinueSendPart(const void *buffer, size_t size) {
 	if (!m_write_op->SendRequest(
 			std::string_view(static_cast<const char *>(buffer), write_size),
 			std::to_string(partNumber), uploadId, m_object_size, is_final)) {
+		m_write_offset = -1;
+		if (m_write_op->getErrorCode() == "E_TIMEOUT") {
+			m_log.Emsg("Write", "Timeout when uploading to S3");
+			m_write_op.reset();
+			return -ETIMEDOUT;
+		}
 		m_log.Emsg("Write", "Upload to S3 failed: ",
 				   m_write_op->getErrorMessage().c_str());
+		m_write_op.reset();
 		return -EIO;
 	}
 	if (is_final) {
@@ -283,13 +316,17 @@ ssize_t S3File::ContinueSendPart(const void *buffer, size_t size) {
 		if (startPos == std::string::npos) {
 			m_log.Emsg("Write", "Result from S3 does not include ETag:",
 					   resultString.c_str());
+			m_write_op.reset();
+			m_write_offset = -1;
 			return -EIO;
 		}
 		std::size_t endPos = resultString.find("\"", startPos + 7);
 		if (startPos == std::string::npos) {
 			m_log.Emsg("Write",
 					   "Result from S3 does not include ETag end-character:",
 					   resultString.c_str());
+			m_write_op.reset();
+			m_write_offset = -1;
 			return -EIO;
 		}
 		eTags.push_back(
@@ -301,6 +338,58 @@ ssize_t S3File::ContinueSendPart(const void *buffer, size_t size) {
 	return write_size;
 }
 
+void S3File::LaunchMonitorThread() {
+	std::call_once(m_monitor_launch, [] {
+		std::thread t(S3File::CleanupTransfers);
+		t.detach();
+	});
+}
+
+void S3File::CleanupTransfers() {
+	while (true) {
+		std::this_thread::sleep_for(HTTPRequest::GetStallTimeout()/3);
+		try {
+			CleanupTransfersOnce();
+		} catch (std::exception &exc) {
+			std::cerr << "Warning: caught unexpected exception when trying to clean transfers: " << exc.what() << std::endl;
+		}
+	}
+}
+
+void S3File::CleanupTransfersOnce() {
+	// Make a list of live transfers; erase any dead ones still on the list.
+	std::vector<std::pair<std::shared_ptr<std::mutex>, std::shared_ptr<AmazonS3SendMultipartPart>>> existing_ops;
+	{
+		std::lock_guard lk(m_pending_lk);
+		existing_ops.reserve(m_pending_ops.size());
+		m_pending_ops.erase(std::remove_if(m_pending_ops.begin(), m_pending_ops.end(),
+			[&](const auto &op) -> bool {
+				auto op_lk = op.first.lock();
+				if (!op_lk) {
+					// In this case, the S3File is no longer open for write.  No need to potentially
+					// clean up the transfer.
+					return true;
+				}
+				auto op_part = op.second.lock();
+				if (!op_part) {
+					// In this case, the S3File object is still open for writes but the upload has completed.
+					// Remove from the list.
+					return true;
+				}
+				// The S3File is open and upload is in-progress; we'll tick the transfer.
+				existing_ops.emplace_back(op_lk, op_part);
+				return false;
+			}
+		), m_pending_ops.end());
+	}
+	// For each live transfer, call `Tick` to advance the clock and possibly time things out.
+	auto now = std::chrono::steady_clock::now();
+	for (auto &info : existing_ops) {
+		std::lock_guard lk(*info.first);
+		info.second->Tick(now);
+	}
+}
+
 int S3File::Close(long long *retsz) {
 	// If we opened the object in create mode but did not actually write
 	// anything, make a quick zero-length file.
@@ -315,6 +404,7 @@ int S3File::Close(long long *retsz) {
 		}
 	}
 	if (m_write_op) {
+		std::lock_guard lk(*m_write_lk);
 		m_part_size = m_part_written;
 		auto written = ContinueSendPart(nullptr, 0);
 		if (written < 0) {
@@ -375,6 +465,7 @@ XrdOss *XrdOssGetStorageSystem2(XrdOss *native_oss, XrdSysLogger *Logger,
 
 	envP->Export("XRDXROOTD_NOPOSC", "1");
 
+	S3File::LaunchMonitorThread();
 	try {
 		AmazonRequest::Init(*log);
 		g_s3_oss = new S3FileSystem(Logger, config_fn, envP);

src/S3File.hh

@@ -27,6 +27,7 @@
 #include <XrdVersion.hh>
 
 #include <memory>
+#include <thread>
 
 #include <fcntl.h>
 
@@ -96,7 +97,21 @@ class S3File : public XrdOssDF {
 	size_t getContentLength() { return content_length; }
 	time_t getLastModified() { return last_modified; }
 
+	// Launch the global monitor thread associated with S3File objects.
+	// Currently, the monitor thread is used to cleanup in-progress file
+	// transfers that have been abandoned.
+	static void LaunchMonitorThread();
+
   private:
+	// Periodic cleanup of in-progress transfers.
+	//
+	// Iterates through the global list of pending multipart uploads
+	// that may be paused.  For each, call `Tick` on the upload and
+	// see if the transfer has aborted.
+	static void CleanupTransfers();
+
+	static void CleanupTransfersOnce();
+
 	ssize_t ContinueSendPart(const void *buffer, size_t size);
 	XrdSysError &m_log;
 	S3FileSystem *m_oss;
@@ -121,6 +136,34 @@ class S3File : public XrdOssDF {
 		-1}; // Expected size of the completed object; -1 if unknown.
 	std::string uploadId; // For creates, upload ID as assigned by t
 	std::vector<std::string> eTags;
-	std::unique_ptr<AmazonS3SendMultipartPart>
+
+	// The mutex protecting write activities.  Writes must currently be serialized
+	// as we aggregate them into large operations and upload them to the S3 endpoint.
+	// The mutex prevents corruption of internal state.
+	//
+	// The periodic cleanup thread may decide to abort the in-progress transfer; to
+	// do so, it'll need a reference to this lock that is independent of the lifetime of
+	// the open file; hence, it's a shared pointer.
+	std::shared_ptr<std::mutex> m_write_lk;
+
+	// The in-progress operation for a multi-part upload; its lifetime may be spread
+	// across multiple write calls.
+	std::shared_ptr<AmazonS3SendMultipartPart>
 		m_write_op; // The in-progress operation for a multi-part upload.
+
+	// The multipart uploads represent an in-progress request and the global cleanup
+	// thread may decide to trigger a failure if the request does not advance after
+	// some time period.
+	//
+	// To do so, we must be able to lock the associated write mutex and then call `Tick`
+	// on the upload.  To avoid prolonging the lifetime of the objects beyond the S3File,
+	// we hold onto a reference via a weak pointer.  Mutable operations on this vector are
+	// protected by the `m_pending_lk`.
+	static std::vector<std::pair<std::weak_ptr<std::mutex>, std::weak_ptr<AmazonS3SendMultipartPart>>> m_pending_ops;
+
+	// Mutex protecting the m_pending_ops variable.
+	static std::mutex m_pending_lk;
+
+	// Flag determining whether the monitoring thread has been launched.
+	static std::once_flag m_monitor_launch;
 };