<commit_before>// Copyright (c) 2015 <NAME>
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

#include "internal.h"

// For GetProcAddress and GetModuleHandle
#ifdef _WIN32
#include <windows.h>
#endif

// for pthread thread_local emulation
#if defined(EMULATE_PTHREAD_THREAD_LOCAL)
# include <pthread.h>
#endif

namespace async {
namespace detail {

// Per-thread data, aligned to cachelines to avoid false sharing
struct LIBASYNC_CACHELINE_ALIGN thread_data_t {
	work_steal_queue queue;
	std::minstd_rand rng;
	task_wait_event event;
	std::thread handle;
};

// Internal data used by threadpool_scheduler
struct threadpool_data {
	threadpool_data(std::size_t num_threads)
		: thread_data(num_threads), shutdown(false), num_waiters(0), waiters(new task_wait_event*[num_threads]) {}

	// Mutex protecting everything except thread_data
	std::mutex lock;

	// Array of per-thread data
	aligned_array<thread_data_t> thread_data;

	// Global queue for tasks from outside the pool
	fifo_queue public_queue;

	// Shutdown request indicator
	bool shutdown;

	// List of threads waiting for tasks to run. num_waiters needs to be atomic
	// because it is sometimes read outside the mutex.
	std::atomic<std::size_t> num_waiters;
	std::unique_ptr<task_wait_event*[]> waiters;

#ifdef BROKEN_JOIN_IN_DESTRUCTOR
	// Shutdown complete event, used instead of thread::join()
	std::size_t shutdown_num_threads;
	std::condition_variable shutdown_complete_event;
#endif
};

// this wrapper encapsulates both the owning_threadpool pointer and the thread id.
// this is done to improve performance on the emulated thread_local reducing the number
// of calls to "pthread_getspecific"
struct threadpool_data_wrapper {
	threadpool_data* owning_threadpool;
	std::size_t thread_id;

	threadpool_data_wrapper(threadpool_data* owning_threadpool, std::size_t thread_id):
		owning_threadpool(owning_threadpool), thread_id(thread_id) { }
};

#if defined(EMULATE_PTHREAD_THREAD_LOCAL)
struct pthread_emulation_threadpool_data_initializer {
	pthread_key_t key;
	
	pthread_emulation_threadpool_data_initializer()
	{
		pthread_key_create(&key, [](void* wrapper_ptr) {
			threadpool_data_wrapper* wrapper = static_cast<threadpool_data_wrapper*>(wrapper_ptr);
			delete wrapper;
		});
	}
		
	~pthread_emulation_threadpool_data_initializer()
	{
		pthread_key_delete(key);
	}
};
	
static pthread_key_t get_local_threadpool_data_key()
{
	static pthread_emulation_threadpool_data_initializer initializer;
	return initializer.key;
}

#else
// Thread pool this thread belongs to, or null if not in pool
static THREAD_LOCAL threadpool_data* owning_threadpool = nullptr;

// Current thread's index in the pool
static THREAD_LOCAL std::size_t thread_id;
#endif
	
static void create_threadpool_data(threadpool_data* owning_threadpool_, std::size_t thread_id_)
{
#if defined(EMULATE_PTHREAD_THREAD_LOCAL)
	// the memory allocated here gets deallocated by the lambda declared on the key creation
	pthread_setspecific(get_local_threadpool_data_key(), new threadpool_data_wrapper(owning_threadpool_, thread_id_));
#else
	owning_threadpool = _owning_threadpool;
	thread_id = _thread_id;
#endif
}
	
static threadpool_data_wrapper get_threadpool_data_wrapper()
{
#if defined(EMULATE_PTHREAD_THREAD_LOCAL)
	threadpool_data_wrapper* wrapper = static_cast<threadpool_data_wrapper*>(pthread_getspecific(get_local_threadpool_data_key()));
	if(wrapper == nullptr) {
		// if, for some reason, the wrapper is not set, this won't cause a crash
		return threadpool_data_wrapper(nullptr, 0);
	}
	return *wrapper;
#else
	return threadpool_data_wrapper(owning_threadpool, thread_id);
#endif
}

// Try to steal a task from another thread's queue
static task_run_handle steal_task(threadpool_data* impl, std::size_t thread_id)
{
	// Make a list of victim thread ids and shuffle it
	std::vector<std::size_t> victims(impl->thread_data.size());
	std::iota(victims.begin(), victims.end(), 0);
	std::shuffle(victims.begin(), victims.end(), impl->thread_data[thread_id].rng);

	// Try to steal from another thread<commit_msg>Fix a build issue on platforms that support thread_local<commit_after>// Copyright (c) 2015 <NAME>
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

#include "internal.h"

// For GetProcAddress and GetModuleHandle
#ifdef _WIN32
#include <windows.h>
#endif

// for pthread thread_local emulation
#if defined(EMULATE_PTHREAD_THREAD_LOCAL)
# include <pthread.h>
#endif

namespace async {
namespace detail {

// Per-thread data, aligned to cachelines to avoid false sharing
struct LIBASYNC_CACHELINE_ALIGN thread_data_t {
	work_steal_queue queue;
	std::minstd_rand rng;
	task_wait_event event;
	std::thread handle;
};

// Internal data used by threadpool_scheduler
struct threadpool_data {
	threadpool_data(std::size_t num_threads)
		: thread_data(num_threads), shutdown(false), num_waiters(0), waiters(new task_wait_event*[num_threads]) {}

	// Mutex protecting everything except thread_data
	std::mutex lock;

	// Array of per-thread data
	aligned_array<thread_data_t> thread_data;

	// Global queue for tasks from outside the pool
	fifo_queue public_queue;

	// Shutdown request indicator
	bool shutdown;

	// List of threads waiting for tasks to run. num_waiters needs to be atomic
	// because it is sometimes read outside the mutex.
	std::atomic<std::size_t> num_waiters;
	std::unique_ptr<task_wait_event*[]> waiters;

#ifdef BROKEN_JOIN_IN_DESTRUCTOR
	// Shutdown complete event, used instead of thread::join()
	std::size_t shutdown_num_threads;
	std::condition_variable shutdown_complete_event;
#endif
};

// this wrapper encapsulates both the owning_threadpool pointer and the thread id.
// this is done to improve performance on the emulated thread_local reducing the number
// of calls to "pthread_getspecific"
struct threadpool_data_wrapper {
	threadpool_data* owning_threadpool;
	std::size_t thread_id;

	threadpool_data_wrapper(threadpool_data* owning_threadpool, std::size_t thread_id):
		owning_threadpool(owning_threadpool), thread_id(thread_id) { }
};

#if defined(EMULATE_PTHREAD_THREAD_LOCAL)
struct pthread_emulation_threadpool_data_initializer {
	pthread_key_t key;
	
	pthread_emulation_threadpool_data_initializer()
	{
		pthread_key_create(&key, [](void* wrapper_ptr) {
			threadpool_data_wrapper* wrapper = static_cast<threadpool_data_wrapper*>(wrapper_ptr);
			delete wrapper;
		});
	}
		
	~pthread_emulation_threadpool_data_initializer()
	{
		pthread_key_delete(key);
	}
};
	
static pthread_key_t get_local_threadpool_data_key()
{
	static pthread_emulation_threadpool_data_initializer initializer;
	return initializer.key;
}

#else
// Thread pool this thread belongs to, or null if not in pool
static THREAD_LOCAL threadpool_data* owning_threadpool = nullptr;

// Current thread's index in the pool
static THREAD_LOCAL std::size_t thread_id;
#endif
	
static void create_threadpool_data(threadpool_data* owning_threadpool_, std::size_t thread_id_)
{
#if defined(EMULATE_PTHREAD_THREAD_LOCAL)
	// the memory allocated here gets deallocated by the lambda declared on the key creation
	pthread_setspecific(get_local_threadpool_data_key(), new threadpool_data_wrapper(owning_threadpool_, thread_id_));
#else
	owning_threadpool = owning_threadpool_;
	thread_id = thread_id_;
#endif
}
	
static threadpool_data_wrapper get_threadpool_data_wrapper()
{
#if defined(EMULATE_PTHREAD_THREAD_LOCAL)
	threadpool_data_wrapper* wrapper = static_cast<threadpool_data_wrapper*>(pthread_getspecific(get_local_threadpool_data_key()));
	if(wrapper == nullptr) {
		// if, for some reason, the wrapper is not set, this won't cause a crash
		return threadpool_data_wrapper(nullptr, 0);
	}
	return *wrapper;
#else
	return threadpool_data_wrapper(owning_threadpool, thread_id);
#endif
}

// Try to steal a task from another thread's queue
static task_run_handle steal_task(threadpool_data* impl, std::size_t thread_id)
{
	// Make a list of victim thread ids and shuffle it
	std::vector<std::size_t> victims(impl->thread_data.size());
	std::iota(victims.begin(), victims.end(), 0);
	std::shuffle(victims.begin(), victims.end(), impl->thread_data[thread_id].rng);

	// Try to steal from another thread
{'fraction_non_alphanumeric': 0.08200098087297696, 'fraction_numerical': 0.001765571358509073, 'mean_word_length': 5.651011089367254, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 25, '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': '7530602', 'n_tokens_mistral': 3191, 'n_tokens_neox': 3061, 'n_words': 989}