src/main/command_executor_portfolio.cpp

   1 /*********************                                                        */
   2 /*! \file command_executor_portfolio.cpp
   3  ** \verbatim
   4  ** Original author: Morgan Deters
   5  ** Major contributors: Kshitij Bansal
   6  ** Minor contributors (to current version): none
   7  ** This file is part of the CVC4 project.
   8  ** Copyright (c) 2009-2013  New York University and The University of Iowa
   9  ** See the file COPYING in the top-level source directory for licensing
  10  ** information.\endverbatim
  11  **
  12  ** \brief An additional layer between commands and invoking them.
  13  **
  14  ** The portfolio executer branches check-sat queries to several
  15  ** threads.
  16  **/
  17
  18 #include <boost/thread.hpp>
  19 #include <boost/thread/condition.hpp>
  20 #include <boost/exception_ptr.hpp>
  21 #include <boost/lexical_cast.hpp>
  22 #include <string>
  23
  24 #include "expr/command.h"
  25 #include "expr/pickler.h"
  26 #include "main/command_executor_portfolio.h"
  27 #include "main/main.h"
  28 #include "main/options.h"
  29 #include "main/portfolio.h"
  30 #include "options/options.h"
  31 #include "smt/options.h"
  32
  33 using namespace std;
  34
  35 namespace CVC4 {
  36 namespace main {
  37
  38 CommandExecutorPortfolio::CommandExecutorPortfolio
  39 (ExprManager &exprMgr, Options &options, vector<Options>& tOpts):
  40   CommandExecutor(exprMgr, options),
  41   d_numThreads(options[options::threads]),
  42   d_smts(),
  43   d_seq(new CommandSequence()),
  44   d_threadOptions(tOpts),
  45   d_vmaps(),
  46   d_lastWinner(0),
  47   d_channelsOut(),
  48   d_channelsIn(),
  49   d_ostringstreams(),
  50   d_statLastWinner("portfolio::lastWinner")
  51 {
  52   assert(d_threadOptions.size() == d_numThreads);
  53
  54   d_statLastWinner.setData(d_lastWinner);
  55   d_stats.registerStat_(&d_statLastWinner);
  56
  57   /* Duplication, Individualisation */
  58   d_exprMgrs.push_back(&d_exprMgr);
  59   for(unsigned i = 1; i < d_numThreads; ++i) {
  60     d_exprMgrs.push_back(new ExprManager(d_threadOptions[i]));
  61   }
  62
  63   // Create the SmtEngine(s)
  64   d_smts.push_back(&d_smtEngine);
  65   for(unsigned i = 1; i < d_numThreads; ++i) {
  66     d_smts.push_back(new SmtEngine(d_exprMgrs[i]));
  67   }
  68
  69   assert(d_vmaps.size() == 0);
  70   for(unsigned i = 0; i < d_numThreads; ++i) {
  71     d_vmaps.push_back(new ExprManagerMapCollection());
  72   }
  73 }
  74
  75 CommandExecutorPortfolio::~CommandExecutorPortfolio()
  76 {
  77   assert(d_seq != NULL);
  78   delete d_seq;
  79
  80   assert(d_smts.size() == d_numThreads);
  81   for(unsigned i = 1; i < d_numThreads; ++i) {
  82     // the 0-th one is responsibility of parent class
  83
  84     delete d_smts[i];
  85     delete d_exprMgrs[i];
  86   }
  87   d_exprMgrs.clear();
  88   d_smts.clear();
  89 }
  90
  91 void CommandExecutorPortfolio::lemmaSharingInit()
  92 {
  93   /* Sharing channels */
  94   assert(d_channelsIn.size() == 0);
  95   assert(d_channelsOut.size() == 0);
  96
  97   if(d_numThreads == 1) {
  98     // Disable sharing
  99     d_threadOptions[0].set(options::sharingFilterByLength, 0);
 100   } else {
 101     // Setup sharing channels
 102     const unsigned int sharingChannelSize = 1000000;
 103
 104     for(unsigned i = 0; i < d_numThreads; ++i){
 105       d_channelsOut.push_back
 106         (new SynchronizedSharedChannel<ChannelFormat>(sharingChannelSize));
 107       d_channelsIn.push_back
 108         (new SynchronizedSharedChannel<ChannelFormat>(sharingChannelSize));
 109     }
 110
 111     /* Lemma I/O channels */
 112     for(unsigned i = 0; i < d_numThreads; ++i) {
 113       string tag = "thread #" +
 114         boost::lexical_cast<string>(d_threadOptions[i][options::thread_id]);
 115       d_threadOptions[i].set
 116         (options::lemmaOutputChannel,
 117          new PortfolioLemmaOutputChannel(tag, d_channelsOut[i], d_exprMgrs[i],
 118                                          d_vmaps[i]->d_from, d_vmaps[i]->d_to));
 119       d_threadOptions[i].set
 120         (options::lemmaInputChannel,
 121          new PortfolioLemmaInputChannel(tag, d_channelsIn[i], d_exprMgrs[i],
 122                                         d_vmaps[i]->d_from, d_vmaps[i]->d_to));
 123     }
 124
 125     /* Output to string stream  */
 126     assert(d_ostringstreams.size() == 0);
 127     for(unsigned i = 0; i < d_numThreads; ++i) {
 128       d_ostringstreams.push_back(new ostringstream);
 129       d_threadOptions[i].set(options::out, d_ostringstreams[i]);
 130
 131       // important even for muzzled builds (to get result output right)
 132       *d_threadOptions[i][options::out]
 133         << Expr::setlanguage(d_threadOptions[i][options::outputLanguage]);
 134     }
 135   }
 136 }/* CommandExecutorPortfolio::lemmaSharingInit() */
 137
 138 void CommandExecutorPortfolio::lemmaSharingCleanup()
 139 {
 140   assert(d_numThreads == d_options[options::threads]);
 141
 142   if(d_numThreads == 1)
 143     return;
 144
 145   // Channel cleanup
 146   assert(d_channelsIn.size() == d_numThreads);
 147   assert(d_channelsOut.size() == d_numThreads);
 148   for(unsigned i = 0; i < d_numThreads; ++i) {
 149     delete d_channelsIn[i];
 150     delete d_channelsOut[i];
 151     d_threadOptions[i].set(options::lemmaInputChannel, NULL);
 152     d_threadOptions[i].set(options::lemmaOutputChannel, NULL);
 153   }
 154   d_channelsIn.clear();
 155   d_channelsOut.clear();
 156
 157   // sstreams cleanup (if used)
 158   if(d_ostringstreams.size() != 0) {
 159     assert(d_ostringstreams.size() == d_numThreads);
 160     for(unsigned i = 0; i < d_numThreads; ++i) {
 161       d_threadOptions[i].set(options::out, d_options[options::out]);
 162       delete d_ostringstreams[i];
 163     }
 164     d_ostringstreams.clear();
 165   }
 166
 167 }/* CommandExecutorPortfolio::lemmaSharingCleanup() */
 168
 169
 170 bool CommandExecutorPortfolio::doCommandSingleton(Command* cmd)
 171 {
 172   /**
 173    * save the command and if check sat or query command, run a
 174    * porfolio of SMT solvers.
 175    */
 176
 177   int mode = 0;
 178   // mode = 0 : run command on lastWinner, saving the command
 179   // to be run on all others
 180   //
 181   // mode = 1 : run a race of the command, update lastWinner
 182   //
 183   // mode = 2 : run _only_ the lastWinner thread, not saving the
 184   // command
 185
 186   if(dynamic_cast<CheckSatCommand*>(cmd) != NULL ||
 187      dynamic_cast<QueryCommand*>(cmd) != NULL) {
 188     mode = 1;
 189   } else if(dynamic_cast<GetValueCommand*>(cmd) != NULL ||
 190             dynamic_cast<GetAssignmentCommand*>(cmd) != NULL ||
 191             dynamic_cast<GetModelCommand*>(cmd) != NULL ||
 192             dynamic_cast<GetProofCommand*>(cmd) != NULL ||
 193             dynamic_cast<GetUnsatCoreCommand*>(cmd) != NULL ||
 194             dynamic_cast<GetAssertionsCommand*>(cmd) != NULL ||
 195             dynamic_cast<GetInfoCommand*>(cmd) != NULL ||
 196             dynamic_cast<GetOptionCommand*>(cmd) != NULL ||
 197             false) {
 198     mode = 2;
 199   }
 200
 201   Debug("portfolio::outputmode") << "Mode is " << mode
 202                                  << "lastWinner is " << d_lastWinner
 203                                  << "d_seq is " << d_seq << std::endl;
 204
 205   if(mode == 0) {
 206     d_seq->addCommand(cmd->clone());
 207     Command* cmdExported =
 208       d_lastWinner == 0 ?
 209       cmd : cmd->exportTo(d_exprMgrs[d_lastWinner], *(d_vmaps[d_lastWinner]) );
 210     bool ret = smtEngineInvoke(d_smts[d_lastWinner],
 211                                cmdExported,
 212                                d_options[options::verbosity] >= -1 ? d_threadOptions[d_lastWinner][options::out] : NULL);
 213     if(d_lastWinner != 0) delete cmdExported;
 214     return ret;
 215   } else if(mode == 1) {               // portfolio
 216
 217     d_seq->addCommand(cmd->clone());
 218
 219     // We currently don't support changing number of threads for each
 220     // command, but things have been architected in a way so that this
 221     // can be achieved without a lot of work.
 222     Command *seqs[d_numThreads];
 223
 224     if(d_lastWinner == 0)
 225       seqs[0] = cmd;
 226     else
 227       seqs[0] = d_seq;
 228
 229     /* variable maps and exporting */
 230     for(unsigned i = 1; i < d_numThreads; ++i) {
 231       /**
 232        * vmaps[i].d_from [x] = y means
 233        *    that in thread #0's expr manager id is y
 234        *    and  in thread #i's expr manager id is x
 235        * opposite for d_to
 236        *
 237        * d_from[x] : in a sense gives the id if converting *from* it to
 238        *             first thread
 239        */
 240       try {
 241         seqs[i] =
 242           int(i) == d_lastWinner ?
 243           cmd->exportTo(d_exprMgrs[i], *(d_vmaps[i])) :
 244           d_seq->exportTo(d_exprMgrs[i], *(d_vmaps[i]) );
 245       } catch(ExportUnsupportedException& e) {
 246         if(d_options[options::fallbackSequential]) {
 247           Notice() << "Unsupported theory encountered, switching to sequential mode.";
 248           return CommandExecutor::doCommandSingleton(cmd);
 249         }
 250         else
 251           throw Exception("Certain theories (e.g., datatypes) are (currently) unsupported in portfolio\n"
 252                           "mode. Please see option --fallback-sequential to make this a soft error.");
 253       }
 254     }
 255
 256     /**
 257      * Create identity variable map for the first thread, with only
 258      * those variables which have a corresponding variable in
 259      * another thread. (TODO: Also assert, all threads have the same
 260      * set of variables mapped.)
 261      */
 262     if(d_numThreads >= 2) {
 263       for(typeof(d_vmaps[1]->d_to.begin()) i=d_vmaps[1]->d_to.begin();
 264           i!=d_vmaps[1]->d_to.end(); ++i) {
 265         (d_vmaps[0]->d_from)[i->first] = i->first;
 266       }
 267       d_vmaps[0]->d_to = d_vmaps[0]->d_from;
 268     }
 269
 270     lemmaSharingInit();
 271
 272     /* Portfolio */
 273     boost::function<bool()>* fns = new boost::function<bool()>[d_numThreads];
 274     for(unsigned i = 0; i < d_numThreads; ++i) {
 275       fns[i] = boost::bind(smtEngineInvoke,
 276                            d_smts[i],
 277                            seqs[i],
 278                            d_options[options::verbosity] >= -1 ? d_threadOptions[i][options::out] : NULL
 279                            );
 280     }
 281
 282     assert(d_channelsIn.size() == d_numThreads
 283            || d_numThreads == 1);
 284     assert(d_channelsOut.size() == d_numThreads
 285            || d_numThreads == 1);
 286     assert(d_smts.size() == d_numThreads);
 287     boost::function<void()>
 288       smFn = d_numThreads <= 1 ? boost::function<void()>() :
 289              boost::bind(sharingManager<ChannelFormat>,
 290                          d_numThreads,
 291                          &d_channelsOut[0],
 292                          &d_channelsIn[0],
 293                          &d_smts[0]);
 294
 295     pair<int, bool> portfolioReturn =
 296         runPortfolio(d_numThreads, smFn, fns,
 297                      d_options[options::waitToJoin]);
 298
 299     delete d_seq;
 300     d_seq = new CommandSequence();
 301
 302     d_lastWinner = portfolioReturn.first;
 303     d_result = d_smts[d_lastWinner]->getStatusOfLastCommand();
 304
 305     if(d_ostringstreams.size() != 0) {
 306       assert(d_numThreads == d_options[options::threads]);
 307       assert(portfolioReturn.first >= 0);
 308       assert(unsigned(portfolioReturn.first) < d_numThreads);
 309
 310       if(Debug.isOn("treat-unknown-error")) {
 311         if(d_ostringstreams[portfolioReturn.first]->str() == "unknown\n") {
 312           *d_options[options::out]
 313             << "portfolioReturn = (" << portfolioReturn.first << ", " << portfolioReturn.second
 314             << ")\n";
 315           for(unsigned i = 0; i < d_numThreads; ++i)
 316             *d_options[options::out]
 317               << "thread " << i << ": " << d_ostringstreams[i]->str() << std::endl;
 318           throw Exception("unknown encountered");
 319         }
 320       }
 321
 322       *d_options[options::out]
 323         << d_ostringstreams[portfolioReturn.first]->str();
 324     }
 325
 326     /* cleanup this check sat specific stuff */
 327     lemmaSharingCleanup();
 328
 329     delete[] fns;
 330     return portfolioReturn.second;
 331   } else if(mode == 2) {
 332     Command* cmdExported =
 333       d_lastWinner == 0 ?
 334       cmd : cmd->exportTo(d_exprMgrs[d_lastWinner], *(d_vmaps[d_lastWinner]) );
 335     bool ret = smtEngineInvoke(d_smts[d_lastWinner],
 336                                cmdExported,
 337                                d_options[options::verbosity] >= -1 ? d_threadOptions[d_lastWinner][options::out] : NULL);
 338     if(d_lastWinner != 0) delete cmdExported;
 339     return ret;
 340   } else {
 341     // Unreachable();
 342     assert(false);
 343     return false;
 344   }
 345
 346 }/* CommandExecutorPortfolio::doCommandSingleton() */
 347
 348 void CommandExecutorPortfolio::flushStatistics(std::ostream& out) const {
 349   assert(d_numThreads == d_exprMgrs.size() && d_exprMgrs.size() == d_smts.size());
 350   for(size_t i = 0; i < d_numThreads; ++i) {
 351     string emTag = "thread#"
 352       + boost::lexical_cast<string>(d_threadOptions[i][options::thread_id]);
 353     Statistics stats = d_exprMgrs[i]->getStatistics();
 354     stats.setPrefix(emTag);
 355     stats.flushInformation(out);
 356
 357     string smtTag = "thread#"
 358       + boost::lexical_cast<string>(d_threadOptions[i][options::thread_id]);
 359     stats = d_smts[i]->getStatistics();
 360     stats.setPrefix(smtTag);
 361     stats.flushInformation(out);
 362   }
 363   d_stats.flushInformation(out);
 364 }
 365
 366 }/* CVC4::main namespace */
 367 }/* CVC4 namespace */