Use oneapi::tbb::task_group instead of tbb::task if available

As tbb::task has been removed in oneTBB 2021.01, we need to replace its use with oneapi::tbb::task_group. Define a wrapper so that tbb::task_group is used for newer versions of oneTBB.

Fixes gazebosim#2867.

Signed-off-by: Alex Dewar <alex.dewar@gmx.co.uk>

gazebo/transport/CMakeLists.txt

@@ -29,6 +29,7 @@ set (headers
   SubscribeOptions.hh
   Subscriber.hh
   SubscriptionTransport.hh
+  TaskGroup.hh
   TopicManager.hh
   TransportIface.hh
   TransportTypes.hh

gazebo/transport/Connection.hh

@@ -17,7 +17,6 @@
 #ifndef _CONNECTION_HH_
 #define _CONNECTION_HH_
 
-#include <tbb/task.h>
 #include <google/protobuf/message.h>
 
 #include <boost/asio.hpp>
@@ -37,6 +36,7 @@
 #include "gazebo/common/Console.hh"
 #include "gazebo/common/Exception.hh"
 #include "gazebo/common/WeakBind.hh"
+#include "gazebo/transport/TaskGroup.hh"
 #include "gazebo/util/system.hh"
 
 #define HEADER_LENGTH 8
@@ -54,7 +54,7 @@ namespace gazebo
     /// \cond
     /// \brief A task instance that is created when data is read from
     /// a socket and used by TBB
-    class GZ_TRANSPORT_VISIBLE ConnectionReadTask : public tbb::task
+    class GZ_TRANSPORT_VISIBLE ConnectionReadTask
     {
       /// \brief Constructor
       /// \param[_in] _func Boost function pointer, which is the function
@@ -68,13 +68,9 @@ namespace gazebo
               {
               }
 
-      /// \bried Overridden function from tbb::task that exectues the data
-      /// callback.
-      public: tbb::task *execute()
-              {
-                this->func(this->data);
-                return NULL;
-              }
+      /// \brief Execute the data callback
+      public: void operator()() const
+              { this->func(this->data); }
 
       /// \brief The boost function pointer
       private: boost::function<void (const std::string &)> func;
@@ -310,12 +306,7 @@ namespace gazebo
 
                 if (!_e && !transport::is_stopped())
                 {
-                  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);
+                  this->taskGroup.run<ConnectionReadTask>(boost::get<0>(_handler), data);
                 }
               }
 
@@ -465,6 +456,10 @@ namespace gazebo
 
       /// \brief True if the connection is open.
       private: bool isOpen;
+
+
+      /// \brief For managing asynchronous tasks with tbb
+      private: TaskGroup taskGroup;
     };
     /// \}
   }

gazebo/transport/ConnectionManager.cc

@@ -28,18 +28,15 @@ using namespace gazebo;
 using namespace transport;
 
 /// TBB task to establish subscriber to publisher connection.
-class TopicManagerConnectionTask : public tbb::task
+class TopicManagerConnectionTask
 {
   /// \brief Constructor.
   /// \param[in] _pub Publish message
   public: explicit TopicManagerConnectionTask(msgs::Publish _pub) : pub(_pub) {}
 
   /// Implements the necessary execute function
-  public: tbb::task *execute()
-          {
-            TopicManager::Instance()->ConnectSubToPub(pub);
-            return NULL;
-          }
+  public: void operator()() const
+          { TopicManager::Instance()->ConnectSubToPub(pub); }
 
   /// \brief Publish message
   private: msgs::Publish pub;
@@ -385,9 +382,7 @@ void ConnectionManager::ProcessMessage(const std::string &_data)
     if (pub.host() != this->serverConn->GetLocalAddress() ||
         pub.port() != this->serverConn->GetLocalPort())
     {
-      TopicManagerConnectionTask *task = new(tbb::task::allocate_root())
-      TopicManagerConnectionTask(pub);
-      tbb::task::enqueue(*task);
+      this->taskGroup.run<TopicManagerConnectionTask>(pub);
     }
   }
   // publisher_subscribe. This occurs when we try to subscribe to a topic, and

gazebo/transport/ConnectionManager.hh

@@ -27,8 +27,9 @@
 #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"
+#include "gazebo/transport/TaskGroup.hh"
 #include "gazebo/util/system.hh"
 
 /// \brief Explicit instantiation for typed SingletonT.
@@ -193,6 +194,9 @@ namespace gazebo
 
       /// \brief Condition used for synchronization
       private: boost::condition_variable namespaceCondition;
+
+      /// \brief For managing asynchronous tasks with tbb
+      private: TaskGroup taskGroup;
 
       // Singleton implementation
       private: friend class SingletonT<ConnectionManager>;

gazebo/transport/Node.hh

@@ -18,14 +18,14 @@
 #ifndef GAZEBO_TRANSPORT_NODE_HH_
 #define GAZEBO_TRANSPORT_NODE_HH_
 
-#include <tbb/task.h>
 #include <boost/bind.hpp>
 #include <boost/enable_shared_from_this.hpp>
 #include <map>
 #include <list>
 #include <string>
 #include <vector>
 
+#include "gazebo/transport/TaskGroup.hh"
 #include "gazebo/transport/TransportTypes.hh"
 #include "gazebo/transport/TopicManager.hh"
 #include "gazebo/util/system.hh"
@@ -36,7 +36,7 @@ namespace gazebo
   {
     /// \cond
     /// \brief Task used by Node::Publish to publish on a one-time publisher
-    class GZ_TRANSPORT_VISIBLE PublishTask : public tbb::task
+    class GZ_TRANSPORT_VISIBLE PublishTask
     {
       /// \brief Constructor
       /// \param[in] _pub Publisher to publish the message on.
@@ -49,16 +49,14 @@ namespace gazebo
         this->msg->CopyFrom(_message);
       }
 
-      /// \brief Overridden function from tbb::task that exectues the
-      /// publish task.
-      public: tbb::task *execute()
+      /// \brief Executes the publish task.
+      public: void operator()()
               {
                 this->pub->WaitForConnection();
                 this->pub->Publish(*this->msg, true);
                 this->pub->SendMessage();
                 delete this->msg;
                 this->pub.reset();
-                return NULL;
               }
 
       /// \brief Pointer to the publisher.
@@ -159,11 +157,7 @@ namespace gazebo
                   const google::protobuf::Message &_message)
               {
                 transport::PublisherPtr pub = this->Advertise<M>(_topic);
-                PublishTask *task = new(tbb::task::allocate_root())
-                  PublishTask(pub, _message);
-
-                tbb::task::enqueue(*task);
-                return;
+                this->taskGroup.run<PublishTask>(pub, _message);
               }
 
       /// \brief Advertise a topic
@@ -418,6 +412,9 @@ namespace gazebo
 
       /// \brief List of newly arrive messages
       private: std::map<std::string, std::list<MessagePtr> > incomingMsgsLocal;
+
+      /// \brief For managing asynchronous tasks with tbb
+      private: TaskGroup taskGroup;
 
       private: boost::mutex publisherMutex;
       private: boost::mutex publisherDeleteMutex;

gazebo/transport/TaskGroup.hh

@@ -0,0 +1,84 @@
+/*
+ * Copyright (C) 2021 Alex Dewar
+ *
+ * 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>
+
+#if __has_include(<oneapi/tbb/task_group.h>)
+
+// Emit is both a macro in Qt and a function defined by tbb
+#undef emit
+#include <oneapi/tbb/task_group.h>
+#define emit
+
+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: oneapi::tbb::task_group taskGroup;
+    };
+  }
+}
+#else
+#include <tbb/task.h>
+
+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)
+              {
+                auto *task = new (tbb::task::allocate_root())
+                    TaskWrapper<Functor>(std::forward<Args>(args)...);
+                tbb::task::enqueue(*task);
+              }
+    };
+  }
+}
+
+#endif
+#endif

gazebo/transport/transport_pch.hh

@@ -46,6 +46,12 @@
 #include <string>
 #include <tbb/blocked_range.h>
 #include <tbb/parallel_for.h>
+#if __has_include(<oneapi/tbb/task_group.h>)
+#undef emit
+#include <oneapi/tbb/task_group.h>
+#define emit
+#else
 #include <tbb/task.h>
+#endif
 #include <utility>
 #include <vector>