controller.cc 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800
  1. extern "C" {
  2. #include <event2/listener.h>
  3. #include <event2/bufferevent.h>
  4. #include <event2/buffer.h>
  5. }
  6. #include <NTL/vec_ZZ.h>
  7. #include <NTL/ZZ.h>
  8. #include <NTL/ZZ_p.h>
  9. #include <sys/socket.h>
  10. #include <netinet/in.h>
  11. #include <arpa/inet.h>
  12. #include <sys/time.h>
  13. #include <fstream>
  14. #include <vector>
  15. #include <set>
  16. #include <map>
  17. #include <string.h>
  18. #include "evutils.h"
  19. #include "subproblem.h"
  20. #include "controller.h"
  21. NTL_CLIENT
  22. #define DEBUG
  23. struct SubproblemProgress;
  24. typedef std::set<struct bufferevent *> BESet;
  25. typedef std::map<struct bufferevent *, SubproblemProgress *> BEMap;
  26. static void besetdump(const BESet &bes, ostream &os)
  27. {
  28. BESet::const_iterator besit;
  29. os << hex << " ";
  30. for (besit = bes.begin(); besit != bes.end(); ++besit) {
  31. os << *besit << " ";
  32. }
  33. os << dec << "\n";
  34. }
  35. static void besetfree(BESet &bes)
  36. {
  37. BESet::iterator besit;
  38. for (besit = bes.begin(); besit != bes.end(); ++besit) {
  39. bufferevent_free(*besit);
  40. }
  41. bes.clear();
  42. }
  43. static void bemapdump(const BEMap &bem, ostream &os)
  44. {
  45. BEMap::const_iterator bemit;
  46. os << hex << " ";
  47. for (bemit = bem.begin(); bemit != bem.end(); ++bemit) {
  48. os << bemit->first << "->" << bemit->second << " ";
  49. }
  50. os << dec << "\n";
  51. }
  52. static void bemapfree(BEMap &bem)
  53. {
  54. BEMap::iterator bemit;
  55. for (bemit = bem.begin(); bemit != bem.end(); ++bemit) {
  56. bufferevent_free(bemit->first);
  57. }
  58. bem.clear();
  59. }
  60. struct Statuses {
  61. BESet idle;
  62. BEMap working;
  63. // Dump the state for debug purposes
  64. void dump(ostream &os) const {
  65. os << " idle (" << idle.size() << "):\n";
  66. besetdump(idle, os);
  67. os << " working (" << working.size() << "):\n";
  68. bemapdump(working, os);
  69. }
  70. void free(void) {
  71. besetfree(idle);
  72. bemapfree(working);
  73. }
  74. };
  75. struct FactorDecomp {
  76. ZZ factor;
  77. vec_ZZ fvec;
  78. };
  79. static void vsppdump(const vector<SubproblemProgress> &spv, ostream &os);
  80. static struct ControllerState {
  81. ZZ rho;
  82. FactorDecomp p, q;
  83. ZZ base, target;
  84. bool working;
  85. struct timeval started_working;
  86. vector<SubproblemProgress> subproblems_p, subproblems_q;
  87. Statuses dpnodes, workers;
  88. unsigned int num_unsolved_subproblems;
  89. unsigned int problems_remaining;
  90. struct evconnlistener *listener;
  91. ControllerState() : working(false), num_unsolved_subproblems(0),
  92. problems_remaining(0), listener(NULL) {}
  93. // Reset the state for a new problem with the same modulus
  94. void reset(void) {
  95. base = 0;
  96. target = 0;
  97. working = false;
  98. subproblems_p.clear();
  99. subproblems_q.clear();
  100. num_unsolved_subproblems = 0;
  101. started_working.tv_sec = 0;
  102. started_working.tv_usec = 0;
  103. }
  104. // Dump the state for debug purposes
  105. void dump(ostream &os) const {
  106. if (!working) {
  107. os << "Not working\n";
  108. return;
  109. }
  110. os << "P:\n";
  111. vsppdump(subproblems_p, os);
  112. os << "Q:\n";
  113. vsppdump(subproblems_q, os);
  114. os << "dpnodes:\n";
  115. dpnodes.dump(os);
  116. os << "workers:\n";
  117. workers.dump(os);
  118. }
  119. } ctrlstate;
  120. struct IPPort {
  121. unsigned char ipport[6];
  122. IPPort(unsigned char *ipp) {
  123. memmove(ipport, ipp, 6);
  124. }
  125. void dump(ostream &os) const {
  126. os << int(ipport[0]) << "." << int(ipport[1]) << "." <<
  127. int(ipport[2]) << "." << int(ipport[3]) << ":" <<
  128. ((ipport[4] << 8) + ipport[5]) << " ";
  129. }
  130. };
  131. typedef vector<IPPort> IPPortSet;
  132. static void ipportsetdump(const IPPortSet &ipps, ostream &os)
  133. {
  134. IPPortSet::const_iterator ippsit;
  135. os << " ";
  136. for (ippsit = ipps.begin(); ippsit != ipps.end(); ++ippsit) {
  137. ippsit->dump(os);
  138. }
  139. os << "\n";
  140. }
  141. struct SubproblemProgress : Subproblem {
  142. // The sets of dpnodes and workers currently working on this subproblem
  143. BESet dpnodes, workers;
  144. // The dpnode IPPorts registered for this subproblem
  145. IPPortSet ipports;
  146. // The desired number of DPnodes for this subproblem
  147. unsigned short desired_dpnodes;
  148. // The maximum number of workers useful for this subproblem
  149. unsigned int max_workers;
  150. // Have we found a solution?
  151. bool solved;
  152. // The solution, if found.
  153. ZZ solution;
  154. SubproblemProgress(unsigned short id, const ZZ &b, const ZZ &t,
  155. const ZZ &m, const ZZ &o) :
  156. Subproblem(id, b, t, m, o, 4294967), solved(false) {
  157. // How many DPnodes should we use for a problem of this size?
  158. desired_dpnodes = 2;
  159. // How many workers would we like to use?
  160. ZZ sorder = SqrRoot(order >> 46);
  161. if (NumBits(sorder) > 30) {
  162. // Just use all the workers we can find
  163. max_workers = 4294967295U; // 2^32 - 1
  164. } else {
  165. max_workers = trunc_long(sorder,31) + 1;
  166. }
  167. // By default, 1 in 1000 points are distinguihed points. The
  168. // number in the next line is 2^32/1000
  169. dpfreq = 4294967;
  170. if (order < 1000) {
  171. // Just make every point a DP
  172. dpfreq = 4294967295U;
  173. } else if (NumBits(order) < 27) {
  174. // The frequency of DPs should be 10/sqrt(order) to avoid
  175. // a DP-free cycle, so dpfreq = (10*2^32)/sqrt(order)
  176. ZZ f = (to_ZZ(10) << 32) / SqrRoot(order);
  177. dpfreq = trunc_long(f, 31);
  178. }
  179. }
  180. // Stop all dpnodes and workers
  181. void stop(void) {
  182. BESet::iterator iter;
  183. unsigned char stopcmd[1] = { 'S' };
  184. for (BESet::iterator iter = dpnodes.begin(); iter != dpnodes.end();
  185. ++iter) {
  186. bufferevent_write(*iter, stopcmd, 1);
  187. ctrlstate.dpnodes.working.erase(*iter);
  188. ctrlstate.dpnodes.idle.insert(*iter);
  189. }
  190. for (BESet::iterator iter = workers.begin(); iter != workers.end();
  191. ++iter) {
  192. bufferevent_write(*iter, stopcmd, 1);
  193. ctrlstate.workers.working.erase(*iter);
  194. ctrlstate.workers.idle.insert(*iter);
  195. }
  196. dpnodes.clear();
  197. workers.clear();
  198. ipports.clear();
  199. }
  200. // Dump for debugging purposes
  201. void dump(ostream &os) const {
  202. os << " Subproblem " << problemid << "\n";
  203. os << " dpnodes (" << dpnodes.size() << "):\n";
  204. besetdump(dpnodes, os);
  205. os << " workers (" << workers.size() << "):\n";
  206. besetdump(workers, os);
  207. os << " ipports (" << ipports.size() << "):\n";
  208. ipportsetdump(ipports, os);
  209. if (solved) {
  210. os << " solution: " << solution << "\n\n";
  211. }
  212. }
  213. void worker_write(struct bufferevent *bev) {
  214. bev_write(bev);
  215. unsigned short num_ipports = ipports.size();
  216. bufferevent_write(bev, &num_ipports, 2);
  217. for (unsigned short i = 0; i < num_ipports; ++i) {
  218. bufferevent_write(bev, ipports[i].ipport, 6);
  219. }
  220. #ifdef DEBUG
  221. cerr << "Added worker " << bev << " to subproblem "
  222. << problemid << "\n";
  223. #endif
  224. }
  225. };
  226. // Dump the state for debug purposes
  227. static void vsppdump(const vector<SubproblemProgress> &spv, ostream &os)
  228. {
  229. vector<SubproblemProgress>::const_iterator spiter;
  230. int count = 0;
  231. for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
  232. ++count;
  233. os << " " << count << ":\n";
  234. spiter->dump(os);
  235. }
  236. os << "\n";
  237. }
  238. // Find a subproblem in the given vector that could use another DPnode,
  239. // and give it one of the idle ones. Only allocate it to a subproblem
  240. // with no current DPnodes if consider_empty is true.
  241. static void find_subproblem_for_dpnode(vector<SubproblemProgress> &spv,
  242. bool consider_empty)
  243. {
  244. vector<SubproblemProgress>::iterator spiter;
  245. for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
  246. if (spiter->solved) continue;
  247. if (spiter->dpnodes.size() == 0 && consider_empty == false) continue;
  248. // How many DPnodes would we like to have for this subproblem?
  249. while (spiter->dpnodes.size() < spiter->desired_dpnodes &&
  250. ctrlstate.dpnodes.idle.size() > 0) {
  251. // Get the first idle DPnode
  252. BESet::iterator beviter = ctrlstate.dpnodes.idle.begin();
  253. struct bufferevent *firstbev = *beviter;
  254. // Allocate it to the subproblem
  255. spiter->dpnodes.insert(firstbev);
  256. ctrlstate.dpnodes.working[firstbev] = &(*spiter);
  257. ctrlstate.dpnodes.idle.erase(firstbev);
  258. // Tell it to start listening for DPs
  259. spiter->bev_write(firstbev);
  260. }
  261. }
  262. }
  263. // Find a subproblem in the given vector that has all of its DPnodes and
  264. // could use another worker, and give it one of the idle ones.
  265. static void find_subproblem_for_worker(vector<SubproblemProgress> &spv)
  266. {
  267. vector<SubproblemProgress>::iterator spiter;
  268. for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
  269. if (spiter->solved) continue;
  270. while (spiter->ipports.size() == spiter->desired_dpnodes &&
  271. spiter->workers.size() < spiter->max_workers &&
  272. ctrlstate.workers.idle.size() > 0) {
  273. // Get the first idle worker
  274. BESet::iterator beviter = ctrlstate.workers.idle.begin();
  275. struct bufferevent *firstbev = *beviter;
  276. // Allocate it to the subproblem
  277. spiter->workers.insert(firstbev);
  278. ctrlstate.workers.working[firstbev] = &(*spiter);
  279. ctrlstate.workers.idle.erase(firstbev);
  280. // Tell it to start working on the subproblem
  281. spiter->worker_write(firstbev);
  282. }
  283. }
  284. }
  285. // See if there are any idle DPnodes or workers we can put to use
  286. static void schedule(void)
  287. {
  288. // Check the DPnodes
  289. // Iterate through the subproblems, looking for one that can use
  290. // another DPnode. First look for subproblems that already have
  291. // some, but not all, of their DPnodes
  292. if (ctrlstate.dpnodes.idle.size() > 0) {
  293. find_subproblem_for_dpnode(ctrlstate.subproblems_p, false);
  294. }
  295. if (ctrlstate.dpnodes.idle.size() > 0) {
  296. find_subproblem_for_dpnode(ctrlstate.subproblems_q, false);
  297. }
  298. // If there are still more dpnodes to place, start assigning them to
  299. // subproblems with no current dpnodes
  300. if (ctrlstate.dpnodes.idle.size() > 0) {
  301. find_subproblem_for_dpnode(ctrlstate.subproblems_p, true);
  302. }
  303. if (ctrlstate.dpnodes.idle.size() > 0) {
  304. find_subproblem_for_dpnode(ctrlstate.subproblems_q, true);
  305. }
  306. // Check the workers
  307. // Iterate through the subproblems, looking for one that can use
  308. // another worker.
  309. if (ctrlstate.workers.idle.size() > 0) {
  310. find_subproblem_for_worker(ctrlstate.subproblems_p);
  311. }
  312. if (ctrlstate.workers.idle.size() > 0) {
  313. find_subproblem_for_worker(ctrlstate.subproblems_q);
  314. }
  315. // cerr << "After schedule:\n"; ctrlstate.dump(cerr);
  316. }
  317. static ZZ computation_complete_p(const vector<SubproblemProgress> &v)
  318. {
  319. ZZ curmodulus, curexp;
  320. curmodulus = 2;
  321. curexp = 0;
  322. vector<SubproblemProgress>::const_iterator vit;
  323. for(vit = v.begin(); vit != v.end(); ++vit) {
  324. CRT(curexp, curmodulus, vit->solution, vit->order);
  325. }
  326. if (curexp < 0) {
  327. curexp += curmodulus;
  328. }
  329. return curexp;
  330. }
  331. // All subproblems are solved. Combine the results.
  332. static void computation_complete(void)
  333. {
  334. ZZ exp_p = computation_complete_p(ctrlstate.subproblems_p);
  335. ZZ exp_q = computation_complete_p(ctrlstate.subproblems_q);
  336. ZZ pm1 = (ctrlstate.p.factor - 1)/2;
  337. ZZ qm1 = (ctrlstate.q.factor - 1)/2;
  338. CRT(exp_p, pm1, exp_q, qm1);
  339. if (exp_p < 0) {
  340. exp_p += pm1;
  341. }
  342. ZZ& expon = exp_p;
  343. struct timeval ended_working;
  344. gettimeofday(&ended_working, NULL);
  345. unsigned long long computation_length_ms =
  346. (ended_working.tv_sec - ctrlstate.started_working.tv_sec) * 1000 +
  347. (ended_working.tv_usec - ctrlstate.started_working.tv_usec) / 1000;
  348. char length_buf[50];
  349. sprintf(length_buf, "%lld.%03lld s", computation_length_ms / 1000,
  350. computation_length_ms % 1000);
  351. cout << "expon = " << expon << "\n";
  352. ZZ base_exp = PowerMod(ctrlstate.base, expon, ctrlstate.rho);
  353. if (base_exp == ctrlstate.target) {
  354. cout << "CORRECT in " << length_buf << "\n";
  355. } else {
  356. cout << "INCORRECT in " << length_buf << ":\n";
  357. cout << "base^exp = " << base_exp << "\n";
  358. cout << "target = " << ctrlstate.target << "\n";
  359. }
  360. ctrlstate.reset();
  361. }
  362. static unsigned short curproblemid = 0;
  363. // Take base and target mod f.factor, then decompose that into small
  364. // subproblems given our knowledge of the factors of phi(f.factor)
  365. static vector<SubproblemProgress> decomp(const ZZ_p &base, const ZZ_p &target,
  366. const FactorDecomp &f)
  367. {
  368. vector<SubproblemProgress> ret;
  369. // Compute phi(factor)
  370. const int fveclen = f.fvec.length();
  371. ZZ phi = to_ZZ(2);
  372. for (int i = 0; i < fveclen; ++i) {
  373. phi *= f.fvec[i];
  374. }
  375. ZZ_p::init(f.factor);
  376. for (int i = 0; i < fveclen; ++i) {
  377. const ZZ& order = f.fvec[i];
  378. ZZ quotient = phi / order;
  379. ZZ_p subgroup_base = to_ZZ_p(rep(base));
  380. subgroup_base = power(subgroup_base, quotient);
  381. ZZ_p subgroup_target = to_ZZ_p(rep(target));
  382. subgroup_target = power(subgroup_target, quotient);
  383. if (subgroup_base == 1) {
  384. // The original base wasn't a generator of the whole group
  385. if (subgroup_target == 1) {
  386. // But the target is in the subgroup. Lucky us.
  387. continue;
  388. } else {
  389. ret.clear();
  390. return ret;
  391. }
  392. }
  393. ++curproblemid;
  394. ret.push_back(SubproblemProgress(curproblemid, rep(subgroup_base),
  395. rep(subgroup_target), f.factor, order));
  396. }
  397. return ret;
  398. }
  399. static int generate_problem(struct event_base *evbase)
  400. {
  401. // If there are no more problems to generate, close the listener
  402. if (ctrlstate.problems_remaining == 0) {
  403. evconnlistener_free(ctrlstate.listener);
  404. ctrlstate.dpnodes.free();
  405. ctrlstate.workers.free();
  406. return -1;
  407. }
  408. // If there's already a problem on the go, don't generate another one
  409. if (ctrlstate.working == true) {
  410. return -1;
  411. }
  412. ctrlstate.working = true;
  413. int num_subproblems_p = 0;
  414. int num_subproblems_q = 0;
  415. // Generate a DLP mod rho (in the large odd-order subgroup)
  416. ZZ_p::init(ctrlstate.rho);
  417. do {
  418. ZZ_p base = power(random_ZZ_p(), 2);
  419. ZZ_p target = power(random_ZZ_p(), 2);
  420. gettimeofday(&ctrlstate.started_working, NULL);
  421. ctrlstate.base = rep(base);
  422. ctrlstate.target = rep(target);
  423. // Decompose it mod p and mod q
  424. ctrlstate.subproblems_p = decomp(base, target, ctrlstate.p);
  425. ctrlstate.subproblems_q = decomp(base, target, ctrlstate.q);
  426. num_subproblems_p = ctrlstate.subproblems_p.size();
  427. num_subproblems_q = ctrlstate.subproblems_q.size();
  428. } while (num_subproblems_p == 0 || num_subproblems_q == 0);
  429. ctrlstate.num_unsolved_subproblems =
  430. num_subproblems_p + num_subproblems_q;
  431. schedule();
  432. --ctrlstate.problems_remaining;
  433. return 0;
  434. }
  435. typedef enum {
  436. CCSTATE_START,
  437. CCSTATE_DPWAITRESP,
  438. CCSTATE_DPLISTENING,
  439. CCSTATE_DPEXPON,
  440. CCSTATE_END
  441. } CCState;
  442. struct ControllerConnInfo {
  443. CCState state;
  444. ControllerConnInfo() : state(CCSTATE_DPWAITRESP) {}
  445. };
  446. static void controller_dpnode_reader(struct bufferevent *bev, void *ctx)
  447. {
  448. struct evbuffer *input = bufferevent_get_input(bev);
  449. ControllerConnInfo *info = (ControllerConnInfo *)ctx;
  450. unsigned char cmd[1];
  451. ZZ expon;
  452. while(1) {
  453. size_t len = evbuffer_get_length(input);
  454. switch (info->state) {
  455. case CCSTATE_START:
  456. case CCSTATE_DPWAITRESP:
  457. if (len < 1) return;
  458. bufferevent_read(bev, cmd, 1);
  459. switch (cmd[0]) {
  460. case 'L':
  461. info->state = CCSTATE_DPLISTENING;
  462. break;
  463. case 'E':
  464. info->state = CCSTATE_DPEXPON;
  465. break;
  466. default:
  467. /* Unknown DPnode command received */
  468. fprintf(stderr, "Unknown command in "
  469. "controller_dpnode_reader: "
  470. "%c\n", cmd[0]);
  471. info->state = CCSTATE_END;
  472. break;
  473. }
  474. break;
  475. case CCSTATE_DPLISTENING:
  476. // Read 6 bytes
  477. if (len < 6) return;
  478. unsigned char ipport[6];
  479. unsigned int DPip;
  480. unsigned short DPport;
  481. bufferevent_read(bev, ipport, 6);
  482. memmove(&DPip, ipport, 4);
  483. memmove(&DPport, ipport+4, 2);
  484. {
  485. struct in_addr DPaddr = { DPip };
  486. #ifdef DEBUG
  487. fprintf(stderr, "DP node at %s:%d\n", inet_ntoa(DPaddr), ntohs(DPport));
  488. #endif
  489. if (ctrlstate.dpnodes.working.count(bev) > 0) {
  490. ctrlstate.dpnodes.working[bev]->ipports.push_back(
  491. IPPort(ipport));
  492. schedule();
  493. }
  494. }
  495. info->state = CCSTATE_DPWAITRESP;
  496. break;
  497. case CCSTATE_DPEXPON:
  498. // Read the subproblemid and the answer to the subproblem
  499. if (len < 2 + 3*sizeof(unsigned int)) return;
  500. unsigned char exponbytes[2 + 3*sizeof(unsigned int)];
  501. unsigned short problemid;
  502. bufferevent_read(bev, exponbytes, 2 + 3*sizeof(unsigned int));
  503. memmove(&problemid, exponbytes, 2);
  504. ZZFromBytes(expon, exponbytes+2, 3*sizeof(unsigned int));
  505. // Find the subproblem and check the answer
  506. if (ctrlstate.dpnodes.working.count(bev) > 0) {
  507. SubproblemProgress *spp = ctrlstate.dpnodes.working[bev];
  508. if (spp->problemid == problemid &&
  509. spp->solved == false &&
  510. spp->target ==
  511. PowerMod(spp->base, expon, spp->modulus)) {
  512. // Subproblem solved!
  513. spp->solution = expon;
  514. spp->solved = true;
  515. spp->stop();
  516. ctrlstate.num_unsolved_subproblems--;
  517. if (ctrlstate.num_unsolved_subproblems == 0) {
  518. computation_complete();
  519. generate_problem(bufferevent_get_base(bev));
  520. }
  521. schedule();
  522. }
  523. }
  524. info->state = CCSTATE_DPWAITRESP;
  525. break;
  526. case CCSTATE_END:
  527. // Shut down the connection
  528. delete info;
  529. bufferevent_free(bev);
  530. return;
  531. }
  532. }
  533. }
  534. static void controller_dpnode_event_cb(struct bufferevent *bev, short events,
  535. void *ctx)
  536. {
  537. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  538. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  539. fprintf(stderr, "Closing dpnode connection\n");
  540. if (ctrlstate.dpnodes.working.count(bev)) {
  541. // If we lose a dpnode from an active computation, the
  542. // computation is useless.
  543. SubproblemProgress *spp = ctrlstate.dpnodes.working[bev];
  544. ctrlstate.dpnodes.working.erase(bev);
  545. spp->dpnodes.erase(bev);
  546. spp->stop();
  547. } else {
  548. ctrlstate.dpnodes.idle.erase(bev);
  549. }
  550. delete info;
  551. bufferevent_free(bev);
  552. schedule();
  553. }
  554. }
  555. static void controller_worker_event_cb(struct bufferevent *bev, short events,
  556. void *ctx)
  557. {
  558. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  559. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  560. fprintf(stderr, "Closing worker connection\n");
  561. if (ctrlstate.workers.working.count(bev)) {
  562. SubproblemProgress *spp = ctrlstate.workers.working[bev];
  563. ctrlstate.workers.working.erase(bev);
  564. spp->workers.erase(bev);
  565. } else {
  566. ctrlstate.workers.idle.erase(bev);
  567. }
  568. delete info;
  569. bufferevent_free(bev);
  570. schedule();
  571. }
  572. }
  573. static void controller_event_cb(struct bufferevent *bev, short events,
  574. void *ctx)
  575. {
  576. if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
  577. fprintf(stderr, "Closing connection\n");
  578. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  579. delete info;
  580. bufferevent_free(bev);
  581. }
  582. }
  583. // We're just going to read a single byte that will tell us whether the
  584. // peer is a DPnode or a Worker
  585. static void controller_master_reader(struct bufferevent *bev, void *ctx)
  586. {
  587. struct evbuffer *input = bufferevent_get_input(bev);
  588. size_t len = evbuffer_get_length(input);
  589. if (len < 1) return;
  590. char indata[1];
  591. bufferevent_read(bev, indata, 1);
  592. switch(indata[0]) {
  593. case 'D':
  594. /* Add this DPnode to the list of available ones */
  595. ctrlstate.dpnodes.idle.insert(bev);
  596. bufferevent_setcb(bev, controller_dpnode_reader, NULL,
  597. controller_dpnode_event_cb, ctx);
  598. controller_dpnode_reader(bev, ctx);
  599. schedule();
  600. return;
  601. case 'W':
  602. ctrlstate.workers.idle.insert(bev);
  603. // We don't actually read anything from workers
  604. bufferevent_enable(bev, EV_WRITE);
  605. bufferevent_setcb(bev, NULL, NULL,
  606. controller_worker_event_cb, ctx);
  607. schedule();
  608. return;
  609. default:
  610. fprintf(stderr, "Unknown command in controller_master_reader: "
  611. "%c\n", indata[0]);
  612. ControllerConnInfo *info = (ControllerConnInfo*)ctx;
  613. delete info;
  614. bufferevent_free(bev);
  615. return;
  616. }
  617. }
  618. static void controller_accept_cb(struct evconnlistener *listener,
  619. evutil_socket_t fd, struct sockaddr *address, int socklen,
  620. void *ctx)
  621. {
  622. // Create the state of the new connection
  623. ControllerConnInfo *info = new ControllerConnInfo();
  624. // Create a bufferevent for the new connection
  625. struct event_base *base = evconnlistener_get_base(listener);
  626. struct bufferevent *bev = bufferevent_socket_new(
  627. base, fd, BEV_OPT_CLOSE_ON_FREE);
  628. bufferevent_setcb(bev, controller_master_reader, NULL,
  629. controller_event_cb, info);
  630. bufferevent_enable(bev, EV_READ|EV_WRITE);
  631. }
  632. // Create a new controller socket. bindport is the port to bind to (in
  633. // host byte order), or 0 if any port will do. ip and boundport are set
  634. // to the IP and port of the socket, in network byte order.
  635. static struct evconnlistener *controller_create(struct event_base *evbase,
  636. unsigned short bindport, unsigned int *ip, unsigned short *boundport)
  637. {
  638. return listener_create(evbase, bindport, controller_accept_cb, NULL,
  639. ip, boundport, false);
  640. }
  641. // Read the modulus (and the factorization of the modulus and its
  642. // totient) from the given file. "-" means cin. Returns true if
  643. // successful.
  644. static bool read_modulus(const char *filename)
  645. {
  646. if (strcmp(filename, "-")) {
  647. ifstream ins(filename);
  648. if (!ins.good()) return false;
  649. ins >> ctrlstate.rho >> ctrlstate.p.factor >> ctrlstate.p.fvec >>
  650. ctrlstate.q.factor >> ctrlstate.q.fvec;
  651. ins.close();
  652. } else {
  653. cin >> ctrlstate.rho >> ctrlstate.p.factor >> ctrlstate.p.fvec >>
  654. ctrlstate.q.factor >> ctrlstate.q.fvec;
  655. }
  656. return true;
  657. }
  658. int controller_main(const char *modulus_file, unsigned short bindport)
  659. {
  660. // Initialize the prng with some randomness from the kernel
  661. unsigned char randbuf[1024];
  662. ifstream urand("/dev/urandom");
  663. urand.read((char *)randbuf, sizeof(randbuf));
  664. urand.close();
  665. ZZ randzz = ZZFromBytes(randbuf, sizeof(randbuf));
  666. SetSeed(randzz);
  667. // Read the modulus and the factorization of its totient from the
  668. // specified file
  669. if (!read_modulus(modulus_file)) {
  670. cerr << "Unable to read file " << modulus_file << "\n";
  671. return 1;
  672. }
  673. struct event_base *evbase = event_base_new();
  674. unsigned int myip;
  675. unsigned short myport;
  676. ctrlstate.listener = controller_create(evbase, bindport, &myip, &myport);
  677. struct in_addr myaddr = { myip };
  678. printf("Bound to %s:%d\n", inet_ntoa(myaddr), ntohs(myport));
  679. ctrlstate.problems_remaining = 2;
  680. // Kick off the first problem to solve
  681. generate_problem(evbase);
  682. event_base_dispatch(evbase);
  683. return 0;
  684. }