Merge branch 'alternate_multithreaded'

inc/genetic.h

@@ -9,6 +9,7 @@
 #include "rand.h"
 
 using namespace sync;
+using namespace std;
 
 namespace genetic {
 
@@ -162,7 +163,11 @@ template <class T> DWORD worker(LPVOID args) {
 		    assert(cell != NULL);
 		    children[i-child_begin] = cell;
 		}
-		strat.crossover(parents, children);
+		CrossoverJob<T> cj = {parents, children};
+		TaggedJob<T> job;
+		job.data.c=cj;
+		job.type=JobType::CROSSOVER;
+		q.jobs[q.write_i++] = job;
 		parent_end += strat.crossover_parent_stride;
 		child_begin -= nchild;
 	    }
@@ -187,14 +192,27 @@ template <class T> DWORD worker(LPVOID args) {
 	start = now();
 	if (strat.test_all) {
 	    for (int i = 0; i < trackers.len; i++) {
-		trackers[i].score = strat.fitness(cells[trackers[i].cellid]);
+		FitnessJob<T> fj = {&cells[trackers[i].cellid], &trackers[i]};
+		TaggedJob<T> job;
+		job.data.f=fj;
+		job.type=JobType::FITNESS;
+		if (i == trackers.len-1) lock(q.m);
+		q.jobs[q.write_i++] = job;
+		if (i == trackers.len-1) { q.done_writing = true; unlock(q.m); }
 	    }
 	} else {
+	    lock(q.m);
 	    for (int i = 0; i < trackers.len; i++) {
 		if (abs(norm_rand(strat.rand_seed)) < strat.test_chance) {
-		    trackers[i].score = strat.fitness(cells[trackers[i].cellid]);
+		    FitnessJob<T> fj = {&cells[trackers[i].cellid], &trackers[i]};
+		    TaggedJob<T> job;
+		    job.data.f=fj;
+		    job.type=JobType::FITNESS;
+		    q.jobs[q.write_i++] = job;
 		}
 	    }
+	    q.done_writing = true;
+	    unlock(q.m);
 	}
 	lock(stats.m);
 	append(stats.fitness_time, now() - start);
@@ -339,4 +357,95 @@ template <class T> T run(Strategy<T> strat) {
     return best_cell;
 }
 
+template<class T> WorkQueue<T> make_work_queue(int len, int batch_size) {
+    return {
+	.jobs=make_array<TaggedJob<T>>(len),
+	.read_i=0,
+	.write_i=0,
+	.batch_size=batch_size,
+	.done_writing=false,
+	.work_complete=false,
+	.m=make_mutex(),
+	.done=make_condition_var(),
+	.jobs_ready=make_condition_var()
+    };
+}
+
+template<class T> bool tryget_job_batch(WorkQueue<T> &q, Array<TaggedJob<T>>* out_batch, bool* out_batch_is_end) {
+    lock(q.m);
+    if (q.stop) {
+	unlock(q.m);
+	return false;
+    }
+
+    // Keep waiting till jobs are available
+    while (q.read_i >= q.write_i) {
+	wait(q.jobs_ready, q.m, infinite_ts); 
+	
+	if (q.stop) {
+	    unlock(q.m);
+	    return false;
+	}
+    }
+
+    // Yay! Let's grab some jobs to do
+
+    // If the batch we're about to grab moves read_i to write_i and the producer
+    // is done writing, we should let our callee know it's handling this gen's last 
+    // batch know that way it sets work_complete and signals done.
+    *out_batch_is_end = q.done_writing && q.read_i + q.batch_size >= q.write_i;
+
+    out_batch->data = &q.jobs[q.read_i];
+    out_batch->len = min(q.batch_size, q.write_i - q.read_i);
+    q.read_i += q.batch_size;
+    unlock(q.m);
+    return true;
+}
+
+template<class T>
+void work_batch(Array<TaggedJob<T>> batch, Strategy<T> &s) {
+    for (int i = 0; i < batch.len; i++) {
+	switch (batch[i].type) {
+	    case JobType::MUTATE: {
+		MutateJob<T> mj = batch[i].data.m;
+		s.mutate(*mj.cell);
+	    } break;
+	    case JobType::CROSSOVER: {
+		CrossoverJob<T> cj = batch[i].data.c;
+		s.crossover(cj.parents, cj.children);
+	    } break;
+	    case JobType::FITNESS: {	
+		FitnessJob<T> fj = batch[i].data.f;
+		fj.track->score = s.fitness(*fj.cell);
+	    } break;
+	    default: {
+		assert(false);
+	    }
+	}
+    }
+}
+
+template<class T>
+DWORD worker(LPVOID args) {
+    WorkerThreadArgs<T>* wa = static_cast<WorkerThreadArgs<T>*>(args);
+    WorkQueue<T> &q = wa->q;
+    Strategy<T> &s = wa->s;
+
+    // These are written by tryget_job_batch
+    bool batch_is_end;
+    Array<TaggedJob<T>> batch;
+
+    while (tryget_job_batch(q, &batch, &batch_is_end)) {
+	work_batch(batch, s);
+	if (batch_is_end) {
+	    lock(q.m);
+	    q.work_complete = true;
+	    wake_one(q.done);
+	    unlock(q.m);
+	}
+    }
+
+    return NULL;
+}
+
 } // namespace genetic