proper thread batching

README.md

@@ -0,0 +1,5 @@
+
+
+
+dependencies:
+ - pthread

src/main/c/includes/electrosphere_FluidSim.h

@@ -129,6 +129,30 @@ JNIEXPORT void JNICALL Java_electrosphere_FluidSim_copyNeighbors
 JNIEXPORT jlong JNICALL Java_electrosphere_FluidSim_createThreadpool
   (JNIEnv *, jclass, jint);
 
+/*
+ * Class:     electrosphere_FluidSim
+ * Method:    unlockThreads
+ * Signature: (J)J
+ */
+JNIEXPORT jlong JNICALL Java_electrosphere_FluidSim_unlockThreads
+  (JNIEnv *, jclass, jlong);
+
+/*
+ * Class:     electrosphere_FluidSim
+ * Method:    submitWork
+ * Signature: ()V
+ */
+JNIEXPORT void JNICALL Java_electrosphere_FluidSim_submitWork
+  (JNIEnv *, jclass);
+
+/*
+ * Class:     electrosphere_FluidSim
+ * Method:    fetchWork
+ * Signature: ()V
+ */
+JNIEXPORT void JNICALL Java_electrosphere_FluidSim_fetchWork
+  (JNIEnv *, jclass);
+
 #ifdef __cplusplus
 }
 #endif

src/main/c/includes/threadpool.h

@@ -5,6 +5,8 @@
 typedef struct {
     int numThreads;
     pthread_t * threads;
+    pthread_barrier_t * barrierMain;
+    pthread_barrier_t * barrierWithParentThread;
 } ThreadPool;
 
 

src/main/c/src/threadpool.c

@@ -4,15 +4,21 @@
 #include <stdint.h>
 #include <unistd.h>
 #include <pthread.h>
+#include <semaphore.h>
 
 #include "../includes/libfluidsim.h"
 #include "../includes/threadpool.h"
 
+typedef struct {
+  ThreadPool * threadPool;
+  int threadIndex; //the index for this thread in all threadpool objects
+} ThreadData;
+
 
 /**
  * Main loops for the thread
 */
-void * main_thread_loop(void * data);
+void * mainThreadLoop(void * data);
 
 /*
  * Class:     electrosphere_FluidSim
@@ -23,23 +29,99 @@ JNIEXPORT jlong JNICALL Java_electrosphere_FluidSim_createThreadpool
   (JNIEnv * env,
   jclass class,
   jint numThreads){
+
+  //init threadpool
   ThreadPool * pool = (ThreadPool *)malloc(sizeof(ThreadPool));
   pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * numThreads);
   pool->numThreads = numThreads;
+
+  //used for storing return codes from pthread calls
   int retCode = 0;
-  for(int i = 0; i < pool->numThreads; i++){
-    retCode = pthread_create(&pool->threads[i], NULL, main_thread_loop, NULL);
-    // printf("thread ret code %d\n",retCode);
+
+  //
+  //create thread barriers
+  printf("Creating barrier for %d threads\n",pool->numThreads);
+  pool->barrierMain = (pthread_barrier_t *)malloc(sizeof(pthread_barrier_t));
+  retCode = pthread_barrier_init(pool->barrierMain,NULL,pool->numThreads);
+  if(retCode != 0){
+    printf("Failed to create main barrier! %d\n",retCode);
   }
-  
-  for(int i = 0; i < pool->numThreads; i++){
-    pthread_join(pool->threads[i], NULL);
-    printf("resolve thread\n");
+
+  //+1 is for the main thread
+  printf("Creating barrier for %d threads\n",pool->numThreads+1);
+  pool->barrierWithParentThread = (pthread_barrier_t *)malloc(sizeof(pthread_barrier_t));
+  retCode = pthread_barrier_init(pool->barrierWithParentThread,NULL,(pool->numThreads)+1);
+  if(retCode != 0){
+    printf("Failed to create barrier with main thread! %d\n",retCode);
   }
+
+  //start threads
+  for(int i = 0; i < pool->numThreads; i++){
+
+    //create data to be sent to thread
+    ThreadData * threadData = (ThreadData *)malloc(sizeof(ThreadData));
+    threadData->threadPool = pool;
+
+    //create thread
+    retCode = pthread_create(&pool->threads[i], NULL, mainThreadLoop, threadData);
+    if(retCode != 0){
+      printf("Failed to create thread!~ %d\n",retCode);
+    }
+  }
+
+  printf("Finished creating threads\n");
+
+  return (jlong)pool;
 }
 
 
-void * main_thread_loop(void * data){
-    printf("thread work\n");
-    sleep(1);
+/*
+ * Class:     electrosphere_FluidSim
+ * Method:    unlockThreads
+ * Signature: ()J
+ */
+JNIEXPORT jlong JNICALL Java_electrosphere_FluidSim_unlockThreads
+  (JNIEnv * env,
+  jclass class,
+  jlong threadPoolPtrRaw){
+  ThreadPool * pool = (ThreadPool *)threadPoolPtrRaw;
+  pthread_barrier_wait(pool->barrierWithParentThread);
+}
+
+
+/**
+ * Main loop for threads
+*/
+void * mainThreadLoop(void * dataRaw){
+  ThreadData * threadData = (ThreadData *)dataRaw;
+  ThreadPool * threadPool = threadData->threadPool;
+  int threadIndex = threadData->threadIndex;
+
+  printf("thread work\n");
+
+  //main thread loop
+  int running = 1;
+  while(running){
+
+    //begin work
+    //wait to begin work until parent signals its ready
+    printf("call barrier 1\n");
+    pthread_barrier_wait(threadPool->barrierWithParentThread);
+
+    //do main work
+
+
+    //call this if the child threads need to wait on one another
+    printf("call barrier 2\n");
+    pthread_barrier_wait(threadPool->barrierMain);
+
+
+
+    //finalize work
+    //the parent thread needs to call the barrier as well with whatever method it uses to grab data from the sim back to java
+    printf("call barrier 3\n");
+    pthread_barrier_wait(threadPool->barrierWithParentThread);
+
+
+  }
 }

src/main/java/electrosphere/FluidSim.java

@@ -851,12 +851,41 @@ public class FluidSim {
     private native void copyNeighbors(int DIM_X, int chunkMask, int x, int vectorDir, ByteBuffer[] neighborMap);
 
 
+    /**
+     * Creates the thread pool for handling main work of the c lib
+     */
     public static long createThreadpoolWrapper(int numThreads){
         return createThreadpool(numThreads);
     }
     private static native long createThreadpool(int numThreads);
 
 
+    /**
+     * Unlocks the C threadpool
+     */
+    public static long unlockThreadsWrapper(){
+        return unlockThreads(threadpool);
+    }
+    private static native long unlockThreads(long threadPoolPtr);
+
+
+    /**
+     * Sends work to the threadpool
+     */
+    public static void submitWorkWrapper(){
+        submitWork();
+    }
+    public static native void submitWork();
+
+    /**
+     * Fetches work from the threadpool
+     */
+    public static void fetchWorkWrapper(){
+        fetchWork();
+    }
+    public static native void fetchWork();
+
+
 
 
 

src/main/java/electrosphere/Main.java

@@ -55,7 +55,9 @@ public class Main {
                 //
                 //Simulate
                 //
+                FluidSim.unlockThreadsWrapper();
                 FluidSim.simChunks(simArray,i,0.01f);
+                FluidSim.unlockThreadsWrapper();
                 time = time + (System.currentTimeMillis() - lastTime);
                 //
                 //Remesh
@@ -84,7 +86,6 @@ public class Main {
 
     private static FluidSim[][][] initFluidSim(int dimx, int dimy, int dimz){
         FluidSim.threadpool = FluidSim.createThreadpoolWrapper(5);
-        
         FluidSim[][][] simArray = new FluidSim[dimx][dimy][dimz];
         for(int x = 0; x < simArray.length; x++){
             for(int y = 0; y < simArray[0].length; y++){