Merge branch 'master' into coverity/remove-not-required-copies-inference

Author: praasz

cmake/dependencies.cmake

@@ -158,10 +158,10 @@ function(ov_download_tbb)
     elseif(APPLE AND AARCH64)
         # build oneTBB with export MACOSX_DEPLOYMENT_TARGET=11.0
         RESOLVE_DEPENDENCY(TBB
-                ARCHIVE_MAC "oneapi-tbb-2021.13.0-mac-arm64-canary.tgz"
+                ARCHIVE_MAC "oneapi-tbb-2021.13.1-mac-arm64-canary.tgz"
                 TARGET_PATH "${TEMP}/${PLATFORM_SUBDIR}/tbb"
                 ENVIRONMENT "TBBROOT"
-                SHA256 "fb4be1dd03044a97475c45a0cf4576e502b4b64048e98e019520b0720fc255aa"
+                SHA256 "031c34cbb25a73246a3076754343997b9e0f61737221a1d7979352d3b3aba898"
                 USE_NEW_LOCATION TRUE)
     elseif(RISCV64)
         RESOLVE_DEPENDENCY(TBB

src/bindings/python/src/pyopenvino/core/properties/properties.cpp

@@ -356,7 +356,8 @@ void regmodule_properties(py::module m) {
 
     py::enum_<ov::intel_npu::CompilerType>(m_intel_npu, "CompilerType", py::arithmetic())
         .value("DRIVER", ov::intel_npu::CompilerType::DRIVER)
-        .value("PLUGIN", ov::intel_npu::CompilerType::PLUGIN);
+        .value("PLUGIN", ov::intel_npu::CompilerType::PLUGIN)
+        .value("PREFER_PLUGIN", ov::intel_npu::CompilerType::PREFER_PLUGIN);
 
     wrap_property_RW(m_intel_npu, ov::intel_npu::compiler_type, "compiler_type");
 }

src/bindings/python/tests/test_runtime/test_properties.py

@@ -512,7 +512,8 @@ def test_properties_ro(ov_property_ro, expected_value):
             intel_npu.compiler_type,
             "NPU_COMPILER_TYPE",
             ((intel_npu.CompilerType.DRIVER, intel_npu.CompilerType.DRIVER),
-             (intel_npu.CompilerType.PLUGIN, intel_npu.CompilerType.PLUGIN),),
+             (intel_npu.CompilerType.PLUGIN, intel_npu.CompilerType.PLUGIN),
+             (intel_npu.CompilerType.PREFER_PLUGIN, intel_npu.CompilerType.PREFER_PLUGIN),),
         ),
         (props.enable_weightless, "ENABLE_WEIGHTLESS", ((True, True), (False, False))),
     ],

src/inference/include/openvino/runtime/intel_npu/properties.hpp

@@ -31,7 +31,7 @@ namespace intel_npu {
  * Type of NPU compiler to be used for compilation of a network
  * @note Configuration API v 2.0
  */
-enum class CompilerType { PLUGIN, DRIVER };
+enum class CompilerType { PLUGIN, DRIVER, PREFER_PLUGIN };
 
 /**
  * @brief Prints a string representation of ov::intel_npu::CompilerType to a stream
@@ -48,6 +48,9 @@ inline std::ostream& operator<<(std::ostream& out, const CompilerType& fmt) {
     case CompilerType::DRIVER: {
         out << "DRIVER";
     } break;
+    case CompilerType::PREFER_PLUGIN: {
+        out << "PREFER_PLUGIN";
+    } break;
     default:
         out << static_cast<uint32_t>(fmt);
         break;

src/inference/src/dev/threading/cpu_streams_executor.cpp

@@ -238,128 +238,110 @@ struct CPUStreamsExecutor::Impl {
     // it's only a workaround for ticket CVS-111490, please be carefully when need to modify
     // CustomeThreadLocal::local(), especially like operations that will affect the count of
     // CustomThreadLocal::ThreadId
-    class CustomThreadLocal : public ThreadLocal<std::shared_ptr<Stream>> {
-        class ThreadTracker {
-        public:
-            explicit ThreadTracker(const std::thread::id& id)
-                : _id(id),
-                  _count_ptr(std::make_shared<std::atomic_int>(1)) {}
-            ~ThreadTracker() {
-                _count_ptr->fetch_sub(1);
-            }
-            std::shared_ptr<ThreadTracker> fetch() {
-                auto new_ptr = std::shared_ptr<ThreadTracker>(new ThreadTracker(*this));
-                auto pre_valule = new_ptr.get()->_count_ptr->fetch_add(1);
-                OPENVINO_ASSERT(pre_valule == 1, "this value must be 1, please check code CustomThreadLocal::local()");
-                return new_ptr;
+    class CustomThreadLocal : public ThreadLocal<std::shared_ptr<Stream>>,
+                              public std::enable_shared_from_this<CustomThreadLocal> {
+    public:
+        CustomThreadLocal(std::function<std::shared_ptr<Stream>()> callback_construct, Impl* impl)
+            : ThreadLocal<std::shared_ptr<Stream>>(std::move(callback_construct)),
+              _impl(impl) {
+            _executor_thread_id = std::this_thread::get_id();
+        }
+        void cleanup(std::thread::id thread_id) {
+            std::lock_guard<std::mutex> guard(_stream_map_mutex);
+            auto item = _stream_map.find(thread_id);
+            if (item != _stream_map.end()) {
+                _stream_map.erase(item);
             }
-            const std::thread::id& get_id() const {
-                return _id;
+        }
+        struct ThreadCleaner {
+            struct ResourceKeeper {
+                // The crt call the atexit released the global variable before thread exit
+                // Need to call the map's clear to set the size to 0
+                // Or the thread exit callback will double free the map's content
+                ~ResourceKeeper() {
+                    std::lock_guard<std::mutex> lock(_mutex);
+                    _mapdata.clear();
+                }
+                std::mutex _mutex;
+                std::multimap<std::thread::id, std::weak_ptr<CustomThreadLocal>> _mapdata;
+            };
+            static ResourceKeeper global_resource_holder;
+
+            ThreadCleaner(std::thread::id thread_id) : _thread_id(thread_id) {}
+            void add(std::weak_ptr<CustomThreadLocal> parent) {
+                std::lock_guard<std::mutex> lock(global_resource_holder._mutex);
+                global_resource_holder._mapdata.insert({_thread_id, parent});
             }
-            int count() const {
-                return *(_count_ptr.get());
+            ~ThreadCleaner() {
+                std::vector<std::shared_ptr<CustomThreadLocal>> parent_ptr;
+                {
+                    // Releaese the global lock as soon as possible
+                    std::lock_guard<std::mutex> lock(global_resource_holder._mutex);
+                    auto range = global_resource_holder._mapdata.equal_range(_thread_id);
+                    for (auto it = range.first; it != range.second; ++it) {
+                        parent_ptr.push_back(it->second.lock());
+                    }
+                    if (!parent_ptr.empty()) {
+                        global_resource_holder._mapdata.erase(range.first, range.second);
+                    }
+                }
+                for (auto& parent : parent_ptr) {
+                    if (parent) {
+                        parent->cleanup(_thread_id);
+                    }
+                }
             }
 
         private:
-            // disable all copy and move semantics, user only can use fetch()
-            // to create a new instance with a shared count num;
-            ThreadTracker(const ThreadTracker&) = default;
-            ThreadTracker(ThreadTracker&&) = delete;
-            ThreadTracker& operator=(const ThreadTracker&) = delete;
-            ThreadTracker& operator=(ThreadTracker&&) = delete;
-            std::thread::id _id;
-            std::shared_ptr<std::atomic_int> _count_ptr;
+            std::thread::id _thread_id;
         };
-
-    public:
-        CustomThreadLocal(std::function<std::shared_ptr<Stream>()> callback_construct, Impl* impl)
-            : ThreadLocal<std::shared_ptr<Stream>>(std::move(callback_construct)),
-              _impl(impl) {}
         std::shared_ptr<Stream> local() {
-            // maybe there are two CPUStreamsExecutors in the same thread.
-            static thread_local std::map<void*, std::shared_ptr<CustomThreadLocal::ThreadTracker>> t_stream_count_map;
-            // fix the memory leak issue that CPUStreamsExecutor is already released,
-            // but still exists CustomThreadLocal::ThreadTracker in t_stream_count_map
-            for (auto it = t_stream_count_map.begin(); it != t_stream_count_map.end();) {
-                if (this != it->first && it->second->count() == 1) {
-                    t_stream_count_map.erase(it++);
-                } else {
-                    it++;
-                }
-            }
             auto id = std::this_thread::get_id();
-            auto search = _thread_ids.find(id);
-            if (search != _thread_ids.end()) {
+            if (id == _executor_thread_id) {
                 return ThreadLocal<std::shared_ptr<Stream>>::local();
             }
-            std::lock_guard<std::mutex> guard(_stream_map_mutex);
-            for (auto& item : _stream_map) {
-                if (item.first->get_id() == id) {
-                    // check if the ThreadTracker of this stream is already in t_stream_count_map
-                    // if not, then create ThreadTracker for it
-                    auto iter = t_stream_count_map.find((void*)this);
-                    if (iter == t_stream_count_map.end()) {
-                        t_stream_count_map[(void*)this] = item.first->fetch();
-                    }
-                    return item.second;
-                }
-            }
-            std::shared_ptr<Impl::Stream> stream = nullptr;
-            for (auto it = _stream_map.begin(); it != _stream_map.end();) {
-                if (it->first->count() == 1) {
-                    if (stream == nullptr) {
-                        stream = it->second;
-                    }
-                    _stream_map.erase(it++);
-                } else {
-                    it++;
+
+            // ensure ThreadCleaner is created only once per thread exit
+            thread_local ThreadCleaner t_cleaner(id);
+            {
+                std::lock_guard<std::mutex> guard(_stream_map_mutex);
+                auto search = _stream_map.find(id);
+                if (search != _stream_map.end()) {
+                    return search->second;
                 }
             }
-            if (stream == nullptr) {
-                stream = std::make_shared<Impl::Stream>(_impl);
+            std::shared_ptr<Impl::Stream> stream = std::make_shared<Impl::Stream>(_impl);
+            t_cleaner.add(this->shared_from_this());
+            {
+                std::lock_guard<std::mutex> guard(_stream_map_mutex);
+                _stream_map[id] = stream;
             }
-            auto tracker_ptr = std::make_shared<CustomThreadLocal::ThreadTracker>(id);
-            t_stream_count_map[(void*)this] = tracker_ptr;
-            auto new_tracker_ptr = tracker_ptr->fetch();
-            _stream_map[new_tracker_ptr] = stream;
             return stream;
         }
 
-        void set_thread_ids_map(std::vector<std::thread>& threads) {
-            for (auto& thread : threads) {
-                _thread_ids.insert(thread.get_id());
-            }
-        }
-
         bool find_thread_id() {
             auto id = std::this_thread::get_id();
-            auto search = _thread_ids.find(id);
-            if (search != _thread_ids.end()) {
+            if (id == _executor_thread_id) {
                 return true;
             }
             std::lock_guard<std::mutex> guard(_stream_map_mutex);
-            for (auto& item : _stream_map) {
-                if (item.first->get_id() == id) {
-                    return true;
-                }
-            }
-            return false;
+            auto item = _stream_map.find(id);
+            return item != _stream_map.end();
         }
 
     private:
-        std::set<std::thread::id> _thread_ids;
         Impl* _impl;
-        std::map<std::shared_ptr<CustomThreadLocal::ThreadTracker>, std::shared_ptr<Impl::Stream>> _stream_map;
+        std::map<std::thread::id, std::shared_ptr<Impl::Stream>> _stream_map;
         std::mutex _stream_map_mutex;
+        std::thread::id _executor_thread_id;
     };
 
-    explicit Impl(const Config& config)
-        : _config{config},
-          _streams(
-              [this] {
-                  return std::make_shared<Impl::Stream>(this);
-              },
-              this) {
+    explicit Impl(const Config& config) : _config{config} {
+        _streams = std::make_shared<CustomThreadLocal>(
+            [this] {
+                return std::make_shared<Impl::Stream>(this);
+            },
+            this);
         auto numaNodes = get_available_numa_nodes();
         int streams_num = _config.get_streams();
         auto processor_ids = _config.get_stream_processor_ids();
@@ -390,12 +372,11 @@ struct CPUStreamsExecutor::Impl {
                         }
                     }
                     if (task) {
-                        Execute(task, *(_streams.local()));
+                        Execute(task, *(_streams->local()));
                     }
                 }
             });
         }
-        _streams.set_thread_ids_map(_threads);
     }
 
     void Enqueue(Task task) {
@@ -424,7 +405,7 @@ struct CPUStreamsExecutor::Impl {
     void pin_stream_to_cpus() {
 #if OV_THREAD == OV_THREAD_SEQ
         if (_config.get_cpu_pinning()) {
-            auto stream = _streams.local();
+            auto stream = _streams->local();
             auto proc_type_table = get_org_proc_type_table();
             std::tie(stream->_mask, stream->_ncpus) = get_process_mask();
             if (get_num_numa_nodes() > 1) {
@@ -443,15 +424,15 @@ struct CPUStreamsExecutor::Impl {
 
     void unpin_stream_to_cpus() {
 #if OV_THREAD == OV_THREAD_SEQ
-        auto stream = _streams.local();
+        auto stream = _streams->local();
         if (stream->_mask) {
             pin_current_thread_by_mask(stream->_ncpus, stream->_mask);
         }
 #endif
     }
 
     void Defer(Task task) {
-        auto& stream = *(_streams.local());
+        auto& stream = *(_streams->local());
         stream._taskQueue.push(std::move(task));
         if (!stream._execute) {
             stream._execute = true;
@@ -476,17 +457,20 @@ struct CPUStreamsExecutor::Impl {
     std::queue<Task> _taskQueue;
     bool _isStopped = false;
     std::vector<int> _usedNumaNodes;
-    CustomThreadLocal _streams;
+    std::shared_ptr<CustomThreadLocal> _streams;
     bool _isExit = false;
     std::vector<int> _cpu_ids_all;
     std::mutex _cpu_ids_mutex;
 };
 
+CPUStreamsExecutor::Impl::CustomThreadLocal::ThreadCleaner::ResourceKeeper
+    CPUStreamsExecutor::Impl::CustomThreadLocal::ThreadCleaner::global_resource_holder;
+
 int CPUStreamsExecutor::get_stream_id() {
-    if (!_impl->_streams.find_thread_id()) {
+    if (!_impl->_streams->find_thread_id()) {
         return 0;
     }
-    auto stream = _impl->_streams.local();
+    auto stream = _impl->_streams->local();
     return stream->_streamId;
 }
 
@@ -495,23 +479,23 @@ int CPUStreamsExecutor::get_streams_num() {
 }
 
 int CPUStreamsExecutor::get_numa_node_id() {
-    if (!_impl->_streams.find_thread_id()) {
+    if (!_impl->_streams->find_thread_id()) {
         return 0;
     }
-    auto stream = _impl->_streams.local();
+    auto stream = _impl->_streams->local();
     return stream->_numaNodeId;
 }
 
 int CPUStreamsExecutor::get_socket_id() {
-    if (!_impl->_streams.find_thread_id()) {
+    if (!_impl->_streams->find_thread_id()) {
         return 0;
     }
-    auto stream = _impl->_streams.local();
+    auto stream = _impl->_streams->local();
     return stream->_socketId;
 }
 
 std::vector<int> CPUStreamsExecutor::get_rank() {
-    auto stream = _impl->_streams.local();
+    auto stream = _impl->_streams->local();
     return stream->_rank;
 }
 

src/plugins/intel_gpu/src/graph/impls/onednn/concatenation_onednn.cpp

@@ -59,9 +59,14 @@ struct concatenation_onednn : typed_primitive_onednn_impl<concatenation, dnnl::c
                                                                                       const int64_t axis) {
         std::vector<dnnl::memory::desc> input_mds;
         for (size_t i = 0; i < impl_params.input_layouts.size(); i++) {
-            input_mds.push_back(onednn::layout_to_memory_desc(impl_params.get_input_layout(i)));
+            auto input_layout = impl_params.get_input_layout(i);
+            input_mds.push_back(onednn::layout_to_memory_desc(input_layout, false, input_layout.format.is_blocked()));
         }
-        auto output_md = onednn::layout_to_memory_desc(impl_params.get_output_layout());
+
+        dnnl::memory::desc output_md;
+        auto output_layout = impl_params.get_output_layout();
+        output_md = onednn::layout_to_memory_desc(output_layout, false, output_layout.format.is_blocked());
+
         return std::make_shared<dnnl::concat::primitive_desc>(
             engine.get_onednn_engine(),
             output_md,