private void StartWorking()
	{
//		try {
//			Thread.sleep((this.joinId+1)*5000);
//		} catch (InterruptedException e1) {
//			// TODO Auto-generated catch block
//			e1.printStackTrace();
//		}
		while (true)
		{
			try 
			{
				
				//DEBUG
				//WriteWorkingTupleToFile (tupleToProcess);
				//System.out.println("There is "+workQueue.size()+" tuples in working quee of join "+joinId);
				
				//Eddy is terminated
				if (this.eddyThreadState==ThreadStatus.Termintaed)
					break;
				
				//Queue is empty report to eddy that you are ready to terminate if you have not reported yet. 
				if ((this.workQueue.isEmpty()) && !(this.stateReportedToEddy) && (this.eddyThreadState==ThreadStatus.ReadyToTerminate))
					ReportReadyToTerminate() ;
				
				Tuple tupleToProcess =workQueue.take() ;
				if (tupleToProcess instanceof SyncPoisonPill)
					ProcessPoison((SyncPoisonPill)tupleToProcess);
				else
					ProcessTuple(tupleToProcess);
				
			} 
			catch (InterruptedException e) 
			{
				throw new OptimizerException(e) ;
			}
			
		}
		myLogger.printInfo("Join" +Thread.currentThread().getName()+ " Finished.");
		//System.out.println("Join" +Thread.currentThread().getName()+ " Finished.");
		myLogger.printInfo("Join "+this.joinId+" Count Results Tuples= "+this.coutResultTuples);
		//System.out.println("Count Results Tuples= "+this.coutResultTuples);
		myLogger.printInfo("Join" +Thread.currentThread().getName()+" "+ this.countReport+" times report to eddy");
		//System.out.println("Join" +Thread.currentThread().getName()+" "+ this.countReport+" times report to eddy" );
	}