| 71 | } |
| 72 | |
| 73 | void DataReader::Body::InternalThreadEntry() { |
| 74 | shared_ptr<db::DB> db(db::GetDB(param_.data_param().backend())); // note data_param and cpmdata_param is conflicting.. |
| 75 | db->Open(param_.data_param().source(), db::READ); |
| 76 | shared_ptr<db::Cursor> cursor(db->NewCursor()); |
| 77 | vector<shared_ptr<QueuePair> > qps; |
| 78 | try { |
| 79 | int solver_count = param_.phase() == TRAIN ? Caffe::solver_count() : 1; |
| 80 | |
| 81 | // To ensure deterministic runs, only start running once all solvers |
| 82 | // are ready. But solvers need to peek on one item during initialization, |
| 83 | // so read one item, then wait for the next solver. |
| 84 | for (int i = 0; i < solver_count; ++i) { |
| 85 | shared_ptr<QueuePair> qp(new_queue_pairs_.pop()); |
| 86 | read_one(cursor.get(), qp.get()); |
| 87 | qps.push_back(qp); |
| 88 | } |
| 89 | // Main loop |
| 90 | while (!must_stop()) { |
| 91 | for (int i = 0; i < solver_count; ++i) { |
| 92 | read_one(cursor.get(), qps[i].get()); |
| 93 | } |
| 94 | // Check no additional readers have been created. This can happen if |
| 95 | // more than one net is trained at a time per process, whether single |
| 96 | // or multi solver. It might also happen if two data layers have same |
| 97 | // name and same source. |
| 98 | CHECK_EQ(new_queue_pairs_.size(), 0); |
| 99 | } |
| 100 | } catch (boost::thread_interrupted&) { |
| 101 | // Interrupted exception is expected on shutdown |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | void DataReader::Body::read_one(db::Cursor* cursor, QueuePair* qp) { |
| 106 | Datum* datum = qp->free_.pop(); |