src/threadgenesorter.cpp

   1 /***************************************************************************
   2  *   Copyright (C) 2008 by Michael Andreen                                 *
   3  *   andreen@student.chalmers.se                                           *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) any later version.                                   *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15  *   You should have received a copy of the GNU General Public License     *
  16  *   along with this program; if not, write to the                         *
  17  *   Free Software Foundation, Inc.,                                       *
  18  *   51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA          *
  19  ***************************************************************************/
  20
  21 #include "threadgenesorter.h"
  22 #include "sortaction.h"
  23
  24 #include "genealgorithms.h"
  25
  26 #include <set>
  27 using namespace std;
  28
  29 /**
  30  * Handles locking and unlocking of mutexes.
  31  *
  32  * RAII style class for mutex acquisition.
  33  */
  34 class Mutex {
  35         pthread_mutex_t* _m;
  36         bool _locked;
  37         public:
  38                 Mutex(pthread_mutex_t* m){
  39                         _m = m;
  40                         pthread_mutex_lock(_m);
  41                         _locked = true;
  42                 }
  43                 void unlock(){
  44                         pthread_mutex_unlock(_m);
  45                         _locked = false;
  46                 }
  47                 ~Mutex(){
  48                         if (_locked)
  49                                 unlock();
  50                 }
  51 };
  52
  53 struct ExecutionStuff {
  54         ExecutionStuff(ThreadGeneSorter* so, const GeneOrder& go): _sorter(so)
  55                         ,_go(go){};
  56         ThreadGeneSorter* _sorter;
  57         GeneOrder _go;
  58 };
  59
  60
  61 ThreadGeneSorter::ThreadGeneSorter(const Model& m, int threads) : _model(m){
  62         _workers = threads;
  63         pthread_mutex_init(&_queuelock, NULL);
  64         pthread_mutex_init(&_solutionslock, NULL);
  65         pthread_cond_init(&_addedSolution,NULL);
  66         pthread_cond_init(&_addedTask,NULL);
  67         pthread_cond_init(&_waiting,NULL);
  68 }
  69
  70 void ThreadGeneSorter::join(){
  71         pthread_join(_tid,NULL);
  72 }
  73
  74 void ThreadGeneSorter::stop(){
  75         _done = true;
  76 }
  77
  78 size_t ThreadGeneSorter::size(){
  79         //TODO: thread safety..
  80         return _solutions.size();
  81 }
  82
  83 double progress(int time, int solutions = 1){
  84 }
  85
  86 ThreadGeneSorter::SolutionsQueue ThreadGeneSorter::solutions(){
  87         //TODO: thread safety..
  88         return _solutions;
  89 }
  90
  91 void ThreadGeneSorter::start(const GeneOrder& go){
  92         pthread_attr_t tattr;
  93         pthread_attr_init(&tattr);
  94         pthread_attr_setscope(&tattr, PTHREAD_SCOPE_SYSTEM);
  95         ExecutionStuff* es = new ExecutionStuff(this,go);
  96         pthread_create(&_tid, &tattr, t_sorter, es);
  97 }
  98
  99 void ThreadGeneSorter::sorter(const GeneOrder& go){
 100         _queue.push(SortUnit(0.0,go,ActionList()));
 101         pthread_attr_t tattr;
 102         pthread_attr_init(&tattr);
 103         pthread_attr_setscope(&tattr, PTHREAD_SCOPE_SYSTEM);
 104         _done = false;
 105         _queue = SortQueue();
 106         _solutions = SolutionsQueue();
 107
 108         queue<pthread_t> threads;
 109
 110         _queue.push(SortUnit(0.0,go,ActionList()));
 111
 112         int workers = _workers;
 113         while (!_done){
 114                 if (workers){
 115                         --workers;
 116                         pthread_t tid;
 117                         pthread_create(&tid, &tattr, t_worker, this);
 118                         threads.push(tid);
 119                 }else{
 120                         Mutex m(&_queuelock);
 121                         while(_workers != 0 && !_done){
 122                                 pthread_cond_wait(&_waiting,&_queuelock);
 123                         }
 124                         if (_workers == 0)
 125                                 _done = true;
 126                 }
 127         }
 128         pthread_cond_broadcast(&_addedTask);
 129         while(threads.size() > 0){
 130                 pthread_join(threads.front(),NULL);
 131                 threads.pop();
 132         }
 133         pthread_exit((void*)0);
 134 }
 135
 136 void ThreadGeneSorter::worker(){
 137         try{
 138         while(!_done){
 139                 Mutex m(&_queuelock);
 140
 141                 while (_queue.size() == 0 && !_done){
 142                         --_workers;
 143                         pthread_cond_signal(&_waiting);
 144                         pthread_cond_wait(&_addedTask,&_queuelock);
 145                         ++_workers;
 146                 }
 147                 if (_done)
 148                         break;
 149
 150                 SortUnit su = _queue.top();
 151                 _queue.pop();
 152                 m.unlock();
 153
 154                 size_t dist = inversionDistance(su._go);
 155                 if (dist == 0){
 156                         Mutex m(&_solutionslock);
 157                         _solutions.push(pair<double,ActionList>(su._score,su._al));
 158                         pthread_cond_broadcast(&_addedSolution);
 159                         continue;;
 160                 }
 161
 162                 ActionList act = safeActions(su._go);
 163                 if (act.size() > 0){
 164                         set<SortAction> safe(act.begin(), act.end());
 165                         for (set<SortAction>::iterator sa = safe.begin(); sa != safe.end(); ++sa){
 166                                 GeneOrder go(su._go);
 167                                 ActionList al(su._al);
 168                                 al.push_back(*sa);
 169                                 (*sa)(go);
 170                                 double score = su._score + _model.score(*sa,su._go);
 171                                 Mutex m(&_queuelock);
 172                                 _queue.push(SortUnit(score,go,al));
 173                         }
 174                         pthread_cond_broadcast(&_addedTask);
 175                 } //TODO: Hurdles..
 176
 177         }
 178         }catch(const bad_alloc& e){
 179                 _done = true;
 180         }
 181 }
 182
 183 void* t_sorter(void* arg){
 184         ExecutionStuff* es = static_cast<ExecutionStuff*>(arg);
 185         es->_sorter->sorter(es->_go);
 186         delete es;
 187         pthread_exit((void*)0);
 188 }
 189
 190 void* t_worker(void* arg){
 191         ThreadGeneSorter* sorter = static_cast<ThreadGeneSorter*>(arg);
 192         sorter->worker();
 193         pthread_exit((void*)0);
 194 }