ALIROOT-5433 Transition to CDHv3 in HLT
[u/mrichter/AliRoot.git] / HLT / BASE / AliHLTTask.cxx
1 // $Id$
2 //**************************************************************************
3 //* This file is property of and copyright by the                          * 
4 //* ALICE Experiment at CERN, All rights reserved.                         *
5 //*                                                                        *
6 //* Primary Authors: Matthias Richter <Matthias.Richter@ift.uib.no>        *
7 //*                  for The ALICE HLT Project.                            *
8 //*                                                                        *
9 //* Permission to use, copy, modify and distribute this software and its   *
10 //* documentation strictly for non-commercial purposes is hereby granted   *
11 //* without fee, provided that the above copyright notice appears in all   *
12 //* copies and that both the copyright notice and this permission notice   *
13 //* appear in the supporting documentation. The authors make no claims     *
14 //* about the suitability of this software for any purpose. It is          *
15 //* provided "as is" without express or implied warranty.                  *
16 //**************************************************************************
17
18 /// @file   AliHLTTask.cxx
19 /// @author Matthias Richter
20 /// @date   
21 /// @brief  Implementation of HLT tasks.
22 ///
23
24 #include <cerrno>
25 #include <cassert>
26 #include <iostream>
27 #include <string>
28 #include <ctime>
29 #include "AliHLTTask.h"
30 #include "AliHLTConfiguration.h"
31 #include "AliHLTConfigurationHandler.h"
32 #include "AliHLTComponent.h"
33 #include "AliHLTComponentHandler.h"
34 #include "TList.h"
35 #include "AliHLTErrorGuard.h"
36
37 using std::cout;
38
39 /** ROOT macro for the implementation of ROOT specific class methods */
40 ClassImp(AliHLTTask)
41
42 AliHLTTask::AliHLTTask()
43   :
44   fpConfiguration(NULL),
45   fpComponent(NULL),
46   fpDataBuffer(NULL),
47   fListTargets(),
48   fListDependencies(),
49   fBlockDataArray()
50 {
51   // see header file for class documentation
52   // or
53   // refer to README to build package
54   // or
55   // visit http://web.ift.uib.no/~kjeks/doc/alice-hlt
56 }
57
58 AliHLTTask::AliHLTTask(AliHLTConfiguration* pConf)
59   :
60   fpConfiguration(pConf),
61   fpComponent(NULL),
62   fpDataBuffer(NULL),
63   fListTargets(),
64   fListDependencies(),
65   fBlockDataArray()
66 {
67   // see header file for function documentation
68 }
69
70 AliHLTTask::~AliHLTTask()
71 {
72   // see header file for function documentation
73   TObjLink* lnk=fListDependencies.FirstLink();
74
75   while (lnk!=NULL) {
76     AliHLTTask* pTask=(AliHLTTask*)lnk->GetObject();
77     pTask->UnsetTarget(this);
78     lnk=lnk->Next();
79   }
80   lnk=fListTargets.FirstLink();
81
82   while (lnk!=NULL) {
83     AliHLTTask* pTask=(AliHLTTask*)lnk->GetObject();
84     pTask->UnsetDependency(this);
85     lnk=lnk->Next();
86   }
87
88   if (fpComponent) delete fpComponent;
89   fpComponent=NULL;
90 }
91
92 int AliHLTTask::Init(AliHLTConfiguration* pConf, AliHLTComponentHandler* pCH)
93 {
94   // see header file for function documentation
95   int iResult=0;
96   if (fpConfiguration!=NULL && pConf!=NULL && fpConfiguration!=pConf) {
97     HLTWarning("overriding existing reference to configuration object %p by %p",
98                fpConfiguration, pConf);
99   }
100   if (pConf!=NULL) fpConfiguration=pConf;
101   iResult=CreateComponent(fpConfiguration, pCH, fpComponent);
102   if (iResult>=0) {
103     iResult=CustomInit(pCH);
104   }
105   return iResult;
106 }
107
108 int AliHLTTask::CreateComponent(AliHLTConfiguration* pConfiguration, AliHLTComponentHandler* pCH, AliHLTComponent*& pComponent) const
109 {
110   // see header file for class documentation
111   int iResult=0;
112   if (!pConfiguration) return -EINVAL;
113
114   const AliHLTConfiguration* pConf=AliHLTConfigurationHandler::FindSubstitution(*pConfiguration);
115   if (!pConf) pConf=pConfiguration;
116   if (pConf) {
117     if (pCH) {
118       int argc=0;
119       const char** argv=NULL;
120       if ((iResult=pConf->GetArguments(&argv))>=0) {
121         argc=iResult; // just to make it clear
122         // TODO: we have to think about the optional environment parameter,
123         // currently just set to NULL.
124         iResult=pCH->CreateComponent(pConf->GetComponentID(), pComponent);
125         if (pComponent && iResult>=0) {
126           TString description;
127           description.Form("chainid=%s", GetName());
128           pComponent->SetComponentDescription(description.Data());
129           const AliHLTAnalysisEnvironment* pEnv=pCH->GetEnvironment();
130           if ((iResult=pComponent->Init(pEnv, NULL, argc, argv))>=0) {
131             //HLTDebug("component %s (%p) created", pComponent->GetComponentID(), pComponent); 
132           } else {
133             HLTError("Initialization of component \"%s\" failed with error %d", pComponent->GetComponentID(), iResult);
134           }
135         } else {
136           //HLTError("can not find component \"%s\" (%d)", pConf->GetComponentID(), iResult);
137         }
138       } else {
139         HLTError("can not get argument list for configuration %s (%s)", pConf->GetName(), pConf->GetComponentID());
140         iResult=-EINVAL;
141       }
142     } else {
143       HLTError("component handler instance needed for task initialization");
144       iResult=-EINVAL;
145     }
146   } else {
147     HLTError("configuration object instance needed for task initialization");
148     iResult=-EINVAL;
149   }
150   return iResult;
151 }
152
153 int AliHLTTask::Deinit()
154 {
155   // see header file for function documentation
156   int iResult=0;
157   CustomCleanup();
158   AliHLTComponent* pComponent=GetComponent();
159   fpComponent=NULL;
160   if (pComponent) {
161     //HLTDebug("delete component %s (%p)", pComponent->GetComponentID(), pComponent); 
162     pComponent->Deinit();
163     delete pComponent;
164   } else {
165     HLTWarning("task doesn't seem to be in initialized");
166   }
167   return iResult;
168 }
169
170 const char *AliHLTTask::GetName() const
171 {
172   // see header file for function documentation
173   if (fpConfiguration)
174     return fpConfiguration->GetName();
175   return TObject::GetName();
176 }
177
178 AliHLTConfiguration* AliHLTTask::GetConf() const
179 {
180   // see header file for function documentation
181   return fpConfiguration;
182 }
183
184 AliHLTComponent* AliHLTTask::GetComponent() const
185 {
186   // see header file for function documentation
187   return fpComponent;
188 }
189
190 AliHLTTask* AliHLTTask::FindDependency(const char* id)
191 {
192   // see header file for function documentation
193   AliHLTTask* pTask=NULL;
194   if (id) {
195     pTask=(AliHLTTask*)fListDependencies.FindObject(id);
196   }
197   return pTask;
198 }
199
200 int AliHLTTask::FollowDependency(const char* id, TList* pTgtList)
201 {
202   // see header file for function documentation
203   int iResult=0;
204   if (id) {
205     AliHLTTask* pDep=NULL;
206     if ((pDep=(AliHLTTask*)fListDependencies.FindObject(id))!=NULL) {
207       if (pTgtList) pTgtList->Add(pDep);
208       iResult++;
209     } else {
210       TObjLink* lnk=fListDependencies.FirstLink();
211       while (lnk && iResult==0) {
212         pDep=(AliHLTTask*)lnk->GetObject();
213         if (pDep) {
214           if ((iResult=pDep->FollowDependency(id, pTgtList))>0) {
215             if (pTgtList) pTgtList->AddFirst(pDep);
216             iResult++;
217           }
218         } else {
219           iResult=-EFAULT;
220         }
221         lnk=lnk->Next();
222       }
223     }
224   } else {
225     iResult=-EINVAL;
226   }
227   return iResult;
228 }
229
230 void AliHLTTask::PrintDependencyTree(const char* id, int bFromConfiguration)
231 {
232   // see header file for function documentation
233   HLTLogKeyword("task dependencies");
234   int iResult=0;
235   TList tgtList;
236   if (bFromConfiguration) {
237     if (fpConfiguration)
238       iResult=fpConfiguration->FollowDependency(id, &tgtList);
239     else
240       iResult=-EFAULT;
241   } else
242     iResult=FollowDependency(id, &tgtList);
243   if (iResult>0) {
244     HLTMessage("     dependency level %d ", iResult);
245     TObjLink* lnk=tgtList.FirstLink();
246     int i=iResult;
247     char* pSpace = new char[iResult+1];
248     if (pSpace) {
249       memset(pSpace, 32, iResult);
250       pSpace[i]=0;
251       while (lnk) {
252         TObject* obj=lnk->GetObject();
253         HLTMessage("     %s^-- %s ", &pSpace[i--], obj->GetName());
254         lnk=lnk->Next();
255       }
256       delete [] pSpace;
257     } else {
258       iResult=-ENOMEM;
259     }
260   }
261 }
262
263 int AliHLTTask::SetDependency(AliHLTTask* pDep)
264 {
265   // see header file for function documentation
266   int iResult=0;
267   if (pDep) {
268     if (FindDependency(pDep->GetName())==NULL) {
269       fListDependencies.Add(pDep);
270     } else {
271       iResult=-EEXIST;
272     }
273   } else {
274     iResult=-EINVAL;
275   }
276   return iResult;
277 }
278
279 int AliHLTTask::UnsetDependency(AliHLTTask* pDep)
280 {
281   // see header file for function documentation
282   fListDependencies.Remove(pDep);
283   if (fpConfiguration) {
284     fpConfiguration->InvalidateSources();
285   }
286   return 0;
287 }
288
289 int AliHLTTask::CheckDependencies()
290 {
291   // see header file for function documentation
292   int iResult=0;
293   AliHLTConfiguration* pSrc=fpConfiguration->GetFirstSource();
294   while (pSrc) {
295     if (FindDependency(pSrc->GetName())==NULL) {
296       //HLTDebug("dependency \"%s\" unresolved", pSrc->GetName());
297       iResult++;
298     }
299     pSrc=fpConfiguration->GetNextSource();
300   }
301   return iResult;
302 }
303
304
305 int AliHLTTask::Depends(AliHLTTask* pTask)
306 {
307   // see header file for function documentation
308   int iResult=0;
309   if (pTask) {
310     if (fpConfiguration) {
311       iResult=fpConfiguration->GetSource(pTask->GetName())!=NULL;
312       if (iResult>0) {
313         //HLTDebug("task  \"%s\" depends on \"%s\"", GetName(), pTask->GetName());
314       } else {
315         //HLTDebug("task  \"%s\" independend of \"%s\"", GetName(), pTask->GetName());
316       }
317     } else {
318       iResult=-EFAULT;
319     }
320   } else {
321     iResult=-EINVAL;
322   }
323   return iResult;
324 }
325
326 AliHLTTask* AliHLTTask::FindTarget(const char* id)
327 {
328   // see header file for function documentation
329   AliHLTTask* pTask=NULL;
330   if (id) {
331     pTask=(AliHLTTask*)fListTargets.FindObject(id);
332   }
333   return pTask;
334 }
335
336 int AliHLTTask::SetTarget(AliHLTTask* pTgt)
337 {
338   // see header file for function documentation
339   int iResult=0;
340   if (pTgt) {
341     if (FindTarget(pTgt->GetName())==NULL) {
342       fListTargets.Add(pTgt);
343     } else {
344       iResult=-EEXIST;
345     }
346   } else {
347     iResult=-EINVAL;
348   }
349   return iResult;
350 }
351
352 int AliHLTTask::UnsetTarget(AliHLTTask* pTarget)
353 {
354   // see header file for function documentation
355   fListTargets.Remove(pTarget);
356   return 0;
357 }
358
359 int AliHLTTask::StartRun()
360 {
361   // see header file for function documentation
362   int iResult=0;
363   int iNofInputDataBlocks=0;
364   AliHLTComponent* pComponent=GetComponent();
365   if (pComponent) {
366     // determine the number of input data blocks provided from the source tasks
367     { // set scope for lnk as a local variable
368     TObjLink* lnk=fListDependencies.FirstLink();
369     while (lnk && iResult>=0) {
370       AliHLTTask* pSrcTask=(AliHLTTask*)lnk->GetObject();
371       if (pSrcTask) {
372         if ((iResult=pSrcTask->GetNofMatchingDataTypes(this))>0) {
373           iNofInputDataBlocks+=iResult;
374         } else if (iResult==0) {
375           HLTWarning("source task %s (%p) does not provide any matching data type for task %s (%p)", pSrcTask->GetName(), pSrcTask, GetName(), this);
376         } else {
377           HLTError("task %s (%p): error getting matching data types for source task %s (%p)", GetName(), this, pSrcTask->GetName(), pSrcTask);
378           iResult=-EFAULT;
379         }
380       }
381       lnk=lnk->Next();
382     }
383     }
384     if (iResult>=0) {
385       if (fBlockDataArray.size()>0) {
386         HLTWarning("block data array for task %s (%p) was not cleaned", GetName(), this);
387         fBlockDataArray.clear();
388       }
389
390       // component init
391       // the initialization of the component is done by the ComponentHandler after creation
392       // of the component.
393       //iResult=Init( AliHLTAnalysisEnvironment* environ, void* environ_param, int argc, const char** argv );
394
395       // allocate the data buffer, which controls the output buffer and subscriptions
396       if (iResult>=0) {
397         fpDataBuffer=new AliHLTDataBuffer;
398         if (fpDataBuffer!=NULL) {
399           fpDataBuffer->SetLocalLoggingLevel(GetLocalLoggingLevel());
400           HLTDebug("created data buffer %p for task %s (%p)", fpDataBuffer, GetName(), this);
401           TObjLink* lnk=fListTargets.FirstLink();
402           while (lnk && iResult>=0) {
403             AliHLTTask* pTgtTask=(AliHLTTask*)lnk->GetObject();
404             if (pTgtTask) {
405               if ((iResult=fpDataBuffer->SetConsumer(pTgtTask->GetComponent()))>=0) {
406               }
407             } else {
408               iResult=-EFAULT;
409               break;
410             }
411             lnk=lnk->Next();
412           }
413         } else {
414           HLTFatal("can not create data buffer object, memory allocation failed");
415           iResult=-ENOMEM;
416         }
417       }
418     }
419     if (iResult>=0) {
420       // send the SOR event
421       
422     }
423   } else {
424     HLTError("task %s (%p) does not have a component", GetName(), this);
425     iResult=-EFAULT;
426   }
427   return iResult;
428 }
429
430 int AliHLTTask::EndRun()
431 {
432   // see header file for function documentation
433   int iResult=0;
434   if (fBlockDataArray.size()>0) {
435     fBlockDataArray.clear();
436   }
437   if (fpDataBuffer) {
438     AliHLTDataBuffer* pBuffer=fpDataBuffer;
439     fpDataBuffer=NULL;
440     delete pBuffer;
441   }
442   return iResult;
443 }
444
445 int AliHLTTask::ProcessTask(Int_t eventNo, AliHLTUInt32_t eventType, AliHLTTriggerMask_t trgMask,
446                             AliHLTUInt32_t timestamp, AliHLTUInt32_t participatingDetectors)
447 {
448   // see header file for function documentation
449   int iResult=0;
450   AliHLTComponent* pComponent=GetComponent();
451   if (pComponent && fpDataBuffer) {
452     HLTDebug("Processing task %s (%p) fpDataBuffer %p", GetName(), this, fpDataBuffer);
453     fpDataBuffer->Reset();
454     int iSourceDataBlock=0;
455     int iInputDataVolume=0;
456
457     AliHLTTask* pSrcTask=NULL;
458     AliHLTTaskPList subscribedTaskList;
459     TObjLink* lnk=fListDependencies.FirstLink();
460
461     // instances of SOR and EOR events to be kept
462     int iSOR=-1;
463     int iEOR=-1;
464     // TODO 2009-09-30
465     // generalize handling of the special blocks to be forwarded on SOR and EOR
466     // just adding a new specific handling for the ECS parameter block as a quick
467     // solution
468     int iECS=-1;
469
470     // subscribe to all source tasks
471     fBlockDataArray.clear();
472     while (lnk && iResult>=0) {
473       pSrcTask=(AliHLTTask*)lnk->GetObject();
474       if (pSrcTask) {
475         int iMatchingDB=pSrcTask->GetNofMatchingDataBlocks(this);
476         if (iMatchingDB<0) {
477           HLTError("task %s (%p): error getting no of matching data blocks from task %s (%p), error %d", GetName(), this, pSrcTask->GetName(), pSrcTask, iMatchingDB);
478           iResult=iMatchingDB;
479           break;
480         } else if (iMatchingDB==0) {
481           HLTDebug("source task %s (%p) does not provide any matching data type for task %s (%p)", pSrcTask->GetName(), pSrcTask, GetName(), this);
482         }
483         if ((iResult=pSrcTask->Subscribe(this, fBlockDataArray))>=0) {
484           iSOR=iEOR=iECS=-1;
485           AliHLTComponentBlockDataList::iterator block=fBlockDataArray.begin();
486           for (int i=0; block!=fBlockDataArray.end(); i++) {
487             bool bRemove=0;
488             bRemove|=(*block).fDataType==kAliHLTDataTypeSOR && !(iSOR<0 && (iSOR=i)>=0);
489             bRemove|=(*block).fDataType==kAliHLTDataTypeEOR && !(iEOR<0 && (iEOR=i)>=0);
490             bRemove|=(*block).fDataType==kAliHLTDataTypeECSParam && !(iECS<0 && (iECS=i)>=0);
491             //HLTInfo("block %d, iSOR=%d iEOR=%d remove=%d", i, iSOR, iEOR, bRemove);
492             if (i<iSourceDataBlock) {
493               assert(!bRemove);
494             } else if (bRemove) {
495               HLTDebug("remove duplicated event %s (%d)", AliHLTComponent::DataType2Text((*block).fDataType).c_str(), i);
496               pSrcTask->Release(&(*block), this);
497               block=fBlockDataArray.erase(block);
498               continue;
499             } else {
500             iInputDataVolume+=(*block).fSize;
501             // put the source task as many times into the list as it provides data blocks
502             // makes the bookkeeping for the data release easier
503             subscribedTaskList.push_back(pSrcTask);
504             }
505             block++;
506           }
507           HLTDebug("Task %s (%p) successfully subscribed to %d data block(s) of task %s (%p)", GetName(), this, iResult, pSrcTask->GetName(), pSrcTask);
508           iSourceDataBlock=fBlockDataArray.size();        
509           iResult=0;
510         } else {
511           HLTError("Task %s (%p): subscription to task %s (%p) failed with error %d", GetName(), this, pSrcTask->GetName(), pSrcTask, iResult);
512           iResult=-EFAULT;
513         }
514       } else {
515         HLTFatal("fatal internal error in ROOT list handling");
516         iResult=-EFAULT;
517       }
518       lnk=lnk->Next();
519     }    
520
521     // process the event
522     int iNofTrial=0; // repeat processing if component returns -ENOSPC
523     AliHLTUInt32_t iLastOutputDataSize=0;
524     if (iResult>=0) {
525     do {
526       long unsigned int iOutputDataSize=0;
527       AliHLTConfiguration* pConf=GetConf();
528       // check if there was a buffer size specified, query output size
529       // estimator from component otherwize
530       if (pConf && pConf->GetOutputBufferSize()>=0) {
531         iOutputDataSize=pConf->GetOutputBufferSize();
532       } else {
533       long unsigned int iConstBase=0;
534       double fInputMultiplier=0;
535       if (pComponent->GetComponentType()!=AliHLTComponent::kSink) {
536         pComponent->GetOutputDataSize(iConstBase, fInputMultiplier);
537         // add a small margin to the buffer to allow optional component
538         // statistics
539         iConstBase+=100;
540 #if defined(__DEBUG) || defined(HLT_COMPONENT_STATISTICS)
541         for (AliHLTComponentBlockDataList::iterator element=fBlockDataArray.begin();
542              element!=fBlockDataArray.end(); element++) {
543           if (element->fDataType==kAliHLTDataTypeComponentStatistics) {
544             iConstBase+=element->fSize;
545           }
546         }
547 #endif
548       }
549       if (fInputMultiplier<0) {
550         HLTWarning("ignoring negative input multiplier");
551         fInputMultiplier=0;
552       }
553       iOutputDataSize=int(fInputMultiplier*iInputDataVolume) + iConstBase;
554       //HLTDebug("task %s: reqired output size %d", GetName(), iOutputDataSize);
555       }
556       if (iNofTrial>0) {
557         // dont process again if the buffer size is the same
558         if (iLastOutputDataSize==iOutputDataSize) break;
559         HLTImportant("processing event %d again with buffer size %d", eventNo, iOutputDataSize);
560       }
561       AliHLTUInt8_t* pTgtBuffer=NULL;
562       if (iOutputDataSize>0) pTgtBuffer=fpDataBuffer->GetTargetBuffer(iOutputDataSize);
563       //HLTDebug("provided raw buffer %p", pTgtBuffer);
564       AliHLTComponentEventData evtData;
565       AliHLTComponent::FillEventData(evtData);
566       if (eventNo>=0)
567         evtData.fEventID=(AliHLTEventID_t)eventNo;
568       if (timestamp < kMaxUInt) evtData.fEventCreation_s=timestamp;
569       else
570       evtData.fEventCreation_s=static_cast<AliHLTUInt32_t>(time(NULL));
571       AliHLTComponentTriggerData trigData;
572       AliHLTEventTriggerData evtTrigData;
573       trigData.fStructSize=sizeof(trigData);
574       trigData.fDataSize=sizeof(AliHLTEventTriggerData);
575       memset(&evtTrigData, 0, trigData.fDataSize);
576       // Setup the CDH in the trigger data, based on the event type, CTP trigger
577       // mask and participating detectors.
578       evtTrigData.fCommonHeaderWordCnt=gkAliHLTCommonHeaderCount;
579       AliHLTUInt8_t l1msg = 0x0;
580       switch (eventType)
581       {
582       case gkAliEventTypeData:        l1msg = 0x00; break;
583       case gkAliEventTypeDataReplay:  l1msg = 0x00; break;
584       case gkAliEventTypeStartOfRun:  l1msg = (0xE << 2) | 0x01; break;
585       case gkAliEventTypeEndOfRun:    l1msg = (0xF << 2) | 0x01; break;
586       case gkAliEventTypeCalibration: l1msg = (0x1 << 6) | 0x01; break;
587       case gkAliEventTypeSoftware:    l1msg = 0x01; break;
588       }
589       evtTrigData.fCommonHeader[1] = AliHLTUInt32_t(l1msg) << 14;
590       evtTrigData.fCommonHeader[3] = ((l1msg & 0x1) == 0x1) ? (participatingDetectors & 0xFFFFFF) : 0x0;
591       evtTrigData.fCommonHeader[5] = (trgMask & AliHLTTriggerMask_t(0xffffffff)).to_ulong();
592       evtTrigData.fCommonHeader[6] = ((trgMask>>32) & AliHLTTriggerMask_t(0x3ffff)).to_ulong();
593       evtTrigData.fCommonHeader[7] = ((trgMask>>50) & AliHLTTriggerMask_t(0xffffffff)).to_ulong();
594       evtTrigData.fCommonHeader[8] = ((trgMask>>72) & AliHLTTriggerMask_t(0x3ffff)).to_ulong();
595
596       trigData.fData=&evtTrigData;
597       iLastOutputDataSize=iOutputDataSize;
598       AliHLTUInt32_t size=iOutputDataSize;
599       AliHLTUInt32_t outputBlockCnt=0;
600       AliHLTComponentBlockData* outputBlocks=NULL;
601       AliHLTComponentEventDoneData* edd=NULL;
602       if (pTgtBuffer!=NULL || iOutputDataSize==0) {
603         // add event type data block
604         // the block is removed immediately after processing from the list
605         AliHLTComponentBlockData eventTypeBlock;
606         AliHLTComponent::FillBlockData(eventTypeBlock);
607         // Note: no payload!
608         eventTypeBlock.fDataType=kAliHLTDataTypeEvent;
609         eventTypeBlock.fSpecification=eventType;
610         fBlockDataArray.push_back(eventTypeBlock);
611
612         if (CheckFilter(kHLTLogDebug)) Print("proc");
613
614         AliHLTUInt32_t iblock=0;
615         // check input and output buffers for consistency
616         // to be enabled after fixing bug with DataBuffer and forwarded SOR/EOR
617         //for (iblock=0; iblock<fBlockDataArray.size(); iblock++) {
618         //  if ((AliHLTUInt8_t*)fBlockDataArray[iblock].fPtr >= pTgtBuffer+size) continue;
619         //  if (pTgtBuffer >= (AliHLTUInt8_t*)fBlockDataArray[iblock].fPtr+fBlockDataArray[iblock].fSize) continue;
620         //  HLTFatal("input and output buffer overlap for block descriptor %d (ptr %p size %d): output buffer %p %d",
621         //         iblock, fBlockDataArray[iblock].fPtr, fBlockDataArray[iblock].fSize,
622         //         pTgtBuffer, size);
623         //}     
624
625         // process
626         evtData.fBlockCnt=fBlockDataArray.size();
627         iResult=pComponent->ProcessEvent(evtData, &fBlockDataArray[0], trigData, pTgtBuffer, size, outputBlockCnt, outputBlocks, edd);
628         HLTDebug("component %s ProcessEvent finnished (%d): size=%d blocks=%d", pComponent->GetComponentID(), iResult, size, outputBlockCnt);
629
630         // EventDoneData is for the moment ignored in AliHLTSystem
631         if (edd) {
632           HLTDebug("got EventDoneData size %d", edd->fDataSize);
633           delete [] reinterpret_cast<char*>(edd);
634           edd=NULL;
635         }
636
637         // remove event data block
638         fBlockDataArray.pop_back();
639
640         // check for forwarded blocks.
641         // loop over all output blocks and check
642         // 1. for duplicate blocks (pointing to same location in output buffer
643         //    or to the same buffer)
644         // 2. for blocks forwarded from the input.
645         if (iResult>=0 && outputBlocks) {
646           if (fListTargets.First()!=NULL) {
647             AliHLTComponentBlockDataList segments;
648             for (AliHLTUInt32_t oblock=0; oblock<outputBlockCnt; oblock++) {
649               // consistency check for data reference
650               if (outputBlocks[oblock].fPtr!=NULL && outputBlocks[oblock].fPtr!=pTgtBuffer &&
651                   outputBlocks[oblock].fOffset!=0) {
652                 HLTWarning("output block %s 0x%08x has inconsistent data reference ptr=%p offset=0x%08x: "
653                            "for new blocks use offset only, forwarded blocks have fPtr set only",
654                            AliHLTComponent::DataType2Text(outputBlocks[oblock].fDataType).c_str(),
655                            outputBlocks[oblock].fSpecification,
656                            outputBlocks[oblock].fPtr, outputBlocks[oblock].fOffset);
657               }
658
659               // check for duplicates in the output
660               // this check applies for forwarded data blocks where
661               // the ptr is not inside the target buffer
662               AliHLTUInt32_t checkblock=0;
663               for (; checkblock<oblock; checkblock++) {
664                 if (outputBlocks[oblock].fPtr!=NULL && outputBlocks[oblock].fPtr!=pTgtBuffer &&
665                     outputBlocks[checkblock].fPtr==outputBlocks[oblock].fPtr) {
666                   if (outputBlocks[checkblock].fSize!=outputBlocks[oblock].fSize ||
667                       outputBlocks[checkblock].fDataType!=outputBlocks[oblock].fDataType) {
668                     HLTWarning("output blocks %d (%s 0x%08x) and %d (%s 0x%08x) have identical data references ptr=%p "
669                                "but differ in data type and/or size: %d vs. %d, ignoring block %d",
670                                oblock,
671                                AliHLTComponent::DataType2Text(outputBlocks[oblock].fDataType).c_str(),
672                                outputBlocks[oblock].fSpecification,
673                                checkblock,
674                                AliHLTComponent::DataType2Text(outputBlocks[checkblock].fDataType).c_str(),
675                                outputBlocks[checkblock].fSpecification,
676                                outputBlocks[oblock].fPtr,
677                                outputBlocks[oblock].fSize,
678                                outputBlocks[checkblock].fSize,
679                                checkblock);
680                   }
681                   // ignore from the second copy
682                   break;
683                 }
684               }
685               if (checkblock<oblock) continue;
686
687               // search for the forwarded data blocks
688               // new data blocks are announced to the data buffer, forwarded data blocks
689               // to the publisher task. The publisher task of a forwarded data block is
690               // removed from the list in order to keep the buffer open. It will be releases
691               // when the subscribing task releases it
692               iblock=0;
693               for (; iblock<fBlockDataArray.size(); iblock++) {
694                 if (outputBlocks[oblock].fDataType==kAliHLTDataTypeEvent) {
695                   // the event type data block is an artificial data block
696                   // ignore if it was forwarded
697                   break;
698                 }
699                 if (fBlockDataArray[iblock].fPtr==outputBlocks[oblock].fPtr) {
700                   assert(subscribedTaskList[iblock]!=NULL);
701                   if (subscribedTaskList[iblock]==NULL) {
702                     ALIHLTERRORGUARD(1, "missing parent task for forwarded data block %s 0x%08x, original data block %s 0x%08x, subsequent errors are suppressed",
703                                      AliHLTComponent::DataType2Text(outputBlocks[oblock].fDataType).c_str(),
704                                      outputBlocks[oblock].fSpecification,
705                                      AliHLTComponent::DataType2Text(outputBlocks[iblock].fDataType).c_str(),
706                                      outputBlocks[iblock].fSpecification);
707                     continue;
708                   }
709                   HLTDebug("forward segment %d (source task %s %p) to data buffer %p", iblock, pSrcTask->GetName(), pSrcTask, fpDataBuffer);
710                   fpDataBuffer->Forward(subscribedTaskList[iblock], &fBlockDataArray[iblock]);
711                   subscribedTaskList[iblock]=NULL; // not to be released in the loop further down
712                   break;
713                 }
714               }
715               if (iblock==fBlockDataArray.size()) segments.push_back(outputBlocks[oblock]);
716             }
717             if (pTgtBuffer && segments.size()>0) {
718               iResult=fpDataBuffer->SetSegments(pTgtBuffer, &segments[0], segments.size());
719             }
720           } else {
721             // no forwarding, actually we dont even need to keep the data, this is a
722             // dead end (fListTargets empty)
723             //iResult=fpDataBuffer->SetSegments(pTgtBuffer, outputBlocks, outputBlockCnt);
724           }
725           delete [] outputBlocks; outputBlocks=NULL; outputBlockCnt=0;
726         } else {
727           fpDataBuffer->Reset();
728         }
729         if (fListTargets.First()!=NULL) {
730           if (iSOR>=0 && subscribedTaskList[iSOR]!=NULL) {
731             HLTDebug("forward SOR event segment %d (source task %s %p) to data buffer %p", iSOR, pSrcTask->GetName(), pSrcTask, fpDataBuffer);
732             fpDataBuffer->Forward(subscribedTaskList[iSOR], &fBlockDataArray[iSOR]);
733             subscribedTaskList[iSOR]=NULL; // not to be released in the loop further down
734           }
735           if (iEOR>=0 && subscribedTaskList[iEOR]!=NULL) {
736             HLTDebug("forward EOR event (%s) segment %d (source task %s %p) to data buffer %p", AliHLTComponent::DataType2Text(fBlockDataArray[iEOR].fDataType).c_str(), iEOR, pSrcTask->GetName(), pSrcTask, fpDataBuffer);
737             fpDataBuffer->Forward(subscribedTaskList[iEOR], &fBlockDataArray[iEOR]);
738             subscribedTaskList[iEOR]=NULL; // not to be released in the loop further down
739           }
740           if (iECS>=0 && subscribedTaskList[iECS]!=NULL) {
741             HLTDebug("forward ECS event (%s) segment %d (source task %s %p) to data buffer %p", AliHLTComponent::DataType2Text(fBlockDataArray[iECS].fDataType).c_str(), iECS, pSrcTask->GetName(), pSrcTask, fpDataBuffer);
742             fpDataBuffer->Forward(subscribedTaskList[iECS], &fBlockDataArray[iECS]);
743             subscribedTaskList[iECS]=NULL; // not to be released in the loop further down
744           }
745         }
746       } else {
747         HLTError("no target buffer available");
748         iResult=-EFAULT;
749       }
750     } while (iResult==-ENOSPC && iNofTrial++<1);
751     }
752
753     fBlockDataArray.clear();
754     if (CheckFilter(kHLTLogDebug)) Print("proc");
755
756     // now release all buffers which we have subscribed to
757     iSourceDataBlock=0;
758     AliHLTTaskPList::iterator element;
759     while ((element=subscribedTaskList.begin())!=subscribedTaskList.end()) {
760       pSrcTask=*element;
761       if (pSrcTask) {
762         int iTempRes=0;
763         if ((iTempRes=pSrcTask->Release(&fBlockDataArray[iSourceDataBlock], this))>=0) {
764           HLTDebug("successfully released segment of task %s (%p)", pSrcTask->GetName(), pSrcTask);
765         } else {
766           HLTError("realease of task %s (%p) failed with error %d", pSrcTask->GetName(), pSrcTask, iTempRes);
767         }
768       }
769       subscribedTaskList.erase(element);
770       iSourceDataBlock++;
771     }
772     if (subscribedTaskList.size()>0) {
773       HLTError("could not release all data buffers");
774     }
775   } else {
776     HLTError("internal failure (not initialized component %p, data buffer %p)", fpComponent, fpDataBuffer);
777     iResult=-EFAULT;
778   }
779   return iResult;
780 }
781
782 int AliHLTTask::SubscribeSourcesAndSkip()
783 {
784   // function carries out the proper cleanup of the source components
785   // by subscribing and releasing
786   int iResult=0;  
787   AliHLTTask* pSrcTask=NULL;
788   AliHLTTaskPList subscribedTaskList;
789
790   // cleanup the data buffer
791   if (fpDataBuffer) fpDataBuffer->Reset();
792
793   // subscribe to all source tasks
794   fBlockDataArray.clear();
795   for (TObjLink* lnk=fListDependencies.FirstLink(); lnk!=NULL; lnk=lnk->Next()) {
796     if (!lnk->GetObject()) continue;
797     pSrcTask=dynamic_cast<AliHLTTask*>(lnk->GetObject());
798     if (!pSrcTask) continue;
799     unsigned iPosition=fBlockDataArray.size();
800     if ((iResult=pSrcTask->Subscribe(this, fBlockDataArray))>0) {
801       for (unsigned i=iPosition; i<fBlockDataArray.size(); i++) {
802         subscribedTaskList.push_back(pSrcTask);
803       }
804       HLTDebug("subscribed to %d blocks of task %s (%p)", iResult, pSrcTask->GetName(), pSrcTask, iResult);
805     } else if (iResult<0) {
806       HLTError("failed to subscribe to task %s (%p) with error %d", pSrcTask->GetName(), pSrcTask, iResult);
807     }
808   }
809
810   unsigned iSourceDataBlock=0;
811   AliHLTTaskPList::iterator element;
812   while ((element=subscribedTaskList.begin())!=subscribedTaskList.end()) {
813     assert(iSourceDataBlock<fBlockDataArray.size());
814     pSrcTask=*element;
815     if (pSrcTask && iSourceDataBlock<fBlockDataArray.size()) {
816       if ((iResult=pSrcTask->Release(&fBlockDataArray[iSourceDataBlock], this))>=0) {
817         HLTDebug("successfully released segment of task %s (%p)", pSrcTask->GetName(), pSrcTask);
818       } else if (iSourceDataBlock>=fBlockDataArray.size()) {
819         HLTError("mismatch between list of subscribed tasks and block list in task %s (%p), can not release task %s (%p)", GetName(), this, pSrcTask->GetName(), pSrcTask);
820       } else {
821         HLTError("realease of task %s (%p) failed with error %d", pSrcTask->GetName(), pSrcTask, iResult);
822       }
823     }
824     subscribedTaskList.erase(element);
825     iSourceDataBlock++;
826   }
827   if (iSourceDataBlock<fBlockDataArray.size()) {
828     HLTWarning("not all subscriptions released for task %s (%p)", GetName(), this);
829   }
830
831   return 0;
832 }
833
834 int AliHLTTask::GetNofMatchingDataBlocks(const AliHLTTask* pConsumerTask) const
835 {
836   // see header file for function documentation
837   int iResult=0;
838   if (pConsumerTask) {
839     if (fpDataBuffer) {
840       iResult=fpDataBuffer->FindMatchingDataBlocks(pConsumerTask->GetComponent(), NULL);
841     } else {
842       HLTFatal("internal data buffer missing");
843       iResult=-EFAULT;
844     }
845   } else {
846     iResult=-EINVAL;
847   }
848   return iResult;
849 }
850
851 int AliHLTTask::GetNofMatchingDataTypes(const AliHLTTask* pConsumerTask) const
852 {
853   // see header file for function documentation
854   int iResult=0;
855   if (pConsumerTask) {
856     AliHLTComponent* pComponent=GetComponent();
857     if (!pComponent) {
858       // init ?
859       HLTError("component not initialized");
860       iResult=-EFAULT;
861     }
862     if (pComponent) {
863       iResult=pComponent->FindMatchingDataTypes(pConsumerTask->GetComponent(), NULL);
864     } else {
865       HLTFatal("task initialization failed");
866       iResult=-EFAULT;
867     }
868   } else {
869     iResult=-EINVAL;
870   }
871   return iResult;
872 }
873
874 int AliHLTTask::Subscribe(const AliHLTTask* pConsumerTask, AliHLTComponentBlockDataList& blockDescList)
875 {
876   // see header file for function documentation
877   int iResult=0;
878   if (pConsumerTask) {
879     if (fpDataBuffer) {
880       iResult=fpDataBuffer->Subscribe(pConsumerTask->GetComponent(), blockDescList);
881     } else {
882       HLTFatal("internal data buffer missing");
883       iResult=-EFAULT;
884     }
885   } else {
886     iResult=-EINVAL;
887   }
888   return iResult;
889 }
890
891 int AliHLTTask::Release(AliHLTComponentBlockData* pBlockDesc, const AliHLTTask* pConsumerTask)
892 {
893   // see header file for function documentation
894   int iResult=0;
895   if (pConsumerTask && pBlockDesc) {
896     if (fpDataBuffer) {
897       iResult=fpDataBuffer->Release(pBlockDesc, pConsumerTask->GetComponent(), this);
898     } else {
899       HLTFatal("internal data buffer missing");
900       iResult=-EFAULT;
901     }
902   } else {
903     iResult=-EINVAL;
904   }
905   return iResult;
906 }
907
908 void AliHLTTask::PrintStatus()
909 {
910   // see header file for function documentation
911   HLTLogKeyword("task properties");
912   AliHLTComponent* pComponent=GetComponent();
913   if (pComponent) {
914     HLTMessage("     component: %s (%p)", pComponent->GetComponentID(), pComponent);
915   } else {
916     HLTMessage("     no component set!");
917   }
918   if (fpConfiguration) {
919     AliHLTConfiguration* pSrc=fpConfiguration->GetFirstSource();
920     while (pSrc) {
921       const char* pQualifier="unresolved";
922       if (FindDependency(pSrc->GetName()))
923         pQualifier="resolved";
924       HLTMessage("     source: %s (%s)", pSrc->GetName(), pQualifier);
925       pSrc=fpConfiguration->GetNextSource();
926     }
927     TObjLink* lnk = fListTargets.FirstLink();
928     while (lnk) {
929       TObject *obj = lnk->GetObject();
930       HLTMessage("     target: %s", obj->GetName());
931       lnk = lnk->Next();
932     }
933   } else {
934     HLTMessage("     task not initialized");
935   }
936 }
937
938 void AliHLTTask::Print(const char* options) const
939 {
940   // Overloaded from TObject
941   if (strcmp(options, "proc")==0) {
942     // print processing info
943     HLTMessage("**********************************************");
944     HLTMessage("******* AliHLTTask Processing info ***********");
945     HLTMessage(" component: %p %s", fpComponent, (fpComponent?fpComponent->GetComponentID():""));
946     HLTMessage(" data buffer: %p", fpDataBuffer);
947     if (fpDataBuffer) fpDataBuffer->Print("");
948     HLTMessage(" input block descriptors: %d", fBlockDataArray.size());
949     for (unsigned i=0; i<fBlockDataArray.size(); i++) {
950       HLTMessage("  %d: %s 0x%08x %p %d", i,
951                  AliHLTComponent::DataType2Text(fBlockDataArray[i].fDataType).c_str(),
952                  fBlockDataArray[i].fSpecification,
953                  fBlockDataArray[i].fPtr,
954                  fBlockDataArray[i].fSize
955                  );
956     }
957     HLTMessage("**** end of AliHLTTask Processing info *******");
958     HLTMessage("**********************************************");
959     return;
960   }
961
962   cout << "AliHLTTask " << GetName() << " " << this
963        << " component " << fpComponent << " "
964        << (fpComponent?fpComponent->GetComponentID():"")
965        << endl;
966 }
967
968
969 int AliHLTTask::CustomInit(AliHLTComponentHandler* /*pCH*/)
970 {
971   // default implementation nothing to do
972   return 0;
973 }
974
975 int AliHLTTask::CustomCleanup()
976 {
977   // default implementation nothing to do
978   return 0;
979 }
980
981 int AliHLTTask::LoggingVarargs(AliHLTComponentLogSeverity severity, 
982                                     const char* originClass, const char* originFunc,
983                                     const char* file, int line, ... ) const
984 {
985   // see header file for function documentation
986   int iResult=0;
987
988   va_list args;
989   va_start(args, line);
990
991   AliHLTLogging::SetLogString(this, " (%p)", "%s_pfmt_: ", GetName());
992   iResult=SendMessage(severity, originClass, originFunc, file, line, AliHLTLogging::BuildLogString(NULL, args, true /*append*/));
993   va_end(args);
994
995   return iResult;
996 }