]> git.uio.no Git - u/mrichter/AliRoot.git/blame - HLT/BASE/HOMER/AliHLTHOMERReader.cxx
git://git.uio.no / u / mrichter / AliRoot.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
correcting compilation warning
[u/mrichter/AliRoot.git] / HLT / BASE / HOMER / AliHLTHOMERReader.cxx
CommitLineData
2be16a33 1/************************************************************************
2**
3**
4** This file is property of and copyright by the Technical Computer
5** Science Group, Kirchhoff Institute for Physics, Ruprecht-Karls-
6** University, Heidelberg, Germany, 2001
7** This file has been written by Timm Morten Steinbeck,
8** timm@kip.uni-heidelberg.de
9**
10**
11** See the file license.txt for details regarding usage, modification,
12** distribution and warranty.
13** Important: This file is provided without any warranty, including
14** fitness for any particular purpose.
15**
16**
17** Newer versions of this file's package will be made available from
18** http://web.kip.uni-heidelberg.de/Hardwinf/L3/
19** or the corresponding page of the Heidelberg Alice Level 3 group.
20**
21*************************************************************************/
22
23/*
24***************************************************************************
25**
26** $Author$ - Initial Version by Timm Morten Steinbeck
27**
28** $Id$
29**
30***************************************************************************
31*/
32
9e91e956 33/** @file AliHLTHOMERReader.cxx
2be16a33 34 @author Timm Steinbeck
35 @date Sep 14 2007
36 @brief HLT Online Monitoring Environment including ROOT - Reader
37 @note migrated from PubSub HLT-stable-20070905.141318 (rev 2375) */
38
04a939f7 39// see header file for class documentation
2be16a33 40// or
41// refer to README to build package
42// or
43// visit http://web.ift.uib.no/~kjeks/doc/alice-hlt
44
45#include "AliHLTHOMERReader.h"
6a8e0bb4 46//#include <stdio.h>
5a1c4a5e 47#ifdef __SUNPRO_CC
48#include <string.h>
49#else
6a8e0bb4 50#include <cstring>
5a1c4a5e 51#endif
6a8e0bb4 52#include <cerrno>
2be16a33 53#include <netdb.h>
54extern int h_errno;
6a8e0bb4 55//#include <sys/types.h>
56//#include <sys/socket.h>
57//#include <netinet/in.h>
58//#include <netinet/tcp.h>
2be16a33 59#include <unistd.h>
72ae28cd 60#ifndef __CYGWIN__
2be16a33 61#include <rpc/types.h>
72ae28cd 62#endif
2be16a33 63#include <fcntl.h>
64#include <sys/stat.h>
65#include <netinet/in.h>
66#include <arpa/inet.h>
67#ifdef USE_ROOT
68#include <Rtypes.h>
69#endif
70
71
72#define MOD_BIN "MOD BIN\n"
73#define MOD_ASC "MOD ASC\n"
74#define GET_ONE "GET ONE\n"
75#define GET_ALL "GET ALL\n"
76
27011446 77// MAXHOSTNAMELEN not defined on macosx
78// 686-apple-darwin9-gcc-4.0.1
79#ifndef MAXHOSTNAMELEN
80#define MAXHOSTNAMELEN 64
81#endif
82
2be16a33 83#ifdef USE_ROOT
fe1edbae 84ClassImp(AliHLTMonitoringReader);
85ClassImp(AliHLTHOMERReader);
2be16a33 86#endif
87
88
89
90
91
92#ifdef USE_ROOT
fe1edbae 93AliHLTHOMERReader::AliHLTHOMERReader()
746d2a94 94 :
95 fCurrentEventType(~(homer_uint64)0),
96 fCurrentEventID(~(homer_uint64)0),
97 fBlockCnt(0),
98 fMaxBlockCnt(0),
99 fBlocks(NULL),
100 fDataSourceCnt(0),
101 fTCPDataSourceCnt(0),
102 fShmDataSourceCnt(0),
103 fDataSourceMaxCnt(0),
04a939f7 104 fDataSources(NULL),
105 fConnectionStatus(0),
106 fErrorConnection(~(unsigned int)0),
107 fEventRequestAdvanceTime(0)
2be16a33 108 {
1d47dbac 109// Reader implementation of the HOMER interface.
110// The HLT Monitoring Environment including ROOT is
111// a native interface to ship out data from the HLT chain.
112// See pdf document shiped with the package
113// for class documentation and tutorial.
2be16a33 114 Init();
115 }
116#endif
117
118
fe1edbae 119AliHLTHOMERReader::AliHLTHOMERReader( const char* hostname, unsigned short port )
746d2a94 120 :
fe1edbae 121 AliHLTMonitoringReader(),
746d2a94 122 fCurrentEventType(~(homer_uint64)0),
123 fCurrentEventID(~(homer_uint64)0),
124 fBlockCnt(0),
125 fMaxBlockCnt(0),
126 fBlocks(NULL),
127 fDataSourceCnt(0),
128 fTCPDataSourceCnt(0),
129 fShmDataSourceCnt(0),
130 fDataSourceMaxCnt(0),
04a939f7 131 fDataSources(NULL),
132 fConnectionStatus(0),
133 fErrorConnection(~(unsigned int)0),
134 fEventRequestAdvanceTime(0)
2be16a33 135 {
04a939f7 136// see header file for class documentation
137// For reading from a TCP port
2be16a33 138 Init();
139 if ( !AllocDataSources(1) )
140 {
141 fErrorConnection = 0;
142 fConnectionStatus = ENOMEM;
143 return;
144 }
145 fConnectionStatus = AddDataSource( hostname, port, fDataSources[0] );
146 if ( fConnectionStatus )
147 fErrorConnection = 0;
148 else
149 {
150 fDataSourceCnt++;
151 fTCPDataSourceCnt++;
152 fDataSources[0].fNdx = 0;
153 }
154 }
155
fe1edbae 156AliHLTHOMERReader::AliHLTHOMERReader( unsigned int tcpCnt, const char** hostnames, unsigned short* ports )
746d2a94 157 :
fe1edbae 158 AliHLTMonitoringReader(),
746d2a94 159 fCurrentEventType(~(homer_uint64)0),
160 fCurrentEventID(~(homer_uint64)0),
161 fBlockCnt(0),
162 fMaxBlockCnt(0),
163 fBlocks(NULL),
164 fDataSourceCnt(0),
165 fTCPDataSourceCnt(0),
166 fShmDataSourceCnt(0),
167 fDataSourceMaxCnt(0),
04a939f7 168 fDataSources(NULL),
169 fConnectionStatus(0),
170 fErrorConnection(~(unsigned int)0),
171 fEventRequestAdvanceTime(0)
2be16a33 172 {
04a939f7 173// see header file for class documentation
174// For reading from multiple TCP ports
2be16a33 175 Init();
176 if ( !AllocDataSources(tcpCnt) )
177 {
178 fErrorConnection = 0;
179 fConnectionStatus = ENOMEM;
180 return;
181 }
182 for ( unsigned int n = 0; n < tcpCnt; n++, fDataSourceCnt++, fTCPDataSourceCnt++ )
183 {
184 fConnectionStatus = AddDataSource( hostnames[n], ports[n], fDataSources[n] );
185 if ( fConnectionStatus )
186 {
187 fErrorConnection = n;
188 return;
189 }
190 fDataSources[n].fNdx = n;
191 }
192 }
193
fe1edbae 194AliHLTHOMERReader::AliHLTHOMERReader( key_t shmKey, int shmSize )
746d2a94 195 :
fe1edbae 196 AliHLTMonitoringReader(),
746d2a94 197 fCurrentEventType(~(homer_uint64)0),
198 fCurrentEventID(~(homer_uint64)0),
199 fBlockCnt(0),
200 fMaxBlockCnt(0),
201 fBlocks(NULL),
202 fDataSourceCnt(0),
203 fTCPDataSourceCnt(0),
204 fShmDataSourceCnt(0),
205 fDataSourceMaxCnt(0),
04a939f7 206 fDataSources(NULL),
207 fConnectionStatus(0),
208 fErrorConnection(~(unsigned int)0),
209 fEventRequestAdvanceTime(0)
2be16a33 210 {
04a939f7 211// see header file for class documentation
212// For reading from a System V shared memory segment
2be16a33 213 Init();
214 if ( !AllocDataSources(1) )
215 {
216 fErrorConnection = 0;
217 fConnectionStatus = ENOMEM;
218 return;
219 }
220 fConnectionStatus = AddDataSource( shmKey, shmSize, fDataSources[0] );
221 if ( fConnectionStatus )
222 fErrorConnection = 0;
223 else
224 {
225 fDataSourceCnt++;
226 fShmDataSourceCnt++;
227 fDataSources[0].fNdx = 0;
228 }
229 }
230
fe1edbae 231AliHLTHOMERReader::AliHLTHOMERReader( unsigned int shmCnt, key_t* shmKeys, int* shmSizes )
746d2a94 232 :
fe1edbae 233 AliHLTMonitoringReader(),
746d2a94 234 fCurrentEventType(~(homer_uint64)0),
235 fCurrentEventID(~(homer_uint64)0),
236 fBlockCnt(0),
237 fMaxBlockCnt(0),
238 fBlocks(NULL),
239 fDataSourceCnt(0),
240 fTCPDataSourceCnt(0),
241 fShmDataSourceCnt(0),
242 fDataSourceMaxCnt(0),
04a939f7 243 fDataSources(NULL),
244 fConnectionStatus(0),
245 fErrorConnection(~(unsigned int)0),
246 fEventRequestAdvanceTime(0)
2be16a33 247 {
04a939f7 248// see header file for class documentation
249// For reading from multiple System V shared memory segments
2be16a33 250 Init();
251 if ( !AllocDataSources(shmCnt) )
252 {
253 fErrorConnection = 0;
254 fConnectionStatus = ENOMEM;
255 return;
256 }
257 for ( unsigned int n = 0; n < shmCnt; n++, fDataSourceCnt++, fShmDataSourceCnt++ )
258 {
259 fConnectionStatus = AddDataSource( shmKeys[n], shmSizes[n], fDataSources[n] );
260 if ( fConnectionStatus )
261 {
262 fErrorConnection = n;
263 return;
264 }
265 fDataSources[n].fNdx = n;
266 }
267 }
268
fe1edbae 269AliHLTHOMERReader::AliHLTHOMERReader( unsigned int tcpCnt, const char** hostnames, unsigned short* ports,
2be16a33 270 unsigned int shmCnt, key_t* shmKeys, int* shmSizes )
746d2a94 271 :
fe1edbae 272 AliHLTMonitoringReader(),
746d2a94 273 fCurrentEventType(~(homer_uint64)0),
274 fCurrentEventID(~(homer_uint64)0),
275 fBlockCnt(0),
276 fMaxBlockCnt(0),
277 fBlocks(NULL),
278 fDataSourceCnt(0),
279 fTCPDataSourceCnt(0),
280 fShmDataSourceCnt(0),
281 fDataSourceMaxCnt(0),
04a939f7 282 fDataSources(NULL),
283 fConnectionStatus(0),
284 fErrorConnection(~(unsigned int)0),
285 fEventRequestAdvanceTime(0)
2be16a33 286 {
04a939f7 287// see header file for class documentation
288// For reading from multiple TCP ports and multiple System V shared memory segments
2be16a33 289 Init();
290 if ( !AllocDataSources(tcpCnt+shmCnt) )
291 {
292 fErrorConnection = 0;
293 fConnectionStatus = ENOMEM;
294 return;
295 }
296 for ( unsigned int n = 0; n < tcpCnt; n++, fDataSourceCnt++, fTCPDataSourceCnt++ )
297 {
298 fConnectionStatus = AddDataSource( hostnames[n], ports[n], fDataSources[n] );
299 if ( fConnectionStatus )
300 {
301 fErrorConnection = n;
302 return;
303 }
304 fDataSources[n].fNdx = n;
305 }
306 for ( unsigned int n = 0; n < shmCnt; n++, fDataSourceCnt++, fShmDataSourceCnt++ )
307 {
308 fConnectionStatus = AddDataSource( shmKeys[n], shmSizes[n], fDataSources[tcpCnt+n] );
309 if ( fConnectionStatus )
310 {
311 fErrorConnection = tcpCnt+n;
312 return;
313 }
314 fDataSources[n].fNdx = n;
315 }
316 }
8c617325 317
29e6fd20 318AliHLTHOMERReader::AliHLTHOMERReader( const void* pBuffer, int size )
8c617325 319 :
320 AliHLTMonitoringReader(),
321 fCurrentEventType(~(homer_uint64)0),
322 fCurrentEventID(~(homer_uint64)0),
323 fBlockCnt(0),
324 fMaxBlockCnt(0),
325 fBlocks(NULL),
326 fDataSourceCnt(0),
327 fTCPDataSourceCnt(0),
328 fShmDataSourceCnt(0),
329 fDataSourceMaxCnt(0),
330 fDataSources(NULL),
331 fConnectionStatus(0),
332 fErrorConnection(~(unsigned int)0),
333 fEventRequestAdvanceTime(0)
334 {
335// see header file for class documentation
336// For reading from a System V shared memory segment
337 Init();
338 if ( !AllocDataSources(1) )
339 {
340 fErrorConnection = 0;
341 fConnectionStatus = ENOMEM;
342 return;
343 }
29e6fd20 344 fConnectionStatus = AddDataSource(const_cast<void*>(pBuffer), size, fDataSources[0] );
8c617325 345 if ( fConnectionStatus )
346 fErrorConnection = 0;
347 else
348 {
349 fDataSourceCnt++;
aa5178e8 350 fShmDataSourceCnt++;
8c617325 351 fDataSources[0].fNdx = 0;
352 }
353 }
354
fe1edbae 355AliHLTHOMERReader::~AliHLTHOMERReader()
2be16a33 356 {
04a939f7 357// see header file for class documentation
2be16a33 358 ReleaseCurrentEvent();
359 FreeDataSources();
242e6536 360
361 if (fDataSources)
362 delete [] fDataSources;
2be16a33 363 }
364
fe1edbae 365int AliHLTHOMERReader::ReadNextEvent()
2be16a33 366 {
04a939f7 367// see header file for class documentation
368// Read in the next available event
2be16a33 369 return ReadNextEvent( false, 0 );
370 }
371
fe1edbae 372int AliHLTHOMERReader::ReadNextEvent( unsigned long timeout )
2be16a33 373 {
04a939f7 374// see header file for class documentation
375// Read in the next available event
2be16a33 376 return ReadNextEvent( true, timeout );
377 }
378
fe1edbae 379unsigned long AliHLTHOMERReader::GetBlockDataLength( unsigned long ndx ) const
2be16a33 380 {
04a939f7 381// see header file for class documentation
382// Return the size (in bytes) of the current event's data
383// block with the given block index (starting at 0).
2be16a33 384 if ( ndx >= fBlockCnt )
385 return 0;
386 return fBlocks[ndx].fLength;
387 }
388
fe1edbae 389const void* AliHLTHOMERReader::GetBlockData( unsigned long ndx ) const
2be16a33 390 {
04a939f7 391// see header file for class documentation
392// Return a pointer to the start of the current event's data
393// block with the given block index (starting at 0).
2be16a33 394 if ( ndx >= fBlockCnt )
395 return NULL;
396 return fBlocks[ndx].fData;
397 }
398
fe1edbae 399const char* AliHLTHOMERReader::GetBlockSendNodeID( unsigned long ndx ) const
2be16a33 400 {
04a939f7 401// see header file for class documentation
402// Return IP address or hostname of node which sent the
403// current event's data block with the given block index
404// (starting at 0).
405// For HOMER this is the ID of the node on which the subscriber
406// that provided this data runs/ran.
2be16a33 407 if ( ndx >= fBlockCnt )
408 return NULL;
409#ifdef DEBUG
410 if ( fBlocks[ndx].fSource >= fDataSourceCnt )
411 {
937cec81 412 fprintf( stderr, "%s:%d: Internal Error: fBlocks[ndx].fSource (%u) >= fDataSourceCnt (%u)\n",
2be16a33 413 __FILE__, __LINE__, fBlocks[ndx].fSource, fDataSourceCnt );
414 return NULL;
415 }
416#endif
417 return fDataSources[ fBlocks[ndx].fSource ].fHostname;
418 //return fBlocks[ndx].fOriginatingNodeID;
419 }
420
fe1edbae 421homer_uint8 AliHLTHOMERReader::GetBlockByteOrder( unsigned long ndx ) const
2be16a33 422 {
04a939f7 423// see header file for class documentation
424// Return byte order of the data stored in the
425// current event's data block with the given block index (starting at 0).
426// 0 is unknown alignment,
427// 1 ist little endian,
428// 2 is big endian. */
2be16a33 429 if ( ndx >= fBlockCnt )
430 return 0;
431 //return ((AliHLTRIBlockDescriptorV1*)fBlocks[ndx].fMetaData)->fType.fID;
432 return *(((homer_uint8*)fBlocks[ndx].fMetaData)+kByteOrderAttribute_8b_Offset);
433 }
04a939f7 434
fe1edbae 435homer_uint8 AliHLTHOMERReader::GetBlockTypeAlignment( unsigned long ndx, homer_uint8 dataType ) const
2be16a33 436 {
04a939f7 437// see header file for class documentation
438// Return the alignment (in bytes) of the given datatype
439// in the data stored in the current event's data block
440// with the given block index (starting at 0).
441// Possible values for the data type are
442// 0: homer_uint64
443// 1: homer_uint32
444// 2: uin16
445// 3: homer_uint8
446// 4: double
447// 5: float
2be16a33 448 if ( ndx >= fBlockCnt )
449 return 0;
450 if ( dataType > (kFloatAlignment_8b_Offset-kAlignment_8b_StartOffset) )
451 return 0;
452 //return ((AliHLTRIBlockDescriptorV1*)fBlocks[ndx].fMetaData)->fType.fID;
453 return *(((homer_uint8*)fBlocks[ndx].fMetaData)+kAlignment_8b_StartOffset+dataType);
454 }
455
fe1edbae 456homer_uint64 AliHLTHOMERReader::GetBlockStatusFlags( unsigned long ndx ) const
2be16a33 457 {
04a939f7 458// see header file for class documentation
2be16a33 459 if ( ndx >= fBlockCnt )
460 return 0;
461 return *(((homer_uint64*)fBlocks[ndx].fMetaData)+kStatusFlags_64b_Offset);
462 }
463
464/* HOMER specific */
465/* Return the type of the data in the current event's data
466 block with the given block index (starting at 0). */
fe1edbae 467homer_uint64 AliHLTHOMERReader::GetBlockDataType( unsigned long ndx ) const
2be16a33 468 {
04a939f7 469// see header file for class documentation
2be16a33 470 if ( ndx >= fBlockCnt )
471 return ~(homer_uint64)0;
472 //return ((AliHLTRIBlockDescriptorV1*)fBlocks[ndx].fMetaData)->fType.fID;
473 return *(((homer_uint64*)fBlocks[ndx].fMetaData)+kType_64b_Offset);
474 }
475
476/* Return the origin of the data in the current event's data
477 block with the given block index (starting at 0). */
fe1edbae 478homer_uint32 AliHLTHOMERReader::GetBlockDataOrigin( unsigned long ndx ) const
2be16a33 479 {
04a939f7 480// see header file for class documentation
2be16a33 481 if ( ndx >= fBlockCnt )
482 return ~(homer_uint32)0;
483 //return (homer_uint32)( ((AliHLTRIBlockDescriptorV1*)fBlocks[ndx].fMetaData)->fSubType1.fID );
484 return (homer_uint32)*(((homer_uint64*)fBlocks[ndx].fMetaData)+kSubType1_64b_Offset);
485 }
486
487/* Return a specification of the data in the current event's data
488 block with the given block index (starting at 0). */
fe1edbae 489homer_uint32 AliHLTHOMERReader::GetBlockDataSpec( unsigned long ndx ) const
2be16a33 490 {
04a939f7 491// see header file for class documentation
2be16a33 492 if ( ndx >= fBlockCnt )
493 return ~(homer_uint32)0;
494 //return (homer_uint32)( ((AliHLTRIBlockDescriptorV1*)fBlocks[ndx].fMetaData)->fSubType2.fID );
495 return (homer_uint32)*(((homer_uint64*)fBlocks[ndx].fMetaData)+kSubType2_64b_Offset);
496 }
497
498/* Find the next data block in the current event with the given
499 data type, origin, and specification. Returns the block's
500 index. */
fe1edbae 501unsigned long AliHLTHOMERReader::FindBlockNdx( homer_uint64 type, homer_uint32 origin,
2be16a33 502 homer_uint32 spec, unsigned long startNdx ) const
503 {
04a939f7 504// see header file for class documentation
2be16a33 505 for ( unsigned long n=startNdx; n < fBlockCnt; n++ )
506 {
507 if ( ( type == 0xFFFFFFFFFFFFFFFFULL || *(((homer_uint64*)fBlocks[n].fMetaData)+kType_64b_Offset)==type ) &&
508 ( origin == 0xFFFFFFFF || (homer_uint32)( *(((homer_uint64*)fBlocks[n].fMetaData)+kSubType1_64b_Offset) )==origin ) &&
509 ( spec == 0xFFFFFFFF || (homer_uint32)( *(((homer_uint64*)fBlocks[n].fMetaData)+kSubType2_64b_Offset) )==spec ) )
510 return n;
511 }
512 return ~(unsigned long)0;
513 }
514
515/* Find the next data block in the current event with the given
516 data type, origin, and specification. Returns the block's
517 index. */
fe1edbae 518unsigned long AliHLTHOMERReader::FindBlockNdx( char type[8], char origin[4],
2be16a33 519 homer_uint32 spec, unsigned long startNdx ) const
520 {
04a939f7 521// see header file for class documentation
2be16a33 522 for ( unsigned long n=startNdx; n < fBlockCnt; n++ )
523 {
524 bool found1=true, found2=true;
525 for ( unsigned i = 0; i < 8; i++ )
526 {
527 if ( type[i] != (char)0xFF )
528 {
529 found1=false;
530 break;
531 }
532 }
533 if ( !found1 )
534 {
535 found1 = true;
536 for ( unsigned i = 0; i < 8; i++ )
537 {
538 //printf( "%u: Comparing type '%c' and '%c'\n", i, ((char*)(((homer_uint64*)fBlocks[n].fMetaData)+kType_64b_Offset))[i], type[i] );
539 if ( ((char*)(((homer_uint64*)fBlocks[n].fMetaData)+kType_64b_Offset))[i] != type[i] )
540 {
541 found1=false;
542 break;
543 }
544 }
545 }
546 for ( unsigned i = 0; i < 4; i++ )
547 {
548 if ( origin[i] != (char)0xFF )
549 {
550 found2 = false;
551 break;
552 }
553 }
554 if ( !found2 )
555 {
556 found2 = true;
557 for ( unsigned i = 0; i < 4; i++ )
558 {
559 //printf( "Comparing origin '%c' and '%c'\n", ((char*)(((homer_uint64*)fBlocks[n].fMetaData)+kSubType1_64b_Offset))[i], origin[i] );
560 if ( ((char*)(((homer_uint64*)fBlocks[n].fMetaData)+kSubType1_64b_Offset))[i] != origin[i] )
561 {
562 found2=false;
563 break;
564 }
565 }
566 }
567 //printf( "Comparing spec '0x%08lX' and '0x%08lX'\n", (homer_uint32)( *(((homer_uint64*)fBlocks[n].fMetaData)+kSubType2_64b_Offset) ), spec );
568 if ( found1 && found2 &&
569 ( spec == 0xFFFFFFFF || ((homer_uint32)( *(((homer_uint64*)fBlocks[n].fMetaData)+kSubType2_64b_Offset)) )==spec ) )
570 return n;
571 }
572 return ~(unsigned long)0;
573 }
574
575/* Return the ID of the node that actually produced this data block.
576 This may be different from the node which sent the data to this
577 monitoring object as returned by GetBlockSendNodeID. */
fe1edbae 578const char* AliHLTHOMERReader::GetBlockCreateNodeID( unsigned long ndx ) const
2be16a33 579 {
04a939f7 580// see header file for class documentation
2be16a33 581 if ( ndx >= fBlockCnt )
582 return NULL;
583 return fBlocks[ndx].fOriginatingNodeID;
584 }
585
586
fe1edbae 587void AliHLTHOMERReader::Init()
2be16a33 588 {
04a939f7 589// see header file for class documentation
2be16a33 590 fCurrentEventType = ~(homer_uint64)0;
591 fCurrentEventID = ~(homer_uint64)0;
592 fMaxBlockCnt = fBlockCnt = 0;
593 fBlocks = NULL;
594
595 fDataSourceMaxCnt = fDataSourceCnt = fTCPDataSourceCnt = fShmDataSourceCnt = 0;
596 fDataSources = NULL;
597
598
599 fConnectionStatus = 0;
600 fErrorConnection = ~(unsigned int)0;
601
04a939f7 602 fEventRequestAdvanceTime = 0;
2be16a33 603 }
604
fe1edbae 605bool AliHLTHOMERReader::AllocDataSources( unsigned int sourceCnt )
2be16a33 606 {
04a939f7 607// see header file for class documentation
2be16a33 608 fDataSources = new DataSource[ sourceCnt ];
609 if ( !fDataSources )
610 return false;
6a8e0bb4 611 memset(fDataSources, 0, sizeof(DataSource)*sourceCnt);
2be16a33 612 fDataSourceCnt = 0;
613 fDataSourceMaxCnt = sourceCnt;
614 return true;
615 }
616
fe1edbae 617int AliHLTHOMERReader::AddDataSource( const char* hostname, unsigned short port, DataSource& source )
2be16a33 618 {
04a939f7 619// see header file for class documentation
2be16a33 620 struct hostent* he;
621 he = gethostbyname( hostname );
622 if ( he == NULL )
623 {
624 //fprintf( stderr, "Unable to determine remote host address from '%s'.\n", hostname );
625 return EADDRNOTAVAIL;
626 }
627
628 struct sockaddr_in remoteAddr;
629 remoteAddr.sin_family = AF_INET; // host byte order
630 remoteAddr.sin_port = htons(port); // short, network byte order
631 remoteAddr.sin_addr = *((struct in_addr *)he->h_addr);
632 memset(&(remoteAddr.sin_zero), '\0', 8); // zero the rest of the struct
633
634 // Create socket and connect to target program on remote node
635 source.fTCPConnection = socket( AF_INET, SOCK_STREAM, 0 );
636 if ( source.fTCPConnection == -1 )
637 {
638 return errno;
639 }
640
641 int ret;
642
643 ret = connect( source.fTCPConnection, (struct sockaddr *)&remoteAddr, sizeof(struct sockaddr) );
644 if ( ret == -1 )
645 {
646 ret=errno;
647 close( source.fTCPConnection );
648 return ret;
649 }
650
651 ret = write( source.fTCPConnection, MOD_BIN, strlen(MOD_BIN) );
652 if ( ret != (int)strlen(MOD_BIN) )
653 {
654 ret=errno;
655 close( source.fTCPConnection );
656 return ret;
657 }
658
659 char* tmpchar = new char[ strlen( hostname )+1 ];
660 if ( !tmpchar )
661 {
662 close( source.fTCPConnection );
663 return ENOMEM;
664 }
665 strcpy( tmpchar, hostname );
666 source.fHostname = tmpchar;
667
668 source.fType = kTCP;
669 source.fTCPPort = port;
670 source.fData = NULL;
671 source.fDataSize = 0;
672 source.fDataRead = 0;
673 return 0;
674 }
675
fe1edbae 676int AliHLTHOMERReader::AddDataSource( key_t shmKey, int shmSize, DataSource& source )
2be16a33 677 {
04a939f7 678// see header file for class documentation
2be16a33 679 int ret;
680 char* tmpchar = new char[ MAXHOSTNAMELEN+1 ];
681 if ( !tmpchar )
682 {
683 return ENOMEM;
684 }
685 gethostname( tmpchar, MAXHOSTNAMELEN );
686 tmpchar[MAXHOSTNAMELEN]=(char)0;
687 source.fHostname = tmpchar;
688
689 source.fShmID = shmget( shmKey, shmSize, 0660 );
690 if ( source.fShmID == -1 )
691 {
692 ret = errno;
693 delete [] source.fHostname;
694 return ret;
695 }
696
697 source.fShmPtr = (void*)shmat( source.fShmID, NULL, 0 );
698
699 if ( !source.fShmPtr )
700 {
701 ret = errno;
702 shmctl( source.fShmID, IPC_RMID, NULL );
703 delete [] source.fHostname;
704 return ret;
705 }
706
707 source.fType = kShm;
708 source.fShmKey = shmKey;
709 source.fShmSize = shmSize;
710 source.fDataSize = 0;
711 source.fDataRead = 0;
712 return 0;
713 }
714
8c617325 715int AliHLTHOMERReader::AddDataSource( void* pBuffer, int size, DataSource& source )
716 {
717// see header file for class documentation
718// a buffer data source is like a shm source apart from the shm attach and detach
719// procedure. Furthermore, the size indicator at the beginning of the buffer is not
720// cleared right before sources are read but after the reading.
d76bc02a 721 //int ret;
8c617325 722 if ( !pBuffer || size<=0) return EINVAL;
723
724 char* tmpchar = new char[ MAXHOSTNAMELEN+1 ];
725 if ( !tmpchar )
726 {
727 return ENOMEM;
728 }
729 gethostname( tmpchar, MAXHOSTNAMELEN );
730 tmpchar[MAXHOSTNAMELEN]=(char)0;
731 source.fHostname = tmpchar;
732
733 source.fShmID = -1;
734 // the data buffer does not contain a size indicator in the first 4 bytes
735 // like the shm source buffer. Still we want to use the mechanism to invalidate/
736 // trigger by clearing the size indicator. Take the source.fShmSize variable.
737 source.fShmPtr = &source.fShmSize;
738 source.fType = kBuf;
739 source.fShmKey = 0;
740 source.fShmSize = size;
741 source.fData = pBuffer;
742 source.fDataSize = 0;
743 source.fDataRead = 0;
744 return 0;
745 }
746
fe1edbae 747void AliHLTHOMERReader::FreeDataSources()
2be16a33 748 {
04a939f7 749// see header file for class documentation
2be16a33 750 for ( unsigned n=0; n < fDataSourceCnt; n++ )
751 {
752 if ( fDataSources[n].fType == kTCP )
753 FreeTCPDataSource( fDataSources[n] );
8c617325 754 else if ( fDataSources[n].fType == kShm )
2be16a33 755 FreeShmDataSource( fDataSources[n] );
242e6536 756 if ( fDataSources[n].fHostname )
757 delete [] fDataSources[n].fHostname;
2be16a33 758 }
242e6536 759 fDataSourceCnt=0;
2be16a33 760 }
761
fe1edbae 762int AliHLTHOMERReader::FreeShmDataSource( DataSource& source )
2be16a33 763 {
04a939f7 764// see header file for class documentation
2be16a33 765 if ( source.fShmPtr )
6001c5d5 766 shmdt( (char*)source.fShmPtr );
2be16a33 767// if ( source.fShmID != -1 )
768// shmctl( source.fShmID, IPC_RMID, NULL );
2be16a33 769 return 0;
770 }
771
fe1edbae 772int AliHLTHOMERReader::FreeTCPDataSource( DataSource& source )
2be16a33 773 {
04a939f7 774// see header file for class documentation
2be16a33 775 if ( source.fTCPConnection )
776 close( source.fTCPConnection );
2be16a33 777 return 0;
778 }
779
fe1edbae 780int AliHLTHOMERReader::ReadNextEvent( bool useTimeout, unsigned long timeout )
2be16a33 781 {
04a939f7 782// see header file for class documentation
2be16a33 783 if ( fDataSourceCnt<=0 )
784 return ENXIO;
785 // Clean up currently active event.
786 ReleaseCurrentEvent();
29e6fd20 787 int ret=0;
2be16a33 788 // Trigger all configured data sources
789 for ( unsigned n = 0; n<fDataSourceCnt; n++ )
790 {
791 if ( fDataSources[n].fType == kTCP )
792 ret = TriggerTCPSource( fDataSources[n], useTimeout, timeout );
8c617325 793 else if ( fDataSources[n].fType == kShm )
2be16a33 794 ret = TriggerShmSource( fDataSources[n], useTimeout, timeout );
795 if ( ret )
796 {
797 fErrorConnection = n;
798 fConnectionStatus=ret;
799 return fConnectionStatus;
800 }
801 }
802 // Now read in data from the configured data source
803 ret = ReadDataFromTCPSources( fTCPDataSourceCnt, fDataSources, useTimeout, timeout );
804 if ( ret )
805 {
806 return ret;
807 }
808 ret = ReadDataFromShmSources( fShmDataSourceCnt, fDataSources+fTCPDataSourceCnt, useTimeout, timeout );
809 if ( ret )
810 {
811 return ret;
812 }
813// for ( unsigned n = 0; n<fDataSourceCnt; n++ )
814// {
815// if ( fDataSources[n].fType == kTCP )
816// ret = ReadDataFromTCPSource( fDataSources[n], useTimeout, timeout );
817// else
818// ret = ReadDataFromShmSource( fDataSources[n], useTimeout, timeout );
819// if ( ret )
820// {
821// fErrorConnection = n;
822// fConnectionStatus=ret;
823// return fConnectionStatus;
824// }
825// }
826 //Check to see that all sources contributed data for the same event
827 homer_uint64 eventID;
828 homer_uint64 eventType;
29e6fd20 829 if (!fDataSources[0].fData)
830 {
831 fErrorConnection = 0;
832 fConnectionStatus=56;//ENOBUF;
833 return fConnectionStatus;
834 }
2be16a33 835 eventID = GetSourceEventID( fDataSources[0] );
836 eventType = GetSourceEventType( fDataSources[0] );
837 for ( unsigned n = 1; n < fDataSourceCnt; n++ )
838 {
29e6fd20 839 if ( !fDataSources[n].fData || GetSourceEventID( fDataSources[n] ) != eventID || GetSourceEventType( fDataSources[n] ) != eventType )
2be16a33 840 {
841 fErrorConnection = n;
746d2a94 842 fConnectionStatus=56;//EBADRQC;
2be16a33 843 return fConnectionStatus;
844 }
845 }
846 // Find all the different data blocks contained in the data from all
847 // the sources.
848 for ( unsigned n = 0; n < fDataSourceCnt; n++ )
849 {
850 ret = ParseSourceData( fDataSources[n] );
851 if ( ret )
852 {
853 fErrorConnection = n;
746d2a94 854 fConnectionStatus=57;//EBADSLT;
aa5178e8 855 return ret;
2be16a33 856 }
857 }
858 fCurrentEventID = eventID;
859 fCurrentEventType = eventType;
860 return 0;
861 }
862
fe1edbae 863void AliHLTHOMERReader::ReleaseCurrentEvent()
2be16a33 864 {
04a939f7 865// see header file for class documentation
2be16a33 866 // sources.fDataRead = 0;
867 // fMaxBlockCnt
868 fCurrentEventID = ~(homer_uint64)0;
869 fCurrentEventType = ~(homer_uint64)0;
870 for ( unsigned n = 0; n < fDataSourceCnt; n++ )
871 {
872 if ( fDataSources[n].fData )
873 {
874 if ( fDataSources[n].fType == kTCP )
875 delete [] (homer_uint8*)fDataSources[n].fData;
29e6fd20 876 // do not reset the data pointer for kBuf sources since this
877 // can not be set again.
878 if ( fDataSources[n].fType != kBuf )
879 fDataSources[n].fData = NULL;
2be16a33 880 }
881 fDataSources[n].fDataSize = fDataSources[n].fDataRead = 0;
882 }
883 if ( fBlocks )
884 {
885 for ( unsigned n = 0; n < fMaxBlockCnt; n++ )
886 {
887 if ( fBlocks[n].fOriginatingNodeID )
888 delete [] fBlocks[n].fOriginatingNodeID;
889 }
890 delete [] fBlocks;
891 fBlocks=0;
892 fMaxBlockCnt = 0;
893 fBlockCnt=0;
894 }
895 }
896
fe1edbae 897int AliHLTHOMERReader::TriggerTCPSource( DataSource& source, bool useTimeout, unsigned long timeoutUsec )
2be16a33 898 {
04a939f7 899// see header file for class documentation
2be16a33 900 int ret;
901 struct timeval oldSndTO, newSndTO;
902 if ( useTimeout )
903 {
904 socklen_t optlen=sizeof(oldSndTO);
905 ret = getsockopt( source.fTCPConnection, SOL_SOCKET, SO_SNDTIMEO, &oldSndTO, &optlen );
906 if ( ret )
907 {
908 return errno;
909 }
910 if ( optlen!=sizeof(oldSndTO) )
911 {
912 return ENXIO;
913 }
3a7c0444 914 newSndTO.tv_sec = timeoutUsec / 1000000;
915 newSndTO.tv_usec = timeoutUsec - (newSndTO.tv_sec*1000000);
2be16a33 916 ret = setsockopt( source.fTCPConnection, SOL_SOCKET, SO_SNDTIMEO, &newSndTO, sizeof(newSndTO) );
917 if ( ret )
918 {
919 return errno;
920 }
921 }
922 // Send one event request
04a939f7 923 if ( !fEventRequestAdvanceTime )
2be16a33 924 {
925 ret = write( source.fTCPConnection, GET_ONE, strlen(GET_ONE) );
926
927 if ( ret != (int)strlen(GET_ONE) )
928 {
929 ret=errno;
930 setsockopt( source.fTCPConnection, SOL_SOCKET, SO_SNDTIMEO, &oldSndTO, sizeof(oldSndTO) );
931 return ret;
932 }
933 }
934 else
935 {
936 char tmpCmd[ 128 ];
937
04a939f7 938 int len = snprintf( tmpCmd, 128, "FIRST ORBIT EVENT 0x%Lu\n", (unsigned long long)fEventRequestAdvanceTime );
2be16a33 939 if ( len>128 || len<0 )
940 {
941 ret=EMSGSIZE;
942 setsockopt( source.fTCPConnection, SOL_SOCKET, SO_SNDTIMEO, &oldSndTO, sizeof(oldSndTO) );
943 return ret;
944 }
945
946 ret = write( source.fTCPConnection, tmpCmd, strlen(tmpCmd) );
947
948 if ( ret != (int)strlen(tmpCmd) )
949 {
950 ret=errno;
951 setsockopt( source.fTCPConnection, SOL_SOCKET, SO_SNDTIMEO, &oldSndTO, sizeof(oldSndTO) );
952 return ret;
953 }
954
955 }
956 return 0;
957 }
958
6a8e0bb4 959int AliHLTHOMERReader::TriggerShmSource( DataSource& source, bool, unsigned long ) const
2be16a33 960 {
04a939f7 961// see header file for class documentation
fe1edbae 962// clear the size indicator in the first 4 bytes of the buffer to request data
963// from the HOMER writer.
2be16a33 964 if ( source.fShmPtr )
965 {
966 *(homer_uint32*)( source.fShmPtr ) = 0;
967 return 0;
968 }
969 else
970 return EFAULT;
971 }
972
fe1edbae 973int AliHLTHOMERReader::ReadDataFromTCPSources( unsigned sourceCnt, DataSource* sources, bool useTimeout, unsigned long timeout )
2be16a33 974 {
3a7c0444 975// see header file for class documentation
2be16a33 976 bool toRead = false;
977 do
978 {
979 fd_set conns;
980 FD_ZERO( &conns );
981 int highestConn=0;
982 toRead = false;
983 unsigned firstConnection=~(unsigned)0;
984 for ( unsigned long n = 0; n < sourceCnt; n++ )
985 {
986 if ( sources[n].fDataSize == 0 // size specifier not yet read
987 || sources[n].fDataRead < sources[n].fDataSize ) // Data not yet read fully
988 {
989 toRead = true;
990 FD_SET( sources[n].fTCPConnection, &conns );
991 if ( sources[n].fTCPConnection > highestConn )
992 highestConn = sources[n].fTCPConnection;
993 fcntl( sources[n].fTCPConnection, F_SETFL, O_NONBLOCK );
994 if ( firstConnection == ~(unsigned)0 )
995 firstConnection = n;
996 }
997 else
998 {
999 fcntl( sources[n].fTCPConnection, F_SETFL, 0 );
1000 }
1001 }
1002 if ( toRead )
1003 {
1004 struct timeval tv, *ptv;
1005 if ( useTimeout )
1006 {
1007 tv.tv_sec = timeout / 1000000;
1008 tv.tv_usec = timeout - (tv.tv_sec*1000000);
1009 ptv = &tv;
1010 }
1011 else
1012 ptv = NULL;
1013 // wait until something is ready to be read
1014 // either for timeout usecs or until eternity
1015 int ret;
1016 ret = select( highestConn+1, &conns, NULL, NULL, ptv );
1017 if ( ret <=0 )
1018 {
1019 fErrorConnection = firstConnection;
1020 if ( errno )
1021 fConnectionStatus = errno;
1022 else
1023 fConnectionStatus = ETIMEDOUT;
1024 return fConnectionStatus;
1025 }
1026 for ( unsigned n = 0; n < sourceCnt; n++ )
1027 {
1028 if ( FD_ISSET( sources[n].fTCPConnection, &conns ) )
1029 {
1030 if ( sources[n].fDataSize == 0 )
1031 {
1032 ret=read( sources[n].fTCPConnection, &(sources[n].fDataSize), sizeof(homer_uint32) );
1033 if ( ret != sizeof(homer_uint32) )
1034 {
1035 fErrorConnection = n;
1036 if ( errno )
1037 fConnectionStatus = errno;
1038 else
1039 fConnectionStatus = ENOMSG;
1040 return fConnectionStatus;
1041 }
1042 sources[n].fDataSize = ntohl( sources[n].fDataSize );
1043 sources[n].fDataRead = 0;
1044 sources[n].fData = new homer_uint8[ sources[n].fDataSize ];
1045 if ( !sources[n].fData )
1046 {
1047 fErrorConnection = n;
1048 fConnectionStatus = ENOMEM;
1049 return fConnectionStatus;
1050 }
1051 }
1052 else if ( sources[n].fData && sources[n].fDataRead < sources[n].fDataSize)
1053 {
1054 ret=read( sources[n].fTCPConnection, ((homer_uint8*)sources[n].fData)+sources[n].fDataRead, sources[n].fDataSize-sources[n].fDataRead );
1055 if ( ret>0 )
1056 sources[n].fDataRead += ret;
1057 else if ( ret == 0 )
1058 {
1059 fErrorConnection = n;
1060 fConnectionStatus = ECONNRESET;
1061 return fConnectionStatus;
1062 }
1063 else
1064 {
1065 fErrorConnection = n;
1066 fConnectionStatus = errno;
1067 return fConnectionStatus;
1068 }
1069 }
1070 else
1071 {
1072 fErrorConnection = n;
1073 fConnectionStatus = ENXIO;
1074 return fConnectionStatus;
1075 }
1076 }
1077 }
1078 }
1079 }
1080 while ( toRead );
1081 return 0;
1082 }
1083
1084/*
fe1edbae 1085int AliHLTHOMERReader::ReadDataFromTCPSources( DataSource& source, bool useTimeout, unsigned long timeout )
2be16a33 1086 {
1087#warning TODO If useTimeout: Set sockets to nonblocking, select + loop around GET_ONE write
1088 // Send one event request
1089 ret = write( source.fTCPConnection, GET_ONE, strlen(GET_ONE) );
1090 if ( ret != strlen(GET_ONE) )
1091 {
1092 return errno;
1093 }
1094 // wait for and read back size specifier
1095 unsigned sizeNBO;
1096 // The value transmitted is binary, in network byte order
1097 ret = read( source.fTCPConnection, &sizeNBO, sizeof(sizeNBO) );
1098 if ( ret != sizeof(sizeNBO) )
1099 {
1100 return errno;
1101 }
1102 // Convert back to host byte order
1103 source.fDataSize = ntohl( sizeNBO );
1104 source.fData = new homer_uint8[ source.fDataSize ];
1105 unsigned long dataRead=0, toRead;
1106 if ( !source.fData )
1107 {
1108 char buffer[1024];
1109 // Read in data into buffer in order not to block connection
1110 while ( dataRead < source.fDataSize )
1111 {
1112 if ( source.fDataSize-dataRead > 1024 )
1113 toRead = 1024;
1114 else
1115 toRead = source.fDataSize-dataRead;
1116 ret = read( source.fTCPConnection, buffer, toRead );
1117 if ( ret > 0 )
1118 dataRead += ret;
1119 else
1120 return errno;
1121 }
1122 return ENOMEM;
1123 }
1124 while ( dataRead < source.fDataSize )
1125 {
1126 toRead = source.fDataSize-dataRead;
1127 ret = read( source.fTCPConnection, source.fData+dataRead, toRead );
1128 if ( ret > 0 )
1129 dataRead += ret;
1130 else if ( ret == 0 && useTimeout )
1131 {
1132 struct timeval tv;
1133 tv.tv_sec = timeout / 1000000;
1134 tv.tv_usec = timeout - (tv.tv_sec*1000000);
1135 fd_set conns;
1136 FD_ZERO( &conns );
1137 FD_SET( source.fTCPConnection, &conns );
1138 ret = select( source.fTCPConnection+1, &conns, NULL, NULL );
1139 if ( ret <=0 )
1140 return errno;
1141 }
1142 else if ( ret == 0 )
1143 {
1144 if ( errno == EOK )
1145 return ECONNRESET;
1146 else
1147 return errno;
1148 }
1149 else
1150 {
1151 return errno;
1152 }
1153 }
1154 return 0;
1155 }
1156*/
1157
1158/*
fe1edbae 1159int AliHLTHOMERReader::ReadDataFromShmSource( DataSource& source, bool useTimeout, unsigned long timeout )
2be16a33 1160 {
1161
1162 }
1163*/
1164
fe1edbae 1165int AliHLTHOMERReader::ReadDataFromShmSources( unsigned sourceCnt, DataSource* sources, bool useTimeout, unsigned long timeout )
2be16a33 1166 {
04a939f7 1167// see header file for class documentation
2be16a33 1168 struct timeval tv1, tv2;
1169 bool found=false;
1170 bool all=true;
1171 if ( useTimeout )
1172 gettimeofday( &tv1, NULL );
1173 do
1174 {
1175 found = false;
1176 all = true;
1177 for ( unsigned n = 0; n < sourceCnt; n++ )
1178 {
1179 if ( !sources[n].fDataSize )
1180 all = false;
1181 if ( sources[n].fShmPtr && *(homer_uint32*)sources[n].fShmPtr>0 && !sources[n].fDataSize )
1182 {
1183 found = true;
1184 sources[n].fDataSize = *(homer_uint32*)sources[n].fShmPtr;
8c617325 1185 if (sources[n].fType==kBuf)
1186 {
1187 // the data buffer is already set to fData, just need to set fDataSize member
1188 // to invalidate after the first reading. Subsequent calls to ReadNextEvent return 0
1189 TriggerShmSource( sources[n], 0, 0 );
1190 } else
1191 {
1192 sources[n].fData = ((homer_uint8*)sources[n].fShmPtr)+sizeof(homer_uint32);
1193 }
2be16a33 1194 }
1195 }
1196 if ( found && useTimeout )
1197 gettimeofday( &tv1, NULL );
1198 if ( !all && useTimeout )
1199 {
1200 gettimeofday( &tv2, NULL );
1201 unsigned long long tdiff;
1202 tdiff = tv2.tv_sec-tv1.tv_sec;
1203 tdiff *= 1000000;
1204 tdiff += tv2.tv_usec-tv1.tv_usec;
1205 if ( tdiff > timeout )
1206 return ETIMEDOUT;
1207 }
1208 if ( !all )
1209 usleep( 0 );
1210 }
1211 while ( !all );
1212 return 0;
1213 }
1214
fe1edbae 1215int AliHLTHOMERReader::ParseSourceData( DataSource& source )
2be16a33 1216 {
04a939f7 1217// see header file for class documentation
2be16a33 1218 if ( source.fData )
1219 {
1220 homer_uint8 sourceByteOrder = ((homer_uint8*)source.fData)[ kByteOrderAttribute_8b_Offset ];
aa5178e8 1221 if (sourceByteOrder!=kHOMERLittleEndianByteOrder && sourceByteOrder!=kHOMERBigEndianByteOrder) return EBADMSG;
2be16a33 1222 homer_uint64 blockCnt = Swap( kHOMERNativeByteOrder, sourceByteOrder, ((homer_uint64*)source.fData)[ kSubType2_64b_Offset ] );
aa5178e8 1223 // block count is not related to size of the data in the way the
1224 // following condition implies. But we can at least limit the block
1225 // count for the case the data is corrupted
1226 if (blockCnt>source.fDataSize) return EBADMSG;
2be16a33 1227 int ret=ReAllocBlocks( fMaxBlockCnt+blockCnt );
1228 if ( ret )
1229 return ret;
1230 homer_uint64 descrOffset = Swap( kHOMERNativeByteOrder, sourceByteOrder, ((homer_uint64*)source.fData)[ kOffset_64b_Offset ] );
1231 for ( homer_uint64 n = 0; n < blockCnt && fBlockCnt < fMaxBlockCnt; n++, fBlockCnt++ )
1232 {
aa5178e8 1233 if (descrOffset+kLength_64b_Offset>=source.fDataSize) return EBADMSG;
2be16a33 1234 homer_uint8* descr = ((homer_uint8*)source.fData)+descrOffset;
1235 unsigned descrLen = Swap( kHOMERNativeByteOrder, sourceByteOrder, ((homer_uint64*)descr)[ kLength_64b_Offset ] );
aa5178e8 1236 if (descrOffset+descrLen>=source.fDataSize) return EBADMSG;
31774e81 1237 if (Swap( kHOMERNativeByteOrder, sourceByteOrder, ((homer_uint64*)descr)[ kID_64b_Offset ] ) != HOMER_BLOCK_DESCRIPTOR_TYPEID) return 126/*ENOKEY*/;
2be16a33 1238 fBlocks[fBlockCnt].fSource = source.fNdx;
1239 fBlocks[fBlockCnt].fData = ((homer_uint8*)source.fData) + Swap( kHOMERNativeByteOrder, sourceByteOrder, ((homer_uint64*)descr)[ kOffset_64b_Offset ] );
1240 fBlocks[fBlockCnt].fLength = Swap( kHOMERNativeByteOrder, sourceByteOrder, ((homer_uint64*)descr)[ kSize_64b_Offset ] );
1241 fBlocks[fBlockCnt].fMetaData = (homer_uint64*)descr;
1242 struct in_addr tmpA;
1243 tmpA.s_addr = (homer_uint32)( ((homer_uint64*)descr)[ kProducerNode_64b_Offset ] );
1244 char* addr = inet_ntoa( tmpA );
1245 char* tmpchar = new char[ strlen(addr)+1 ];
1246 if ( !tmpchar )
1247 return ENOMEM;
1248 strcpy( tmpchar, addr );
1249 fBlocks[fBlockCnt].fOriginatingNodeID = tmpchar;
1250 descrOffset += descrLen;
1251 }
1252 return 0;
1253 }
1254 return EFAULT;
1255 }
1256
fe1edbae 1257int AliHLTHOMERReader::ReAllocBlocks( unsigned long newCnt )
2be16a33 1258 {
04a939f7 1259// see header file for class documentation
2be16a33 1260 DataBlock* newBlocks;
1261 newBlocks = new DataBlock[ newCnt ];
1262 if ( !newBlocks )
1263 return ENOMEM;
1264 unsigned long cpCnt = (newCnt > fMaxBlockCnt) ? fMaxBlockCnt : newCnt;
1265 memcpy( newBlocks, fBlocks, cpCnt*sizeof(DataBlock) );
1266 if ( newCnt > fMaxBlockCnt )
1267 memset( newBlocks+fMaxBlockCnt, 0, (newCnt-fMaxBlockCnt)*sizeof(DataBlock) );
1268 if ( fBlocks )
1269 delete [] fBlocks;
1270 fBlocks = newBlocks;
1271 fMaxBlockCnt = newCnt;
1272 return 0;
1273 }
1274
fe1edbae 1275homer_uint64 AliHLTHOMERReader::GetSourceEventID( DataSource& source )
2be16a33 1276 {
04a939f7 1277// see header file for class documentation
2be16a33 1278 homer_uint8 sourceByteOrder = ((homer_uint8*)source.fData)[ kByteOrderAttribute_8b_Offset ];
1279 return Swap( kHOMERNativeByteOrder, sourceByteOrder, ((homer_uint64*)source.fData)[ kSubType1_64b_Offset ] );
1280 }
1281
fe1edbae 1282homer_uint64 AliHLTHOMERReader::GetSourceEventType( DataSource& source )
2be16a33 1283 {
04a939f7 1284// see header file for class documentation
2be16a33 1285 homer_uint8 sourceByteOrder = ((homer_uint8*)source.fData)[ kByteOrderAttribute_8b_Offset ];
1286 return Swap( kHOMERNativeByteOrder, sourceByteOrder, ((homer_uint64*)source.fData)[ kType_64b_Offset ] );
1287 }
1288
fe1edbae 1289homer_uint64 AliHLTHOMERReader::Swap( homer_uint8 destFormat, homer_uint8 sourceFormat, homer_uint64 source ) const
3a7c0444 1290 {
1291// see header file for class documentation
1292 if ( destFormat == sourceFormat )
1293 return source;
1294 else
1295 return ((source & 0xFFULL) << 56) |
1296 ((source & 0xFF00ULL) << 40) |
1297 ((source & 0xFF0000ULL) << 24) |
1298 ((source & 0xFF000000ULL) << 8) |
1299 ((source & 0xFF00000000ULL) >> 8) |
1300 ((source & 0xFF0000000000ULL) >> 24) |
1301 ((source & 0xFF000000000000ULL) >> 40) |
1302 ((source & 0xFF00000000000000ULL) >> 56);
1303 }
2be16a33 1304
fe1edbae 1305homer_uint32 AliHLTHOMERReader::Swap( homer_uint8 destFormat, homer_uint8 sourceFormat, homer_uint32 source ) const
3a7c0444 1306 {
1307// see header file for class documentation
1308 if ( destFormat == sourceFormat )
1309 return source;
1310 else
1311 return ((source & 0xFFUL) << 24) |
1312 ((source & 0xFF00UL) << 8) |
1313 ((source & 0xFF0000UL) >> 8) |
1314 ((source & 0xFF000000UL) >> 24);
1315 }
fe1edbae 1316
a183f221 1317AliHLTHOMERReader* AliHLTHOMERReaderCreateFromTCPPort(const char* hostname, unsigned short port )
fe1edbae 1318 {
a183f221 1319 // see header file for function documentation
1320 return new AliHLTHOMERReader(hostname, port);
1321 }
1322
1323AliHLTHOMERReader* AliHLTHOMERReaderCreateFromTCPPorts(unsigned int tcpCnt, const char** hostnames, unsigned short* ports)
1324 {
1325 // see header file for function documentation
1326 return new AliHLTHOMERReader(tcpCnt, hostnames, ports);
1327 }
1328
1329AliHLTHOMERReader* AliHLTHOMERReaderCreateFromBuffer(const void* pBuffer, int size)
1330 {
1331 // see header file for function documentation
8c617325 1332 return new AliHLTHOMERReader(pBuffer, size);
fe1edbae 1333 }
1334
1335void AliHLTHOMERReaderDelete(AliHLTHOMERReader* pInstance)
1336 {
a183f221 1337 // see header file for function documentation
fe1edbae 1338 if (pInstance) delete pInstance;
1339 }
1340
2be16a33 1341/*
1342***************************************************************************
1343**
1344** $Author$ - Initial Version by Timm Morten Steinbeck
1345**
1346** $Id$
1347**
1348***************************************************************************
1349*/