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,8 @@ 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 +237,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 +277,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 +312,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 +643,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.
@@ -100,7 +114,6 @@ class HTTPRequest {
 							 const std::string_view payload, off_t payload_size,
 							 bool final);
 
-
 	// Called by the curl handler thread that the request has been finished.
 	virtual void Notify();
 
@@ -118,6 +131,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 +207,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 +216,15 @@ 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,15 +32,18 @@
 
 #include <curl/curl.h>
 
+#include <algorithm>
 #include <charconv>
 #include <filesystem>
+#include <iostream>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <sstream>
 #include <stdlib.h>
 #include <string>
 #include <string_view>
+#include <thread>
 #include <vector>
 
 using namespace XrdHTTPServer;
@@ -49,6 +52,12 @@ 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 +70,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 +218,29 @@ 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 +265,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 +300,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 +319,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 +341,67 @@ 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 +416,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 +477,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);