Ticket #1707: ParallelFor_rangemanager.patch

ps/Parallel.cpp

@@ +1 @@
+/* Copyright (C) 2013 Wildfire Games.
+ * This file is part of 0 A.D.
+ *
+ * 0 A.D. is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * 0 A.D. is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with 0 A.D.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#include "precompiled.h"
+
+#include "Parallel.h"
+#include <lib/posix/posix_pthread.h>
+#include <lib/sysdep/os_cpu.h>
+
+namespace ps
+{
+    #define MAX_WORKER_THREADS 8
+
+    struct worker
+    {
+        pthread_t thread;
+        sem_t jobPending;
+        sem_t workDone;
+
+        void (*job)(void* param);
+        void* arg;
+    };
+    static worker   g_Workers[MAX_WORKER_THREADS];
+    static bool     g_Initialized   = false;
+    static int      g_NumWorkers    = 0; // default workers 0, incase we have a 1-core CPU
+
+
+
+    extern "C" static void* _worker(void* arg)
+    {
+        worker* w = (worker*)arg;
+        while (true)
+        {
+            sem_wait(&w->jobPending);   // wait for job signal
+            w->job(w->arg);             // launch the job
+            sem_post(&w->workDone);     // post done signal
+        }
+
+    }
+    static void uninit_workers()
+    {
+        for (int i = 0; i < g_NumWorkers; ++i)
+        {
+            pthread_cancel(g_Workers[i].thread);
+            sem_destroy(&g_Workers[i].jobPending);
+            sem_destroy(&g_Workers[i].workDone);
+        }
+    }
+    static int init_workers()
+    {
+        if (!g_Initialized)
+        {
+            int numCPUS = os_cpu_NumProcessors();
+            if (numCPUS > 1)
+            {
+                if (numCPUS > MAX_WORKER_THREADS)
+                    numCPUS = MAX_WORKER_THREADS;
+
+                g_NumWorkers = numCPUS;
+                for (int i = 0; i < numCPUS; ++i)
+                {
+                    sem_init(&g_Workers[i].jobPending, 0, 0);
+                    sem_init(&g_Workers[i].workDone, 0, 0);
+                    pthread_create(&g_Workers[i].thread, NULL, _worker, &g_Workers[i]);
+                }
+                atexit(uninit_workers);
+            }
+            g_Initialized = true;
+        }
+        return 1;
+    }
+
+
+
+    struct pfor
+    {
+        int startIndex;
+        int endIndex;
+        int step;
+        pfor_functor* callback;
+    };
+    static void pfor_threadfunc(pfor* arg)
+    {
+        int i            = arg->startIndex;
+        int end          = arg->endIndex;
+        int step         = arg->step;
+        pfor_functor* cb = arg->callback;
+        for (; i < end; i += step)
+            if ((*cb)(i) == false)
+                break;
+    }
+    void parallel_for(int startIndex, int endIndex, pfor_functor& functor)
+    {
+        static int initialized = init_workers(); // init once
+
+        // special case: we don't have any workers (1-core CPU)
+        if (g_NumWorkers == 0)
+        {
+            // just a normal for-loop then
+            for (int i = startIndex; i < endIndex; ++i)
+                if (functor(i) == false)
+                    break;
+            return;
+        }
+
+        pfor params[MAX_WORKER_THREADS];
+
+        // launch threads
+        for (int i = 0; i < g_NumWorkers; ++i)
+        {
+            params[i].startIndex = startIndex;
+            params[i].endIndex = endIndex;
+            params[i].step = g_NumWorkers;
+            params[i].callback = &functor;
+            g_Workers[i].job = (void(*)(void*))&pfor_threadfunc;
+            g_Workers[i].arg = &params[i];
+
+            sem_post(&g_Workers[i].jobPending); // notify worker - new job pending
+        }
+
+        // wait for them to finish
+        for (int i = 0; i < g_NumWorkers; ++i)
+        {
+            sem_wait(&g_Workers[i].workDone); // wait for work to finish
+        }
+    }
+
+}

ps/Parallel.h

@@ +1 @@
+/* Copyright (C) 2013 Wildfire Games.
+ * This file is part of 0 A.D.
+ *
+ * 0 A.D. is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * 0 A.D. is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with 0 A.D.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef INCLUDED_PARALLEL
+#define INCLUDED_PARALLEL
+
+
+namespace ps
+{
+
+    struct pfor_functor
+    {
+        /**
+         * Parallel - For callback.
+         * @return TRUE to continue iteration, FALSE to break from the loop.
+         */
+        virtual bool operator()(int atIndex) = 0;
+
+        // make sure the correct destructors get used
+        virtual ~pfor_functor() {}
+    };
+
+    /**
+     * A simple parallel FOR loop.
+     * for (int i = 0; i < 10; i++)  ->  parallel_for(0, 10, pfor_functor())
+     *
+     * @param startIndex Starting index (inclusive)
+     * @param endIndex Ending index (exclusive)
+     * @param functor Functor to call on every index
+     */
+    void parallel_for(int startIndex, int endIndex, pfor_functor& functor);
+}
+#endif
+ No newline at end of file

simulation2/components/CCmpRangeManager.cpp

@@ -39 +39 @@
 #include "ps/Profile.h"
 #include "renderer/Scene.h"
 #include "lib/ps_stl.h"
+#include "ps/Parallel.h"
 
 
 #define DEBUG_RANGE_MANAGER_BOUNDS 0
@@ -814 +815 @@
             m_DebugOverlayLines.clear();
     }
 
-    /**
-     * Update all currently-enabled active queries.
-     */
-    void ExecuteActiveQueries()
+
+
+
+
+    struct ExecuteParallelQuery : ps::pfor_functor
     {
-        PROFILE3("ExecuteActiveQueries");
-
         // Store a queue of all messages before sending any, so we can assume
         // no entities will move until we've finished checking all the ranges
         std::vector<std::pair<entity_id_t, CMessageRangeUpdate> > messages;
-        std::vector<entity_id_t> results;
-        std::vector<entity_id_t> added;
-        std::vector<entity_id_t> removed;
+        std::vector<std::pair<tag_t, Query*> > queries;
+        CCmpRangeManager* rangeMan;
+        CMutex mutex;
 
-        for (std::map<tag_t, Query>::iterator it = m_Queries.begin(); it != m_Queries.end(); ++it)
+
+        ExecuteParallelQuery(CCmpRangeManager* rangeMan) : rangeMan(rangeMan) 
         {
-            Query& query = it->second;
+        }
 
+        virtual bool operator()(int atIndex)
+        {
+            tag_t tag = queries[atIndex].first;
+            Query& query = *queries[atIndex].second;
+
             if (!query.enabled)
-                continue;
+                return true; // continue
 
             CmpPtr<ICmpPosition> cmpSourcePosition(query.source);           
             if (!cmpSourcePosition || !cmpSourcePosition->IsInWorld())
-                continue;
+                return true; // continue
 
-            results.clear();
+            std::vector<entity_id_t> added;
+            std::vector<entity_id_t> removed;
+            std::vector<entity_id_t> results;
             results.reserve(query.lastMatch.size());
-            PerformQuery(query, results);
+            
+            rangeMan->PerformQuery(query, results);
 
             // Compute the changes vs the last match
-            added.clear();
-            removed.clear();
             // Return the 'added' list sorted by distance from the entity
             // (Don't bother sorting 'removed' because they might not even have positions or exist any more)
-            std::set_difference(results.begin(), results.end(), query.lastMatch.begin(), query.lastMatch.end(), 
-                std::back_inserter(added));
-            std::set_difference(query.lastMatch.begin(), query.lastMatch.end(), results.begin(), results.end(), 
-                std::back_inserter(removed));
+            std::set_difference(results.begin(), results.end(), 
+                query.lastMatch.begin(), query.lastMatch.end(), std::back_inserter(added));
+            std::set_difference(query.lastMatch.begin(), query.lastMatch.end(), 
+                results.begin(), results.end(), std::back_inserter(removed));
+            
             if (added.empty() && removed.empty())
-                continue;
+                return true; // continue
 
-            std::stable_sort(added.begin(), added.end(), EntityDistanceOrdering(m_EntityData, cmpSourcePosition->GetPosition2D()));
+            std::stable_sort(added.begin(), added.end(), 
+                EntityDistanceOrdering(rangeMan->m_EntityData, cmpSourcePosition->GetPosition2D()));
 
-            messages.resize(messages.size() + 1);
-            std::pair<entity_id_t, CMessageRangeUpdate>& back = messages.back();
-            back.first = query.source.GetId();
-            back.second.tag = it->first;
-            back.second.added.swap(added);
-            back.second.removed.swap(removed);
-            it->second.lastMatch.swap(results);
+            {
+                CScopeLock lock(mutex);
+                messages.resize(messages.size() + 1);
+                std::pair<entity_id_t, CMessageRangeUpdate>& back = messages.back();
+                back.first = query.source.GetId();
+                back.second.tag = tag;
+                back.second.added.swap(added);
+                back.second.removed.swap(removed);
+            
+            }
+            query.lastMatch.swap(results);
+
+            return true; // continue
         }
+    };
 
+    /**
+     * Update all currently-enabled active queries.
+     */
+    void ExecuteActiveQueries()
+    {
+        PROFILE3("ExecuteActiveQueries");
+
+        int size = m_Queries.size();
+        if (size == 0)
+            return;
+
+        ExecuteParallelQuery pquery(this);
+
+        // prepare queries into a random-access vector, so we can use Parallel For
+        pquery.queries.reserve(size);
+        for (std::map<tag_t, Query>::iterator it = m_Queries.begin(); it != m_Queries.end(); ++it)
+            pquery.queries.push_back(std::make_pair(it->first, &it->second));
+
+        // run the parallel for loop
+        ps::parallel_for(0, size, pquery);
+
         CComponentManager& cmpMgr = GetSimContext().GetComponentManager();
-        for (size_t i = 0; i < messages.size(); ++i)
-            cmpMgr.PostMessage(messages[i].first, messages[i].second);
+        for (size_t i = 0; i < pquery.messages.size(); ++i)
+            cmpMgr.PostMessage(pquery.messages[i].first, pquery.messages[i].second);
     }
 
     /**