Implement async dispatcher #221

src/client/localplayer.h

@@ -29,7 +29,7 @@
 class LocalPlayer : public Player
 {
     enum {
-        PREWALK_TIMEOUT = 5000
+        PREWALK_TIMEOUT = 1000
     };
 
 public:

src/framework/CMakeLists.txt

@@ -64,6 +64,8 @@ set(framework_SOURCES ${framework_SOURCES}
     ${CMAKE_CURRENT_LIST_DIR}/core/application.h
     ${CMAKE_CURRENT_LIST_DIR}/core/adaptativeframecounter.cpp
     ${CMAKE_CURRENT_LIST_DIR}/core/adaptativeframecounter.h
+    ${CMAKE_CURRENT_LIST_DIR}/core/asyncdispatcher.cpp
+    ${CMAKE_CURRENT_LIST_DIR}/core/asyncdispatcher.h
     ${CMAKE_CURRENT_LIST_DIR}/core/binarytree.cpp
     ${CMAKE_CURRENT_LIST_DIR}/core/binarytree.h
     ${CMAKE_CURRENT_LIST_DIR}/core/clock.cpp
@@ -194,6 +196,7 @@ message(STATUS "Build revision: ${BUILD_REVISION}")
 add_definitions(-D"BUILD_REVISION=\\\"${BUILD_REVISION}\\\"")
 
 # find boost
+set(framework_DEFINITIONS ${framework_DEFINITIONS} -DBOOST_THREAD_PROVIDES_FUTURE) # enable boost::future
 set(REQUIRED_BOOST_COMPONENTS system thread chrono)
 if(WIN32)
     set(Boost_THREADAPI win32)

src/framework/core/application.cpp

@@ -27,6 +27,7 @@
 #include <framework/core/modulemanager.h>
 #include <framework/core/eventdispatcher.h>
 #include <framework/core/configmanager.h>
+#include "asyncdispatcher.h"
 #include <framework/luaengine/luainterface.h>
 #include <framework/platform/crashhandler.h>
 #include <framework/platform/platform.h>
@@ -76,6 +77,8 @@ void Application::init(std::vector<std::string>& args)
     // process args encoding
     g_platform.processArgs(args);
 
+    g_asyncDispatcher.init();
+
     std::string startupOptions;
     for(uint i=1;i<args.size();++i) {
         const std::string& arg = args[i];
@@ -109,6 +112,8 @@ void Application::deinit()
     // poll remaining events
     poll();
 
+    g_asyncDispatcher.terminate();
+
     // disable dispatcher events
     g_dispatcher.shutdown();
 }

src/framework/core/asyncdispatcher.cpp

@@ -0,0 +1,71 @@
+/*
+ * Copyright (c) 2010-2013 OTClient <https://github.com/edubart/otclient>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+#include "asyncdispatcher.h"
+
+AsyncDispatcher g_asyncDispatcher;
+
+void AsyncDispatcher::init()
+{
+    spawn_thread();
+}
+
+void AsyncDispatcher::terminate()
+{
+    stop();
+    m_tasks.clear();
+}
+
+void AsyncDispatcher::spawn_thread()
+{
+    m_running = true;
+    m_threads.push_back(std::thread(std::bind(&AsyncDispatcher::exec_loop, this)));
+}
+
+void AsyncDispatcher::stop()
+{
+    m_mutex.lock();
+    m_running = false;
+    m_condition.notify_all();
+    m_mutex.unlock();
+    for(std::thread& thread : m_threads)
+        thread.join();
+    m_threads.clear();
+};
+
+void AsyncDispatcher::exec_loop() {
+    std::unique_lock<std::mutex> lock(m_mutex);
+    while(true) {
+        while(m_tasks.size() == 0 && m_running)
+            m_condition.wait(lock);
+
+        if(!m_running)
+            return;
+
+        std::function<void()> task = m_tasks.front();
+        m_tasks.pop_front();
+
+        lock.unlock();
+        task();
+        lock.lock();
+    }
+}

src/framework/core/asyncdispatcher.h

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2010-2013 OTClient <https://github.com/edubart/otclient>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+#ifndef ASYNCDISPATCHER_H
+#define ASYNCDISPATCHER_H
+
+#include "declarations.h"
+#include <framework/stdext/thread.h>
+
+class AsyncDispatcher {
+public:
+    void init();
+    void terminate();
+
+    void spawn_thread();
+    void stop();
+
+    template<class F>
+    std::future<typename std::result_of<F()>::type> schedule(const F& task) {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        auto prom = std::make_shared<std::promise<typename std::result_of<F()>::type>>();
+        m_tasks.push_back([=]() { prom->set_value(task()); });
+        m_condition.notify_all();
+        return prom->get_future();
+    }
+
+protected:
+    void exec_loop();
+
+private:
+    std::list<std::function<void()>> m_tasks;
+    std::mutex m_mutex;
+    std::list<std::thread> m_threads;
+    std::condition_variable m_condition;
+    stdext::boolean<false> m_running;
+};
+
+extern AsyncDispatcher g_asyncDispatcher;
+
+#endif

src/framework/core/logger.cpp

@@ -35,6 +35,8 @@ Logger g_logger;
 
 void Logger::log(Fw::LogLevel level, const std::string& message)
 {
+    std::lock_guard<std::recursive_mutex> lock(m_mutex);
+
 #ifdef NDEBUG
     if(level == Fw::LogDebug)
         return;
@@ -95,6 +97,8 @@ void Logger::log(Fw::LogLevel level, const std::string& message)
 
 void Logger::logFunc(Fw::LogLevel level, const std::string& message, std::string prettyFunction)
 {
+    std::lock_guard<std::recursive_mutex> lock(m_mutex);
+
     prettyFunction = prettyFunction.substr(0, prettyFunction.find_first_of('('));
     if(prettyFunction.find_last_of(' ') != std::string::npos)
         prettyFunction = prettyFunction.substr(prettyFunction.find_last_of(' ') + 1);
@@ -114,6 +118,8 @@ void Logger::logFunc(Fw::LogLevel level, const std::string& message, std::string
 
 void Logger::fireOldMessages()
 {
+    std::lock_guard<std::recursive_mutex> lock(m_mutex);
+
     if(m_onLog) {
         auto backup = m_logMessages;
         for(const LogMessage& logMessage : backup) {
@@ -124,6 +130,8 @@ void Logger::fireOldMessages()
 
 void Logger::setLogFile(const std::string& file)
 {
+    std::lock_guard<std::recursive_mutex> lock(m_mutex);
+
     m_outFile.open(stdext::utf8_to_latin1(file.c_str()).c_str(), std::ios::out | std::ios::app);
     if(!m_outFile.is_open() || !m_outFile.good()) {
         g_logger.error(stdext::format("Unable to save log to '%s'", file));

src/framework/core/logger.h

@@ -25,6 +25,7 @@
 
 #include "../global.h"
 
+#include <framework/stdext/thread.h>
 #include <fstream>
 
 struct LogMessage {
@@ -61,6 +62,7 @@ private:
     std::list<LogMessage> m_logMessages;
     OnLogCallback m_onLog;
     std::ofstream m_outFile;
+    std::recursive_mutex m_mutex;
 };
 
 extern Logger g_logger;

src/framework/sound/soundfile.cpp

@@ -31,11 +31,10 @@ SoundFile::SoundFile(const FileStreamPtr& fileStream)
 
 SoundFilePtr SoundFile::loadSoundFile(const std::string& filename)
 {
+    stdext::timer t;
     FileStreamPtr file = g_resources.openFile(filename);
-    if(!file) {
-        g_logger.traceError(stdext::format("unable to open %s", filename));
-        return nullptr;
-    }
+    if(!file)
+        stdext::throw_exception(stdext::format("unable to open %s", filename));
 
     // cache file buffer to avoid lags from hard drive
     file->cache();
@@ -50,7 +49,7 @@ SoundFilePtr SoundFile::loadSoundFile(const std::string& filename)
         if(oggSoundFile->prepareOgg())
             soundFile = oggSoundFile;
     } else
-        g_logger.error(stdext::format("unknown sound file format %s", filename));
+        stdext::throw_exception(stdext::format("unknown sound file format %s", filename));
 
     return soundFile;
 }

src/framework/sound/soundmanager.cpp

@@ -30,6 +30,8 @@
 #include <framework/core/clock.h>
 #include <framework/core/eventdispatcher.h>
 #include <framework/core/resourcemanager.h>
+#include <framework/core/asyncdispatcher.h>
+#include <thread>
 
 SoundManager g_sounds;
 
@@ -57,9 +59,14 @@ void SoundManager::terminate()
 {
     ensureContext();
 
+    for(auto it = m_streamFiles.begin(); it != m_streamFiles.end();++it)
+        it->second.wait();
+    m_streamFiles.clear();
+
     m_sources.clear();
     m_buffers.clear();
     m_channels.clear();
+
     m_audioEnabled = false;
 
     alcMakeContextCurrent(nullptr);
@@ -86,6 +93,23 @@ void SoundManager::poll()
     lastUpdate = now;
 
     ensureContext();
+
+    for(auto it = m_streamFiles.begin(); it != m_streamFiles.end();) {
+        StreamSoundSourcePtr source = it->first;
+        std::future<SoundFilePtr>& future = it->second;
+
+        if(std::is_ready(future)) {
+            SoundFilePtr sound = future.get();
+            if(sound)
+                source->setSoundFile(sound);
+            else
+                source->stop();
+            it = m_streamFiles.erase(it);
+        } else {
+            ++it;
+        }
+    }
+
     for(auto it = m_sources.begin(); it != m_sources.end();) {
         SoundSourcePtr source = *it;
 
@@ -161,9 +185,8 @@ SoundSourcePtr SoundManager::play(std::string filename, float fadetime, float ga
     soundSource->setRelative(true);
     soundSource->setGain(gain);
 
-    if(fadetime > 0) {
+    if(fadetime > 0)
         soundSource->setFading(StreamSoundSource::FadingOn, fadetime);
-    }
 
     soundSource->play();
 
@@ -202,46 +225,56 @@ SoundSourcePtr SoundManager::createSoundSource(const std::string& filename)
             source = SoundSourcePtr(new SoundSource);
             source->setBuffer(it->second);
         } else {
-            SoundFilePtr soundFile = SoundFile::loadSoundFile(filename);
-            if(!soundFile)
-                return nullptr;
-
-            if(soundFile->getSize() <= MAX_CACHE_SIZE) {
-                source = SoundSourcePtr(new SoundSource);
-                SoundBufferPtr buffer = SoundBufferPtr(new SoundBuffer);
-                buffer->fillBuffer(soundFile);
-                source->setBuffer(buffer);
-                m_buffers[filename] = buffer;
-                g_logger.warning(stdext::format("uncached sound '%s' requested to be played", filename));
-            } else {
-                StreamSoundSourcePtr streamSource(new StreamSoundSource);
-                streamSource->setSoundFile(soundFile);
-                source = streamSource;
-
 #if defined __linux && !defined OPENGL_ES
             // due to OpenAL implementation bug, stereo buffers are always downmixed to mono on linux systems
             // this is hack to work around the issue
             // solution taken from http://opensource.creative.com/pipermail/openal/2007-April/010355.html
-                if(soundFile->getSampleFormat() == AL_FORMAT_STEREO16) {
             CombinedSoundSourcePtr combinedSource(new CombinedSoundSource);
+            StreamSoundSourcePtr streamSource;
 
+            streamSource = StreamSoundSourcePtr(new StreamSoundSource);
             streamSource->downMix(StreamSoundSource::DownMixLeft);
             streamSource->setRelative(true);
             streamSource->setPosition(Point(-128, 0));
             combinedSource->addSource(streamSource);
+            m_streamFiles[streamSource] = g_asyncDispatcher.schedule([=]() -> SoundFilePtr {
+                stdext::timer a;
+                try {
+                    return SoundFile::loadSoundFile(filename); 
+                } catch(std::exception& e) {
+                    g_logger.error(e.what());
+                    return nullptr;
+                }
+            });
 
             streamSource = StreamSoundSourcePtr(new StreamSoundSource);
-                    streamSource->setSoundFile(SoundFile::loadSoundFile(filename));
             streamSource->downMix(StreamSoundSource::DownMixRight);
             streamSource->setRelative(true);
             streamSource->setPosition(Point(128,0));
             combinedSource->addSource(streamSource);
+            m_streamFiles[streamSource] = g_asyncDispatcher.schedule([=]() -> SoundFilePtr {
+                try {
+                    return SoundFile::loadSoundFile(filename); 
+                } catch(std::exception& e) {
+                    g_logger.error(e.what());
+                    return nullptr;
+                }
+            });
 
             source = combinedSource;
+#else
+            StreamSoundSourcePtr streamSource(new StreamSoundSource);
+            m_streamFiles[streamSource] = m_loadJobs [=]() -> SoundFilePtr {
+                try {
+                    return SoundFile::loadSoundFile(filename); 
+                } catch(std::exception& e) {
+                    g_logger.error(e.what());
+                    return nullptr;
                 }
+            });
+            source = streamSource;
 #endif
         }
-        }
     } catch(std::exception& e) {
         g_logger.error(stdext::format("failed to load sound source: '%s'", e.what()));
         return nullptr;

src/framework/sound/soundmanager.h

@@ -57,6 +57,7 @@ private:
     ALCdevice *m_device;
     ALCcontext *m_context;
 
+    std::map<StreamSoundSourcePtr, std::future<SoundFilePtr>> m_streamFiles;
     std::unordered_map<std::string, SoundBufferPtr> m_buffers;
     std::vector<SoundSourcePtr> m_sources;
     stdext::boolean<true> m_audioEnabled;

src/framework/sound/streamsoundsource.cpp

@@ -25,6 +25,7 @@
 #include "soundfile.h"
 
 #include <framework/util/databuffer.h>
+#include <boost/concept_check.hpp>
 
 StreamSoundSource::StreamSoundSource()
 {
@@ -41,22 +42,26 @@ StreamSoundSource::~StreamSoundSource()
 void StreamSoundSource::setSoundFile(const SoundFilePtr& soundFile)
 {
     m_soundFile = soundFile;
+    if(m_waitingFile) {
+        m_waitingFile = false;
+        play();
+    }
 }
 
 void StreamSoundSource::play()
 {
     m_playing = true;
 
+    if(!m_soundFile) {
+        m_waitingFile = true;
+        return;
+    }
+
     if(m_eof) {
         m_soundFile->reset();
         m_eof = false;
     }
 
-    if(!m_soundFile) {
-        g_logger.error("there is not sound file to play the stream");
-        return;
-    }
-
     queueBuffers();
 
     SoundSource::play();
@@ -64,9 +69,13 @@ void StreamSoundSource::play()
 
 void StreamSoundSource::stop()
 {
+    m_playing = false;
+
+    if(m_waitingFile)
+        return;
+
     SoundSource::stop();
     unqueueBuffers();
-    m_playing = false;
 }
 
 void StreamSoundSource::queueBuffers()
@@ -91,6 +100,9 @@ void StreamSoundSource::unqueueBuffers()
 
 void StreamSoundSource::update()
 {
+    if(m_waitingFile)
+        return;
+
     SoundSource::update();
 
     int processed = 0;
@@ -118,6 +130,9 @@ void StreamSoundSource::update()
 
 bool StreamSoundSource::fillBufferAndQueue(uint buffer)
 {
+    if(m_waitingFile)
+        return false;
+
     // fill buffer
     static DataBuffer<char> bufferData(2*STREAM_FRAGMENT_SIZE);
     ALenum format = m_soundFile->getSampleFormat();
@@ -170,15 +185,5 @@ bool StreamSoundSource::fillBufferAndQueue(uint buffer)
 
 void StreamSoundSource::downMix(StreamSoundSource::DownMix downMix)
 {
-    if(!m_soundFile) {
-        g_logger.error("down mix must be set after setting a sound file");
-        return;
-    }
-
-    if(m_soundFile->getSampleFormat() != AL_FORMAT_STEREO16) {
-        g_logger.error("can only downmix 16 bit stereo audio files");
-        return;
-    }
-
     m_downMix = downMix;
 }

src/framework/sound/streamsoundsource.h

@@ -61,6 +61,7 @@ private:
     stdext::boolean<false> m_looping;
     stdext::boolean<false> m_playing;
     stdext::boolean<false> m_eof;
+    stdext::boolean<false> m_waitingFile;
 };
 
 #endif

src/framework/stdext/thread.h

@@ -29,9 +29,13 @@
 #include <boost/thread/mutex.hpp>
 #include <boost/thread/recursive_mutex.hpp>
 #include <boost/thread/locks.hpp>
+#include <boost/thread/future.hpp>
 #include <boost/thread/condition_variable.hpp>
 namespace std {
     using boost::thread;
+    using boost::future;
+    using boost::future_status;
+    using boost::promise;
 
     using boost::mutex;
     using boost::timed_mutex;
@@ -43,11 +47,24 @@ namespace std {
 
     using boost::condition_variable;
     using boost::condition_variable_any;
+
+    template<typename R>
+    bool is_ready(std::future<R>& f)
+    { return f.wait_for(boost::chrono::seconds(0)) == future_status::ready; }
 }
+  
 #else
 #include <thread>
 #include <condition_variable>
 #include <mutex>
+#include <future>
+
+namespace std {
+    template<typename R>
+    bool is_ready(std::future<R>& f)
+    { return f.wait_for(chrono::seconds(0)) == future_status::ready; }
+};
+
 #endif
 
 #endif