added AlgorithmWrapper.h Buffer.h BufferDispatcher.h and removed authro and date from templates

Author: brichards64

UserTools/template/MyTool.h

@@ -12,8 +12,8 @@
  *
  * This is a balnk template for a Tool used by the script to generate a new custom tool. Please fill out the descripton and author information.
 *
-* $Author: B.Richards $
-* $Date: 2019/05/28 10:44:00 $
+* $Author:  $
+* $Date:  $
 */
 
 class MyTool: public Tool {

UserTools/template/MyToolDynamicMultiThread.h

@@ -13,8 +13,8 @@
  * This is a struct to place data you want your thread to acess or exchange with it. The idea is the datainside is only used by the threa\
 d and so will be thread safe
 *
-* $Author: B.Richards $
-* $Date: 2019/05/28 10:44:00 $
+* $Author:  $
+* $Date:  $
 */
 
 struct MyToolDynamicMultiThread_args:Thread_args{

UserTools/template/MyToolMultiThread.h

@@ -12,8 +12,8 @@
  *
  * This is a struct to place data you want your thread to acess or exchange with it. The idea is the datainside is only used by the thread and so will be thread safe
  *
- * $Author: B.Richards $
- * $Date: 2019/05/28 10:44:00 $
+ * $Author:  $
+ * $Date:  $
  */
 
 struct MyToolMultiThread_args:Thread_args{

UserTools/template/MyToolThread.h

@@ -12,8 +12,8 @@
  *
  * This is a struct to place data you want your thread to access or exchange with it. The idea is the datainside is only used by the threa\d and so will be thread safe
  *
- * $Author: B.Richards $
- * $Date: 2019/05/28 10:44:00 $
+ * $Author:  $
+ * $Date:  $
  */
 
 struct MyToolThread_args:Thread_args{

src/DataModelBase/AlgorithmWrapper.h

@@ -0,0 +1,30 @@
+#ifndef ALGORITHM_WRAPPER_H
+#define ALGORITHM_WRAPPER_H
+
+#include <string>
+#include <functional>
+
+namespace ToolFramework{
+  /**
+   * \struct Trigger_struct
+   *
+   * This tricts acts as a generator for trigger jobs with fucntionpointers
+   */
+  
+  template<class T> struct AlgorithmWrapper{
+    
+    AlgorithmWrapper(std::string in_name, bool (*in_algo)(void*&), std::function<void*(T)> in_setup_func,  void (*in_fail_func)(void*&)){
+      name = in_name;
+      algo = in_algo;
+      setup_func = in_setup_func;
+      fail_func = in_fail_func;
+      
+    }
+    std::string name; ///< name of algorihtm
+    bool (*algo)(void*&); ///< algorithm to run on data
+    void (*fail_func)(void*&); ///< fail funciton if algroithm fails
+    std::function<void*(T)> setup_func; ///< setup function to create arguments 
+    
+  };
+}
+#endif

src/DataModelBase/Buffer.h

@@ -0,0 +1,59 @@
+#ifndef BUFFER_H
+#define BUFFER_H
+
+#include <vector>
+#include <mutex>
+#include <SerialisableObject.h>
+#include <BinaryStream.h>
+
+namespace ToolFramework{
+  
+  template<class T> class Buffer: SerialisableObject{
+    
+  public:
+    
+    Buffer(){;}
+    void Add(T &in){
+      std::lock_guard<std::mutex> lock(mtx);    
+      data.push_back(in);
+    };
+    void Swap(std::vector<T> &in){
+      std::lock_guard<std::mutex> lock(mtx);
+      if(data.size()) std::swap (data, in);
+    
+    }
+    size_t Size(){
+      
+      std::lock_guard<std::mutex> lock(mtx);
+      return data.size();
+      
+    }
+    void Clear(){
+      std::lock_guard<std::mutex> lock(mtx);
+      data.clear();
+    }
+  
+    
+    std::string GetVersion(){ return "1";}
+    
+    bool Print(){
+
+      return true;
+    }
+    
+    bool Serialise(BinaryStream& bs){
+      
+      bs & data;
+      return true;
+    }
+    
+  
+  private:
+    std::vector<T> data;
+    std::mutex mtx;
+    
+  };
+  
+}
+
+#endif

src/DataModelBase/BufferDispatcher.h

@@ -0,0 +1,114 @@
+#ifndef BUFFER_DISPATCHER_H
+#define BUFFER_DISPATCHER_H
+
+#include <atomic>
+#include <vector>
+#include <Utilities.h>
+#include <JobQueue.h>
+#include <Pool.h>
+#include <Buffer.h>
+#include <AlgorithmWrapper.h>
+
+namespace ToolFramework{
+
+   template<class T> struct BufferDispatcher_args:Thread_args{
+     BufferDispatcher_args(){;}
+     ~BufferDispatcher_args(){;}
+      
+      Buffer<T>* buffer = 0;
+      std::vector<T> local_buffer; 
+      std::vector<AlgorithmWrapper<T> >* algorithms = 0;
+      Job* job = 0;
+
+      JobQueue* job_queue = 0;      
+      Pool<Job>* job_pool = 0;
+      
+      std::atomic<uint64_t>* counter = 0;
+      
+    };  
+
+    template<class T> class BufferDispatcher{
+      
+    public:
+      
+      BufferDispatcher(){;}
+      ~BufferDispatcher(){Close();}
+      bool Init(Buffer<T>* buffer, std::vector<AlgorithmWrapper<T> >* algorithms, JobQueue* job_queue, Pool<Job>* job_pool){
+
+	args.buffer = buffer;
+	args.algorithms = algorithms;
+	args.job_queue = job_queue;
+	args.job_pool = job_pool;
+	counter = 0;
+	args.counter = &counter;
+
+	if(buffer == 0 || algorithms == 0 || job_queue == 0 || job_pool == 0) return false; 	
+		
+	m_util.CreateThread("BufferDispatcher", &Thread, &args);
+	
+      }
+      
+      void Close(){
+	if(args.buffer == 0 || args.algorithms == 0 || args.job_queue == 0 || args.job_pool == 0) return;
+	m_util.KillThread(&args);
+	
+	args.buffer = 0;
+	args.algorithms = 0;
+	args.job_queue = 0;
+	args.job_pool = 0;
+	
+      }
+      
+      std::atomic<uint64_t> counter;
+      
+      
+    private:
+      
+      static void Thread(Thread_args* arg){  
+	BufferDispatcher_args<T>* args=reinterpret_cast<BufferDispatcher_args<T>*>(arg);
+	if(args->algorithms->size()==0){
+	  usleep(100); 
+	  return;	
+	}
+
+	args->buffer->Swap(args->local_buffer);
+	
+	if(args->local_buffer.size() == 0){
+	  usleep(100);
+	  return;
+	}
+	
+	for(size_t i = 0; i < args->local_buffer.size(); i++){
+	  
+	  for(size_t j = 0; j < args->algorithms->size(); j++){
+	    
+	    args->job = args->job_pool->GetNew(args->algorithms->at(j).name);
+	    args->job->m_id = args->algorithms->at(j).name;
+	    args->job->func = args->algorithms->at(j).algo;
+	    args->job->fail_func = args->algorithms->at(j).fail_func;
+	    args->job->data = args->algorithms->at(j).setup_func(args->local_buffer.at(i));
+	    args->job->out_pool = args->job_pool;
+	    
+	    args->job_queue->AddJob(args->job);
+	    args->job = 0;
+	    (*args->counter)++;
+	  }
+	  
+	  args->local_buffer.at(i) = 0;
+	  
+	}
+	
+	args->local_buffer.clear();
+	
+	return;	
+      }    
+    
+      BufferDispatcher_args<T> args;
+      Utilities m_util;
+    
+    
+    };
+    
+}
+
+#endif

src/DataModelBase/DataModelBase.h

@@ -7,6 +7,7 @@
 #include "Utilities.h"
 #include "WorkerPoolManager.h"
 #include "Pool.h"
+#include "BufferDispatcher.h"
 
 #include "Store.h"
 #include "BStore.h"