Replace deprecated tbb task for tbb >= 2021

AUTHORS

@@ -1,2 +1,3 @@
+Alex Dewar <alex.dewar@gmx.co.uk>
 Nate Koenig <nkoenig@osrfoundation.org>
 John Hsu <hsu@osrfoundation.org>

cmake/SearchForStuff.cmake

@@ -301,7 +301,7 @@ if (PKG_CONFIG_FOUND)
 
   #################################################
   # Find TBB
-  pkg_check_modules(TBB tbb<2021)
+  pkg_check_modules(TBB tbb)
   set (TBB_PKG_CONFIG "tbb")
   if (NOT TBB_FOUND)
     message(STATUS "TBB not found, attempting to detect manually")

gazebo/transport/CMakeLists.txt

@@ -29,6 +29,7 @@ set (headers
   SubscribeOptions.hh
   Subscriber.hh
   SubscriptionTransport.hh
+  TaskGroup.hh
   TopicManager.hh
   TransportIface.hh
   TransportTypes.hh
@@ -70,6 +71,12 @@ if (WIN32)
   target_link_libraries(gazebo_transport ws2_32 Iphlpapi)
 endif()
 
+if(${CMAKE_VERSION} VERSION_LESS "3.13.0")
+  link_directories(${TBB_LIBRARY_DIRS})
+else()
+  target_link_directories(gazebo_transport PUBLIC ${TBB_LIBRARY_DIRS})
+endif()
+
 if (USE_PCH)
     add_pch(gazebo_transport transport_pch.hh ${Boost_PKGCONFIG_CFLAGS} "-I${PROTOBUF_INCLUDE_DIR}" "-I${TBB_INCLUDEDIR}")
 endif()

gazebo/transport/Connection.hh

@@ -17,7 +17,16 @@
 #ifndef _CONNECTION_HH_
 #define _CONNECTION_HH_
 
+#undef emit
 #include <tbb/task.h>
+#define emit
+
+// If TBB_VERSION_MAJOR is not defined, this means that
+// tbb >= 2021 and we can include the tbb/version.h header
+#ifndef TBB_VERSION_MAJOR
+#include <tbb/version.h>
+#endif
+
 #include <google/protobuf/message.h>
 
 // This fixes compiler warnings, see #3147 and #3160
@@ -41,6 +50,9 @@
 #include "gazebo/common/Console.hh"
 #include "gazebo/common/Exception.hh"
 #include "gazebo/common/WeakBind.hh"
+#if TBB_VERSION_MAJOR >= 2021
+#include "gazebo/transport/TaskGroup.hh"
+#endif
 #include "gazebo/util/system.hh"
 
 #define HEADER_LENGTH 8
@@ -58,7 +70,11 @@ namespace gazebo
     /// \cond
     /// \brief A task instance that is created when data is read from
     /// a socket and used by TBB
+#if TBB_VERSION_MAJOR < 2021
     class GZ_TRANSPORT_VISIBLE ConnectionReadTask : public tbb::task
+#else
+    class GZ_TRANSPORT_VISIBLE ConnectionReadTask
+#endif
     {
       /// \brief Constructor
       /// \param[_in] _func Boost function pointer, which is the function
@@ -72,14 +88,19 @@ namespace gazebo
               {
               }
 
+#if TBB_VERSION_MAJOR < 2021
       /// \bried Overridden function from tbb::task that exectues the data
       /// callback.
       public: tbb::task *execute()
               {
                 this->func(this->data);
                 return NULL;
               }
-
+#else
+      /// \brief Execute the data callback
+      public: void operator()() const
+              { this->func(this->data); }
+#endif
       /// \brief The boost function pointer
       private: boost::function<void (const std::string &)> func;
 
@@ -314,12 +335,16 @@ namespace gazebo
 
                 if (!_e && !transport::is_stopped())
                 {
+#if TBB_VERSION_MAJOR < 2021
                   ConnectionReadTask *task = new(tbb::task::allocate_root())
                         ConnectionReadTask(boost::get<0>(_handler), data);
                   tbb::task::enqueue(*task);
 
                   // Non-tbb version:
                   // boost::get<0>(_handler)(data);
+#else
+                  this->taskGroup.run<ConnectionReadTask>(boost::get<0>(_handler), data);
+#endif
                 }
               }
 
@@ -376,7 +401,7 @@ namespace gazebo
       private: boost::asio::ip::tcp::endpoint GetRemoteEndpoint() const;
 
       /// \brief Gets hostname
-      /// \param[in] _ep The end point to get the hostename of
+      /// \param[in] _ep The end point to get the hostname of
       private: static std::string GetHostname(
                    boost::asio::ip::tcp::endpoint _ep);
 
@@ -469,6 +494,11 @@ namespace gazebo
 
       /// \brief True if the connection is open.
       private: bool isOpen;
+
+#if TBB_VERSION_MAJOR >= 2021
+      /// \brief For managing asynchronous tasks with tbb
+      private: TaskGroup taskGroup;
+#endif
     };
     /// \}
   }

gazebo/transport/ConnectionManager.cc

@@ -27,6 +27,7 @@
 using namespace gazebo;
 using namespace transport;
 
+#if TBB_VERSION_MAJOR < 2021
 /// TBB task to process nodes.
 class TopicManagerProcessTask : public tbb::task
 {
@@ -37,20 +38,30 @@ class TopicManagerProcessTask : public tbb::task
             return NULL;
           }
 };
+#endif
 
 /// TBB task to establish subscriber to publisher connection.
+#if TBB_VERSION_MAJOR < 2021
 class TopicManagerConnectionTask : public tbb::task
+#else
+class TopicManagerConnectionTask
+#endif
 {
   /// \brief Constructor.
   /// \param[in] _pub Publish message
   public: explicit TopicManagerConnectionTask(msgs::Publish _pub) : pub(_pub) {}
 
   /// Implements the necessary execute function
+#if TBB_VERSION_MAJOR < 2021
   public: tbb::task *execute()
           {
             TopicManager::Instance()->ConnectSubToPub(pub);
             return NULL;
           }
+#else
+  public: void operator()() const
+          { TopicManager::Instance()->ConnectSubToPub(pub); }
+#endif
 
   /// \brief Publish message
   private: msgs::Publish pub;
@@ -273,11 +284,6 @@ void ConnectionManager::RunUpdate()
   if (this->masterConn)
     this->masterConn->ProcessWriteQueue();
 
-  // Use TBB to process nodes. Need more testing to see if this makes
-  // a difference.
-  // TopicManagerProcessTask *task = new(tbb::task::allocate_root())
-  //   TopicManagerProcessTask();
-  // tbb::task::enqueue(*task);
   boost::recursive_mutex::scoped_lock lock(this->connectionMutex);
 
   TopicManager::Instance()->ProcessNodes();
@@ -403,9 +409,13 @@ void ConnectionManager::ProcessMessage(const std::string &_data)
     if (pub.host() != this->serverConn->GetLocalAddress() ||
         pub.port() != this->serverConn->GetLocalPort())
     {
+#if TBB_VERSION_MAJOR < 2021
       TopicManagerConnectionTask *task = new(tbb::task::allocate_root())
       TopicManagerConnectionTask(pub);
       tbb::task::enqueue(*task);
+#else
+      this->taskGroup.run<TopicManagerConnectionTask>(pub);
+#endif
     }
   }
   // publisher_subscribe. This occurs when we try to subscribe to a topic, and

gazebo/transport/ConnectionManager.hh

@@ -27,8 +27,11 @@
 #include "gazebo/msgs/msgs.hh"
 #include "gazebo/common/SingletonT.hh"
 
-#include "gazebo/transport/Publisher.hh"
 #include "gazebo/transport/Connection.hh"
+#include "gazebo/transport/Publisher.hh"
+#if TBB_VERSION_MAJOR >= 2021
+#include "gazebo/transport/TaskGroup.hh"
+#endif
 #include "gazebo/util/system.hh"
 
 /// \brief Explicit instantiation for typed SingletonT.
@@ -194,6 +197,11 @@ namespace gazebo
       /// \brief Condition used for synchronization
       private: boost::condition_variable namespaceCondition;
 
+#if TBB_VERSION_MAJOR >= 2021
+      /// \brief For managing asynchronous tasks with tbb
+      private: TaskGroup taskGroup;
+#endif
+
       // Singleton implementation
       private: friend class SingletonT<ConnectionManager>;
     };

gazebo/transport/Node.hh

@@ -18,19 +18,27 @@
 #ifndef GAZEBO_TRANSPORT_NODE_HH_
 #define GAZEBO_TRANSPORT_NODE_HH_
 
+#undef emit
 #include <tbb/task.h>
+#define emit
+#ifndef TBB_VERSION_MAJOR
+#include <tbb/version.h>
+#endif
 
 // This fixes compiler warnings, see #3147 and #3160
 #ifndef BOOST_BIND_GLOBAL_PLACEHOLDERS
 #define BOOST_BIND_GLOBAL_PLACEHOLDERS
 #endif
+
 #include <boost/bind.hpp>
 #include <boost/enable_shared_from_this.hpp>
 #include <map>
 #include <list>
 #include <string>
 #include <vector>
-
+#if TBB_VERSION_MAJOR >= 2021
+#include "gazebo/transport/TaskGroup.hh"
+#endif
 #include "gazebo/transport/TransportTypes.hh"
 #include "gazebo/transport/TopicManager.hh"
 #include "gazebo/util/system.hh"
@@ -41,7 +49,11 @@ namespace gazebo
   {
     /// \cond
     /// \brief Task used by Node::Publish to publish on a one-time publisher
+#if TBB_VERSION_MAJOR < 2021
     class GZ_TRANSPORT_VISIBLE PublishTask : public tbb::task
+#else
+    class GZ_TRANSPORT_VISIBLE PublishTask
+#endif
     {
       /// \brief Constructor
       /// \param[in] _pub Publisher to publish the message on.
@@ -54,16 +66,23 @@ namespace gazebo
         this->msg->CopyFrom(_message);
       }
 
+#if TBB_VERSION_MAJOR < 2021
       /// \brief Overridden function from tbb::task that exectues the
       /// publish task.
       public: tbb::task *execute()
+#else
+      /// \brief Executes the publish task.
+      public: void operator()()
+#endif
               {
                 this->pub->WaitForConnection();
                 this->pub->Publish(*this->msg, true);
                 this->pub->SendMessage();
                 delete this->msg;
                 this->pub.reset();
+#if TBB_VERSION_MAJOR < 2021
                 return NULL;
+#endif
               }
 
       /// \brief Pointer to the publisher.
@@ -164,11 +183,15 @@ namespace gazebo
                   const google::protobuf::Message &_message)
               {
                 transport::PublisherPtr pub = this->Advertise<M>(_topic);
-                PublishTask *task = new(tbb::task::allocate_root())
+#if TBB_VERSION_MAJOR < 2021
+                PublishTask *task = new(tbb::task::allocate_root()) 
                   PublishTask(pub, _message);
 
                 tbb::task::enqueue(*task);
                 return;
+#else
+                this->taskGroup.run<PublishTask>(pub, _message);
+#endif
               }
 
       /// \brief Advertise a topic
@@ -425,6 +448,11 @@ namespace gazebo
 
       /// \brief List of newly arrive messages
       private: std::map<std::string, std::list<MessagePtr> > incomingMsgsLocal;
+
+#if TBB_VERSION_MAJOR >= 2021
+      /// \brief For managing asynchronous tasks with tbb
+      private: TaskGroup taskGroup;
+#endif
 
       private: boost::mutex publisherMutex;
       private: boost::mutex publisherDeleteMutex;

gazebo/transport/TaskGroup.hh

@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) 2021 Open Source Robotics Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+*/
+#ifndef _TASK_GROUP_HH_
+#define _TASK_GROUP_HH_
+
+#include <utility>
+
+// Emit is both a macro in Qt and a function defined by tbb
+#undef emit
+#include <tbb/tbb.h>
+#define emit
+
+// tbb::task was removed in v2021.01, so we need a workaround
+#if TBB_VERSION_MAJOR >= 2021
+namespace gazebo {
+  namespace transport {
+    class TaskGroup
+    {
+      public: ~TaskGroup() noexcept
+              {
+                // Wait for running tasks to finish
+                this->taskGroup.wait();
+              }
+
+      public: template<class Functor, class... Args> void run(Args&&... args)
+      {
+        this->taskGroup.run(Functor(std::forward<Args>(args)...));
+      }
+
+      private: tbb::task_group taskGroup;
+    };
+  }
+}
+#else
+namespace gazebo {
+  namespace transport {
+    class TaskGroup
+    {
+      /// \brief A helper class which provides the requisite execute() method
+      /// required by tbb.
+      private: template<class T> class TaskWrapper : public tbb::task
+      {
+        public: template<class... Args> TaskWrapper<T>(Args&&... args)
+          : functor(std::forward<Args>(args)...)
+        {
+        }
+
+        public: tbb::task *execute()
+                {
+                  this->functor();
+                  return nullptr;
+                }
+
+        private: T functor;
+      };
+
+      public: template<class Functor, class... Args> void run(Args&&... args)
+              {
+                TaskWrapper<Functor> *task = new (tbb::task::allocate_root())
+                    TaskWrapper<Functor>(std::forward<Args>(args)...);
+                tbb::task::enqueue(*task);
+              }
+    };
+  }
+}
+
+#endif
+#endif