src/main/portfolio_util.h

   1 /*********************                                                        */
   2 /*! \file portfolio_util.h
   3  ** \verbatim
   4  ** Original author: Kshitij Bansal
   5  ** Major contributors: none
   6  ** Minor contributors (to current version): Morgan Deters
   7  ** This file is part of the CVC4 project.
   8  ** Copyright (c) 2009-2014  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 Code relevant only for portfolio builds
  13  **/
  14
  15 #ifndef __CVC4__PORTFOLIO_UTIL_H
  16 #define __CVC4__PORTFOLIO_UTIL_H
  17
  18 #include <queue>
  19
  20 #include "base/output.h"
  21 #include "expr/pickler.h"
  22 #include "options/main_options.h"
  23 #include "smt/smt_engine.h"
  24 #include "smt_util/lemma_input_channel.h"
  25 #include "smt_util/lemma_output_channel.h"
  26 #include "util/channel.h"
  27
  28 namespace CVC4 {
  29
  30 typedef expr::pickle::Pickle ChannelFormat;
  31
  32 class PortfolioLemmaOutputChannel : public LemmaOutputChannel {
  33 private:
  34   std::string d_tag;
  35   SharedChannel<ChannelFormat>* d_sharedChannel;
  36   expr::pickle::MapPickler d_pickler;
  37
  38 public:
  39   int cnt;
  40   PortfolioLemmaOutputChannel(std::string tag,
  41                               SharedChannel<ChannelFormat> *c,
  42                               ExprManager* em,
  43                               VarMap& to,
  44                               VarMap& from) :
  45     d_tag(tag),
  46     d_sharedChannel(c),
  47     d_pickler(em, to, from),
  48     cnt(0)
  49   {}
  50
  51   ~PortfolioLemmaOutputChannel() throw() { }
  52
  53   void notifyNewLemma(Expr lemma);
  54 };/* class PortfolioLemmaOutputChannel */
  55
  56 class PortfolioLemmaInputChannel : public LemmaInputChannel {
  57 private:
  58   std::string d_tag;
  59   SharedChannel<ChannelFormat>* d_sharedChannel;
  60   expr::pickle::MapPickler d_pickler;
  61
  62 public:
  63   PortfolioLemmaInputChannel(std::string tag,
  64                              SharedChannel<ChannelFormat>* c,
  65                              ExprManager* em,
  66                              VarMap& to,
  67                                VarMap& from);
  68
  69   ~PortfolioLemmaInputChannel() throw() { }
  70
  71   bool hasNewLemma();
  72   Expr getNewLemma();
  73
  74 };/* class PortfolioLemmaInputChannel */
  75
  76 std::vector<Options> parseThreadSpecificOptions(Options opts);
  77
  78 template<typename T>
  79 void sharingManager(unsigned numThreads,
  80                     SharedChannel<T> *channelsOut[], // out and in with respect
  81                     SharedChannel<T> *channelsIn[],
  82                     SmtEngine *smts[])  // to smt engines
  83 {
  84   Trace("sharing") << "sharing: thread started " << std::endl;
  85   std::vector <int> cnt(numThreads); // Debug("sharing")
  86
  87   std::vector< std::queue<T> > queues;
  88   for(unsigned i = 0; i < numThreads; ++i){
  89     queues.push_back(std::queue<T>());
  90   }
  91
  92   const unsigned int sharingBroadcastInterval = 1;
  93
  94   boost::mutex mutex_activity;
  95
  96   /* Disable interruption, so that we can check manually */
  97   boost::this_thread::disable_interruption di;
  98
  99   while(not boost::this_thread::interruption_requested()) {
 100
 101     boost::this_thread::sleep
 102       (boost::posix_time::milliseconds(sharingBroadcastInterval));
 103
 104     for(unsigned t = 0; t < numThreads; ++t) {
 105
 106       /* No activity on this channel */
 107       if(channelsOut[t]->empty()) continue;
 108
 109       /* Alert if channel full, so that we increase sharingChannelSize
 110          or decrease sharingBroadcastInterval */
 111       assert(not channelsOut[t]->full());
 112
 113       T data = channelsOut[t]->pop();
 114
 115       if(Trace.isOn("sharing")) {
 116         ++cnt[t];
 117         Trace("sharing") << "sharing: Got data. Thread #" << t
 118                          << ". Chunk " << cnt[t] << std::endl;
 119       }
 120
 121       for(unsigned u = 0; u < numThreads; ++u) {
 122         if(u != t){
 123           Trace("sharing") << "sharing: adding to queue " << u << std::endl;
 124           queues[u].push(data);
 125         }
 126       }/* end of inner for: broadcast activity */
 127
 128     } /* end of outer for: look for activity */
 129
 130     for(unsigned t = 0; t < numThreads; ++t){
 131       /* Alert if channel full, so that we increase sharingChannelSize
 132          or decrease sharingBroadcastInterval */
 133       assert(not channelsIn[t]->full());
 134
 135       while(!queues[t].empty() && !channelsIn[t]->full()){
 136         Trace("sharing") << "sharing: pushing on channel " << t << std::endl;
 137         T data = queues[t].front();
 138         channelsIn[t]->push(data);
 139         queues[t].pop();
 140       }
 141     }
 142   } /* end of infinite while */
 143
 144   Trace("interrupt")
 145     << "sharing thread interrupted, interrupting all smtEngines" << std::endl;
 146
 147   for(unsigned t = 0; t < numThreads; ++t) {
 148     Trace("interrupt") << "Interrupting thread #" << t << std::endl;
 149     try{
 150       smts[t]->interrupt();
 151     }catch(ModalException &e){
 152       // It's fine, the thread is probably not there.
 153       Trace("interrupt") << "Could not interrupt thread #" << t << std::endl;
 154     }
 155   }
 156
 157   Trace("sharing") << "sharing: Interrupted, exiting." << std::endl;
 158 }/* sharingManager() */
 159
 160 }/* CVC4 namespace */
 161
 162 #endif   /* __CVC4__PORTFOLIO_UTIL_H */