Output number of kernel launches for each subproblem.

The workers now send the number of kernel launches to the controller, which aggregates them and outputs the total for each subproblem.

controller.cc

@@ -106,6 +106,8 @@ struct FactorDecomp {
 
 static void vsppdump(const vector<SubproblemProgress> &spv, ostream &os);
 
+static unsigned short curproblemid = 0;
+
 static struct ControllerState {
     ZZ rho;
     FactorDecomp p, q;
@@ -117,9 +119,10 @@ static struct ControllerState {
     unsigned int num_unsolved_subproblems;
     Worklist worklist;
     struct evconnlistener *listener;
+    unsigned int problemid;
 
     ControllerState() : working(false), num_unsolved_subproblems(0),
-	listener(NULL) {}
+	listener(NULL), problemid(0) {}
 
     // Reset the state for a new problem with the same modulus
     void reset(void) {
@@ -131,6 +134,7 @@ static struct ControllerState {
 	num_unsolved_subproblems = 0;
 	started_working.tv_sec = 0;
 	started_working.tv_usec = 0;
+	problemid = curproblemid++;
     }
 
     // Dump the state for debug purposes
@@ -197,11 +201,14 @@ struct SubproblemProgress : Subproblem {
     // The solution, if found.
     ZZ solution;
 
+    unsigned int kernel_launch_count;
+    unsigned int num_workers_replied;
+
     SubproblemProgress(unsigned short id, const ZZ &b, const ZZ &t,
 	    const ZZ &m, const ZZ &o) :
 	    // By default, 1 in 1000 points are distinguihed points.  The
 	    // number in the next line is 2^32/1000
-	    Subproblem(id, b, t, m, o, 4294967), solved(false) {
+	    Subproblem(id, b, t, m, o, 4294967), solved(false), kernel_launch_count(0), num_workers_replied(0) {
 
 	desired_resources(order, desired_dpnodes, max_workers, dpfreq);
 
@@ -218,12 +225,23 @@ struct SubproblemProgress : Subproblem {
 	for (BESet::iterator iter = dpnodes.begin(); iter != dpnodes.end();
 		++iter) {
 	    bufferevent_write(*iter, stopcmd, 1);
+	}
+	for (BESet::iterator iter = workers.begin(); iter != workers.end();
+		++iter) {
+	    bufferevent_write(*iter, stopcmd, 1);
+	}
+    }
+
+    void reset(void) {
+	BESet::iterator iter;
+
+	for (BESet::iterator iter = dpnodes.begin(); iter != dpnodes.end();
+		++iter) {
 	    ctrlstate.dpnodes.working.erase(*iter);
 	    ctrlstate.dpnodes.idle.insert(*iter);
 	}
 	for (BESet::iterator iter = workers.begin(); iter != workers.end();
 		++iter) {
-	    bufferevent_write(*iter, stopcmd, 1);
 	    ctrlstate.workers.working.erase(*iter);
 	    ctrlstate.workers.idle.insert(*iter);
 	}
@@ -428,7 +446,7 @@ static void computation_complete(void)
     ctrlstate.reset();
 }
 
-static unsigned short curproblemid = 0;
+static unsigned short cursubproblemid = 0;
 
 // Take base and target mod f.factor, then decompose that into small
 // subproblems given our knowledge of the factors of phi(f.factor)
@@ -464,8 +482,8 @@ static vector<SubproblemProgress> decomp(const ZZ_p &base, const ZZ_p &target,
 	    }
 	}
 
-	++curproblemid;
-	ret.push_back(SubproblemProgress(curproblemid, rep(subgroup_base),
+	++cursubproblemid;
+	ret.push_back(SubproblemProgress(cursubproblemid, rep(subgroup_base),
 				    rep(subgroup_target), f.factor, order));
     }
 
@@ -490,8 +508,13 @@ static bool read_modulus(const char *filename)
     return true;
 }
 
-static int generate_problem(struct event_base *evbase)
+static int generate_problem()
 {
+    // If there's already a problem on the go, don't generate another one
+    if (ctrlstate.working == true) {
+	return -1;
+    }
+
     while (ctrlstate.worklist[0].second == 0) {
 	ctrlstate.worklist.erase(ctrlstate.worklist.begin());
 	if (ctrlstate.worklist.size() > 0) {
@@ -509,10 +532,6 @@ static int generate_problem(struct event_base *evbase)
 	return -1;
     }
 
-    // If there's already a problem on the go, don't generate another one
-    if (ctrlstate.working == true) {
-	return -1;
-    }
     ctrlstate.working = true;
 
     int num_subproblems_p = 0;
@@ -628,14 +647,10 @@ static void controller_dpnode_reader(struct bufferevent *bev, void *ctx)
 			    spp->target ==
 				PowerMod(spp->base, expon, spp->modulus)) {
 			// Subproblem solved!
+			ctrlstate.num_unsolved_subproblems--;
 			spp->solution = expon;
 			spp->solved = true;
 			spp->stop();
-			ctrlstate.num_unsolved_subproblems--;
-			if (ctrlstate.num_unsolved_subproblems == 0) {
-			    computation_complete();
-			    generate_problem(bufferevent_get_base(bev));
-			}
 			schedule();
 		    }
 		}
@@ -652,6 +667,47 @@ static void controller_dpnode_reader(struct bufferevent *bev, void *ctx)
 
 }
 
+
+static void controller_worker_reader(struct bufferevent *bev, void *ctx)
+{
+    struct evbuffer *input = bufferevent_get_input(bev);
+    ControllerConnInfo *info = (ControllerConnInfo *)ctx;
+
+    while(1) {
+	unsigned int kernel_launch_count = 0;
+
+	size_t len = evbuffer_get_length(input);
+	if (len < sizeof(kernel_launch_count)) return;
+
+	bufferevent_read(bev, &kernel_launch_count, sizeof(kernel_launch_count));
+
+	// this assumes that workers never leave a subproblem (never crash or get re-assigned)
+	// otherwise we'll miss out on the kernel_launch_count for some workers
+
+	SubproblemProgress *spp = ctrlstate.workers.working[bev];
+
+	spp->kernel_launch_count += kernel_launch_count;
+	spp->num_workers_replied += 1;
+
+	if (spp->num_workers_replied == spp->workers.size()) {
+	    cout << "Timing (name, problemid, subproblemid, launches): "
+	         << ctrlstate.worklist[0].first << ", "
+	         << ctrlstate.problemid << ", "
+	         << spp->problemid << ", "
+	         << spp->kernel_launch_count << "\n";
+	    cout.flush();
+
+	    spp->reset();
+	    schedule();
+	}
+
+	if (ctrlstate.num_unsolved_subproblems == 0 && ctrlstate.workers.working.size() == 0) {
+	    computation_complete();
+	    generate_problem();
+	}
+    }
+}
+
 static void controller_dpnode_event_cb(struct bufferevent *bev, short events,
     void *ctx)
 {
@@ -665,6 +721,7 @@ static void controller_dpnode_event_cb(struct bufferevent *bev, short events,
 	    ctrlstate.dpnodes.working.erase(bev);
 	    spp->dpnodes.erase(bev);
 	    spp->stop();
+	    spp->reset();
 	} else {
 	    ctrlstate.dpnodes.idle.erase(bev);
 	}
@@ -727,10 +784,10 @@ static void controller_master_reader(struct bufferevent *bev, void *ctx)
 	    return;
 	case 'W':
 	    ctrlstate.workers.idle.insert(bev);
-	    // We don't actually read anything from workers
 	    bufferevent_enable(bev, EV_WRITE);
-	    bufferevent_setcb(bev, NULL, NULL,
+	    bufferevent_setcb(bev, controller_worker_reader, NULL,
 		    controller_worker_event_cb, ctx);
+	    controller_worker_reader(bev, ctx);
 	    schedule();
 	    return;
 	default:
@@ -858,7 +915,7 @@ int controller_main(const Worklist &worklist, unsigned short bindport,
     }
 
     // Kick off the first problem to solve
-    generate_problem(evbase);
+    generate_problem();
 
     event_base_dispatch(evbase);
 

cudadl.h

@@ -26,6 +26,6 @@
 NTL_CLIENT
 
 void cuda_dl(const ZZ_p &base, const ZZ_p &target, const ZZ &order,
-		const ZZ &modulus, unsigned int dpfreq, void *data);
+		const ZZ &modulus, unsigned int dpfreq, void *data, unsigned int* launch_count);
 
 #endif

dlrho.cc

@@ -162,7 +162,8 @@ static int p_dl(const ZZ_p &target, const ZZ_p &base, ZZ &exp,
 	    dpfreq = 4294967295; // 2^32-1 : every point is a DP
 	}
 	gettimeofday(&st, NULL);
-	cuda_dl(subgroup_base, subgroup_target, fvec[i], md, dpfreq, &cbdata);
+	unsigned int launch_count = 0;
+	cuda_dl(subgroup_base, subgroup_target, fvec[i], md, dpfreq, &cbdata, &launch_count);
 	ZZ subgroup_dl = cbdata.expon;
 	gettimeofday(&et, NULL);
 	unsigned long us_elapsed = (et.tv_sec-st.tv_sec)*1000000 +

parrhoasm.cu

@@ -325,7 +325,8 @@ int nthreads = 25600;
 int nblocks = 50;
 
 void cuda_dl(const ZZ_p &base, const ZZ_p &target, const ZZ &order,
-		const ZZ &modulus, unsigned int dpfreq, void *cbdata)
+		const ZZ &modulus, unsigned int dpfreq, void *cbdata,
+		unsigned int* launch_count)
 {
     unsigned long long totmicros = 0;
     ZZ_pBak pbak;
@@ -458,7 +459,7 @@ void cuda_dl(const ZZ_p &base, const ZZ_p &target, const ZZ &order,
     cudaMemcpy(d_ts, l_ts, nthreads * sizeof(GlobalThreadState), cudaMemcpyHostToDevice);
     dim3 tpb(threadDimx, threadsPerBlock/threadDimx);
     gettimeofday(&st, NULL);
-    int launchcount = 0;
+    int local_launchcount = 0;
     bool stop_computing = false;
 #ifdef CHECK_RESULTS
     for (int ln=0; ln<nlaunch && stop_computing == false; ++ln)
@@ -466,9 +467,9 @@ void cuda_dl(const ZZ_p &base, const ZZ_p &target, const ZZ &order,
     while(stop_computing == false)
 #endif
     {
-	launchcount++;
+	local_launchcount++;
 #ifdef VERBOSE
-	cerr << getpid() << " Launch " << launchcount << "...\n";
+	cerr << getpid() << " Launch " << local_launchcount << "...\n";
 #endif
 	cudaMulmod<<< nblocks, nthreads/nblocks /*tpb*/, 0 >>>(d_ts,
 	    order_0, order_1, order_2, dpfreq);
@@ -503,8 +504,8 @@ void cuda_dl(const ZZ_p &base, const ZZ_p &target, const ZZ &order,
 	btemp = bstart[t];
 	zz_p_xr_temp = zz_p_xr_start[t];
 
-	// launchcount should be <= nlaunch
-	for (int l = 0; l < nmult * launchcount; ++l)
+	// local_launchcount should be <= nlaunch
+	for (int l = 0; l < nmult * local_launchcount; ++l)
 	{
 	    unsigned int w = trunc_long(rep(zz_p_xr_temp), 32);
 	    if (w < TWO_32_DIV_3) {
@@ -567,7 +568,7 @@ cout << i << ": " << l_Z[i] << " != " << l_z[i + t * WORDS] << "\n";
     sprintf(timestamp, "%d.%06d", et.tv_sec, et.tv_usec);
 
     unsigned long long totnanos = totmicros * 1000;
-    cout << timestamp << ":" << output_prefix << ": " << totmicros << " us / " << nthreads << " = " << totmicros / nthreads << " us / " << (nmult*launchcount) << " = " << totnanos / ((unsigned long long)nthreads * nmult * launchcount) << " ns\n";
+    cout << timestamp << ":" << output_prefix << ": " << totmicros << " us / " << nthreads << " = " << totmicros / nthreads << " us / " << (nmult*local_launchcount) << " = " << totnanos / ((unsigned long long)nthreads * nmult * local_launchcount) << " ns\n";
     cout.flush();
     //cudaPrintfEnd();
     DPstreamEnd();
@@ -580,6 +581,8 @@ cout << i << ": " << l_Z[i] << " != " << l_z[i + t * WORDS] << "\n";
     free(l_z);
 
     pbak.restore();
+
+    *launch_count = local_launchcount;
 }
 
 #ifdef TEST_CUDA

worker.cc

@@ -53,6 +53,7 @@ static struct WrkControllerState {
     unsigned short num_connected_dpnodes;
     WTState worker_thread_state;
     pthread_t worker_thread;
+    unsigned int kernel_launch_count;
 
     WrkControllerState(): current_problem(NULL),
 	worker_thread_state(WT_NOT_RUNNING) {}
@@ -104,7 +105,7 @@ static void *worker_thread_start(void *data)
 	    to_ZZ_p(wrkctrlstate.current_problem->target),
 	    wrkctrlstate.current_problem->order,
 	    wrkctrlstate.current_problem->modulus,
-	    wrkctrlstate.current_problem->dpfreq, NULL);
+	    wrkctrlstate.current_problem->dpfreq, NULL, &wrkctrlstate.kernel_launch_count);
 
     return NULL;
 }
@@ -154,6 +155,7 @@ static void start_working(void)
 
     cout << timestamp << ":" << output_prefix << ": Subproblem " << wrkctrlstate.current_problem->problemid << "\n";
     cout.flush();
+    wrkctrlstate.kernel_launch_count = 0;
     wrkctrlstate.worker_thread_state = WT_RUNNING;
     if (pthread_create(&wrkctrlstate.worker_thread, NULL,
 			worker_thread_start, NULL)) {
@@ -213,6 +215,9 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 			break;
 		    case 'S':
 			stop_working();
+			bufferevent_write(bev, &(wrkctrlstate.kernel_launch_count), sizeof(wrkctrlstate.kernel_launch_count));
+			cout << "Launch count: " << wrkctrlstate.kernel_launch_count << "\n";
+			cout.flush();
 			break;
 		    default:
 			/* Unknown command received */