hazy_sgd.hxx

Go to the documentation of this file.

template <class T>
Hazy_Sgd<T>::Hazy_Sgd(Storage_Manager<T> *sm, IncrementalSGD<T> *sgd, Skiing &_ski, hazy_model::strategy s) 
  : ski(_ski), hm(sgd->getModel(),s) {
  
  // set _dim
  pthread_rwlock_init(&_rwlock, NULL);
  this->sgd = sgd;
  _high_low_water_valid = true;
  _last_sort_model = sgd->getModel();
  _dim = _last_sort_model.dim;
  st_man = sm;
  if(hazy_model::isHazy(s))
     st_man->resort(hm);
}

template<class T>
void
Hazy_Sgd<T>::updateModel() {
  //pthread_mutex_lock( &_mutex );
  pthread_rwlock_wrlock(&_rwlock);
  
  hm._model = sgd->getModel();
  
  // If we are lazy, there is nothing more to do:
  if( !hm.isEager() ) {
    
    if(hm._strategy == hazy_model::LAZY_HAZY) { _high_low_water_valid = false; }
    //pthread_mutex_unlock( &_mutex );
    pthread_rwlock_unlock(&_rwlock);
    return;
  }
  
  if( hm._strategy == hazy_model::EAGER_NAIVE) {
    assert(hm._strategy == hazy_model::EAGER_NAIVE);
    double _notused = 0.0;
    st_man->incrementalUpdate(hm, _notused);
    //pthread_mutex_unlock( &_mutex );
    pthread_rwlock_unlock(&_rwlock);
    return;
  }

  assert(hm._strategy == hazy_model::EAGER_HAZY);
  // Here, we may assume that it is eager, hazy
  
  update_low_high_water();
  
  if(ski.shouldResort()) {
    Timer t(true);
    st_man->resort(hm);
    _last_sort_model = model(hm._model);
    
    ski.doResort(t.stop());
    hm.low_water = 0.0, hm.high_water = 0.0;
  } else {
    LOGGING_ONLY(std::cout << "storage manager update with eager hazy" << std::endl;);
    double waste_time = 0.0;
    st_man->incrementalUpdate(hm, waste_time);
    ski.updateAccCost(waste_time);
    LOGGING_ONLY(std::cout << "acc cost become: " << ski.getAccCost() << std::endl;);
  }
  
  //pthread_mutex_unlock( &_mutex );
  pthread_rwlock_unlock(&_rwlock);
}

template <class T>
void
Hazy_Sgd<T>::updateModel(T featureArray, int classOfExample) {
  LOGGING_ONLY(std::cout << "update model called" << std::endl;);
  
  LOGGING_ONLY(std::cout << "[hazy_sgd] ski.acc_cost = " << ski.getAccCost() << ", resort cost: " << ski.getResortCost() << ", baseline: " << ski.getSVMBaselineUpdate() << std::endl;);
  
  //pthread_mutex_lock( &_mutex );
  pthread_rwlock_wrlock(&_rwlock);
  // If the model doesn't change, we simply return.
  bool model_changed = sgd->addExample(classOfExample, featureArray); 

  if(model_changed) {    
    // Model changed and so is invalid
    LOGGING_ONLY(std::cout << "model changed" << std::endl;);
    hm.invalidate_db_model(); 
  } else {
    LOGGING_ONLY(std::cout << "model didn't change" << std::endl;);
    //   pthread_mutex_unlock( &_mutex );
    pthread_rwlock_unlock(&_rwlock);
    return; 
  } 
  //pthread_mutex_unlock( &_mutex );
  pthread_rwlock_unlock(&_rwlock);
  
  updateModel();
  LOGGING_ONLY(std::cout << "[hazy_sgd] {after update} ski.acc_cost = " << ski.getAccCost() << ", resort cost: " << ski.getResortCost() << ", baseline: " << ski.getSVMBaselineUpdate() << std::endl;);
}

template<class T>
struct _id_label_vector_entry {
  int id;
  T x;
  int label;
  static int parse_tuple(PGresult *res, int index, _id_label_vector_entry &c) {
    c.id = atoi(PQgetvalue(res, index, 0));
    c.label = atoi(PQgetvalue(res, index, 1));
    std::string f_vec = PQgetvalue(res, index, 2);
    int r = c.x.fromPSQL(f_vec);
    return r;
  }
  static bool _entity_compare(const _id_label_vector_entry &x, const _id_label_vector_entry &y) {
    return x.eps < y.eps;
  }
};

template <class T>
void
Hazy_Sgd<T>::deleteModel(std::string db_name, std::string table_name) {
  _connection_map::iterator i = connections.find(db_name);
  Hazy_Database *db_conn;
  //there is no connection to specified db
  if(i == connections.end()) {
    db_conn = new Hazy_Database(db_name);
    connections[db_name] = db_conn;
  }
  else
    db_conn = i->second;

  std::vector< _id_label_vector_entry<T> > _tuples;
  LOGGING_ONLY(Timer retrieve_parse_timer(true););
  std::string retrieve_query = "SELECT id, label, feature_vector FROM " + table_name + ";";
  db_conn->postgresTupleParser<_id_label_vector_entry <T> >(retrieve_query.c_str(), _id_label_vector_entry<T>::parse_tuple, 
                                                            _tuples);
  LOGGING_ONLY(std::cout << "[deleteModel] time elapsed for retrieve & parse from training table: " << retrieve_parse_timer.stop() << std::endl;);
  LOGGING_ONLY(std::cout << "[deleteModel] entity size: " << _tuples.size() << std::endl;);

  LOGGING_ONLY(std::cout << "delete model called" << std::endl;);
  
  LOGGING_ONLY(std::cout << "[hazy_sgd] ski.acc_cost = " << ski.getAccCost() << ", resort cost: " << ski.getResortCost() << ", baseline: " << ski.getSVMBaselineUpdate() << std::endl;);
  
  //pthread_mutex_lock( &_mutex );
  pthread_rwlock_wrlock(&_rwlock);
  // If the model doesn't change, we simply return.
  bool model_changed = false;

  //reset model
  LOGGING_ONLY(Timer reset_model_timer(true););
  sgd->resetModel();
  LOGGING_ONLY(std::cout << "[deleteModel] reset model timer: " << reset_model_timer.stop() << std::endl;);

  LOGGING_ONLY(Timer learn_timer(true););
  for(unsigned int i = 0; i < _tuples.size(); i ++) {
    int classOfExample = _tuples[i].label;
    T featureArray = _tuples[i].x;
    model_changed = sgd->addExample(classOfExample, featureArray); 
  }
  LOGGING_ONLY(std::cout << "[deleteModel] time for retraining: " << learn_timer.stop() << std::endl;);
  
  if(model_changed) {    
    // Model changed and so is invalid
    LOGGING_ONLY(std::cout << "model changed" << std::endl;);
    hm.invalidate_db_model(); 
  } else {
    LOGGING_ONLY(std::cout << "model didn't change" << std::endl;);
    //   pthread_mutex_unlock( &_mutex );
    pthread_rwlock_unlock(&_rwlock);
    return; 
  } 
  //pthread_mutex_unlock( &_mutex );
  pthread_rwlock_unlock(&_rwlock);
  
  updateModel();
  LOGGING_ONLY(std::cout << "[hazy_sgd] {after update} ski.acc_cost = " << ski.getAccCost() << ", resort cost: " << ski.getResortCost() << ", baseline: " << ski.getSVMBaselineUpdate() << std::endl;);
}

template <class T>
void
Hazy_Sgd<T>::readEntityClass(key entity_id, sClass &c) {
  pthread_rwlock_rdlock(&_rwlock);
  if(hm._strategy == hazy_model::LAZY_HAZY && !_high_low_water_valid) {  update_low_high_water(); _high_low_water_valid = true; }
  st_man->getEntityClass(entity_id, c, hm);
  pthread_rwlock_unlock(&_rwlock);
}

template <class T>
void
Hazy_Sgd<T>::readEntityClass(T vec, sClass &c) {
  // This only called on LAZY approaches.. is this a good assert?
  assert(hm.isLazy()); 
  //pthread_mutex_lock( &_mutex );
  pthread_rwlock_rdlock(&_rwlock);
  c = sgd->classifyExample(vec) ? 1 : 0;
  //pthread_mutex_unlock( &_mutex );
  pthread_rwlock_unlock(&_rwlock);
}

template <class T>
void
Hazy_Sgd<T>::readEntityClass(T vec, double eps, sClass &c) {
  // This only called on LAZY HAZY
  // THIS DOESN'T NEED TO LOCK!
  assert(hm._strategy == hazy_model::LAZY_HAZY);
  if(!_high_low_water_valid) {  update_low_high_water(); _high_low_water_valid = true; }
  if(eps > hm.high_water) { c = 1; return; }
  if(eps < hm.low_water)  { c = 0; return; }
  // AND LEVENT, IF YOU LOCK THIS IT WILL DEADLOCK.
  readEntityClass(vec, c);
}

template <class T>
void
Hazy_Sgd<T>::readNumInClass(sClass c, int &nClass) {
  double waste_time = 0;
  //pthread_mutex_lock( &_mutex );
  pthread_rwlock_rdlock(&_rwlock);
  if(!_high_low_water_valid) {  update_low_high_water(); _high_low_water_valid = true; }
  
  st_man->getNumInClass(c, nClass, hm, waste_time);
  
  if(hm._strategy == hazy_model::LAZY_HAZY) {
    ski.updateAccCost(waste_time);
  }
  //pthread_mutex_unlock( &_mutex );
  pthread_rwlock_unlock(&_rwlock);
}

template <class T>
void
Hazy_Sgd<T>::holder_difference_models(double &delta_w, double &delta_b) {// outputs
  struct model current = sgd->getModel();
  delta_b = current.bias - _last_sort_model.bias;
  delta_w = 0.0;
  for(int d = 1; d < _dim; d ++) {
    delta_w = std::max(delta_w, fabs((current.w.get(d) * current.wscale - _last_sort_model.w.get(d)*_last_sort_model.wscale)));
  }
  // Multiply with the max-1 norm of the dataset (why is this stored in skiing?)
  // Seems like this should be stored here....
  delta_w *= ski.getMFactor();
}

---