vendor: Update vendored sources to duckdb/duckdb@28b8a39

Add methods to retrieve current task scheduler busyness (duckdb/duckdb#16465)

Author: krlmlr

src/duckdb/src/function/table/version/pragma_version.cpp

@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "1-dev221"
+#define DUCKDB_PATCH_VERSION "1-dev226"
 #endif
 #ifndef DUCKDB_MINOR_VERSION
 #define DUCKDB_MINOR_VERSION 2
@@ -8,10 +8,10 @@
 #define DUCKDB_MAJOR_VERSION 1
 #endif
 #ifndef DUCKDB_VERSION
-#define DUCKDB_VERSION "v1.2.1-dev221"
+#define DUCKDB_VERSION "v1.2.1-dev226"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "a0b525cb6d"
+#define DUCKDB_SOURCE_ID "28b8a398a4"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"

src/duckdb/src/include/duckdb/parallel/task_scheduler.hpp

@@ -68,6 +68,10 @@ class TaskScheduler {
 	//! Returns the number of threads
 	DUCKDB_API int32_t NumberOfThreads();
 
+	idx_t GetNumberOfTasks() const;
+	idx_t GetProducerCount() const;
+	idx_t GetTaskCountForProducer(ProducerToken &token) const;
+
 	//! Send signals to n threads, signalling for them to wake up and attempt to execute a task
 	void Signal(idx_t n);
 

src/duckdb/src/parallel/task_scheduler.cpp

@@ -284,6 +284,39 @@ int32_t TaskScheduler::NumberOfThreads() {
 	return current_thread_count.load();
 }
 
+idx_t TaskScheduler::GetNumberOfTasks() const {
+#ifndef DUCKDB_NO_THREADS
+	return queue->q.size_approx();
+#else
+	idx_t task_count = 0;
+	for (auto &producer : queue->q) {
+		task_count += producer.second.size();
+	}
+	return task_count;
+#endif
+}
+
+idx_t TaskScheduler::GetProducerCount() const {
+#ifndef DUCKDB_NO_THREADS
+	return queue->q.size_producers_approx();
+#else
+	return queue->q.size();
+#endif
+}
+
+idx_t TaskScheduler::GetTaskCountForProducer(ProducerToken &token) const {
+#ifndef DUCKDB_NO_THREADS
+	lock_guard<mutex> producer_lock(token.producer_lock);
+	return queue->q.size_producer_approx(token.token->queue_token);
+#else
+	const auto it = queue->q.find(std::ref(*token.token));
+	if (it == queue->q.end()) {
+		return 0;
+	}
+	return it->second.size();
+#endif
+}
+
 void TaskScheduler::SetThreads(idx_t total_threads, idx_t external_threads) {
 	if (total_threads == 0) {
 		throw SyntaxException("Number of threads must be positive!");

src/duckdb/third_party/concurrentqueue/concurrentqueue.h

@@ -1254,6 +1254,23 @@ class ConcurrentQueue
 		return size;
 	}
 
+
+	// Returns the number of producers currently associated with the queue.
+	size_t size_producers_approx() const
+	{
+		size_t size = 0;
+		for (auto ptr = producerListTail.load(std::memory_order_acquire); ptr != nullptr; ptr = ptr->next_prod()) {
+			size += 1;
+		}
+		return size;
+	}
+
+	// Returns the number of elements currently in the queue for a specific producer.
+	size_t size_producer_approx(producer_token_t const& producer) const
+	{
+		return static_cast<ExplicitProducer*>(producer.producer)->size_approx();
+	}
+
 
 	// Returns true if the underlying atomic variables used by
 	// the queue are lock-free (they should be on most platforms).