Commit ab357197 authored by Mathieu Faverge's avatar Mathieu Faverge

Apply Francois's patch

parent 6f7488f8
......@@ -89,6 +89,9 @@ void ParserOTF::parse(Trace &trace,
OTF_FileManager *manager;
OTF_Reader *reader;
int ret = 0;
const std::string filename = get_next_file_to_parse();
if(filename == "")
return;
manager = OTF_FileManager_open(VITE_OTF_MAXFILES_OPEN);
if(manager == NULL) {
......@@ -96,7 +99,7 @@ void ParserOTF::parse(Trace &trace,
return;
}
reader = OTF_Reader_open(_file_to_parse.c_str(), manager);
reader = OTF_Reader_open(filename.c_str(), manager);
if(reader == NULL) {
cerr << QObject::tr(VITE_ERR_OTF_OPENREADER).toStdString() << endl;
OTF_FileManager_close(manager);
......
......@@ -88,9 +88,12 @@ void mt_ParserOTF::parse(Trace &trace,
QSemaphore freeSlots(5);
QSemaphore linesProduced(5);
OTFTraceBuilderThread *parserevent2 = new OTFTraceBuilderThread(&ended, &freeSlots, &mutex);
const std::string filename = get_next_file_to_parse();
if(filename == "")
return ;
/* OTF_MasterControl* mc;
mc = OTF_MasterControl_new( manager );
OTF_MasterControl_read( mc, _file_to_parse.c_str());
OTF_MasterControl_read( mc, filename.c_str());
uint32_t streamcount = OTF_MasterControl_getCount( mc );
......@@ -117,7 +120,7 @@ void mt_ParserOTF::parse(Trace &trace,
parserevent2->moveToThread(&traceThread);
traceThread.start();
reader = OTF_Reader_open(_file_to_parse.c_str(), manager);
reader = OTF_Reader_open(filename.c_str(), manager);
if(reader == NULL) {
cerr << QObject::tr(VITE_ERR_OTF_OPENREADER).toStdString() << endl;
OTF_FileManager_close(manager);
......
......@@ -91,7 +91,17 @@ public:
*/
void parse(Trace &trace,
bool finish_trace_after_parse = true);
/*!
* \fn parse(const string filename, Trace &trace, bool finish_trace_after_parse = true)
* \param filename the name of the file to parse
* \param trace the structure of data to fill
* \param finish_trace_after_parse boolean set if we do not have to finish the trace after parsing
*/
void parse(const std::string filename,
Trace &trace,
bool finish_trace_after_parse = true);
/*!
* \fn get_percent_loaded() const
* \brief return the size of the file already read.
......
......@@ -58,59 +58,52 @@
#include "parser/PajeParser/BuilderThread.hpp"
#include "trace/TraceBuilderThread.hpp"
BuilderThread::BuilderThread(mt_ParserEventPaje *p, Trace * trace,QWaitCondition* cond,QWaitCondition* trace_cond, QSemaphore *sem1, QSemaphore *sem2, QMutex *mutex, QMutex *mutex2){
_parser = p;
_cond=cond;
_trace_cond=trace_cond;
_freeSlots=sem1;
_mutex=mutex;
_mutex2=mutex2;
_linesProduced=sem2;
_trace=trace;
_is_finished=false;
BuilderThread::BuilderThread(mt_ParserEventPaje *p, Trace *trace,
QWaitCondition* cond,QWaitCondition* trace_cond,
QSemaphore *sem1, QSemaphore *sem2, QMutex *mutex,
QMutex *mutex2, Parser *parser) {
_event_parser = p;
_parser = parser;
_cond = cond;
_trace_cond = trace_cond;
_freeSlots = sem1;
_mutex = mutex;
_mutex2 = mutex2;
_linesProduced= sem2;
_trace = trace;
_is_finished = false;
}
void BuilderThread::run(int n, PajeLine* line) {
_freeSlots->acquire(); //do not produce too fast (5 blocks max at a time)
void BuilderThread::run(int n, PajeLine* line) {
_freeSlots->acquire(); //do not produce too fast (5 blocks max at a time)
int n_without_errors=0;
Trace_builder_struct* tb_structs= new Trace_builder_struct[n];
int j;
for(j=0; j<n; j++){
if( _parser->store_event(&line[j],*_trace, &tb_structs[n_without_errors])==0)n_without_errors++;
free(line[j]._tokens); // release tokens allocated in the PajeFileManager
}
emit(build_trace(n_without_errors, tb_structs));
_linesProduced->release();
free(line);
int n_without_errors=0;
Trace_builder_struct* tb_structs= new Trace_builder_struct[n];
int j;
for(j=0; j<n; j++){
tb_structs[n_without_errors]._parser = _parser;
if( _event_parser->store_event(&line[j],*_trace, &tb_structs[n_without_errors]) == 0)
n_without_errors++;
free(line[j]._tokens); // release tokens allocated in the PajeFileManager
}
emit(build_trace(n_without_errors, tb_structs));
_linesProduced->release();
free(line);
}
bool BuilderThread::is_finished(){return _is_finished;}
void BuilderThread::finish_build(){
//quit();
//finish the TraceBuilderThread before closing this one
//locks the mutex and automatically unlocks it when going out of scope
QMutexLocker locker2(_mutex2);
emit build_finished();
_trace_cond->wait(_mutex2);
bool BuilderThread::is_finished() { return _is_finished; }
void BuilderThread::finish_build() {
//quit();
//finish the TraceBuilderThread before closing this one
//locks the mutex and automatically unlocks it when going out of scope
QMutexLocker locker2(_mutex2);
emit build_finished();
_trace_cond->wait(_mutex2);
locker2.unlock();
_is_finished=true;
//locks the mutex and automatically unlocks it when going out of scope
QMutexLocker locker(_mutex);
_cond->wakeAll();
}
_is_finished=true;
//locks the mutex and automatically unlocks it when going out of scope
QMutexLocker locker(_mutex);
_cond->wakeAll();
}
......@@ -64,7 +64,8 @@ class BuilderThread : public QObject {
Q_OBJECT
private:
mt_ParserEventPaje *_parser;
mt_ParserEventPaje *_event_parser;
Parser* _parser;
std::queue<PajeLine*>* _q;
QWaitCondition* _cond;
QWaitCondition* _trace_cond;
......@@ -72,18 +73,22 @@ private:
QMutex* _mutex2;
QSemaphore *_freeSlots;
QSemaphore *_linesProduced;
Trace * _trace;
Trace *_trace;
bool _is_finished;
public:
/*!
* \fn ParsingThread(Parser *p, Trace *t)
* \param p the parser used to parse the file.
* \param p the event parser used to parse the file.
* \param t the trace where we store data.
* \param parser the parser used to parse the file.
*/
//BuilderThread(ParserEventPaje *p, Trace * trace,std::queue<PajeLine*>* q ,QWaitCondition *cond, QMutex *mutex);
BuilderThread(mt_ParserEventPaje *p, Trace * trace,QWaitCondition *cond,QWaitCondition* trace_cond, QSemaphore *freeBytes, QSemaphore* linesProduced, QMutex *mutex, QMutex *mutex2);
BuilderThread(mt_ParserEventPaje *p, Trace * trace, QWaitCondition *cond,
QWaitCondition* trace_cond, QSemaphore *freeBytes,
QSemaphore* linesProduced, QMutex *mutex, QMutex *mutex2,
Parser *parser);
bool is_finished();
public slots:
......
......@@ -82,68 +82,71 @@ ParserPaje::~ParserPaje() {
void ParserPaje::parse(Trace &trace,
bool finish_trace_after_parse) {
string filename = get_next_file_to_parse();
static const string PERCENT = "%";
PajeLine_t line;
#ifdef DBG_PARSER_PAJE
int lineid = 0;
#endif
// Open the trace
try {
_file = new PajeFileManager(_file_to_parse.c_str());
} catch (const char *) {
delete _file;
_file = NULL;
_is_canceled = true;
finish();
trace.finish();
std::cerr << "Cannot open file " << _file_to_parse.c_str() << std::endl;
Error::set(Error::VITE_ERR_OPEN, 0, Error::VITE_ERRCODE_WARNING);
return;
}
while(filename != "") {
// Open the trace
try {
_file = new PajeFileManager(filename.c_str());
} catch (const char *) {
delete _file;
_file = NULL;
_is_canceled = true;
finish();
trace.finish();
std::cerr << "Cannot open file " << filename.c_str() << std::endl;
Error::set(Error::VITE_ERR_OPEN, 0, Error::VITE_ERRCODE_WARNING);
return;
}
while( (!(_file->eof())) && !(_is_canceled) ) {
PajeLine_t line;
#ifdef DBG_PARSER_PAJE
int lineid = 0;
#endif
while( (!(_file->eof())) && !(_is_canceled) ) {
try {
try {
#ifdef DBG_PARSER_PAJE
if ( (lineid+1) == _file->get_line(&line) )
{
_file->print_line();
lineid++;
}
if ( (lineid+1) == _file->get_line(&line) )
{
_file->print_line();
lineid++;
}
#else
_file->get_line(&line);
_file->get_line(&line);
#endif
}
catch(char *){
Error::set(Error::VITE_ERR_EXPECT_ID_DEF, 0, Error::VITE_ERRCODE_ERROR);
continue;
}
}
catch(char *){
Error::set(Error::VITE_ERR_EXPECT_ID_DEF, 0, Error::VITE_ERRCODE_ERROR);
continue;
}
// If it's an empty line
if (line._nbtks == 0) {
continue;
}
// The line starts by a '%' : it's a definition
else if(line._tokens[0][0] == '%') {
_ParserDefinition->store_definition(&line);
}
// It's an event
else {
_ParserEvent->store_event(&line, trace);
// If it's an empty line
if (line._nbtks == 0) {
continue;
}
// The line starts by a '%' : it's a definition
else if(line._tokens[0][0] == '%') {
_ParserDefinition->store_definition(&line);
}
// It's an event
else {
_ParserEvent->store_event(&line, trace);
}
}
delete _file;
_file = NULL;
filename = get_next_file_to_parse();
}
if(finish_trace_after_parse) {
finish();
trace.finish();
}
delete _file;
_file = NULL;
}
float ParserPaje::get_percent_loaded() const {
if (_file != NULL)
return _file->get_percent_loaded();
......
......@@ -76,89 +76,91 @@ ParserVite::~ParserVite() {}
void ParserVite::parse(Trace &trace,
bool finish_trace_after_parse){
ParserPaje parserpaje(_file_to_parse);
QString name;
stack<Container *> CTstack;
const Container::Vector *root_containers;
const map<std::string, Value *> *extra_fields;
map<string, Value *>::const_iterator fnamefield;
// Store the absolute directory of the first file for relative path in the others
QString absdir = QFileInfo(_file_to_parse.c_str()).absolutePath();
// Parse the first file with definitions
try {
parserpaje.parse(trace, false);
}
catch(...) {
finish();
trace.finish();
return ;
}
const std::string filename = get_next_file_to_parse();
while(filename != "") {
ParserPaje parserpaje(filename);
QString name;
stack<Container *> CTstack;
const Container::Vector *root_containers;
const map<std::string, Value *> *extra_fields;
map<string, Value *>::const_iterator fnamefield;
// Store the absolute directory of the first file for relative path in the others
QString absdir = QFileInfo(filename.c_str()).absolutePath();
// Parse the first file with definitions
try {
parserpaje.parse(trace, false);
}
catch(...) {
finish();
trace.finish();
return ;
}
#ifdef DBG_PARSER_VITE
std::cerr << "First file parsed" << std::endl;
std::cerr << "Add container : ";
std::cerr << "First file parsed" << std::endl;
std::cerr << "Add container : ";
#endif
/* Loop over root containers to add them in the stack */
root_containers = trace.get_root_containers();
/* Loop over root containers to add them in the stack */
root_containers = trace.get_root_containers();
for (Container::VectorIt it = root_containers->begin();
it != root_containers->end(); it++) {
for (Container::VectorIt it = root_containers->begin();
it != root_containers->end(); it++) {
#ifdef DBG_PARSER_VITE
std::cerr << "+";
std::cerr << "+";
#endif
CTstack.push(*it);
}
CTstack.push(*it);
}
#ifdef DBG_PARSER_VITE
std::cerr << std::endl;
std::cerr << std::endl;
#endif
/* Deep-First search over container to parse extra files */
while ( ! CTstack.empty()) {
Container *c = CTstack.top();
CTstack.pop();
/* Deep-First search over container to parse extra files */
while ( ! CTstack.empty()) {
Container *c = CTstack.top();
CTstack.pop();
extra_fields = c->get_extra_fields();
fnamefield = extra_fields->find(string("FileName"));
extra_fields = c->get_extra_fields();
fnamefield = extra_fields->find(string("FileName"));
// Search the filename
if (fnamefield != extra_fields->end()) {
name = QString::fromStdString( ((String *)(*fnamefield).second)->to_string() );
}
else {
name = "";
}
// Search the filename
if (fnamefield != extra_fields->end()) {
name = QString::fromStdString( ((String *)(*fnamefield).second)->to_string() );
}
else {
name = "";
}
if(name != "") {
if(name != "") {
#ifdef DBG_PARSER_VITE
std::cerr << ( (absdir + QDir::separator() + name).toStdString() ) << std::endl;
std::cerr << ( (absdir + QDir::separator() + name).toStdString() ) << std::endl;
#endif
parserpaje.set_file_to_parse( (absdir + QDir::separator() + name).toStdString() );
try {
parserpaje.parse(trace, false);
parserpaje.set_file_to_parse( (absdir + QDir::separator() + name).toStdString() );
try {
parserpaje.parse(trace, false);
}
catch(...) {
finish();
trace.finish();
return ;
}
}
catch(...) {
finish();
trace.finish();
return ;
// We add the children
Container::VectorIt children_end = c->get_children()->end();
for (Container::VectorIt i = c->get_children()->begin();
i != children_end; ++ i) {
CTstack.push(*i);
}
}
// We add the children
Container::VectorIt children_end = c->get_children()->end();
for (Container::VectorIt i = c->get_children()->begin();
i != children_end; ++ i) {
CTstack.push(*i);
}
finish();
}
finish();
if(finish_trace_after_parse) {
trace.finish();
}
......
......@@ -75,8 +75,8 @@ mt_PajeFileManager::mt_PajeFileManager( const char * filename) :
{
//_tokens = (char**)malloc(sizeof(char*)*_PAJE_NBMAXTKS);
_offset=0;
open(filename);
_eof = false;
_eof = false;
open(filename);
}
mt_PajeFileManager::~mt_PajeFileManager() {
......@@ -100,29 +100,29 @@ void mt_PajeFileManager::open ( const char * filename ) {
_filename = filename;
#ifdef WIN32
_fd = CreateFile (filename, GENERIC_READ, FILE_SHARE_READ, NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL);
_fd = CreateFile (filename, GENERIC_READ, FILE_SHARE_READ, NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL);
#else
_fd=fopen(filename, "r");
if (fail())
throw "Fail to open file";
throw "Fail to open file";
#endif
#ifndef WIN32
struct stat buf; // for the file size
#ifndef WIN32
struct stat buf; // for the file size
if (-1 == stat(filename, &buf)) // Getq the size
{
Error::set(Error::VITE_ERR_FSTAT, 0, Error::VITE_ERRCODE_ERROR);
if(_fd)fclose(_fd); // Close it if wrong
if(_fd)fclose(_fd); // Close it if wrong
throw "Fail to stat file";
}
_filesize=(long)buf.st_size;
#else
LARGE_INTEGER size;
GetFileSizeEx(_fd, &size);
_filesize=size.QuadPart;
LARGE_INTEGER size;
GetFileSizeEx(_fd, &size);
_filesize=size.QuadPart;
#endif
#ifdef WIN32
......@@ -130,43 +130,45 @@ void mt_PajeFileManager::open ( const char * filename ) {
GetSystemInfo(&si);
_pageSize = (int) si.dwAllocationGranularity;
if (_filesize>MAX_SIZE_UNCUT){
_current_chunk=0;
_mapped_size=MAX_SIZE_UNCUT;
if (_filesize>MAX_SIZE_UNCUT){
_current_chunk=0;
_mapped_size=MAX_SIZE_UNCUT;
}else{
_mapped_size=_filesize;
}
}
_fm = CreateFileMapping(_fd, NULL, PAGE_READONLY, 0, 0, NULL);
_line = (char *)MapViewOfFile(_fm, FILE_MAP_COPY, 0, 0, _mapped_size);
_fm = CreateFileMapping(_fd, NULL, PAGE_READONLY, 0, 0, NULL);
_line = (char *)MapViewOfFile(_fm, FILE_MAP_COPY, 0, 0, _mapped_size);
#else
_pageSize=getpagesize();
if (_filesize>MAX_SIZE_UNCUT){
_current_chunk=0;
_mapped_size=MAX_SIZE_UNCUT;
_pageSize=getpagesize();
if (_filesize>MAX_SIZE_UNCUT){
_current_chunk=0;
_mapped_size=MAX_SIZE_UNCUT;
}else{
_mapped_size=_filesize;
}
_line = (char*)mmap(0,_mapped_size, PROT_READ|PROT_WRITE, MAP_PRIVATE, fileno(_fd),0); // Mapping of the file
madvise(_line, _mapped_size , MADV_SEQUENTIAL);//advise kernel that we want to read sequentially in the file
//printf("mmaping %d bytes for chunk %d\n",_mapped_size, _current_chunk);
}
if(_mapped_size > 0) {
_line = (char*)mmap(0,_mapped_size, PROT_READ|PROT_WRITE, MAP_PRIVATE, fileno(_fd),0); // Mapping of the file
madvise(_line, _mapped_size , MADV_SEQUENTIAL);//advise kernel that we want to read sequentially in the file
#endif
} else {
// empty file
_eof=true;
}
if ((char*)-1 == _line)
if ((char*)-1 == _line)
{
#ifndef WIN32
if(_fd)fclose(_fd); // Close it if wrong
#endif
Error::set(Error::VITE_ERR_OPEN, 0, Error::VITE_ERRCODE_ERROR);
throw "Fail to open file";
#ifndef WIN32
if(_fd)fclose(_fd); // Close it if wrong
#endif
Error::set(Error::VITE_ERR_OPEN, 0, Error::VITE_ERRCODE_ERROR);
throw "Fail to open file";
}
// For windows, clear the eof byte for ept files after reading the first file
clear();
}
......@@ -197,10 +199,7 @@ int mt_PajeFileManager::get_line(PajeLine *lineptr) {
long itks = 0;
char c;
_tokens = (char**)calloc(_PAJE_NBMAXTKS,sizeof(char*));
_lineid++;
// Set the adress of the first token
......@@ -441,7 +440,9 @@ if(_old_line!=NULL){
}
}
bool mt_PajeFileManager::eof(){return _eof;}
bool mt_PajeFileManager::eof(){
return _eof;
}
void mt_PajeFileManager::print_line() {
int i;
......
......@@ -106,11 +106,17 @@ unsigned long long getCurrentTime (){
using namespace std;
mt_ParserPaje::mt_ParserPaje() : _ParserDefinition(new ParserDefinitionPaje()),
_ParserEvent(new mt_ParserEventPaje(_ParserDefinition)), _file(NULL){}
_ParserEvent(new mt_ParserEventPaje(_ParserDefinition)),
_file(NULL),
_being_processed(false) {
}
mt_ParserPaje::mt_ParserPaje(const string &filename) : Parser(filename),
_ParserDefinition(new ParserDefinitionPaje()),
_ParserEvent(new mt_ParserEventPaje(_ParserDefinition)), _file(NULL) {}
_ParserDefinition(new ParserDefinitionPaje()),
_ParserEvent(new mt_ParserEventPaje(_ParserDefinition)),
_file(NULL),
_being_processed(false) {
}
mt_ParserPaje::~mt_ParserPaje() {
delete _ParserDefinition;
......@@ -119,103 +125,108 @@ mt_ParserPaje::~mt_ParserPaje() {
delete _file;
}
void mt_ParserPaje::parse(Trace &trace,
bool finish_trace_after_parse) {
static const string PERCENT = "%";
#ifdef DBG_PARSER_PAJE
int lineid = 0;
#endif
// Open the trace
try {
_file = new mt_PajeFileManager(_file_to_parse.c_str());
} catch (const char *) {
delete _file;
_file = NULL;
_is_canceled = true;
finish();
trace.finish();
std::cerr << "Cannot open file " << _file_to_parse.c_str() << std::endl;
Error::set(Error::VITE_ERR_OPEN, 0, Error::VITE_ERRCODE_WARNING);
return;
bool finish_trace_after_parse) {
// Make sure noone is currently parsing
//
// While parsing a file, we may find a createContainer event that requires
// to parse another file. The _being_processed trick fixes the reentrance
// problem that may arise.
if(_being_processed) {
return;
}
_being_processed = true;
QWaitCondition ended;
QWaitCondition trace_ended;
_ParserEvent->setTrace(&trace);
QMutex mutex;
QMutex mutex2;
QWaitCondition ended;
QWaitCondition trace_ended;
QSemaphore freeSlots(5);
QSemaphore linesProduced(5);
_ParserEvent->setTrace(&trace);
//this object will handle syntaxic verification of the events
BuilderThread bt(_ParserEvent, &trace, &ended, &trace_ended, &freeSlots, &linesProduced, &mutex, &mutex2);
bt.connect((const QObject*)this,
SIGNAL(produced(int, PajeLine*)),
SLOT(run(int, PajeLine*)));
bt.connect((const QObject*)this,
SIGNAL(build_finish()),
SLOT(finish_build()));
QThread consumerThread;
bt.moveToThread(&consumerThread);
//this object will handle all trace operations
TraceBuilderThread T_bt(&trace_ended, &freeSlots, &mutex2);
T_bt.connect((const QObject*)&bt,
SIGNAL(build_trace(int, Trace_builder_struct*)),
SLOT(build_trace(int, Trace_builder_struct*)));
T_bt.connect((const QObject*)&bt,
SIGNAL(build_finished()),
SLOT(build_finished()));
BuilderThread _bt(_ParserEvent, &trace, &ended, &trace_ended,
&freeSlots, &linesProduced, &mutex, &mutex2,
this);
QThread consumerThread;
_bt.connect((const QObject*)this,
SIGNAL(produced(int, PajeLine*)),