Import thread/task_specific variables

CMakeLists.txt

@@ -96,6 +96,7 @@ set( fc_sources
      src/exception.cpp
      src/variant_object.cpp
      src/thread/thread.cpp
+     src/thread/thread_specific.cpp
      src/thread/future.cpp
      src/thread/task.cpp
      src/thread/spin_lock.cpp 

include/fc/thread/task.hpp

@@ -8,6 +8,25 @@ namespace fc {
   struct context;
   class spin_lock;
 
+   namespace detail
+   {
+      struct specific_data_info
+      {
+         void* value;
+         void (*cleanup)(void*);
+         specific_data_info() :
+            value(0),
+            cleanup(0)
+            {}
+         specific_data_info(void* value, void (*cleanup)(void*)) :
+            value(value),
+            cleanup(cleanup)
+         {}
+      };
+      void* get_task_specific_data(unsigned slot);
+      void set_task_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
+   }
+
   class task_base : virtual public promise_base {
     public:
               void run(); 
@@ -23,6 +42,12 @@ namespace fc {
       context*    _active_context;
       task_base*  _next;
 
+      // support for task-specific data
+      std::vector<detail::specific_data_info> *_task_specific_data;
+
+      friend void* detail::get_task_specific_data(unsigned slot);
+      friend void detail::set_task_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
+
       task_base(void* func);
       // opaque internal / private data used by
       // thread/thread_private
@@ -37,6 +62,8 @@ namespace fc {
       void          (*_run_functor)(void*, void* );
 
       void          run_impl(); 
+
+      void cleanup_task_specific_data();
   };
 
   namespace detail {

include/fc/thread/thread.hpp

@@ -7,6 +7,15 @@ namespace fc {
   class time_point;
   class microseconds;
 
+   namespace detail
+   {
+      void* get_thread_specific_data(unsigned slot);
+      void set_thread_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
+      unsigned get_next_unused_task_storage_slot();
+      void* get_task_specific_data(unsigned slot);
+      void set_task_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
+   }
+
   class thread {
     public:
       thread( const std::string& name = "" );
@@ -121,6 +130,11 @@ namespace fc {
       friend class promise_base;
       friend class thread_d;
       friend class mutex;
+      friend void* detail::get_thread_specific_data(unsigned slot);
+      friend void detail::set_thread_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
+      friend unsigned detail::get_next_unused_task_storage_slot();
+      friend void* detail::get_task_specific_data(unsigned slot);
+      friend void detail::set_task_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
 #ifndef NDEBUG
       friend class non_preemptable_scope_check;
 #endif

include/fc/thread/thread_specific.hpp

@@ -0,0 +1,84 @@
+#pragma once
+#include "thread.hpp"
+
+namespace fc
+{
+  namespace detail
+  {
+    unsigned get_next_unused_thread_storage_slot();
+    unsigned get_next_unused_task_storage_slot();
+  }
+
+  template <typename T>
+  class thread_specific_ptr
+  {
+  private:
+    unsigned slot;
+  public:
+    thread_specific_ptr() :
+      slot(detail::get_next_unused_thread_storage_slot())
+      {}
+
+    T* get() const
+    {
+      return static_cast<T*>(detail::get_thread_specific_data(slot));
+    }
+    T* operator->() const
+    {
+      return get();
+    }
+    T& operator*() const
+    {
+      return *get();
+    }
+    operator bool() const
+    {
+      return get() != static_cast<T*>(0);
+    }
+    static void cleanup_function(void* obj) 
+    {
+      delete static_cast<T*>(obj);
+    }
+    void reset(T* new_value = 0)
+    {
+      detail::set_thread_specific_data(slot, new_value, cleanup_function);
+    }
+  };
+
+  template <typename T>
+  class task_specific_ptr
+  {
+  private:
+    unsigned slot;
+  public:
+    task_specific_ptr() :
+      slot(detail::get_next_unused_task_storage_slot())
+      {}
+
+    T* get() const
+    {
+      return static_cast<T*>(detail::get_task_specific_data(slot));
+    }
+    T* operator->() const
+    {
+      return get();
+    }
+    T& operator*() const
+    {
+      return *get();
+    }
+    operator bool() const
+    {
+      return get() != static_cast<T*>(0);
+    }
+    static void cleanup_function(void* obj) 
+    {
+      delete static_cast<T*>(obj);
+    }
+    void reset(T* new_value = 0)
+    {
+      detail::set_task_specific_data(slot, new_value, cleanup_function);
+    }
+  };
+
+}

src/thread/task.cpp

@@ -20,6 +20,7 @@ namespace fc {
   _posted_num(0),
   _active_context(nullptr),
   _next(nullptr),
+  _task_specific_data(nullptr),
   _promise_impl(nullptr),
   _functor(func){
   }
@@ -63,6 +64,7 @@ namespace fc {
   }
 
   task_base::~task_base() {
+    cleanup_task_specific_data();
     _destroy_functor( _functor );
   }
 
@@ -71,4 +73,17 @@ namespace fc {
         _active_context = c; 
       }
   }
+
+  void task_base::cleanup_task_specific_data()
+  {
+    if (_task_specific_data)
+    {
+      for (auto iter = _task_specific_data->begin(); iter != _task_specific_data->end(); ++iter)
+        if (iter->cleanup)
+            iter->cleanup(iter->value);
+      delete _task_specific_data;
+      _task_specific_data = nullptr;
+    }
+  }
+
 }

src/thread/thread.cpp

@@ -201,6 +201,7 @@ namespace fc {
         my->check_for_timeouts();
       }
       my->clear_free_list();
+      my->cleanup_thread_specific_data();
    }
      
    void thread::exec() {

src/thread/thread_d.hpp

@@ -26,7 +26,8 @@ namespace fc {
              pt_head(0),
              ready_head(0),
              ready_tail(0),
-             blocked(0)
+             blocked(0),
+             next_unused_task_storage_slot(0)
 #ifndef NDEBUG
              ,non_preemptable_scope_count(0)
 #endif
@@ -87,6 +88,15 @@ namespace fc {
            fc::context*             ready_tail;
 
            fc::context*             blocked;
+
+           // values for thread specific data objects for this thread
+           std::vector<detail::specific_data_info> thread_specific_data;
+           // values for task_specific data for code executing on a thread that's
+           // not a task launched by async (usually the default task on the main 
+           // thread in a process)
+           std::vector<detail::specific_data_info> non_task_specific_data;
+           unsigned next_unused_task_storage_slot;
+
 #ifndef NDEBUG
            unsigned                 non_preemptable_scope_count;
 #endif
@@ -570,5 +580,17 @@ namespace fc {
 
           check_fiber_exceptions();
         }
+
+        void cleanup_thread_specific_data()
+        {
+          for (auto iter = non_task_specific_data.begin(); iter != non_task_specific_data.end(); ++iter)
+            if (iter->cleanup)
+              iter->cleanup(iter->value);
+
+          for (auto iter = thread_specific_data.begin(); iter != thread_specific_data.end(); ++iter)
+            if (iter->cleanup)
+              iter->cleanup(iter->value);
+        }
+
     };
 } // namespace fc

src/thread/thread_specific.cpp

@@ -0,0 +1,65 @@
+#include <fc/log/logger.hpp>
+#include <fc/thread/thread_specific.hpp>
+#include "thread_d.hpp"
+#include <boost/atomic.hpp>
+
+namespace fc
+{
+  namespace detail
+  {
+    boost::atomic<unsigned> thread_specific_slot_counter;
+    unsigned get_next_unused_thread_storage_slot()
+    {
+      return thread_specific_slot_counter.fetch_add(1);
+    }
+
+    void* get_specific_data(std::vector<detail::specific_data_info> *specific_data, unsigned slot)
+    {
+      return slot < specific_data->size() ?
+        (*specific_data)[slot].value : nullptr;
+    }
+    void set_specific_data(std::vector<detail::specific_data_info> *specific_data, unsigned slot, void* new_value, void(*cleanup)(void*))
+    {
+      if (slot + 1 > specific_data->size())
+        specific_data->resize(slot + 1);
+      (*specific_data)[slot] = std::move(detail::specific_data_info(new_value, cleanup));
+    }
+
+    void* get_thread_specific_data(unsigned slot)
+    {
+      return get_specific_data(&thread::current().my->thread_specific_data, slot);
+    }
+    void set_thread_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*))
+    {
+      return set_specific_data(&thread::current().my->thread_specific_data, slot, new_value, cleanup);
+    }
+
+    unsigned get_next_unused_task_storage_slot()
+    {
+      return thread::current().my->next_unused_task_storage_slot++;
+    }
+    void* get_task_specific_data(unsigned slot)
+    {
+      context* current_context = thread::current().my->current;
+      if (!current_context ||
+          !current_context->cur_task)
+        return get_specific_data(&thread::current().my->non_task_specific_data, slot);
+      if (current_context->cur_task->_task_specific_data)
+        return get_specific_data(current_context->cur_task->_task_specific_data, slot);
+      return nullptr;
+    }
+    void set_task_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*))
+    {
+      context* current_context = thread::current().my->current;
+      if (!current_context ||
+          !current_context->cur_task)
+        set_specific_data(&thread::current().my->non_task_specific_data, slot, new_value, cleanup);
+      else
+      {
+        if (!current_context->cur_task->_task_specific_data)
+            current_context->cur_task->_task_specific_data = new std::vector<detail::specific_data_info>;
+        set_specific_data(current_context->cur_task->_task_specific_data, slot, new_value, cleanup);
+      }
+    }
+  }
+} // end namespace fc