ian@0: //  Copyright 2011, Ian Hobson.
ian@0: //
ian@0: //  This file is part of gpsynth.
ian@0: //
ian@0: //  gpsynth is free software: you can redistribute it and/or modify
ian@0: //  it under the terms of the GNU General Public License as published by
ian@0: //  the Free Software Foundation, either version 3 of the License, or
ian@0: //  (at your option) any later version.
ian@0: //
ian@0: //  gpsynth is distributed in the hope that it will be useful,
ian@0: //  but WITHOUT ANY WARRANTY; without even the implied warranty of
ian@0: //  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
ian@0: //  GNU General Public License for more details.
ian@0: //
ian@0: //  You should have received a copy of the GNU General Public License
ian@0: //  along with gpsynth in the file COPYING. 
ian@0: //  If not, see http://www.gnu.org/licenses/.
ian@0: 
ian@0: #include "sc_evaluator.hpp"
ian@0: 
ian@0: #include "boost_ex.hpp"
ian@0: #include "logger.hpp"
ian@0: #include "sc_converter.hpp"
ian@0: #include "statistics.hpp"
ian@0: 
ian@0: #include "boost/filesystem.hpp"
ian@0: #include "boost/format.hpp"
ian@0: #include "boost/lexical_cast.hpp"
ian@0: #include "boost/process.hpp"
ian@0: #include "boost/thread.hpp"
ian@0: #include "boost/math/special_functions/fpclassify.hpp"
ian@0: 
ian@0: #include <algorithm>
ian@0: #include <cmath>
ian@0: #include <functional>
ian@0: #include <numeric>
ian@0: 
ian@0: namespace bp = boost::process;
ian@0: 
ian@0: namespace sc {
ian@0:   
ian@0:   const int g_max_file_wait_time_ms = 10000;
ian@0:   const int g_file_wait_increment_ms = 500;
ian@0:   const int g_sclang_timeout_s = 10;
ian@0:   const double g_fitness_failure_value = 100;
ian@0:   const std::string g_score_variable_name = "x";
ian@0:  
ian@0: // Worker class, runs in own thread
ian@0: class EvaluatorWorker {
ian@0:   sc::Converter converter_;
ian@0:   int thread_id_;
ian@0:   bool quit_threads_;
ian@0:   Evaluator* parent_;
ian@0:   dsp::FileComparer file_comparer_;
ian@0:   // work thread communicates with parent, handles work queue
ian@0:   boost::thread work_thread_;
ian@0:   // render thread launches child processes
ian@0:   boost::thread render_thread_;
ian@0:   boost::mutex render_mutex_;
ian@0:   // render duration in seconds
ian@0:   double render_length_;
ian@0:   std::string render_command_path_;
ian@0:   std::string render_osc_path_;
ian@0:   std::string render_output_path_;
ian@0:   // flag for render thread that a synth is ready for rendering
ian@0:   bool do_render_;
ian@0:   // condition to notify render thread that work is available
ian@0:   boost::condition_variable render_start_;
ian@0:   // condition to notify work thread that work has been done
ian@0:   boost::condition_variable render_finished_;
ian@0:   // stores handle to active child process
ian@0:   boost::shared_ptr<bp::child> render_process_;
ian@0:   
ian@0: public:
ian@0:   EvaluatorWorker(Evaluator* parent,
ian@0:                   int thread_id,
ian@0:                   const dsp::FileComparer& target)
ian@0:   : parent_(parent),
ian@0:     thread_id_(thread_id),
ian@0:     converter_(parent->grammar_),
ian@0:     quit_threads_(false),
ian@0:     file_comparer_(target),
ian@0:     render_length_(file_comparer_.TargetDuration()),
ian@0:     do_render_(false)
ian@0:   {
ian@0:     work_thread_ = boost::thread(&EvaluatorWorker::DoWork, this);
ian@0:     render_thread_ = boost::thread(&EvaluatorWorker::DoSCRender, this);
ian@0:   }
ian@0:   
ian@0:   ~EvaluatorWorker() {
ian@0:     // tell threads to quit
ian@0:     quit_threads_ = true;
ian@0:     // work thread
ian@0:     try {
ian@0:       boost::system_time timeout = boost::get_system_time();
ian@0:       timeout += boost::posix_time::seconds(g_sclang_timeout_s);
ian@0:       parent_->queue_condition_.notify_all();
ian@0:       work_thread_.timed_join(timeout);
ian@0:       // render thread
ian@0:       timeout = boost::get_system_time();
ian@0:       timeout += boost::posix_time::seconds(g_sclang_timeout_s);
ian@0:       render_start_.notify_one();
ian@0:       render_thread_.timed_join(timeout);      
ian@0:     }
ian@0:     catch (const std::exception& e) {
ian@0:       std::stringstream message;
ian@0:       message << "~sc::EvaluatorWorker - Exception: " << e.what() << '\n';
ian@0:       WorkerLog(message.str());
ian@0:     }
ian@0:   }
ian@0:   
ian@0: private:
ian@0:   
ian@0:   void DoWork() {
ian@0:     while (!quit_threads_) {
ian@0:       // wait for some work
ian@0:       {
ian@0:         boost::mutex::scoped_lock lock(parent_->work_mutex_);
ian@0:         while (parent_->work_queue_.empty() && !quit_threads_) {
ian@0:           parent_->queue_condition_.wait(lock);
ian@0:         }
ian@0:       }
ian@0:       // keep going until the work queue is empty 
ian@0:       while (!quit_threads_) {
ian@0:         sg::Graph* graph = NULL;
ian@0:         {
ian@0:           // check there's still work to do
ian@0:           boost::mutex::scoped_lock lock(parent_->work_mutex_);
ian@0:           if (parent_->work_queue_.empty()) {
ian@0:             // wait for all threads to reach this point before continuing
ian@0:             lock.unlock();
ian@0:             parent_->work_done_barrier_->wait();
ian@0:             lock.lock();
ian@0:             // notify main thread that work's finished
ian@0:             parent_->work_done_condition_.notify_one();
ian@0:             break;
ian@0:           }
ian@0:           // get the next graph from the queue
ian@0:           graph = parent_->work_queue_.front();
ian@0:           parent_->work_queue_.pop();
ian@0:         }
ian@0:         // get the graph properties
ian@0:         sg::Graph& graph_ref = *graph;
ian@0:         sg::GraphProperties& graph_properties = graph_ref[boost::graph_bundle];
ian@0:         double fitness;
ian@0:         try {
ian@0:           // rate the graph
ian@0:           fitness = RateGraph(graph_ref);
ian@0:         } catch (const std::exception& e) {
ian@0:           std::stringstream message;
ian@0:           message << "Exception: " << e.what() << '\n';
ian@0:           WorkerLog(message.str());
ian@0:           fitness = g_fitness_failure_value;
ian@0:         }
ian@0:         if (boost::math::isinf(fitness)) {
ian@0:           fitness = g_fitness_failure_value;
ian@0:         } else if (boost::math::isnan(fitness)) {
ian@0:           fitness = g_fitness_failure_value;
ian@0:         }
ian@0:         graph_properties.fitness_ = fitness;
ian@0: //        std::stringstream message;
ian@0: //        message << "Rated " << graph_properties.id_ << ":" << fitness << '\n';
ian@0: //        WorkerLog(message.str());
ian@0:         {
ian@0:           // decrement the graph queue count
ian@0:           boost::mutex::scoped_lock lock(parent_->work_mutex_);
ian@0:           parent_->graphs_to_rate_--;
ian@0:           parent_->work_done_condition_.notify_one();
ian@0:         }
ian@0:       }
ian@0:     }
ian@0:   }
ian@0:   
ian@0:   void WorkerLog(const std::string& message) {
ian@0:     std::stringstream formatter;
ian@0:     formatter << "\n[" << thread_id_ << "] " << message;
ian@0:     logger::Log(formatter.str());
ian@0:   }
ian@0:   
ian@0:   double RateGraph(sg::Graph& graph) {
ian@0:     int id = graph[boost::graph_bundle].id_;
ian@0:     std::stringstream name;
ian@0:     name << "gpsynth_"
ian@0:          << std::setw(parent_->synth_name_zero_pad_length_) << std::setfill('0')
ian@0:          << id;
ian@0:     std::string synth_name = name.str();
ian@0:     // convert the graph to a synthdef and score
ian@0:     std::string synthdef = converter_.ToSynthDef(graph,
ian@0:                                                  synth_name,
ian@0:                                                  false,
ian@0:                                                  parent_->synthdef_folder_);
ian@0:     std::string score = converter_.ToScore(graph,
ian@0:                                            synth_name,
ian@0:                                            g_score_variable_name,
ian@0:                                            false, // prepare for non-realtime
ian@0:                                            parent_->synthdef_folder_,
ian@0:                                            render_length_);
ian@0:     {
ian@0:       // prepare the synthdef and score for the render thread
ian@0:       boost::mutex::scoped_lock lock(render_mutex_);
ian@0:       render_command_path_ = parent_->sc_folder_ + synth_name + ".sc";
ian@0:       render_osc_path_ = parent_->osc_folder_ + synth_name + ".osc";
ian@0:       render_output_path_ = parent_->audio_folder_+ synth_name + ".aiff";
ian@0:       // prepare render script for sclang
ian@0:       std::ofstream command_file(render_command_path_.c_str());
ian@0:       // write the synthdef and the score
ian@0:       command_file << synthdef << '\n' << score;
ian@0:       // tell sclang to write the score as OSC data
ian@0:       command_file << "Score.write(" << g_score_variable_name << ", \"" 
ian@0:                    << render_osc_path_ << "\");\n";
ian@0:       // tell sclang to quit at end of script
ian@0:       command_file << "\n0.exit;";
ian@0:       // finished, close the file
ian@0:       command_file.close();
ian@0:       // tell the render thread that work is available
ian@0:       do_render_ = true;
ian@0:       boost::system_time timeout = boost::get_system_time();
ian@0:       timeout += boost::posix_time::seconds(g_sclang_timeout_s);
ian@0:       render_start_.notify_one();
ian@0:       // wait for the render thread to complete the job
ian@0:       if (!render_finished_.timed_wait(lock, timeout)) {
ian@0:         // sclang or scsynth timed out, kill process and set fitness to failure
ian@0:         // WorkerLog("Render thread timed out.\n");
ian@0:         try {
ian@0:           render_process_->terminate();
ian@0:         }
ian@0:         catch (boost::system::system_error& e) {
ian@0:           // don't need to do anything here, process may already be terminated.
ian@0:         }
ian@0:         do_render_ = false;
ian@0:         // maximum error
ian@0:         return g_fitness_failure_value;
ian@0:       }
ian@0:     }
ian@0:     // check that a rendered file has been produced
ian@0:     if (!boost::filesystem::exists(render_output_path_)) {
ian@0:       // file missing, something's gone wrong..
ian@0:       WorkerLog("Render output missing.\n");
ian@0:       return g_fitness_failure_value;
ian@0:     }
ian@0:     // store render path with graph for future reference
ian@0:     graph[boost::graph_bundle].render_path_ = render_output_path_;
ian@0:     // compare the rendered file to the target
ian@0:     return CompareToTarget(render_output_path_);
ian@0:   }
ian@0:   
ian@0:   void DoSCRender() {
ian@0:     std::vector<std::string> args;
ian@0:     while (!quit_threads_) {
ian@0:       // wait for a command to render
ian@0:       {
ian@0:         boost::mutex::scoped_lock lock(render_mutex_);
ian@0:         do_render_ = false;
ian@0:         while (!do_render_ && !quit_threads_) {
ian@0:           render_start_.wait(lock);
ian@0:         }
ian@0:         if (quit_threads_) {
ian@0:           return;
ian@0:         }
ian@0:       }
ian@0:       // Call sclang to get OSC command for scsynth
ian@0:       args.clear();
ian@0:       args.push_back("-d");
ian@0:       args.push_back(parent_->sc_app_path_);
ian@0:       args.push_back(render_command_path_);
ian@0:       // replace process pipes to suppress output
ian@0:       bp::context closed_streams;
ian@0:       closed_streams.streams[bp::stdout_id] = bp::behavior::close();
ian@0:       closed_streams.streams[bp::stderr_id] = bp::behavior::close();
ian@0:       bp::child process = bp::create_child(parent_->sclang_path_,
ian@0:                                            args,
ian@0:                                            closed_streams);
ian@0:       // store process handle to allow work thread to kill process if it's hung
ian@0:       {
ian@0:         boost::mutex::scoped_lock lock(render_mutex_);
ian@0:         render_process_.reset(new bp::child(process));
ian@0:       }
ian@0:       // wait for sclang to finish
ian@0:       process.wait();
ian@0:       if (!do_render_) {
ian@0:         // if do_render_ is false, process was killed by work thread 
ian@0:         continue;
ian@0:       }
ian@0:       // check that sclang generated an osc command
ian@0:       if (!boost::filesystem::exists(render_osc_path_)) {
ian@0:         // sclang failed, notify work thread and continue
ian@0:         boost::mutex::scoped_lock lock(render_mutex_);
ian@0:         render_finished_.notify_one();
ian@0:         continue;
ian@0:       }
ian@0:       // render osc with scsynth
ian@0:       args.clear();
ian@0:       args.push_back("-N");
ian@0:       args.push_back(render_osc_path_);
ian@0:       args.push_back("_"); // no input file
ian@0:       args.push_back(render_output_path_);
ian@0:       args.push_back("44100");
ian@0:       args.push_back("AIFF");
ian@0:       args.push_back("int16");
ian@0:       // output channels
ian@0:       args.push_back("-o");
ian@0:       args.push_back("1");
ian@0:       // don't publish to rendevous
ian@0:       args.push_back("-R");
ian@0:       args.push_back("0");
ian@0:       // don't load synthdefs
ian@0:       args.push_back("-D");
ian@0:       args.push_back("0");
ian@0:       // plugins path
ian@0:       args.push_back("-U");
ian@0:       args.push_back(parent_->sc_plugins_path_);
ian@0:       // call scsynth
ian@0:       process = bp::create_child(parent_->scsynth_path_, args, closed_streams);
ian@0:       {
ian@0:         boost::mutex::scoped_lock lock(render_mutex_);
ian@0:         render_process_.reset(new bp::child(process));
ian@0:       }
ian@0:       // wait for scsynth to finish rendering
ian@0:       process.wait();
ian@0:       if (!do_render_) {
ian@0:         // process timed out, killed by work thread
ian@0:         continue;
ian@0:       }
ian@0:       // job done, notify work thread
ian@0:       {
ian@0:         boost::mutex::scoped_lock lock(render_mutex_);
ian@0:         render_finished_.notify_one();
ian@0:       }
ian@0:     }
ian@0:   }
ian@0:   
ian@0:   double CompareToTarget(const std::string& file_path) {
ian@0:     return file_comparer_.CompareFile(file_path);
ian@0:   }
ian@0: };
ian@0: 
ian@0: 
ian@0: ////////////////////////////////////////////////////////////////////////////////
ian@0: ////////////////////////////////////////////////////////////////////////////////
ian@0: 
ian@0: Evaluator::Evaluator(const sc::Grammar& grammar,
ian@0:                      const std::string& sc_app_path,
ian@0:                      const dsp::FileComparer& target,
ian@0:                      int core_limit /* = 0 */)
ian@0: : grammar_(grammar),
ian@0:   listener_(NULL),
ian@0:   sc_app_path_(sc_app_path),
ian@0:   sc_plugins_path_(sc_app_path + "/plugins"),
ian@0:   sclang_path_(sc_app_path + "/sclang"),
ian@0:   scsynth_path_(sc_app_path + "/scsynth")
ian@0: {
ian@0:   SetWorkFolder("/tmp/");
ian@0:   int workers = boost::thread::hardware_concurrency();
ian@0:   if (core_limit > 0 && core_limit < workers) {
ian@0:     workers = core_limit;
ian@0:   }
ian@0:   work_done_barrier_.reset(new boost::barrier(workers));
ian@0:   for (int worker_id = 0; worker_id < workers; worker_id++) {
ian@0:     workers_.push_back(EvaluatorWorkerPtr(new EvaluatorWorker(this,
ian@0:                                                               worker_id,
ian@0:                                                               target)));
ian@0:   }
ian@0: }
ian@0: 
ian@0: Evaluator::~Evaluator()
ian@0: {}
ian@0: 
ian@0: 
ian@0: void Evaluator::RateGraphs(std::vector<sg::Graph>& graphs) {
ian@0:   // make sure necessary subfolders exist in work folder
ian@0:   audio_folder_ = work_folder_ + "/audio/";
ian@0:   osc_folder_ = work_folder_ + "/osc/";
ian@0:   sc_folder_ = work_folder_ + "/sc/";
ian@0:   synthdef_folder_ = work_folder_ + "/synthdefs/";
ian@0:   boost::filesystem::create_directories(audio_folder_);
ian@0:   boost::filesystem::create_directories(osc_folder_);
ian@0:   boost::filesystem::create_directories(sc_folder_);
ian@0:   boost::filesystem::create_directories(synthdef_folder_);
ian@0:   // find zero padding length for synth names
ian@0:   synth_name_zero_pad_length_ = std::log10(graphs.size()) + 1;
ian@0:   // store the graphs in the work queue
ian@0:   boost::mutex::scoped_lock lock(work_mutex_);
ian@0:   graphs_to_rate_ = static_cast<int>(graphs.size());
ian@0:   for (int i = 0; i < graphs_to_rate_; i++) {
ian@0:     work_queue_.push(&graphs[i]);
ian@0:   }
ian@0:   // wait for graphs to be rated
ian@0:   std::size_t last_rated_count = 0;
ian@0:   while (graphs_to_rate_ > 0) {
ian@0:     queue_condition_.notify_all();
ian@0:     work_done_condition_.wait(lock);
ian@0:     if (listener_ != NULL) {
ian@0:       std::size_t rated_count = graphs.size() - graphs_to_rate_;
ian@0:       if (rated_count != last_rated_count) {
ian@0:         last_rated_count = rated_count;
ian@0:         listener_->GraphRatedNotification(rated_count);
ian@0:       }
ian@0:     }
ian@0:   }
ian@0:   // remove OSC folder, no longer required
ian@0:   boost::filesystem::remove_all(osc_folder_);
ian@0: }
ian@0:   
ian@0: void Evaluator::SetWorkFolder(const std::string& path) {
ian@0:   work_folder_ = path;
ian@0: }
ian@0:   
ian@0: } // sc namespace