f2a4e1bb9c5c5ccc38cf708a54ef630948e982fc
[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, AliHLTUInt64_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&0xffffffff;
592       evtTrigData.fCommonHeader[6]=(trgMask>>32)&0x3ffff;
593       trigData.fData=&evtTrigData;
594       iLastOutputDataSize=iOutputDataSize;
595       AliHLTUInt32_t size=iOutputDataSize;
596       AliHLTUInt32_t outputBlockCnt=0;
597       AliHLTComponentBlockData* outputBlocks=NULL;
598       AliHLTComponentEventDoneData* edd=NULL;
599       if (pTgtBuffer!=NULL || iOutputDataSize==0) {
600         // add event type data block
601         // the block is removed immediately after processing from the list
602         AliHLTComponentBlockData eventTypeBlock;
603         AliHLTComponent::FillBlockData(eventTypeBlock);
604         // Note: no payload!
605         eventTypeBlock.fDataType=kAliHLTDataTypeEvent;
606         eventTypeBlock.fSpecification=eventType;
607         fBlockDataArray.push_back(eventTypeBlock);
608
609         if (CheckFilter(kHLTLogDebug)) Print("proc");
610
611         AliHLTUInt32_t iblock=0;
612         // check input and output buffers for consistency
613         // to be enabled after fixing bug with DataBuffer and forwarded SOR/EOR
614         //for (iblock=0; iblock<fBlockDataArray.size(); iblock++) {
615         //  if ((AliHLTUInt8_t*)fBlockDataArray[iblock].fPtr >= pTgtBuffer+size) continue;
616         //  if (pTgtBuffer >= (AliHLTUInt8_t*)fBlockDataArray[iblock].fPtr+fBlockDataArray[iblock].fSize) continue;
617         //  HLTFatal("input and output buffer overlap for block descriptor %d (ptr %p size %d): output buffer %p %d",
618         //         iblock, fBlockDataArray[iblock].fPtr, fBlockDataArray[iblock].fSize,
619         //         pTgtBuffer, size);
620         //}     
621
622         // process
623         evtData.fBlockCnt=fBlockDataArray.size();
624         iResult=pComponent->ProcessEvent(evtData, &fBlockDataArray[0], trigData, pTgtBuffer, size, outputBlockCnt, outputBlocks, edd);
625         HLTDebug("component %s ProcessEvent finnished (%d): size=%d blocks=%d", pComponent->GetComponentID(), iResult, size, outputBlockCnt);
626
627         // EventDoneData is for the moment ignored in AliHLTSystem
628         if (edd) {
629           HLTDebug("got EventDoneData size %d", edd->fDataSize);
630           delete [] reinterpret_cast<char*>(edd);
631           edd=NULL;
632         }
633
634         // remove event data block
635         fBlockDataArray.pop_back();
636
637         // check for forwarded blocks.
638         // loop over all output blocks and check
639         // 1. for duplicate blocks (pointing to same location in output buffer
640         //    or to the same buffer)
641         // 2. for blocks forwarded from the input.
642         if (iResult>=0 && outputBlocks) {
643           if (fListTargets.First()!=NULL) {
644             AliHLTComponentBlockDataList segments;
645             for (AliHLTUInt32_t oblock=0; oblock<outputBlockCnt; oblock++) {
646               // consistency check for data reference
647               if (outputBlocks[oblock].fPtr!=NULL && outputBlocks[oblock].fPtr!=pTgtBuffer &&
648                   outputBlocks[oblock].fOffset!=0) {
649                 HLTWarning("output block %s 0x%08x has inconsistent data reference ptr=%p offset=0x%08x: "
650                            "for new blocks use offset only, forwarded blocks have fPtr set only",
651                            AliHLTComponent::DataType2Text(outputBlocks[oblock].fDataType).c_str(),
652                            outputBlocks[oblock].fSpecification,
653                            outputBlocks[oblock].fPtr, outputBlocks[oblock].fOffset);
654               }
655
656               // check for duplicates in the output
657               // this check applies for forwarded data blocks where
658               // the ptr is not inside the target buffer
659               AliHLTUInt32_t checkblock=0;
660               for (; checkblock<oblock; checkblock++) {
661                 if (outputBlocks[oblock].fPtr!=NULL && outputBlocks[oblock].fPtr!=pTgtBuffer &&
662                     outputBlocks[checkblock].fPtr==outputBlocks[oblock].fPtr) {
663                   if (outputBlocks[checkblock].fSize!=outputBlocks[oblock].fSize ||
664                       outputBlocks[checkblock].fDataType!=outputBlocks[oblock].fDataType) {
665                     HLTWarning("output blocks %d (%s 0x%08x) and %d (%s 0x%08x) have identical data references ptr=%p "
666                                "but differ in data type and/or size: %d vs. %d, ignoring block %d",
667                                oblock,
668                                AliHLTComponent::DataType2Text(outputBlocks[oblock].fDataType).c_str(),
669                                outputBlocks[oblock].fSpecification,
670                                checkblock,
671                                AliHLTComponent::DataType2Text(outputBlocks[checkblock].fDataType).c_str(),
672                                outputBlocks[checkblock].fSpecification,
673                                outputBlocks[oblock].fPtr,
674                                outputBlocks[oblock].fSize,
675                                outputBlocks[checkblock].fSize,
676                                checkblock);
677                   }
678                   // ignore from the second copy
679                   break;
680                 }
681               }
682               if (checkblock<oblock) continue;
683
684               // search for the forwarded data blocks
685               // new data blocks are announced to the data buffer, forwarded data blocks
686               // to the publisher task. The publisher task of a forwarded data block is
687               // removed from the list in order to keep the buffer open. It will be releases
688               // when the subscribing task releases it
689               iblock=0;
690               for (; iblock<fBlockDataArray.size(); iblock++) {
691                 if (outputBlocks[oblock].fDataType==kAliHLTDataTypeEvent) {
692                   // the event type data block is an artificial data block
693                   // ignore if it was forwarded
694                   break;
695                 }
696                 if (fBlockDataArray[iblock].fPtr==outputBlocks[oblock].fPtr) {
697                   assert(subscribedTaskList[iblock]!=NULL);
698                   if (subscribedTaskList[iblock]==NULL) {
699                     ALIHLTERRORGUARD(1, "missing parent task for forwarded data block %s 0x%08x, original data block %s 0x%08x, subsequent errors are suppressed",
700                                      AliHLTComponent::DataType2Text(outputBlocks[oblock].fDataType).c_str(),
701                                      outputBlocks[oblock].fSpecification,
702                                      AliHLTComponent::DataType2Text(outputBlocks[iblock].fDataType).c_str(),
703                                      outputBlocks[iblock].fSpecification);
704                     continue;
705                   }
706                   HLTDebug("forward segment %d (source task %s %p) to data buffer %p", iblock, pSrcTask->GetName(), pSrcTask, fpDataBuffer);
707                   fpDataBuffer->Forward(subscribedTaskList[iblock], &fBlockDataArray[iblock]);
708                   subscribedTaskList[iblock]=NULL; // not to be released in the loop further down
709                   break;
710                 }
711               }
712               if (iblock==fBlockDataArray.size()) segments.push_back(outputBlocks[oblock]);
713             }
714             if (pTgtBuffer && segments.size()>0) {
715               iResult=fpDataBuffer->SetSegments(pTgtBuffer, &segments[0], segments.size());
716             }
717           } else {
718             // no forwarding, actually we dont even need to keep the data, this is a
719             // dead end (fListTargets empty)
720             //iResult=fpDataBuffer->SetSegments(pTgtBuffer, outputBlocks, outputBlockCnt);
721           }
722           delete [] outputBlocks; outputBlocks=NULL; outputBlockCnt=0;
723         } else {
724           fpDataBuffer->Reset();
725         }
726         if (fListTargets.First()!=NULL) {
727           if (iSOR>=0 && subscribedTaskList[iSOR]!=NULL) {
728             HLTDebug("forward SOR event segment %d (source task %s %p) to data buffer %p", iSOR, pSrcTask->GetName(), pSrcTask, fpDataBuffer);
729             fpDataBuffer->Forward(subscribedTaskList[iSOR], &fBlockDataArray[iSOR]);
730             subscribedTaskList[iSOR]=NULL; // not to be released in the loop further down
731           }
732           if (iEOR>=0 && subscribedTaskList[iEOR]!=NULL) {
733             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);
734             fpDataBuffer->Forward(subscribedTaskList[iEOR], &fBlockDataArray[iEOR]);
735             subscribedTaskList[iEOR]=NULL; // not to be released in the loop further down
736           }
737           if (iECS>=0 && subscribedTaskList[iECS]!=NULL) {
738             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);
739             fpDataBuffer->Forward(subscribedTaskList[iECS], &fBlockDataArray[iECS]);
740             subscribedTaskList[iECS]=NULL; // not to be released in the loop further down
741           }
742         }
743       } else {
744         HLTError("no target buffer available");
745         iResult=-EFAULT;
746       }
747     } while (iResult==-ENOSPC && iNofTrial++<1);
748     }
749
750     fBlockDataArray.clear();
751     if (CheckFilter(kHLTLogDebug)) Print("proc");
752
753     // now release all buffers which we have subscribed to
754     iSourceDataBlock=0;
755     AliHLTTaskPList::iterator element;
756     while ((element=subscribedTaskList.begin())!=subscribedTaskList.end()) {
757       pSrcTask=*element;
758       if (pSrcTask) {
759         int iTempRes=0;
760         if ((iTempRes=pSrcTask->Release(&fBlockDataArray[iSourceDataBlock], this))>=0) {
761           HLTDebug("successfully released segment of task %s (%p)", pSrcTask->GetName(), pSrcTask);
762         } else {
763           HLTError("realease of task %s (%p) failed with error %d", pSrcTask->GetName(), pSrcTask, iTempRes);
764         }
765       }
766       subscribedTaskList.erase(element);
767       iSourceDataBlock++;
768     }
769     if (subscribedTaskList.size()>0) {
770       HLTError("could not release all data buffers");
771     }
772   } else {
773     HLTError("internal failure (not initialized component %p, data buffer %p)", fpComponent, fpDataBuffer);
774     iResult=-EFAULT;
775   }
776   return iResult;
777 }
778
779 int AliHLTTask::SubscribeSourcesAndSkip()
780 {
781   // function carries out the proper cleanup of the source components
782   // by subscribing and releasing
783   int iResult=0;  
784   AliHLTTask* pSrcTask=NULL;
785   AliHLTTaskPList subscribedTaskList;
786
787   // cleanup the data buffer
788   if (fpDataBuffer) fpDataBuffer->Reset();
789
790   // subscribe to all source tasks
791   fBlockDataArray.clear();
792   for (TObjLink* lnk=fListDependencies.FirstLink(); lnk!=NULL; lnk=lnk->Next()) {
793     if (!lnk->GetObject()) continue;
794     pSrcTask=dynamic_cast<AliHLTTask*>(lnk->GetObject());
795     if (!pSrcTask) continue;
796     unsigned iPosition=fBlockDataArray.size();
797     if ((iResult=pSrcTask->Subscribe(this, fBlockDataArray))>0) {
798       for (unsigned i=iPosition; i<fBlockDataArray.size(); i++) {
799         subscribedTaskList.push_back(pSrcTask);
800       }
801       HLTDebug("subscribed to %d blocks of task %s (%p)", iResult, pSrcTask->GetName(), pSrcTask, iResult);
802     } else if (iResult<0) {
803       HLTError("failed to subscribe to task %s (%p) with error %d", pSrcTask->GetName(), pSrcTask, iResult);
804     }
805   }
806
807   unsigned iSourceDataBlock=0;
808   AliHLTTaskPList::iterator element;
809   while ((element=subscribedTaskList.begin())!=subscribedTaskList.end()) {
810     assert(iSourceDataBlock<fBlockDataArray.size());
811     pSrcTask=*element;
812     if (pSrcTask && iSourceDataBlock<fBlockDataArray.size()) {
813       if ((iResult=pSrcTask->Release(&fBlockDataArray[iSourceDataBlock], this))>=0) {
814         HLTDebug("successfully released segment of task %s (%p)", pSrcTask->GetName(), pSrcTask);
815       } else if (iSourceDataBlock>=fBlockDataArray.size()) {
816         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);
817       } else {
818         HLTError("realease of task %s (%p) failed with error %d", pSrcTask->GetName(), pSrcTask, iResult);
819       }
820     }
821     subscribedTaskList.erase(element);
822     iSourceDataBlock++;
823   }
824   if (iSourceDataBlock<fBlockDataArray.size()) {
825     HLTWarning("not all subscriptions released for task %s (%p)", GetName(), this);
826   }
827
828   return 0;
829 }
830
831 int AliHLTTask::GetNofMatchingDataBlocks(const AliHLTTask* pConsumerTask) const
832 {
833   // see header file for function documentation
834   int iResult=0;
835   if (pConsumerTask) {
836     if (fpDataBuffer) {
837       iResult=fpDataBuffer->FindMatchingDataBlocks(pConsumerTask->GetComponent(), NULL);
838     } else {
839       HLTFatal("internal data buffer missing");
840       iResult=-EFAULT;
841     }
842   } else {
843     iResult=-EINVAL;
844   }
845   return iResult;
846 }
847
848 int AliHLTTask::GetNofMatchingDataTypes(const AliHLTTask* pConsumerTask) const
849 {
850   // see header file for function documentation
851   int iResult=0;
852   if (pConsumerTask) {
853     AliHLTComponent* pComponent=GetComponent();
854     if (!pComponent) {
855       // init ?
856       HLTError("component not initialized");
857       iResult=-EFAULT;
858     }
859     if (pComponent) {
860       iResult=pComponent->FindMatchingDataTypes(pConsumerTask->GetComponent(), NULL);
861     } else {
862       HLTFatal("task initialization failed");
863       iResult=-EFAULT;
864     }
865   } else {
866     iResult=-EINVAL;
867   }
868   return iResult;
869 }
870
871 int AliHLTTask::Subscribe(const AliHLTTask* pConsumerTask, AliHLTComponentBlockDataList& blockDescList)
872 {
873   // see header file for function documentation
874   int iResult=0;
875   if (pConsumerTask) {
876     if (fpDataBuffer) {
877       iResult=fpDataBuffer->Subscribe(pConsumerTask->GetComponent(), blockDescList);
878     } else {
879       HLTFatal("internal data buffer missing");
880       iResult=-EFAULT;
881     }
882   } else {
883     iResult=-EINVAL;
884   }
885   return iResult;
886 }
887
888 int AliHLTTask::Release(AliHLTComponentBlockData* pBlockDesc, const AliHLTTask* pConsumerTask)
889 {
890   // see header file for function documentation
891   int iResult=0;
892   if (pConsumerTask && pBlockDesc) {
893     if (fpDataBuffer) {
894       iResult=fpDataBuffer->Release(pBlockDesc, pConsumerTask->GetComponent(), this);
895     } else {
896       HLTFatal("internal data buffer missing");
897       iResult=-EFAULT;
898     }
899   } else {
900     iResult=-EINVAL;
901   }
902   return iResult;
903 }
904
905 void AliHLTTask::PrintStatus()
906 {
907   // see header file for function documentation
908   HLTLogKeyword("task properties");
909   AliHLTComponent* pComponent=GetComponent();
910   if (pComponent) {
911     HLTMessage("     component: %s (%p)", pComponent->GetComponentID(), pComponent);
912   } else {
913     HLTMessage("     no component set!");
914   }
915   if (fpConfiguration) {
916     AliHLTConfiguration* pSrc=fpConfiguration->GetFirstSource();
917     while (pSrc) {
918       const char* pQualifier="unresolved";
919       if (FindDependency(pSrc->GetName()))
920         pQualifier="resolved";
921       HLTMessage("     source: %s (%s)", pSrc->GetName(), pQualifier);
922       pSrc=fpConfiguration->GetNextSource();
923     }
924     TObjLink* lnk = fListTargets.FirstLink();
925     while (lnk) {
926       TObject *obj = lnk->GetObject();
927       HLTMessage("     target: %s", obj->GetName());
928       lnk = lnk->Next();
929     }
930   } else {
931     HLTMessage("     task not initialized");
932   }
933 }
934
935 void AliHLTTask::Print(const char* options) const
936 {
937   // Overloaded from TObject
938   if (strcmp(options, "proc")==0) {
939     // print processing info
940     HLTMessage("**********************************************");
941     HLTMessage("******* AliHLTTask Processing info ***********");
942     HLTMessage(" component: %p %s", fpComponent, (fpComponent?fpComponent->GetComponentID():""));
943     HLTMessage(" data buffer: %p", fpDataBuffer);
944     if (fpDataBuffer) fpDataBuffer->Print("");
945     HLTMessage(" input block descriptors: %d", fBlockDataArray.size());
946     for (unsigned i=0; i<fBlockDataArray.size(); i++) {
947       HLTMessage("  %d: %s 0x%08x %p %d", i,
948                  AliHLTComponent::DataType2Text(fBlockDataArray[i].fDataType).c_str(),
949                  fBlockDataArray[i].fSpecification,
950                  fBlockDataArray[i].fPtr,
951                  fBlockDataArray[i].fSize
952                  );
953     }
954     HLTMessage("**** end of AliHLTTask Processing info *******");
955     HLTMessage("**********************************************");
956     return;
957   }
958
959   cout << "AliHLTTask " << GetName() << " " << this
960        << " component " << fpComponent << " "
961        << (fpComponent?fpComponent->GetComponentID():"")
962        << endl;
963 }
964
965
966 int AliHLTTask::CustomInit(AliHLTComponentHandler* /*pCH*/)
967 {
968   // default implementation nothing to do
969   return 0;
970 }
971
972 int AliHLTTask::CustomCleanup()
973 {
974   // default implementation nothing to do
975   return 0;
976 }
977
978 int AliHLTTask::LoggingVarargs(AliHLTComponentLogSeverity severity, 
979                                     const char* originClass, const char* originFunc,
980                                     const char* file, int line, ... ) const
981 {
982   // see header file for function documentation
983   int iResult=0;
984
985   va_list args;
986   va_start(args, line);
987
988   AliHLTLogging::SetLogString(this, " (%p)", "%s_pfmt_: ", GetName());
989   iResult=SendMessage(severity, originClass, originFunc, file, line, AliHLTLogging::BuildLogString(NULL, args, true /*append*/));
990   va_end(args);
991
992   return iResult;
993 }