SHS/pip
4
3
Fork 0
Code Issues 67 Pull Requests 1 Actions Releases Wiki Activity

git-svn-id: svn://db.shs.com.ru/pip@810 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5

Browse Source
This commit is contained in:
Пелипенко Иван
2019-06-25 09:42:46 +00:00
parent fdd6c22c94
commit 17b37409d5
2 changed files with 36 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src_main/io_devices/pibinarylog.cpp
View File

@@ -443,7 +443,9 @@ PIBinaryLog::BinLogRecord PIBinaryLog::readRecord() {
} else br.id = 0; } else br.id = 0;
lastrecord = br; lastrecord = br;
if (br.id == 0) fileError(); if (br.id == 0) fileError();
file_mutex.lock();
moveIndex(index_pos.value(file.pos(), -1)); moveIndex(index_pos.value(file.pos(), -1));
file_mutex.unlock();
logmutex.unlock(); logmutex.unlock();
// piCoutObj << "readRecord done"; // piCoutObj << "readRecord done";
return br; return br;
@@ -451,7 +453,7 @@ PIBinaryLog::BinLogRecord PIBinaryLog::readRecord() {
void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<PIBinaryLog::BinLogIndex> * index) { void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<PIBinaryLog::BinLogIndex> * index, PIMutex & file_mutex) {
BinLogInfo * bi = info; BinLogInfo * bi = info;
bool ginfo = info != 0; bool ginfo = info != 0;
bool gindex = index != 0; bool gindex = index != 0;
@@ -464,6 +466,7 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
if (gindex) index->clear(); if (gindex) index->clear();
if (f == 0) return; if (f == 0) return;
if (!f->canRead()) return; if (!f->canRead()) return;
file_mutex.lock();
if (ginfo) { if (ginfo) {
bi->path = f->path(); bi->path = f->path();
bi->log_size = f->size(); bi->log_size = f->size();
@@ -476,6 +479,7 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
if (read_sig[i] != __S__PIBinaryLog::binlog_sig[i]) {if (ginfo) bi->records_count = -2; ok = false;} if (read_sig[i] != __S__PIBinaryLog::binlog_sig[i]) {if (ginfo) bi->records_count = -2; ok = false;}
uchar read_version = 0; uchar read_version = 0;
if (f->read(&read_version, 1) < 0) {if (ginfo) bi->records_count = -3; ok = false;} if (f->read(&read_version, 1) < 0) {if (ginfo) bi->records_count = -3; ok = false;}
file_mutex.unlock();
if (read_version == 0) {if (ginfo) bi->records_count = -4; ok = false;} if (read_version == 0) {if (ginfo) bi->records_count = -4; ok = false;}
if (read_version < PIBINARYLOG_VERSION) {if (ginfo) bi->records_count = -5; ok = false;} if (read_version < PIBINARYLOG_VERSION) {if (ginfo) bi->records_count = -5; ok = false;}
if (read_version > PIBINARYLOG_VERSION) {if (ginfo) bi->records_count = -6; ok = false;} if (read_version > PIBINARYLOG_VERSION) {if (ginfo) bi->records_count = -6; ok = false;}
@@ -487,16 +491,24 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
ba.resize(hdr_size); ba.resize(hdr_size);
while (1) { while (1) {
ba.resize(hdr_size); ba.resize(hdr_size);
{
PIMutexLocker _ml(file_mutex);
if (f->read(ba.data(), ba.size_s()) > 0) { if (f->read(ba.data(), ba.size_s()) > 0) {
ba >> br.id >> br.size >> br.timestamp; ba >> br.id >> br.size >> br.timestamp;
} else break; } else
if (f->size() - f->pos() >= br.size) f->seek(f->pos() + br.size); break;
else break; if (f->size() - f->pos() >= br.size)
f->seek(f->pos() + br.size);
else
break;
}
if (br.id > 0) { if (br.id > 0) {
if (gindex) { if (gindex) {
BinLogIndex bl_ind; BinLogIndex bl_ind;
bl_ind.id = br.id; bl_ind.id = br.id;
file_mutex.lock();
bl_ind.pos = f->pos() - br.size - hdr_size; bl_ind.pos = f->pos() - br.size - hdr_size;
file_mutex.unlock();
bl_ind.timestamp = br.timestamp; bl_ind.timestamp = br.timestamp;
index->append(bl_ind); index->append(bl_ind);
} }
@@ -538,7 +550,8 @@ PIBinaryLog::BinLogInfo PIBinaryLog::getLogInfo(const PIString & path) {
bi.records_count = 0; bi.records_count = 0;
PIFile tfile; PIFile tfile;
if (!tfile.open(path, PIIODevice::ReadOnly)) return bi; if (!tfile.open(path, PIIODevice::ReadOnly)) return bi;
parseLog(&tfile, &bi, 0); PIMutex _m;
parseLog(&tfile, &bi, 0, _m);
return bi; return bi;
} }
@@ -568,12 +581,16 @@ bool PIBinaryLog::cutBinLog(const PIBinaryLog::BinLogInfo & src, const PIString
bool PIBinaryLog::createIndex() { bool PIBinaryLog::createIndex() {
file_mutex.lock();
llong cp = file.pos(); llong cp = file.pos();
file.seekToBegin(); file.seekToBegin();
file_mutex.unlock();
index.clear(); index.clear();
index_pos.clear(); index_pos.clear();
parseLog(&file, &binfo, &index); parseLog(&file, &binfo, &index, file_mutex);
file_mutex.lock();
file.seek(cp); file.seek(cp);
file_mutex.unlock();
is_indexed = !index.isEmpty(); is_indexed = !index.isEmpty();
for (uint i=0; i<index.size(); i++) index_pos[index[i].pos] = i; for (uint i=0; i<index.size(); i++) index_pos[index[i].pos] = i;
return is_indexed; return is_indexed;
@@ -585,7 +602,9 @@ void PIBinaryLog::seekTo(int rindex) {
logmutex.lock(); logmutex.lock();
if (rindex < index.size_s() && rindex >= 0) { if (rindex < index.size_s() && rindex >= 0) {
file.seek(index[rindex].pos); file.seek(index[rindex].pos);
file_mutex.lock();
moveIndex(index_pos.value(file.pos(), -1)); moveIndex(index_pos.value(file.pos(), -1));
file_mutex.unlock();
play_time = index[rindex].timestamp.toMilliseconds(); play_time = index[rindex].timestamp.toMilliseconds();
lastrecord.timestamp = index[rindex].timestamp; lastrecord.timestamp = index[rindex].timestamp;
} }

12
src_main/io_devices/pibinarylog.h
View File

@@ -187,7 +187,10 @@ public:
int readBinLog(int id, void * read_to, int max_size, PISystemTime * time = 0); int readBinLog(int id, void * read_to, int max_size, PISystemTime * time = 0);
//! Returns binary log file size //! Returns binary log file size
llong logSize() const {return file.size();} llong logSize() const {PIMutexLocker _ml(file_mutex); return file.size();}
//! Return position in current binlog file
llong logPos() const {PIMutexLocker _ml(file_mutex); return file.pos();}
//! Return true, if position at the end of BinLog file //! Return true, if position at the end of BinLog file
bool isEnd() const {if (isClosed()) return true; return file.isEnd();} bool isEnd() const {if (isClosed()) return true; return file.isEnd();}
@@ -228,6 +231,9 @@ public:
//! Create index of current binlog file //! Create index of current binlog file
bool createIndex(); bool createIndex();
//! Return if current binlog file is indexed
bool isIndexed() {return is_indexed;}
//! Go to record #index //! Go to record #index
void seekTo(int rindex); void seekTo(int rindex);
@@ -297,7 +303,7 @@ private:
bool writeFileHeader(); bool writeFileHeader();
bool checkFileHeader(); bool checkFileHeader();
BinLogRecord readRecord(); BinLogRecord readRecord();
static void parseLog(PIFile *f, BinLogInfo *info, PIVector<BinLogIndex> * index); static void parseLog(PIFile *f, BinLogInfo *info, PIVector<BinLogIndex> * index, PIMutex & file_mutex);
void moveIndex(int i); void moveIndex(int i);
PIString getLogfilePath() const; PIString getLogfilePath() const;
@@ -310,7 +316,7 @@ private:
PIFile file; PIFile file;
BinLogRecord lastrecord; BinLogRecord lastrecord;
PISystemTime startlogtime, play_delay, split_time, pause_time; PISystemTime startlogtime, play_delay, split_time, pause_time;
PIMutex logmutex, pausemutex; mutable PIMutex logmutex, pausemutex, file_mutex;
double play_time, play_speed; double play_time, play_speed;
llong split_size; llong split_size;
int write_count, split_count, default_id, current_index; int write_count, split_count, default_id, current_index;