First Implementation of Pipeline

(1) Changes to DFGraph.h for identifying streaming graphs and edges
(2) Added/modified launch, wait, push, pop intrinsics for streaming graphs
(3) Changes to DFG2LLVM_X86.cpp and DFG2LLVM.h pass to support code gen for streaming
(4) BuildDFG.cpp modified to create streaming edges
(5) Modified visc-rt.cpp and visc-rt.h to add runtime support for streaming

llvm/include/llvm/IR/DFGraph.h

@@ -135,6 +135,8 @@ public:
     return Parent;
   }
 
+  // Child graph is streaming if any of the edges in the edge list is streaming
+  bool isStreaming();
 };
 
 // DFNode represents a single VISC Dataflow Node in LLVM.
@@ -389,6 +391,10 @@ public:
     return childGraph;
   }
 
+  bool isChildGraphStreaming() {
+    return childGraph->isStreaming();
+  }
+
   void applyDFNodeVisitor(DFNodeVisitor &V); /*virtual*/
 //  void applyDFEdgeVisitor(DFEdgeVisitor &V); /*virtual*/
 };
@@ -448,23 +454,24 @@ private:
                                 ///< DFnode
   unsigned DestPosition;        ///< Position of data in the input of
                                 ///< destination DFnode
-  Type* ArgType;
+  Type* ArgType;                ///< Type of the argument
+  bool isStreaming;             ///< Is this an streaming edge
 
   // Functions
   DFEdge(DFNode* _SrcDF, DFNode* _DestDF, bool _EdgeType,
-         unsigned _SourcePosition, unsigned _DestPosition, Type* _ArgType)
+         unsigned _SourcePosition, unsigned _DestPosition, Type* _ArgType, bool _isStreaming)
        : SrcDF(_SrcDF), DestDF(_DestDF), EdgeType(_EdgeType),
          SourcePosition(_SourcePosition), DestPosition(_DestPosition),
-         ArgType(_ArgType) {}
+         ArgType(_ArgType), isStreaming(_isStreaming) {}
 
 public:
   //TODO: Decide whether we need this type
 //  typedef enum {ONE_TO_ONE = false, ALL_TO_ALL} DFEdgeType;
 
   static DFEdge *Create(DFNode* SrcDF, DFNode* DestDF, bool EdgeType,
-                        unsigned SourcePosition, unsigned DestPosition, Type* ArgType) {
+                        unsigned SourcePosition, unsigned DestPosition, Type* ArgType, bool isStreaming = false) {
     return new DFEdge(SrcDF, DestDF, EdgeType, SourcePosition, DestPosition,
-                      ArgType);
+                      ArgType, isStreaming);
 
   }
 
@@ -492,6 +499,10 @@ public:
     return ArgType;
   }
 
+  bool isStreamingEdge() {
+    return isStreaming;
+  }
+
 };
 
 
@@ -513,6 +524,14 @@ bool DFGraph::compareRank(DFNode* A, DFNode* B) {
   return A->getRank() < B->getRank();
 }
 
+bool DFGraph::isStreaming() {
+  for (auto E: DFEdgeList) {
+    if(E->isStreamingEdge())
+      return true;
+  }
+  return false;
+}
+
 //===--------------------- DFNode Outlined Functions --------------===//
 DFNode::DFNode(IntrinsicInst* _II, Function* _FuncPointer, visc::Target _Hint,
     DFInternalNode* _Parent, unsigned _NumOfDim, std::vector<Value*> _DimLimits,

llvm/include/llvm/IR/IntrinsicsVISC.td

@@ -22,11 +22,21 @@ let TargetPrefix = "visc" in {
    */
   def int_visc_init : Intrinsic<[], [], []>;
 
-  /* Launch intrinsic -
-   * i8* llvm.visc.launch(i8* , ArgList*);
+  /* Launch intrinsic - with streaming argument
+   * i8* llvm.visc.launch(i8*, ArgList*, i1);
    */
   def int_visc_launch : Intrinsic<[llvm_ptr_ty], [llvm_ptr_ty,
-                                  llvm_ptr_ty], []>;
+                                  llvm_ptr_ty, llvm_i1_ty], []>;
+
+  /* Push intrinsic - push data on streaming pipeline
+   * void llvm.visc.push(i8*, ArgList*);
+   */
+  def int_visc_push : Intrinsic<[], [llvm_ptr_ty, llvm_ptr_ty], []>;
+
+  /* Pop intrinsic - pop data from streaming pipeline
+   * i8* llvm.visc.pop(i8*);
+   */
+  def int_visc_pop : Intrinsic<[llvm_ptr_ty], [llvm_ptr_ty], []>;
 
   /* Cleanup intrinsic -
    * void llvm.visc.cleanup(i8*);
@@ -78,23 +88,24 @@ let TargetPrefix = "visc" in {
                                         []>;
 
   /* Create dataflow edge intrinsic -
-   * i8* llvm.visc.createEdge(i8*, i8*, i1, i32, i32);
+   * i8* llvm.visc.createEdge(i8*, i8*, i1, i32, i32, i1);
    */
   def int_visc_createEdge : Intrinsic<[llvm_ptr_ty], [llvm_ptr_ty, llvm_ptr_ty,
-                                      llvm_i1_ty, llvm_i32_ty, llvm_i32_ty],
+                                      llvm_i1_ty, llvm_i32_ty, llvm_i32_ty,
+                                      llvm_i1_ty],
                                       []>;
 
   /* Create bind input intrinsic -
    * void llvm.visc.bind.input(i8*, i32, i32);
    */
   def int_visc_bind_input : Intrinsic<[], [llvm_ptr_ty, llvm_i32_ty,
-                                      llvm_i32_ty], []>;
+                                      llvm_i32_ty, llvm_i1_ty], []>;
 
   /* Create bind output intrinsic -
    * void llvm.visc.bind.output(i8*, i32, i32);
    */
   def int_visc_bind_output : Intrinsic<[], [llvm_ptr_ty, llvm_i32_ty,
-                                       llvm_i32_ty], []>;
+                                       llvm_i32_ty, llvm_i1_ty], []>;
 
   /* Find associated dataflow node intrinsic -
    * i8* llvm.visc.getNode();

llvm/include/llvm/SupportVISC/DFG2LLVM.h

@@ -82,7 +82,10 @@ protected:
 
   // Functions
   Value* getStringPointer(const Twine& S, Instruction* InsertBefore, const Twine& Name = "");
+  void addArgument(Function*, Type*, const Twine& Name = "");
   void addIdxDimArgs(Function* F);
+  std::vector<Value*> extractElements(Value*, std::vector<Type*>,
+      std::vector<std::string>, Instruction*);
   Argument* getArgumentAt(Function* F, unsigned offset);
   void initTimerAPI();
 
@@ -148,13 +151,10 @@ Value* CodeGenTraversal::getStringPointer(const Twine& S, Instruction* IB, const
   return SPtr;
 }
 
-// Change the argument list of function F to add index and limit arguments
-void CodeGenTraversal::addIdxDimArgs(Function* F) {
-  // Add Index and Dim arguments
-  std::string names[] = {"idx_x", "idx_y", "idx_z", "dim_x", "dim_y", "dim_z"};
-  for (int i = 0; i < 6; ++i) {
-    new Argument(Type::getInt32Ty(F->getContext()), names[i], F);
-  }
+// Add an argument of type Ty to the given function F
+void CodeGenTraversal::addArgument(Function* F, Type* Ty, const Twine& name) {
+  // Add the argument to argument list
+  new Argument(Ty, name, F);
 
   // Create the argument type list with added argument types
   std::vector<Type*> ArgTypes;
@@ -172,6 +172,56 @@ void CodeGenTraversal::addIdxDimArgs(Function* F) {
   F->mutateType(PTy);
 }
 
+// Change the argument list of function F to add index and limit arguments
+void CodeGenTraversal::addIdxDimArgs(Function* F) {
+  // Add Index and Dim arguments
+  std::string names[] = {"idx_x", "idx_y", "idx_z", "dim_x", "dim_y", "dim_z"};
+  for (int i = 0; i < 6; ++i) {
+    addArgument(F, Type::getInt32Ty(F->getContext()), names[i]);
+  }
+}
+
+// Extract elements from an aggregate value. TyList contains the type of each
+// element, and names vector contains a name. IB is the instruction before which
+// all the generated code would be inserted.
+std::vector<Value*> CodeGenTraversal::extractElements(Value* Aggregate,
+    std::vector<Type*> TyList, std::vector<std::string> names, Instruction* IB) {
+  // Extract input data from i8* Aggregate.addr and store them in a vector.
+  // For each argument
+  std::vector<Value*> Elements;
+  GetElementPtrInst* GEP;
+  unsigned argNum = 0;
+  for(Type* Ty: TyList) {
+    // BitCast: %arg.addr = bitcast i8* Aggregate.addr to <pointer-to-argType>
+    CastInst* BI = BitCastInst::CreatePointerCast(Aggregate,
+                   Ty->getPointerTo(),
+                   names[argNum]+".addr",
+                   IB);
+    // Load: %arg = load <pointer-to-argType> %arg.addr
+    LoadInst* LI = new LoadInst(BI, names[argNum], IB);
+    // Patch argument to call instruction
+    Elements.push_back(LI);
+    //errs() << "Pushing element " << *LI << "\n";
+    //CI->setArgOperand(argNum, LI);
+
+    // TODO: Minor Optimization - The last GEP statement can/should be left out
+    // as no more arguments left
+    // Increment using GEP: %nextArg = getelementptr <ptr-to-argType> %arg.addr, i64 1
+    // This essentially takes us to the next argument in memory
+    Constant* IntOne = ConstantInt::get(Type::getInt64Ty(M.getContext()), 1);
+    if (argNum < TyList.size()-1)
+      GEP = GetElementPtrInst::Create(BI,
+                                                        ArrayRef<Value*>(IntOne),
+                                                        "nextArg",
+                                                        IB);
+    // Increment argNum and for the next iteration use result of this GEP to
+    // extract next argument
+    argNum++;
+    Aggregate = GEP;
+  }
+  return Elements;
+}
+
 // Traverse the function F argument list to get argument at offset
 Argument* CodeGenTraversal::getArgumentAt(Function* F, unsigned offset) {
   DEBUG(errs() << "Finding argument " << offset << ":\n");

llvm/lib/Transforms/BuildDFG/BuildDFG.cpp

@@ -202,6 +202,8 @@ void BuildDFG::handleCreateEdge (DFInternalNode* N, IntrinsicInst* II) {
   unsigned SourcePosition = cast<ConstantInt>(II->getOperand(3))->getZExtValue();
   unsigned DestPosition = cast<ConstantInt>(II->getOperand(4))->getZExtValue();
 
+  bool isStreaming = !cast<ConstantInt>(II->getOperand(5))->isZero();
+
   Type *SrcTy, *DestTy;
 
   // Get destination type
@@ -225,7 +227,8 @@ void BuildDFG::handleCreateEdge (DFInternalNode* N, IntrinsicInst* II) {
                                      EdgeType,
                                      SourcePosition,
                                      DestPosition,
-                                     DestTy);
+                                     DestTy,
+                                     isStreaming);
 
   HandleToDFEdgeMap[II] = newDFEdge;
 
@@ -244,6 +247,8 @@ void BuildDFG::handleBindInput(DFInternalNode* N, IntrinsicInst* II) {
   unsigned SourcePosition = cast<ConstantInt>(II->getOperand(1))->getZExtValue();
   unsigned DestPosition = cast<ConstantInt>(II->getOperand(2))->getZExtValue();
 
+  bool isStreaming = !cast<ConstantInt>(II->getOperand(3))->isZero();
+  
   // Get destination type
   FunctionType *FT = DestDF->getFuncPointer()->getFunctionType();
   assert((FT->getNumParams() > DestPosition)
@@ -266,7 +271,8 @@ void BuildDFG::handleBindInput(DFInternalNode* N, IntrinsicInst* II) {
                                      false,
                                      SourcePosition,
                                      DestPosition,
-                                     DestTy);
+                                     DestTy,
+                                     isStreaming);
 
   HandleToDFEdgeMap[II] = newDFEdge;
 
@@ -285,6 +291,8 @@ void BuildDFG::handleBindOutput(DFInternalNode* N, IntrinsicInst* II) {
   unsigned SourcePosition = cast<ConstantInt>(II->getOperand(1))->getZExtValue();
   unsigned DestPosition = cast<ConstantInt>(II->getOperand(2))->getZExtValue();
 
+  bool isStreaming = !cast<ConstantInt>(II->getOperand(3))->isZero();
+  
   // Get destination type
   StructType* OutTy = DestDF->getOutputType();
   assert((OutTy->getNumElements() > DestPosition)
@@ -307,7 +315,8 @@ void BuildDFG::handleBindOutput(DFInternalNode* N, IntrinsicInst* II) {
                                      false,
                                      SourcePosition,
                                      DestPosition,
-                                     DestTy);
+                                     DestTy,
+                                     isStreaming);
 
   HandleToDFEdgeMap[II] = newDFEdge;
 

llvm/projects/visc-rt/visc-rt.cpp

@@ -23,6 +23,17 @@ using namespace std;
 
 typedef struct {
   pthread_t threadID;
+  std::vector<pthread_t> threads;
+  //unsigned numInBuffers, numOutBuffers;
+  //void (* PushFunc)(void*, unsigned);
+  //void (* PopFunc)(void*, unsigned);
+  std::vector<uint64_t> BindInSizes;
+  std::vector<uint64_t> BindOutSizes;
+  std::vector<uint64_t> EdgeSizes;
+  std::vector<CircularBuffer<uint64_t>*> BindInputBuffers;
+  std::vector<CircularBuffer<uint64_t>*> BindOutputBuffers;
+  std::vector<CircularBuffer<uint64_t>*> EdgeBuffers;
+  std::vector<CircularBuffer<uint64_t>*> isLastInputBuffers;
 } DFNodeContext_X86;
 
 typedef struct {
@@ -51,7 +62,7 @@ static inline void checkErr(cl_int err, cl_int success, const char * name) {
 
 /************************* Depth Stack Routines ***************************/
 
-void llvm_visc_x86_push(unsigned n, unsigned limitX, unsigned iX, unsigned limitY,
+void llvm_visc_x86_dstack_push(unsigned n, unsigned limitX, unsigned iX, unsigned limitY,
     unsigned iY, unsigned limitZ, unsigned iZ) {
   DEBUG(cout << "Pushing node information on stack:\n");
   DEBUG(cout << "\tNumDim = " << n << "\t Limit(" << limitX << ", " << limitY << ", "<< limitZ <<")\n");
@@ -61,7 +72,7 @@ void llvm_visc_x86_push(unsigned n, unsigned limitX, unsigned iX, unsigned limit
   DEBUG(cout << "DStack size = " << DStack.size() << "\n");
 }
 
-void llvm_visc_x86_pop() {
+void llvm_visc_x86_dstack_pop() {
   DEBUG(cout << "Popping from depth stack\n");
   DStack.pop_back();
   DEBUG(cout << "DStack size = " << DStack.size() << "\n");
@@ -949,14 +960,153 @@ void visc_DestroyTimerSet(struct visc_TimerSet * timers)
   }
 }
 
+/**************************** Pipeline API ************************************/
+#define BUFFER_SIZE 1
+
+// Launch API for a streaming dataflow graph
+void* llvm_visc_streamLaunch(void(*LaunchFunc)(void*, void*), void* args) {
+  DFNodeContext_X86* Context = (DFNodeContext_X86*) malloc(sizeof(DFNodeContext_X86));
+  LaunchFunc(args, Context);
+  return Context;
+}
+
+// Push API for a streaming dataflow graph
+void llvm_visc_streamPush(void* graphID, void* args) {
+  DEBUG(cout << "StreamPush -- Graph: " << graphID << ", Arguments: " << args << "\n");
+  DFNodeContext_X86* Ctx = (DFNodeContext_X86*) graphID;
+  unsigned offset = 0;
+  for (unsigned i=0; i< Ctx->BindInputBuffers.size(); i++) {
+    uint64_t element;
+    memcpy(&element, (char*)args+offset, Ctx->BindInSizes[i]);
+    offset += Ctx->BindInSizes[i];
+    DEBUG(cout << "Pushing Value " << element << " to buffer\n");
+    llvm_visc_bufferPush(&Ctx->BindInputBuffers[i], element);
+  }
+  // Push 0 in isLastInput buffers of all child nodes
+  for (auto buffer: Ctx->isLastInputBuffers)
+    llvm_visc_bufferPush(&buffer, 0);
+}
+
+// Pop API for a streaming dataflow graph
+void* llvm_visc_streamPop(void* graphID) {
+  DEBUG(cout << "StreamPop -- Graph: " << graphID << "\n");
+  DFNodeContext_X86* Ctx = (DFNodeContext_X86*) graphID;
+  unsigned totalBytes = 0;
+  for(auto size:  Ctx->BindOutSizes)
+    totalBytes+= size;
+  void* output = malloc(totalBytes);
+  unsigned offset = 0;
+  for (unsigned i=0; i< Ctx->BindOutputBuffers.size(); i++) {
+    uint64_t element = llvm_visc_bufferPop(Ctx->BindOutputBuffers[i]);
+    DEBUG(cout << "Popped Value " << element << " from buffer\n");
+    memcpy((char*)output+offset, &element, Ctx->BindOutSizes[i]);
+    offset += Ctx->BindOutSizes[i];
+  }
+  return output;
+}
+
+// Wait API for a streaming dataflow graph
+void llvm_visc_streamWait(void* graphID) {
+  DEBUG(cout << "StreamWait -- Graph: " << graphID << ", Arguments: " << args << "\n");
+  DFNodeContext_X86* Ctx = (DFNodeContext_X86*) graphID;
+  // Push garbage to all other input buffers
+  for (unsigned i=0; i< Ctx->BindInputBuffers.size(); i++) {
+    uint64_t element = 0;
+    DEBUG(cout << "Pushing Value " << element << " to buffer\n");
+    llvm_visc_bufferPush(&Ctx->BindInputBuffers[i], element);
+  }
+  // Push 1 in isLastInput buffers of all child nodes
+  for (unsigned i=0; i < Ctx->isLastInputBuffers.size(); i++)
+    llvm_visc_bufferPush(&Ctx->isLastInputBuffers[i], 1);
+
+  llvm_visc_freeThreads(graphID);
+}
+
+// Create a buffer and return the bufferID
+void* llvm_visc_createBindInBuffer(void* graphID, uint64_t size) {
+  DEBUG(cout << "Create BindInBuffer -- Graph: " << graphID << ", Size: " << size << "\n");
+  CircularBuffer<uint64_t> *bufferID = new CircularBuffer<uint64_t>(BUFFER_SIZE);
+  DFNodeContext_X86* Context = (DFNodeContext_X86*) graphID;
+  Context->BindInputBuffers.push_back(bufferID);
+  Context->BindInSizes.push_back(size);
+  return bufferID;
+}
+
+void* llvm_visc_createBindOutBuffer(void* graphID, uint64_t size) {
+  DEBUG(cout << "Create BindOutBuffer -- Graph: " << graphID << ", Size: " << size << "\n");
+  CircularBuffer<uint64_t> *bufferID = new CircularBuffer<uint64_t>(BUFFER_SIZE);
+  DFNodeContext_X86* Context = (DFNodeContext_X86*) graphID;
+  Context->BindOutputBuffers.push_back(bufferID);
+  Context->BindOutSizes.push_back(size);
+  return bufferID;
+}
+void* llvm_visc_createEdgeBuffer(void* graphID, uint64_t size) {
+  DEBUG(cout << "Create EdgeBuffer -- Graph: " << graphID << ", Size: " << size << "\n");
+  CircularBuffer<uint64_t> *bufferID = new CircularBuffer<uint64_t>(BUFFER_SIZE);
+  DFNodeContext_X86* Context = (DFNodeContext_X86*) graphID;
+  Context->EdgeBuffers.push_back(bufferID);
+  Context->EdgeSizes.push_back(size);
+  return bufferID;
+}
+
+void* llvm_visc_createLastInputBuffer(void* graphID, uint64_t size) {
+  DEBUG(cout << "Create isLastInputBuffer -- Graph: " << graphID << ", Size: " << size << "\n");
+  CircularBuffer<uint64_t> *bufferID = new CircularBuffer<uint64_t>(BUFFER_SIZE);
+  DFNodeContext_X86* Context = (DFNodeContext_X86*) graphID;
+  Context->isLastInputBuffers.push_back(bufferID);
+  return bufferID;
+}
+
+// Free buffers 
+void llvm_visc_freeBuffers(void* graphID) {
+  DFNodeContext_X86* Context = (DFNodeContext_X86*) graphID;
+  for(CircularBuffer<uint64_t>* bufferID: Context->BindInputBuffers)
+    delete bufferID;
+  for(CircularBuffer<uint64_t>* bufferID: Context->BindOutputBuffers)
+    delete bufferID;
+  for(CircularBuffer<uint64_t>* bufferID: Context->EdgeBuffers)
+    delete bufferID;
+  for(CircularBuffer<uint64_t>* bufferID: Context->isLastInputBuffers)
+    delete bufferID;
+}
+
+// Pop an element from the buffer
+uint64_t llvm_visc_bufferPop(void* bufferID) {
+  CircularBuffer<uint64_t>* buffer = (CircularBuffer<uint64_t>*) bufferID;
+  return buffer->pop();
+}
+
+// Push an element into the buffer
+void llvm_visc_bufferPush(void* bufferID, uint64_t element) {
+  CircularBuffer<uint64_t>* buffer = (CircularBuffer<uint64_t>*) bufferID;
+  buffer->push(element);
+}
+
+// Create a thread
+void llvm_visc_createThread(void* graphID, void* (*Func)(void*), void* arguments) {
+  DFNodeContext_X86* Ctx = (DFNodeContext_X86*) graphID;
+  int err;
+  pthread_t threadID;
+  if((err = pthread_create(&threadID, NULL, Func, arguments)) != 0)
+    cout << "Failed to create thread. Error code = " << err << "\n";
+
+  Ctx->threads.push_back(threadID);
+}
+
+// Wait for thread to finish
+void llvm_visc_freeThreads(void* graphID) {
+  DFNodeContext_X86* Ctx = (DFNodeContext_X86*) graphID;
+  for(pthread_t thread: Ctx->threads)
+    pthread_join(thread, NULL);
+}
 
 /************************ OPENCL & PTHREAD API ********************************/
 
 void* llvm_visc_x86_launch(void* (*rootFunc)(void*), void* arguments) {
-  DFNodeContext_X86 *Context = (DFNodeContext_X86 *) malloc(sizeof(DFNodeContext_X86));
+  DFNodeContext_X86 *Context = (DFNodeContext_X86*) malloc(sizeof(DFNodeContext_X86));
   int err;
   if((err = pthread_create(&Context->threadID, NULL, rootFunc, arguments)) != 0)
-  cout << "Failed to create pthread. Error code = " << err << "\n";
+    cout << "Failed to create pthread. Error code = " << err << "\n";
   //rootFunc(arguments);
   return Context;
 }

llvm/projects/visc-rt/visc-rt.h

@@ -8,6 +8,10 @@
 #include <iostream>
 #include <map>
 #include <ctime>
+#include <vector>
+#include <pthread.h>
+//#include <condition_variable>
+
 #include "llvm/Support/CommandLine.h"
 #include "llvm/SupportVISC/VISCHint.h"
 #include "llvm/SupportVISC/VISCTimer.h"
@@ -50,26 +54,12 @@ class DFGDepth {
     }
 };
 
-void llvm_visc_x86_push(unsigned n, unsigned limitX = 0, unsigned iX = 0,
+void llvm_visc_x86_dstack_push(unsigned n, unsigned limitX = 0, unsigned iX = 0,
     unsigned limitY = 0, unsigned iY = 0, unsigned limitZ = 0, unsigned iZ = 0);
-void llvm_visc_x86_pop();
+void llvm_visc_x86_dstack_pop();
 unsigned llvm_visc_x86_getDimLimit(unsigned level, unsigned dim);
 unsigned llvm_visc_x86_getDimInstance(unsigned level, unsigned dim);
 
-//class DFGDepthStack {
-  //private:
-    //vector<DFGDepth> Stack;
-  //public:
-    //DFGDepthStack() {
-    //}
-
-    //void push(DFGDepth D) {
-      //Stack.push_back(D);
-    //}
-    //void pop() {
-      //Stack.pop_back();
-    //}
-//};
 
 /********************* Memory Tracker **********************************/
 class MemTrackerEntry {
@@ -175,4 +165,103 @@ void llvm_visc_printTimerSet(void** timerSet, char* timerName = NULL);
 void* llvm_visc_initializeTimerSet();
 
 }
+
+/*************************** Pipeline API ******************************/
+// Circular Buffer class
+template <class ElementType>
+class CircularBuffer {
+private:
+    int numElements;
+    int bufferSize;
+    int Head;
+    int Tail;
+    pthread_mutex_t mtx;
+    pthread_cond_t cv;
+    vector<ElementType> buffer;
+
+public:
+    CircularBuffer(int maxElements) {
+        Head = 0;
+        Tail = 0;
+        numElements = 0;
+        bufferSize = maxElements+1;
+        buffer.reserve(bufferSize);
+        pthread_mutex_init(&mtx, NULL);
+        pthread_cond_init(&cv, NULL);
+    }
+
+    bool push(ElementType E);
+    ElementType pop();
+
+};
+
+template <class ElementType>
+bool CircularBuffer<ElementType>::push(ElementType E) {
+    //unique_lock<mutex> lk(mtx);
+    pthread_mutex_lock(&mtx);
+    if((Head +1) % bufferSize == Tail) {
+        //cv.wait(lk);
+        pthread_cond_wait(&cv, &mtx);
+    }
+    buffer[Head] = E;
+    Head = (Head+1) % bufferSize;
+    numElements++;
+    //DEBUG(cout << "[" << caller << "]: " << name << " Total Elements = " << numElements << "\n");
+    //lk.unlock();
+    pthread_mutex_unlock(&mtx);
+    //cv.notify_one();
+    pthread_cond_signal(&cv);
+    return true;
+}
+
+template <class ElementType>
+ElementType CircularBuffer<ElementType>::pop() {
+    //unique_lock<mutex> lk(mtx);
+    pthread_mutex_lock(&mtx);
+    if(Tail == Head) {
+        //DEBUG(cout << "[C] Going to sleep ...\n");
+        //cv.wait(lk);
+        pthread_cond_wait(&cv, &mtx);
+        //DEBUG(cout << "[C] Waking up\n");
+    }
+    ElementType E = buffer[Tail];
+    Tail = (Tail + 1) % bufferSize;
+    numElements--;
+    //DEBUG(cout << "[" << caller << "]: " << name << " Total Elements = " << numElements << "\n");
+    //lk.unlock();
+    pthread_mutex_unlock(&mtx);
+    //cv.notify_one();
+    pthread_cond_signal(&cv);
+    return E;
+}
+
+extern "C" {
+// Functions to push and pop values from pipeline buffers
+uint64_t llvm_visc_bufferPop(void*);
+void llvm_visc_bufferPush(void*, uint64_t);
+
+// Functions to create and destroy buffers
+void* llvm_visc_createBindInBuffer(void*, uint64_t);
+void* llvm_visc_createBindOutBuffer(void*, uint64_t);
+void* llvm_visc_createEdgeBuffer(void*, uint64_t);
+void* llvm_visc_createLastInputBuffer(void*, uint64_t);
+
+void llvm_visc_freeBuffers(void*);
+
+// Functions to create and destroy threads
+void llvm_visc_createThread(void* graphID, void*(*Func)(void*), void*);
+void llvm_visc_freeThreads(void*);
+
+// Launch API for a streaming graph.
+// Arguments:
+// (1) Launch Function: void* (void*, void*)
+// (2) Push Function:   void (void*, std::vector<uint64_t>**, unsgined)
+// (3) Pop Function:    void* (std::vector<uint64_t>**, unsigned)
+void* llvm_visc_streamLaunch(void(*LaunchFunc)(void*, void*), void*);
+void llvm_visc_streamPush(void* graphID, void* args);
+void* llvm_visc_streamPop(void* graphID);
+void llvm_visc_streamWait(void* graphID);
+
+}
+
 #endif //VISC_RT_HEADER